Adding different directional layouts for graph view

airflow/macros/hive.py

@@ -1,7 +1,7 @@
 from airflow import settings
-from airflow.hooks.hive_hook import HiveHook
 
 def max_partition(
+    from airflow.hooks.hive_hook import HiveHook
         table, schema="default", hive_dbid=settings.HIVE_DEFAULT_DBID):
     hh = HiveHook(hive_dbid=hive_dbid)
     return hh.max_partition(schema=schema, table=table)

airflow/models.py

@@ -16,7 +16,6 @@ from sqlalchemy import (
 )
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import relationship
-from airflow import macros
 from airflow.executors import DEFAULT_EXECUTOR
 from airflow import settings
 from airflow import utils
@@ -407,6 +406,7 @@ class TaskInstance(Base):
                 logging.info(msg.format(self=self))
             try:
                 if not mark_success:
+                    from airflow import macros
                     jinja_context = {
                         'ti': self,
                         'execution_date': self.execution_date,

airflow/utils.py

@@ -32,10 +32,10 @@ def validate_key(k, max_length=250):
     elif len(k) > max_length:
         raise Exception("The key has to be less than {0} characters".format(
             max_length))
-    elif not re.match(r'^[A-Za-z0-9_]+$', k):
+    elif not re.match(r'^[A-Za-z0-9_-]+$', k):
         raise Exception(
-            "The key has to be made of alphanumeric characters and "
-            "undersairflows exclusively")
+            "The key has to be made of alphanumeric characters, dashes "
+            "and underscores exclusively")
     else:
         return True
 

airflow/www/app.py

@@ -7,7 +7,7 @@ from flask import Flask, url_for, Markup, Blueprint, redirect, flash
 from flask.ext.admin import Admin, BaseView, expose, AdminIndexView
 from flask.ext.admin.contrib.sqla import ModelView
 from flask import request
-from wtforms import Form, DateTimeField
+from wtforms import Form, DateTimeField, SelectField
 
 from pygments import highlight
 from pygments.lexers import PythonLexer
@@ -41,6 +41,15 @@ app.jinja_env.add_extension("chartkick.ext.charts")
 class DateTimeForm(Form):
     execution_date = DateTimeField("Execution date")
 
+class GraphForm(Form):
+    execution_date = DateTimeField("Execution date")
+    arrange = SelectField("Layout", choices=(
+        ('LR', "Left->Right"),
+        ('RL', "Right->Left"),
+        ('TB', "Top->Bottom"),
+        ('BT', "Bottom->Top"),
+    ))
+
 
 @app.route('/')
 def index():
@@ -101,6 +110,28 @@ class Airflow(BaseView):
         return self.render(
             'admin/code.html', html_code=html_code, dag=dag, title=title)
 
+    @expose('/task')
+    def task(self):
+        dag_id = request.args.get('dag_id')
+        task_id = request.args.get('task_id')
+        dag = dagbag.dags[dag_id]
+        task = dag.get_task(task_id)
+
+        s = ''
+        for attr_name in dir(task):
+            if not attr_name.startswith('_'):
+                attr = getattr(task, attr_name)
+                if type(attr) != type(self.task):
+                     s += attr_name + ': ' + str(attr) + '\n'
+
+        s = s.replace('<', '&lt')
+        s = s.replace('>', '&gt')
+        html_code = "<pre><code>" + s + "</code></pre>"
+        title = "Task Details for {task_id}".format(**locals())
+
+        return self.render(
+            'admin/code.html', html_code=html_code, dag=dag, title=title)
+
     @expose('/tree')
     def tree(self):
         dag_id = request.args.get('dag_id')
@@ -205,6 +236,7 @@ class Airflow(BaseView):
     def graph(self):
         session = settings.Session()
         dag_id = request.args.get('dag_id')
+        arrange = request.args.get('arrange', "LR")
         dag = dagbag.dags[dag_id]
 
         nodes = []
@@ -234,7 +266,7 @@ class Airflow(BaseView):
         else:
             dttm = dag.latest_execution_date or datetime.now().date()
 
