Remove `:enqueued` counter
This commit is contained in:
parent
75290cbfbd
commit
26693292f4
|
@ -5,9 +5,9 @@
|
||||||
defmodule Pleroma.JobQueueMonitor do
|
defmodule Pleroma.JobQueueMonitor do
|
||||||
use GenServer
|
use GenServer
|
||||||
|
|
||||||
@initial_state %{workers: %{}, queues: %{}, processed_jobs: 0, enqueued: 0}
|
@initial_state %{workers: %{}, queues: %{}, processed_jobs: 0}
|
||||||
@queue %{processed_jobs: 0, success: 0, failure: 0, enqueued: 0}
|
@queue %{processed_jobs: 0, success: 0, failure: 0}
|
||||||
@operation %{processed_jobs: 0, success: 0, failure: 0, enqueued: 0}
|
@operation %{processed_jobs: 0, success: 0, failure: 0}
|
||||||
|
|
||||||
def start_link(_) do
|
def start_link(_) do
|
||||||
GenServer.start_link(__MODULE__, @initial_state, name: __MODULE__)
|
GenServer.start_link(__MODULE__, @initial_state, name: __MODULE__)
|
||||||
|
@ -25,15 +25,6 @@ def stats do
|
||||||
GenServer.call(__MODULE__, :stats)
|
GenServer.call(__MODULE__, :stats)
|
||||||
end
|
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
|
def handle_event([:oban, status], %{duration: duration}, meta, _) do
|
||||||
GenServer.cast(__MODULE__, {:process_event, status, duration, meta})
|
GenServer.cast(__MODULE__, {:process_event, status, duration, meta})
|
||||||
end
|
end
|
||||||
|
@ -43,24 +34,6 @@ def handle_call(:stats, _from, state) do
|
||||||
{:reply, state, state}
|
{:reply, state, state}
|
||||||
end
|
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
|
@impl true
|
||||||
def handle_cast({:process_event, status, duration, meta}, state) do
|
def handle_cast({:process_event, status, duration, meta}, state) do
|
||||||
state =
|
state =
|
||||||
|
@ -76,12 +49,11 @@ def handle_cast({:process_event, status, duration, meta}, state) do
|
||||||
|> Map.update!(meta.queue, &update_queue(&1, status, meta, duration))
|
|> Map.update!(meta.queue, &update_queue(&1, status, meta, duration))
|
||||||
end)
|
end)
|
||||||
|> Map.update!(:processed_jobs, &(&1 + 1))
|
|> Map.update!(:processed_jobs, &(&1 + 1))
|
||||||
|> decr_enqueued()
|
|
||||||
|
|
||||||
{:noreply, state}
|
{:noreply, state}
|
||||||
end
|
end
|
||||||
|
|
||||||
defp update_worker(worker, status, meta, duration \\ 0) do
|
defp update_worker(worker, status, meta, duration) do
|
||||||
worker
|
worker
|
||||||
|> Map.put_new(meta.args["op"], @operation)
|
|> Map.put_new(meta.args["op"], @operation)
|
||||||
|> Map.update!(meta.args["op"], &update_op(&1, status, meta, duration))
|
|> Map.update!(meta.args["op"], &update_op(&1, status, meta, duration))
|
||||||
|
@ -96,20 +68,11 @@ defp update_op(op, status, _meta, _duration) do
|
||||||
op
|
op
|
||||||
|> Map.update!(:processed_jobs, &(&1 + 1))
|
|> Map.update!(:processed_jobs, &(&1 + 1))
|
||||||
|> Map.update!(status, &(&1 + 1))
|
|> Map.update!(status, &(&1 + 1))
|
||||||
|> decr_enqueued()
|
|
||||||
end
|
end
|
||||||
|
|
||||||
defp update_queue(queue, status, _meta, _duration) do
|
defp update_queue(queue, status, _meta, _duration) do
|
||||||
queue
|
queue
|
||||||
|> Map.update!(:processed_jobs, &(&1 + 1))
|
|> Map.update!(:processed_jobs, &(&1 + 1))
|
||||||
|> Map.update!(status, &(&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
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -40,7 +40,6 @@ def enqueue(op, params, worker_args \\ []) do
|
||||||
unquote(caller_module)
|
unquote(caller_module)
|
||||||
|> apply(:new, [params, worker_args])
|
|> apply(:new, [params, worker_args])
|
||||||
|> Pleroma.Repo.insert()
|
|> Pleroma.Repo.insert()
|
||||||
|> Pleroma.JobQueueMonitor.enqueue()
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
Loading…
Reference in New Issue