[AIRFLOW-2657] Add ability to delete dag from web UI

Closes apache#3531 from Noremac201/master

Author: Noremac201

airflow/www/templates/airflow/dag.html

@@ -100,6 +100,13 @@ <h4 class="pull-right">
           Refresh
         </a>
       </li>
+      <li>
+        <a href="{{ url_for("airflow.delete", dag_id=dag.dag_id, root=root) }}"
+        onclick="return confirmDeleteDag('{{ dag.safe_dag_id }}')">
+        <span class="glyphicon glyphicon-remove-circle" style="color:red" aria-hidden="true"></span>
+        Delete
+        </a>
+      </li>
     </ul>
   </div>
   <hr>
@@ -302,6 +309,12 @@ <h4 class="modal-title" id="dagModalLabel">
       $("#dagModal").css("margin-top","0px");
     }
 
+    function confirmDeleteDag(dag_id){
+        return confirm("Are you sure you want to delete '"+dag_id+"' now?\n\
+          This option will delete ALL metadata, DAG runs, etc.\n\
+          This cannot be undone.");
+    }
+
     $("#btn_rendered").click(function(){
       url = "{{ url_for('airflow.rendered') }}" +
         "?task_id=" + encodeURIComponent(task_id) +

airflow/www/templates/airflow/dags.html

@@ -190,6 +190,12 @@ <h2>DAGs</h2>
                   <span class="glyphicon glyphicon-refresh" aria-hidden="true" data-original-title="Refresh"></span>
                 </a>
 
+                <!-- Delete -->
+                <a href="{{ url_for('airflow.delete', dag_id=dag.dag_id) }}"
+                  onclick="return confirmDeleteDag('{{ dag.safe_dag_id }}')">
+                   <span class="glyphicon glyphicon-remove-circle" style="color:red" aria-hidden="true" data-original-title="Delete Dag"></span>
+                </a>
+
                 </td>
             </tr>
         {% endfor %}
@@ -244,6 +250,12 @@ <h2>DAGs</h2>
       function confirmTriggerDag(dag_id){
           return confirm("Are you sure you want to run '"+dag_id+"' now?");
       }
+
+      function confirmDeleteDag(dag_id){
+          return confirm("Are you sure you want to delete '"+dag_id+"' now?\n\
+          This option will delete ALL metadata, DAG runs, etc.\n\
+          This cannot be undone.");
+      }
       all_dags = $("[id^=toggle]");
       $.each(all_dags, function(i,v) {
         $(v).change (function() {

airflow/www/views.py

@@ -1044,6 +1044,32 @@ def run(self):
             "it should start any moment now.".format(ti))
         return redirect(origin)
 
+    @expose('/delete')
+    @login_required
+    @wwwutils.action_logging
+    @wwwutils.notify_owner
+    def delete(self):
+        from airflow.api.common.experimental import delete_dag
+        from airflow.exceptions import DagNotFound, DagFileExists
+
+        dag_id = request.args.get('dag_id')
+        origin = request.args.get('origin') or "/admin/"
+
+        try:
+            delete_dag.delete_dag(dag_id)
+        except DagNotFound:
+            flash("DAG with id {} not found. Cannot delete".format(dag_id))
+            return redirect(request.referrer)
+        except DagFileExists:
+            flash("Dag id {} is still in DagBag. "
+                  "Remove the DAG file first.".format(dag_id))
+            return redirect(request.referrer)
+
+        flash("Deleting DAG with id {}. May take a couple minutes to fully"
+              " disappear.".format(dag_id))
+        # Upon successful delete return to origin
+        return redirect(origin)
+
     @expose('/trigger')
     @login_required
     @wwwutils.action_logging
@@ -1283,6 +1309,10 @@ def tree(self, session=None):
         dag_id = request.args.get('dag_id')
         blur = conf.getboolean('webserver', 'demo_mode')
         dag = dagbag.get_dag(dag_id)
+        if dag_id not in dagbag.dags:
+            flash('DAG "{0}" seems to be missing.'.format(dag_id), "error")
+            return redirect('/admin/')
+
         root = request.args.get('root')
         if root:
             dag = dag.sub_dag(

airflow/www_rbac/templates/airflow/dag.html

@@ -99,6 +99,13 @@ <h4 class="pull-right">
           Refresh
         </a>
       </li>
+      <li>
+        <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id, root=root) }}"
+        onclick="return confirmDeleteDag('{{ dag.safe_dag_id }}')">
+        <span class="glyphicon glyphicon-remove-circle" style="color:red" aria-hidden="true"></span>
+        Delete
+        </a>
+      </li>
     </ul>
   </div>
   <hr>
@@ -300,6 +307,12 @@ <h4 class="modal-title" id="dagModalLabel">
       $("#dagModal").css("margin-top","0px");
     }
 
+    function confirmDeleteDag(dag_id){
+        return confirm("Are you sure you want to delete '"+dag_id+"' now?\n\
+          This option will delete ALL metadata, DAG runs, etc.\n\
+          This cannot be undone.");
+    }
+
     $("#btn_rendered").click(function(){
       url = "{{ url_for('Airflow.rendered') }}" +
         "?task_id=" + encodeURIComponent(task_id) +

airflow/www_rbac/templates/airflow/dags.html

@@ -191,6 +191,11 @@ <h2>DAGs</h2>
                   <span class="glyphicon glyphicon-refresh" aria-hidden="true" data-original-title="Refresh"></span>
                 </a>
 
+                <!-- Delete -->
+                <a href="{{ url_for('Airflow.delete', dag_id=dag.dag_id) }}"
+                   onclick="return confirmDeleteDag('{{ dag.safe_dag_id }}')">
+                  <span class="glyphicon glyphicon-remove-circle" style="color:red" aria-hidden="true" data-original-title="Delete Dag"></span>
+                </a>
                 </td>
             </tr>
         {% endfor %}
@@ -242,6 +247,12 @@ <h2>DAGs</h2>
         window.location = DAGS_INDEX + "?page_size=" + p_size;
       });
 
+      function confirmDeleteDag(dag_id){
+          return confirm("Are you sure you want to delete '"+dag_id+"' now?\n\
+          This option will delete ALL metadata, DAG runs, etc.\n\
+          This cannot be undone.");
+      }
+
       function confirmTriggerDag(dag_id){
           return confirm("Are you sure you want to run '"+dag_id+"' now?");
       }

airflow/www_rbac/views.py

@@ -728,6 +728,32 @@ def run(self):
             "it should start any moment now.".format(ti))
         return redirect(origin)
 
+    @expose('/delete')
+    @action_logging
+    @has_access
+    def delete(self):
+        from airflow.api.common.experimental import delete_dag
+        from airflow.exceptions import DagNotFound, DagFileExists
+
+        dag_id = request.args.get('dag_id')
+        origin = request.args.get('origin') or "/"
+
+        try:
+            delete_dag.delete_dag(dag_id)
+        except DagNotFound:
+            flash("DAG with id {} not found. Cannot delete".format(dag_id))
+            return redirect(request.referrer)
+        except DagFileExists:
+            flash("Dag id {} is still in DagBag. "
+                  "Remove the DAG file first.".format(dag_id))
+            return redirect(request.referrer)
+
+        flash("Deleting DAG with id {}. May take a couple minutes to fully"
+              " disappear.".format(dag_id))
+
+        # Upon success return to origin.
+        return redirect(origin)
+
     @expose('/trigger')
     @has_access
     @action_logging
@@ -961,6 +987,10 @@ def tree(self, session=None):
         dag_id = request.args.get('dag_id')
         blur = conf.getboolean('webserver', 'demo_mode')
         dag = dagbag.get_dag(dag_id)
+        if dag_id not in dagbag.dags:
+            flash('DAG "{0}" seems to be missing.'.format(dag_id), "error")
+            return redirect('/')
+
         root = request.args.get('root')
         if root:
             dag = dag.sub_dag(