-        form = DateTimeForm(data={'execution_date': dttm})
+        form = GraphForm(data={'execution_date': dttm, 'arrange': arrange})
 
         task_instances = {
             ti.task_id: utils.alchemy_to_dict(ti)
@@ -252,6 +284,7 @@ class Airflow(BaseView):
             dag=dag,
             form=form,
             execution_date=dttm.isoformat(),
+            arrange=arrange,
             task_instances=json.dumps(task_instances, indent=2),
             tasks=json.dumps(tasks, indent=2),
             nodes=json.dumps(nodes, indent=2),

airflow/www/static/graph.css

@@ -15,7 +15,7 @@ g.node.running rect{
 g.node.failed rect {
     stroke: red;
 }
-input#execution_date {
+input, select {
     margin-bottom: 0px;
 }
 div#svg_container {

airflow/www/static/main.css

@@ -30,3 +30,12 @@ input#execution_date {
 .modal{
     margin-top:5000px;
 }
+pre {
+  overflow: auto;
+  word-wrap: normal;
+  white-space: pre;
+}
+pre code {
+    overflow-wrap: normal;
+    white-space: pre;
+}

airflow/www/templates/admin/dag.html

@@ -58,6 +58,9 @@
           <button id="btn_log" type="button" class="btn btn-primary">
             View Log
           </button>
+          <button id="btn_task" type="button" class="btn btn-primary">
+            Task Details
+          </button>
           <hr/>
           <button id="btn_clear" type="button" class="btn btn-primary">
             Clear
@@ -122,6 +125,13 @@
       window.location = url;
     });
 
+    $("#btn_task").click(function(){
+      url = "{{ url_for('airflow.task') }}" +
+        "?task_id=" + task_id +
+        "&dag_id=" + dag_id;
+      window.location = url;
+    });
+
     $("#btn_clear").click(function(){
       url = "{{ url_for('airflow.tree') }}" +
         "?action=clear" +

airflow/www/templates/admin/graph.html

@@ -18,11 +18,13 @@
     <form method="get">
         Run:<input type="hidden" value="{{ dag.dag_id }}" name="dag_id">
         {{ form.execution_date | safe }}
+        Layout:
+        {{ form.arrange | safe }}
         <input type="submit" value="Go" class="btn btn-default"
         action="" method="get">
     </form>
     <div id="svg_container">
-      <svg height=500 width="100%">
+      <svg height=800 width="100%">
           <g id='dig' transform="translate(20,20)"/>
       </svg>
     </div>
@@ -43,9 +45,12 @@
     var tasks = {{ tasks|safe }};
     var task_instances = {{ task_instances|safe }};
     var execution_date = "{{ execution_date }}";
+    var arrange = "{{ arrange }}";
     var g = dagreD3.json.decode(nodes, edges);
+
+    var layout = dagreD3.layout().rankDir(arrange);
     var renderer = new dagreD3.Renderer();
-    renderer.run(g, d3.select("#dig"));
+    renderer.layout(layout).run(g, d3.select("#dig"));
 
     // Activating the date time picker widget
     $('#execution_date').datetimepicker();
@@ -55,9 +60,6 @@
       $("tspan:contains(" + task_id + ")")
         .parent().parent().parent()
         .attr("class", "node enter " + ti.state)
-        .click(function(){
-          call_modal(task_id, execution_date);
-        })
         .attr("data-toggle", "tooltip")
         .attr("title", function(d) {
           // Tooltip
@@ -70,6 +72,11 @@
           return tt;
         });
     });
+    
+    d3.selectAll("g.node").on("click", function(d){
+        call_modal(d, execution_date);
+    });
+
     $("g.node").tooltip({
       html: true,
       container: "body",

airflow/www/templates/admin/tree.html

@@ -99,7 +99,7 @@ function update(source) {
   // Compute the flattened node list. TODO use d3.layout.hierarchy.
   var nodes = tree.nodes(root);
 
-  var height = Math.min(500, nodes.length * barHeight + margin.top + margin.bottom);
+  var height = Math.max(500, nodes.length * barHeight + margin.top + margin.bottom);
   var width = square_x + (num_square * (square_size + square_spacing)) + margin.left + margin.right + 50;
   d3.select("svg").transition()
       .duration(duration)