minor updates to script

Author: Ishankoradia

Diff for: ddpui/ddpprefect/prefect_service.py

@@ -923,11 +923,7 @@ def estimate_time_for_next_queued_run_of_dataflow(
     run_time_type = "wt_avg_run_time"
     queue_no = 1
     queue_time_in_seconds = (
-        dataflow.meta[run_time_type]
-        if dataflow.meta
-        and run_time_type in dataflow.meta
-        and dataflow.meta[run_time_type] > 0  # we should have a positive time
-        else 0
+        5  # 5 secs of approx latency of the flow run going from pos no 1 to the prefect worker
     )
     for flow_run in queued_late_flow_runs:
         deployment_meta = (

Diff for: ddpui/management/commands/estimate-time-for-queued-runs.py

@@ -14,7 +14,7 @@ class Command(BaseCommand):
     help = "Estimate time for queued runs"
 
     def add_arguments(self, parser):
-        parser.add_argument("org", type=str, help="Org slug")
+        parser.add_argument("org", type=str, help="Org slug; use 'all' to run for all orgs at once")
         parser.add_argument(
             "--deployment_id", type=str, help="Dataflows's deployment id", required=False
         )
@@ -24,39 +24,44 @@ def add_arguments(self, parser):
             help="No of last x flow runs to look for computing run times",
             default=20,
         )
+        parser.add_argument("--compute-runtimes", action="store_true")
 
     def handle(self, *args, **options):
-        org = Org.objects.filter(slug=options["org"]).first()
-        if org is None:
-            print(f"Org with slug {options['org']} does not exist")
-            return
-
-        dataflows = OrgDataFlowv1.objects.filter(org=org)
-        if "deployment_id" in options and options["deployment_id"]:
-            dataflows = dataflows.filter(deployment_id=options["deployment_id"])
-
-        limit = options["limit"] or 20
-
-        for dataflow in dataflows:
-            print(
-                f"Computing the runs times over last {limit} flow runs for dataflow {dataflow.name}"
-            )
-
-            run_times: DeploymentRunTimes = compute_dataflow_run_times_from_history(
-                dataflow, limit=limit
-            )
-
-            print(
-                f"Run times for {dataflow.name} for last seven days in seconds : {run_times.dict()} "
-            )
-
-        print("Computing the current queue position and time for each dataflow")
-        for dataflow in dataflows:
-            try:
-                current_queue: DeploymentCurrentQueueTime = (
-                    estimate_time_for_next_queued_run_of_dataflow(dataflow)
-                )
-
-                print(f"Current queue time for {dataflow.name} : {current_queue}")
-            except Exception as err:
-                print("Failed to compute current queue time " + str(err))
+        orgs = Org.objects.all()
+        if options["org"] != "all":
+            orgs = orgs.filter(slug=options["org"])
+
+        for org in orgs:
+            print("=" * 40 + org.slug + "=" * 40)
+
+            dataflows = OrgDataFlowv1.objects.filter(org=org)
+            if "deployment_id" in options and options["deployment_id"]:
+                dataflows = dataflows.filter(deployment_id=options["deployment_id"])
+
+            limit = options["limit"] or 20
+
+            if options["compute_runtimes"]:
+                print("Compute run times is set to true")
+                for dataflow in dataflows:
+                    print(
+                        f"Computing the runs times over last {limit} flow runs for dataflow {dataflow.name}"
+                    )
+
+                    run_times: DeploymentRunTimes = compute_dataflow_run_times_from_history(
+                        dataflow, limit=limit
+                    )
+
+                    print(
+                        f"Run times for {dataflow.name} for last seven days in seconds : {run_times.dict()} "
+                    )
+
+            print("Computing the current queue position and time for each dataflow")
+            for dataflow in dataflows:
+                try:
+                    current_queue: DeploymentCurrentQueueTime = (
+                        estimate_time_for_next_queued_run_of_dataflow(dataflow)
+                    )
+
+                    print(f"Current queue time for {dataflow.name} : {current_queue}")
+                except Exception as err:
+                    print("Failed to compute current queue time " + str(err))