Page MenuHomePhorge
Files F85710910

No OneTemporary
Actions

Size
5 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lib/pleroma/job_queue_monitor.ex b/lib/pleroma/job_queue_monitor.ex
index 685ba2ead..3feea8381 100644
--- a/lib/pleroma/job_queue_monitor.ex
+++ b/lib/pleroma/job_queue_monitor.ex
@@ -1,115 +1,78 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 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__)
end
@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)
{:ok, state}
end
def stats 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})
end
@impl true
def handle_call(:stats, _from, state) 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 =
state
|> Map.update!(:workers, fn workers ->
workers
|> Map.put_new(meta.worker, %{})
|> Map.update!(meta.worker, &update_worker(&1, status, meta, duration))
end)
|> Map.update!(:queues, fn workers ->
workers
|> Map.put_new(meta.queue, @queue)
|> 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))
end
defp update_op(op, :enqueue, _meta, _duration) do
op
|> Map.update!(:enqueued, &(&1 + 1))
end
defp update_op(op, status, _meta, _duration) 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
end
diff --git a/lib/pleroma/workers/worker_helper.ex b/lib/pleroma/workers/worker_helper.ex
index a43ce8bc0..358efa14a 100644
--- a/lib/pleroma/workers/worker_helper.ex
+++ b/lib/pleroma/workers/worker_helper.ex
@@ -1,47 +1,46 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Workers.WorkerHelper do
alias Pleroma.Config
alias Pleroma.Workers.WorkerHelper
def worker_args(queue) do
case Config.get([:workers, :retries, queue]) do
nil -> []
max_attempts -> [max_attempts: max_attempts]
end
end
def sidekiq_backoff(attempt, pow \\ 4, base_backoff \\ 15) do
backoff =
:math.pow(attempt, pow) +
base_backoff +
:rand.uniform(2 * base_backoff) * attempt
trunc(backoff)
end
defmacro __using__(opts) do
caller_module = __CALLER__.module
queue = Keyword.fetch!(opts, :queue)
quote do
# Note: `max_attempts` is intended to be overridden in `new/2` call
use Oban.Worker,
queue: unquote(queue),
max_attempts: 1
def enqueue(op, params, worker_args \\ []) do
params = Map.merge(%{"op" => op}, params)
queue_atom = String.to_atom(unquote(queue))
worker_args = worker_args ++ WorkerHelper.worker_args(queue_atom)
unquote(caller_module)
|> apply(:new, [params, worker_args])
|> Pleroma.Repo.insert()
- |> Pleroma.JobQueueMonitor.enqueue()
end
end
end
end

File Metadata

Mime Type
text/x-diff
Expires
Sat, Sep 19, 3:08 AM (15 h, 37 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1768706
Default Alt Text
(5 KB)

Event Timeline