X-Git-Url: https://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fjob_queue_monitor.ex;h=b5f124923101d9c54b6557034e4eeb78ee6bf3ac;hb=9a320ba8149a3efae1235b3b68dbf1bbad423e0d;hp=685ba2eada6454b01b7e032c52c0f211cc9744ba;hpb=75290cbfbd0f93cf3ecec5f44c4624b8c8601c51;p=akkoma

diff --git a/lib/pleroma/job_queue_monitor.ex b/lib/pleroma/job_queue_monitor.ex
index 685ba2ead..b5f124923 100644
--- a/lib/pleroma/job_queue_monitor.ex
+++ b/lib/pleroma/job_queue_monitor.ex
@@ -1,13 +1,13 @@
 # Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.JobQueueMonitor do
   use GenServer
 
-  @initial_state %{workers: %{}, queues: %{}, processed_jobs: 0, enqueued: 0}
-  @queue %{processed_jobs: 0, success: 0, failure: 0, enqueued: 0}
-  @operation %{processed_jobs: 0, success: 0, failure: 0, enqueued: 0}
+  @initial_state %{workers: %{}, queues: %{}, processed_jobs: 0}
+  @queue %{processed_jobs: 0, success: 0, failure: 0}
+  @operation %{processed_jobs: 0, success: 0, failure: 0}
 
   def start_link(_) do
     GenServer.start_link(__MODULE__, @initial_state, name: __MODULE__)
@@ -15,8 +15,8 @@ defmodule Pleroma.JobQueueMonitor do
 
   @impl true
   def init(state) do
-    :telemetry.attach("oban-monitor-failure", [:oban, :failure], &handle_event/4, nil)
-    :telemetry.attach("oban-monitor-success", [:oban, :success], &handle_event/4, nil)
+    :telemetry.attach("oban-monitor-failure", [:oban, :job, :exception], &handle_event/4, nil)
+    :telemetry.attach("oban-monitor-success", [:oban, :job, :stop], &handle_event/4, nil)
 
     {:ok, state}
   end
@@ -25,17 +25,11 @@ defmodule Pleroma.JobQueueMonitor do
     GenServer.call(__MODULE__, :stats)
   end
 
-  def enqueue({:ok, job}) do
-    meta = Map.take(job, [:args, :queue, :worker])
-    GenServer.cast(__MODULE__, {:process_enqueue, meta})
-
-    {:ok, job}
-  end
-
-  def enqueue(result), do: result
-
-  def handle_event([:oban, status], %{duration: duration}, meta, _) do
-    GenServer.cast(__MODULE__, {:process_event, status, duration, meta})
+  def handle_event([:oban, :job, event], %{duration: duration}, meta, _) do
+    GenServer.cast(
+      __MODULE__,
+      {:process_event, mapping_status(event), duration, meta}
+    )
   end
 
   @impl true
@@ -43,24 +37,6 @@ defmodule Pleroma.JobQueueMonitor do
     {:reply, state, state}
   end
 
-  def handle_cast({:process_enqueue, meta}, state) do
-    state =
-      state
-      |> Map.update!(:workers, fn workers ->
-        workers
-        |> Map.put_new(meta.worker, %{})
-        |> Map.update!(meta.worker, &update_worker(&1, :enqueue, meta))
-      end)
-      |> Map.update!(:queues, fn workers ->
-        workers
-        |> Map.put_new(meta.queue, @queue)
-        |> Map.update!(meta.queue, fn queue -> Map.update!(queue, :enqueued, &(&1 + 1)) end)
-      end)
-      |> Map.update!(:enqueued, &(&1 + 1))
-
-    {:noreply, state}
-  end
-
   @impl true
   def handle_cast({:process_event, status, duration, meta}, state) do
     state =
@@ -76,12 +52,11 @@ defmodule Pleroma.JobQueueMonitor do
         |> Map.update!(meta.queue, &update_queue(&1, status, meta, duration))
       end)
       |> Map.update!(:processed_jobs, &(&1 + 1))
-      |> decr_enqueued()
 
     {:noreply, state}
   end
 
-  defp update_worker(worker, status, meta, duration \\ 0) do
+  defp update_worker(worker, status, meta, duration) do
     worker
     |> Map.put_new(meta.args["op"], @operation)
     |> Map.update!(meta.args["op"], &update_op(&1, status, meta, duration))
@@ -96,20 +71,14 @@ defmodule Pleroma.JobQueueMonitor do
     op
     |> Map.update!(:processed_jobs, &(&1 + 1))
     |> Map.update!(status, &(&1 + 1))
-    |> decr_enqueued()
   end
 
   defp update_queue(queue, status, _meta, _duration) do
     queue
     |> Map.update!(:processed_jobs, &(&1 + 1))
     |> Map.update!(status, &(&1 + 1))
-    |> decr_enqueued()
   end
 
-  defp decr_enqueued(map) do
-    Map.update!(map, :enqueued, fn
-      0 -> 0
-      enqueued -> enqueued - 1
-    end)
-  end
+  defp mapping_status(:stop), do: :success
+  defp mapping_status(:exception), do: :failure
 end