Workflow health: bucketed run counts behind the window

lib/lightning/workflows/stats.ex

@@ -41,13 +41,8 @@ defmodule Lightning.Workflows.Stats do
   def outcomes(%Workflow{id: workflow_id}, days_back \\ @default_days_back)
       when days_back > 0 do
     cached({:outcomes, workflow_id, days_back}, fn ->
-      to = DateTime.utc_now()
-      since = DateTime.add(to, -days_back, :day)
-
-      %{
-        window: %{from: since, to: to},
-        counts: count_work_orders(workflow_id, since)
-      }
+      window = window(days_back)
+      %{window: window, counts: count_work_orders(workflow_id, window.from)}
     end)
   end
 
@@ -84,16 +79,137 @@ defmodule Lightning.Workflows.Stats do
       )
       when days_back > 0 do
     cached({:failures, workflow_id, days_back}, fn ->
-      to = DateTime.utc_now()
-      since = DateTime.add(to, -days_back, :day)
+      window = window(days_back)
+
+      %{
+        window: window,
+        signatures: group_by_signature(workflow_id, window.from)
+      }
+    end)
+  end
+
+  # How the runs chart slices each window: `{bucket_seconds, bucket_count}`.
+  # Every width divides a day evenly, so a window whose `from` sits on the grid
+  # keeps every bucket boundary on the clock — 2-hourly, AM/PM, midnight.
+  #
+  # One bucket more than the width divides into: `now` sits mid-bucket, so a
+  # grid of exactly `days_back / width` bars would start *after*
+  # `now - days_back` and leave the oldest hours of the window undrawn while
+  # the donuts beside it counted them. The oldest bar instead reaches back
+  # past `from`, and `window` reports the range actually covered.
+  @buckets %{1 => {7_200, 13}, 7 => {43_200, 15}, 30 => {86_400, 31}}
+
+  @zero_run_counts Map.new(Run.final_states(), &{&1, 0})
+
+  @typedoc "One bar of the runs chart: when its slot starts, and its counts."
+  @type run_bucket :: %{
+          :at => DateTime.t(),
+          optional(atom()) => non_neg_integer()
+        }
+
+  @doc """
+  Final run counts per state, bucketed across the last `days_back` days:
+  2-hourly over a day, AM/PM over a week, daily over a month.
+
+  Bucketed here rather than in the browser, because the alternative is shipping
+  every run in the window — six figures of rows on a busy workflow, cached
+  whole and JSON-encoded — to draw thirty bars.
+
+  Buckets are counted on `inserted_at` — when the attempt started, not when it
+  settled — so a run stays in the bar the traffic arrived in. Every bucket and
+  every state is present, zero-filled: the chart draws a flat window without
+  reasoning about which bars are missing.
+
+  The last bucket is the one `now` falls in, so it is still filling; the first
+  reaches back past `now - days_back`, so nothing in the window goes undrawn.
+  `window` is the range the bars actually cover.
+
+  A bucket is `at` alongside one key per state, flat rather than nested, which
+  is the row shape Recharts takes as `data` — the list goes to the chart
+  untouched, and each `Bar` names the state it draws.
+  """
+  @spec runs(Workflow.t(), 1 | 7 | 30) :: %{
+          window: %{from: DateTime.t(), to: DateTime.t()},
+          buckets: [run_bucket()]
+        }
+  def runs(%Workflow{id: workflow_id}, days_back \\ @default_days_back)
+      when is_map_key(@buckets, days_back) do
+    cached({:runs, workflow_id, days_back}, fn ->
+      {seconds, count} = Map.fetch!(@buckets, days_back)
+      window = bucket_window(seconds, count)
 
       %{
-        window: %{from: since, to: to},
-        signatures: group_by_signature(workflow_id, since)
+        window: window,
+        buckets: bucket_runs(workflow_id, window.from, seconds, count)
+      }
+    end)
+  end
+
+  # Anchored on the bucket `now` is in, not on `now` itself: a window that ends
+  # mid-bucket would put every boundary at whatever minute the request landed
+  # on, and the labels the chart draws — "2am", "PM", a date — would be lies.
+  defp bucket_window(seconds, count) do
+    to = DateTime.utc_now()
+    current = DateTime.from_unix!(div(DateTime.to_unix(to), seconds) * seconds)
+
+    %{from: DateTime.add(current, -(count - 1) * seconds, :second), to: to}
+  end
+
+  defp bucket_runs(workflow_id, from, seconds, count) do
+    tallies = tally_runs(workflow_id, from, seconds)
+
+    Enum.map(0..(count - 1), fn index ->
+      tallies
+      |> Map.get(index, [])
+      |> Enum.into(@zero_run_counts)
+      |> Map.put(:at, DateTime.add(from, index * seconds, :second))
+    end)
+  end
+
+  # `wo.last_activity` is redundant against the run filter — a work order is
+  # touched every time one of its runs is created or settles, so its activity
+  # is never older than its newest run. It is here for the planner: it lets the
+  # `work_orders(workflow_id, last_activity)` index cut the work orders down to
+  # the window before the nested loop into `runs(work_order_id, inserted_at)`,
+  # instead of probing every work order the workflow ever had.
+  #
+  # The grid is aligned, so integer division by the bucket width is the whole
+  # of the bucketing — no `date_trunc` special case per width. `floor` before
+  # the cast because `extract` yields `numeric` and `numeric::bigint` rounds:
+  # without it a run at 01:59:59.7 is counted in the 02:00 bar.
+  defp tally_runs(workflow_id, from, seconds) do
+    from(r in Run,
+      join: wo in WorkOrder,
+      on: wo.id == r.work_order_id,
+      where:
+        wo.workflow_id == ^workflow_id and wo.last_activity >= ^from and
+          r.inserted_at >= ^from and r.state in ^Run.final_states(),
+      group_by: [selected_as(:bucket), r.state],
+      select: {
+        selected_as(
+          fragment(
+            "div(floor(extract(epoch from ? - ?))::bigint, ?)::int",
+            r.inserted_at,
+            type(^from, :utc_datetime_usec),
+            type(^seconds, :integer)
+          ),
+          :bucket
+        ),
+        r.state,
+        count(r.id)
       }
+    )
+    |> Repo.all()
+    |> Enum.group_by(fn {bucket, _, _} -> bucket end, fn {_, state, count} ->
+      {state, count}
     end)
   end
 
