Failure backend

.gitignore

@@ -17,3 +17,6 @@ erl_crash.dump
 *.ez
 
 /logs
+
+# asdf version file
+.tool-versions

README.md

@@ -416,6 +416,29 @@ All workers will restart automatically once the new connection is established.
 TaskBunny aims to provide zero hassle and recover automatically regardless how
 long the host takes to come back and accessible.
 
+#### Failure backends
+
+By default, when the error occurs during the job execution TaskBunny reports it
+to Logger. If you want to report the error to different services, you can configure
+your custom failure backend.
+
+```elixir
+config :task_bunny, failure_backend: [YourApp.CustomFailureBackend]
+```
+
+You can also report the errors to the multiple backends. For example, if you
+want to use our default Logger backend with your custom backend you can
+configure like below:
+
+```elixir
+config :task_bunny, failure_backend: [
+  TaskBunny.FailureBackend.Logger,
+  YourApp.CustomFailureBackend
+]
+```
+
+Check out the implementation of [TaskBunny.FailureBackend.Logger](https://github.com/shinyscorpion/task_bunny/blob/master/lib/task_bunny/failure_backend/logger.ex) to learn how to write your custom failure backend.
+
 ## Monitoring
 
 #### RabbitMQ plugins

lib/task_bunny/config.ex

@@ -185,4 +185,18 @@ defmodule TaskBunny.Config do
       _ -> []
     end
   end
+
+  @doc """
+  Returns the list of failure backends.
+
+  It returns [TaskBunny.FailureBackend.Logger] by default.
+  """
+  @spec failure_backend :: [atom]
+  def failure_backend do
+    case Application.fetch_env(:task_bunny, :failure_backend) do
+      {:ok, list} when is_list(list) -> list
+      {:ok, atom} when is_atom(atom) -> [atom]
+      _ -> [TaskBunny.FailureBackend.Logger]
+    end
+  end
 end

lib/task_bunny/failure_backend.ex

@@ -0,0 +1,55 @@
+defmodule TaskBunny.FailureBackend do
+  @moduledoc """
+  A behaviour module to implment the your own failure backend.
+
+  Note the backend is called only for the errors caught during job processing.
+  Any other errors won't be reported to the backend.
+
+  ## Configuration
+
+  By default, TaskBunny reports the job failures to Logger.
+  If you want to report the error to different services, you can configure
+  your custom failure backend.
+
+      config :task_bunny, failure_backend: [YourApp.CustomFailureBackend]
+
+  You can also report the errors to the multiple backends. For example, if you
+  want to use our default Logger backend with your custom backend you can
+  configure like below:
+
+      config :task_bunny, failure_backend: [
+        TaskBunny.FailureBackend.Logger,
+        YourApp.CustomFailureBackend
+      ]
+
+  ## Example
+
+  See the implmentation of `TaskBunny.FailureBackend.Logger`.
+
+  ## Argument
+
+  See `TaskBunny.JobError` for the details.
+
+  """
+  alias TaskBunny.{JobError, Config, FailureBackend}
+
+  @doc """
+  Callback to report a job error.
+  """
+  @callback report_job_error(JobError.t) :: any
+
+  defmacro __using__(_options \\ []) do
+    quote do
+      @behaviour FailureBackend
+    end
+  end
+
+  @doc false
+  @spec report_job_error(JobError.t) :: :ok
+  def report_job_error(job_error = %JobError{}) do
+    Config.failure_backend()
+    |> Enum.each(&(&1.report_job_error(job_error)))
+
+    :ok
+  end
+end

lib/task_bunny/failure_backend/logger.ex

@@ -0,0 +1,99 @@
+defmodule TaskBunny.FailureBackend.Logger do
+  @moduledoc """
+  Default failure backend that reports job errors to Logger.
+  """
+  use TaskBunny.FailureBackend
+  require Logger
+  alias TaskBunny.JobError
+
+  def report_job_error(error = %JobError{error_type: :exception}) do
+    message = """
+    TaskBunny - #{error.job} failed for an exception.
+
+    Exception:
+    #{my_inspect error.exception}
+
+    #{common_message error}
+
+    Stacktrace:
+    #{Exception.format_stacktrace(error.stacktrace)}
+    """
+
+    do_report(message, error.reject)
+  end
+
+  def report_job_error(error = %JobError{error_type: :return_value}) do
+    message = """
+    TaskBunny - #{error.job} failed for an invalid return value.
+
+    Return value:
+    #{my_inspect error.return_value}
+
+    #{common_message error}
+    """
+
+    do_report(message, error.reject)
+  end
+
+  def report_job_error(error = %JobError{error_type: :exit}) do
+    message = """
+    TaskBunny - #{error.job} failed for EXIT signal.
+
+    Reason:
+    #{my_inspect error.reason}
+
+    #{common_message error}
+    """
+
+    do_report(message, error.reject)
+  end
+
+  def report_job_error(error = %JobError{error_type: :timeout}) do
+    message = """
+    TaskBunny - #{error.job} failed for timeout.
+
+    #{common_message error}
+    """
+
+    do_report(message, error.reject)
+  end
+
+  def report_job_error(error) do
+    message = """
+    TaskBunny - Failed with the unknown error type.
+
+    Error dump:
+    #{my_inspect error}
+    """
+
+    do_report(message, true)
+  end
+
+  defp do_report(message, rejected) do
+    if rejected do
+      Logger.error message
+    else
+      Logger.warn message
+    end
+  end
+
+  defp common_message(error) do
+    """
+    Payload:
+      #{my_inspect error.payload}
+
+    History:
+      - Failed count: #{error.failed_count}
+      - Reject: #{error.reject}
+
+    Worker:
+      - Queue: #{error.queue}
+      - Concurrency: #{error.concurrency}
+      - PID: #{inspect error.pid}
+    """
+  end
+
+  defp my_inspect(arg) do
+    inspect arg, pretty: true, width: 100
+  end
+end

lib/task_bunny/job_error.ex

@@ -0,0 +1,97 @@
+defmodule TaskBunny.JobError do
+  @moduledoc """
+  A struct that holds an error information occured during the job processing.
+
+  ## Attributes
+
+  - job: the job module failed
+  - payload: the payload(arguments) for the job execution
+  - error_type: the type of the error. :exception, :return_value, :timeout or :exit
+  - exception: the inner exception (option)
+  - stacktrace: the stacktrace (only available for the exception)
+  - return_value: the return value from the job (only available for the return value error)
+  - reason: the reason information passed with EXIT signal (only available for exit error)
+  - raw_body: the raw body for the message
+  - meta: the meta data given by RabbitMQ
+  - failed_count: the number of failures for the job processing request
+  - queue: the name of the queue
+  - concurrency: the number of concurrent job processing of the worker
+  - pid: the process ID of the worker
+  - reject: sets true if the job is rejected for the failure (means it won't be retried again)
+
+  """
+
+  @type t :: %__MODULE__{
+    job: atom,
+    payload: any,
+    error_type: :exception | :return_value | :timeout | :exit | nil,
+    exception: struct | nil,
+    stacktrace: list(tuple) | nil,
+    return_value: any,
+    reason: any,
+    raw_body: String.t,
+    meta: map,
+    failed_count: integer,
+    queue: String.t,
+    concurrency: integer,
+    pid: pid,
+    reject: boolean
+  }
+
+  defstruct [
+    job: nil,
+    payload: nil,
+    error_type: nil,
+    exception: nil,
+    stacktrace: nil,
+    return_value: nil,
+    reason: nil,
+    raw_body: "",
+    meta: %{},
+    failed_count: 0,
+    queue: "",
+    concurrency: 1,
+    pid: nil,
+    reject: false
+  ]
+
+  @doc false
+  def handle_exception(job, payload, exception) do
+    %__MODULE__{
+      job: job,
+      payload: payload,
+      error_type: :exception,
+      exception: exception,
+      stacktrace: System.stacktrace()
+    }
+  end
+
+  @doc false
+  def handle_exit(job, payload, reason) do
+    %__MODULE__{
+      job: job,
+      payload: payload,
+      error_type: :exit,
+      reason: reason
+    }
+  end
+
+  @doc false
+  def handle_return_value(job, payload, return_value) do
+    %__MODULE__{
+      job: job,
+      payload: payload,
+      error_type: :return_value,
+      return_value: return_value
+    }
+  end
+
+  @doc false
+  def handle_timeout(job, payload) do
+    %__MODULE__{
+      job: job,
+      payload: payload,
+      error_type: :timeout
+    }
+  end
+end

lib/task_bunny/job_runner.ex

@@ -22,6 +22,7 @@ defmodule TaskBunny.JobRunner do
   @moduledoc false
 
   require Logger
+  alias TaskBunny.JobError
 
   @doc ~S"""
   Invokes the given job with the given payload.
@@ -33,10 +34,10 @@ defmodule TaskBunny.JobRunner do
   def invoke(job, payload, message) do
     caller = self()
 
-    time_out_error = {:error, "#{inspect job} timed out with #{job.timeout}"}
+    timeout_error = {:error, JobError.handle_timeout(job, payload)}
     timer = Process.send_after(
       caller,
-      {:job_finished, time_out_error, message},
+      {:job_finished, timeout_error, message},
       job.timeout
     )
 
@@ -52,14 +53,18 @@ defmodule TaskBunny.JobRunner do
   # Any raises or throws in the perform are caught and turned into an :error tuple.
   @spec run_job(atom, any) :: :ok | {:ok, any} | {:error, any}
   defp run_job(job, payload) do
-    job.perform(payload)
+    case job.perform(payload) do
+      :ok -> :ok
+      {:ok, something} -> {:ok, something}
+      error -> {:error, JobError.handle_return_value(job, payload, error)}
+    end
   rescue
     error ->
-      Logger.error "TaskBunny.JobRunner - Runner rescued #{inspect error}"
-      {:error, error}
+      Logger.debug "TaskBunny.JobRunner - Runner rescued #{inspect error}"
+      {:error, JobError.handle_exception(job, payload, error)}
   catch
     _, reason ->
-      Logger.error "TaskBunny.JobRunner - Runner caught reason: #{inspect reason}"
-      {:error, reason}
+      Logger.debug "TaskBunny.JobRunner - Runner caught reason: #{inspect reason}"
+      {:error, JobError.handle_exit(job, payload, reason)}
   end
 end