+  defp window(days_back) do
+    to = DateTime.utc_now()
+    %{from: DateTime.add(to, -days_back, :day), to: to}
+  end
+
   # Cached whole, `window` included — that is what stops the window rolling per
   # request.
   #

lib/lightning_web/controllers/api/workflow_health_controller.ex

@@ -37,6 +37,13 @@ defmodule LightningWeb.API.WorkflowHealthController do
     )
   end
 
+  def runs(conn, _params) do
+    json(
+      conn,
+      Workflows.Stats.runs(conn.assigns.workflow, conn.assigns.days_back)
+    )
+  end
+
   # Closed set, string-matched — no free integer, no parse to defend.
   @days %{"1" => 1, "7" => 7, "30" => 30}
   @default_days "30"

lib/lightning_web/router.ex

@@ -116,6 +116,10 @@ defmodule LightningWeb.Router do
     get "/projects/:project_id/workflows/:workflow_id/health/failures",
         API.WorkflowHealthController,
         :error_signatures
+
+    get "/projects/:project_id/workflows/:workflow_id/health/runs",
+        API.WorkflowHealthController,
+        :runs
   end
 
   ## Collections

test/lightning/workflows/stats_test.exs

@@ -3,6 +3,7 @@ defmodule Lightning.Workflows.StatsTest do
 
   import Lightning.Factories
 
+  alias Lightning.Run
   alias Lightning.Workflows.Snapshot
   alias Lightning.Workflows.Stats
   alias Lightning.Workflows.Workflow
@@ -683,4 +684,114 @@ defmodule Lightning.Workflows.StatsTest do
       end
     end
   end
+
+  describe "runs/2" do
+    # A run whose `inserted_at` we choose, on a work order active at the same
+    # moment — the shape the window needs and `insert_run/4` can't give.
+    defp run_at(workflow, trigger, state, at) do
+      # A work order has no `:available`, so an in-flight run hangs off a
+      # running one.
+      wo_state = if state in WorkOrder.final_states(), do: state, else: :running
+
+      insert(:run,
+        work_order:
+          work_order(workflow, trigger, state: wo_state, last_activity: at),
+        starting_trigger: trigger,
+        dataclip: insert(:dataclip),
+        state: state,
+        inserted_at: at
+      )
+    end
+
+    # Written out rather than a fixed index, because the grid moves with the
+    # clock: at 09:59 the last 2-hourly bucket is 08:00, at 10:01 it is 10:00.
+    defp bucket_of(%{window: %{from: from}, buckets: [a, b | _]}, at) do
+      DateTime.diff(at, from, :second)
+      |> div(DateTime.diff(b.at, a.at, :second))
+    end
+
+    test "counts each state into the bucket its run started in", ctx do
+      %{workflow: workflow, trigger: trigger} = ctx
+
+      older = DateTime.add(DateTime.utc_now(), -5, :hour)
+      newer = DateTime.add(DateTime.utc_now(), -90, :minute)
+
+      # A hair inside a bucket, not in the next one: `extract(epoch ...)` is
+      # `numeric` and casting it rounds, so the query has to floor first.
+      edge =
+        DateTime.utc_now()
+        |> DateTime.to_unix()
+        |> div(7_200)
+        |> Kernel.*(7_200)
+        |> DateTime.from_unix!()
+        |> DateTime.add(-4, :hour)
+        |> DateTime.add(-100, :millisecond)
+
+      run_at(workflow, trigger, :failed, newer)
+      run_at(workflow, trigger, :success, older)
+      run_at(workflow, trigger, :success, older)
+      run_at(workflow, trigger, :crashed, edge)
+
+      assert %{buckets: buckets} = result = Stats.runs(workflow, 1)
+
+      assert %{success: 2, failed: 0} =
+               Enum.at(buckets, bucket_of(result, older))
+
+      assert %{success: 0, failed: 1} =
+               Enum.at(buckets, bucket_of(result, newer))
+
+      assert %{crashed: 1} = Enum.at(buckets, bucket_of(result, edge))
+    end
+
+    # Boundaries on the clock, so the chart can label a bar "2am" or "Tuesday"
+    # and be telling the truth. And a bar chart with holes in it is a different
+    # chart, so every bucket carries every final state, zero-filled.
+    test "cuts each window into clock-aligned, zero-filled buckets", ctx do
+      %{workflow: workflow} = ctx
+
+      zeroed = Map.new(Run.final_states(), &{&1, 0})
+
+      for {days, seconds, count} <- [
+            {1, 7_200, 13},
+            {7, 43_200, 15},
+            {30, 86_400, 31}
+          ] do
+        assert %{buckets: buckets, window: window} = Stats.runs(workflow, days)
+
+        assert length(buckets) == count
+        assert rem(DateTime.to_unix(window.from), seconds) == 0
+
+        assert DateTime.compare(
+                 window.from,
+                 DateTime.add(window.to, -days, :day)
+               ) != :gt
+
+        assert DateTime.diff(Enum.at(buckets, 1).at, hd(buckets).at) == seconds
+
+        for bucket <- buckets, do: assert(Map.delete(bucket, :at) == zeroed)
+
+        # The window ends inside the last bucket, which is still filling.
+        last = List.last(buckets).at
+        assert DateTime.compare(last, window.to) == :lt
+        assert DateTime.diff(window.to, last, :second) < seconds
+      end
+    end
+
+    test "skips runs outside the window, in flight, or on another workflow",
+         ctx do
+      %{workflow: workflow, trigger: trigger} = ctx
+
+      run_at(workflow, trigger, :success, days_ago(2))
+      run_at(workflow, trigger, :available, DateTime.utc_now())
+
+      other = insert(:simple_workflow)
+      run_at(other, hd(other.triggers), :success, DateTime.utc_now())
+
+      assert %{buckets: buckets} = Stats.runs(workflow, 1)
+
+      assert Enum.sum_by(buckets, fn bucket ->
+               bucket |> Map.delete(:at) |> Map.values() |> Enum.sum()
+             end) == 0
+    end
+  end
 end

test/lightning_web/controllers/api/workflow_health_controller_test.exs

@@ -33,6 +33,15 @@ defmodule LightningWeb.API.WorkflowHealthControllerTest do
     )
   end
 
+  defp get_runs(conn, user, project_id, workflow_id, params \\ %{}) do
+    conn
+    |> log_in_user(user)
+    |> get(
+      ~p"/api/projects/#{project_id}/workflows/#{workflow_id}/health/runs",
+      params
+    )
+  end
+
   describe "authorization" do
     test "a project member is served", %{
       conn: conn,
@@ -196,15 +205,19 @@ defmodule LightningWeb.API.WorkflowHealthControllerTest do
       assert %{status: 404} = get_outcomes(conn, user, project.id, "not-a-uuid")
     end
 
-    # Both actions share one plug, so this only has to prove the plug runs on
-    # the second one too.
-    test "the failures slice is guarded by the same check", %{
+    # Every action shares one plug, so this only has to prove the plug runs on
+    # the others too.
+    test "the other slices are guarded by the same check", %{
       conn: conn,
       project: project,
       workflow: workflow
     } do
+      stranger = insert(:user)
+
       assert %{status: 404} =
-               get_failures(conn, insert(:user), project.id, workflow.id)
+               get_failures(conn, stranger, project.id, workflow.id)
+
+      assert %{status: 404} = get_runs(conn, stranger, project.id, workflow.id)
     end
   end
 
@@ -345,6 +358,43 @@ defmodule LightningWeb.API.WorkflowHealthControllerTest do
     end
   end
 
+  describe "GET /health/runs" do
+    test "returns every bucket in the window, counted by state", %{
+      conn: conn,
+      user: user,
+      project: project,
+      workflow: workflow
+    } do
+      trigger = hd(workflow.triggers)
+      at = DateTime.add(DateTime.utc_now(), -90, :minute)
+
+      work_order =
+        insert(:workorder,
+          workflow: workflow,
+          trigger: trigger,
+          dataclip: insert(:dataclip),
+          state: :failed,
+          last_activity: at
+        )
+
+      insert(:run,
+        work_order: work_order,
+        starting_trigger: trigger,
+        dataclip: insert(:dataclip),
+        state: :failed,
+        inserted_at: at
+      )
+
+      response =
+        conn
+        |> get_runs(user, project.id, workflow.id, %{"days" => "1"})
+        |> json_response(200)
+
+      assert Enum.sum_by(response["buckets"], & &1["failed"]) == 1
+      assert Enum.sum_by(response["buckets"], & &1["success"]) == 0
+    end
+  end
+
   describe "?days=" do
     @accepted_days [1, 7, 30]