Filtering Job Arguments

Adds a new `filter_arguments` option that allows you to filter out sensitive data from job arguments.
This option can be configured globally or per application.

Note: Currently, only root-level hash keys are supported.
If needed, support for key paths, procs, and regular expressions could be added in the future.

Author: intrip

README.md

@@ -114,6 +114,8 @@ Besides `base_controller_class`, you can also set the following for `MissionCont
 - `scheduled_job_delay_threshold`: the time duration before a scheduled job is considered delayed. Defaults to `1.minute` (a job is considered delayed if it hasn't transitioned from the `scheduled` status 1 minute after the scheduled time).
 - `show_console_help`: whether to show the console help. If you don't want the console help message, set this to `false`—defaults to `true`.
 - `backtrace_cleaner`: a backtrace cleaner used for optionally filtering backtraces on the Failed Jobs detail page. Defaults to `Rails::BacktraceCleaner.new`. See the [Advanced configuration](#advanced-configuration) section for how to configure/override this setting on a per application/server basis.
+- `filter_arguments`: an array of job argument keys that you want to filter out in the UI. This is useful for hiding sensitive user data. You can also override this option for each application. Currently, only root-level hash keys are supported, and it only works with the `Resque` and `SolidQueue` adapters. See the [Advanced configuration](#advanced-configuration) section for an example.
+```ruby
 
 This library extends Active Job with a querying interface and the following setting:
 - `config.active_job.default_page_size`: the internal batch size that Active Job will use when sending queries to the underlying adapter and the batch size for the bulk operations defined above—defaults to `1000`.
@@ -184,7 +186,9 @@ SERVERS_BY_APP.each do |app, servers|
     # [ server, [ queue_adapter, BacktraceCleaner.new ]]  # with optional backtrace cleaner
   end.to_h
 
-  MissionControl::Jobs.applications.add(app, queue_adapters_by_name)
+  filter_arguments = %i[ author ]
+
+  MissionControl::Jobs.applications.add(app, queue_adapters_by_name, filter_arguments)
 end
 ```
 

app/helpers/mission_control/jobs/jobs_helper.rb

@@ -67,7 +67,8 @@ def as_renderable_hash(argument)
       elsif argument["_aj_serialized"]
         ActiveJob::Arguments.deserialize([ argument ]).first
       else
-        argument.without("_aj_symbol_keys", "_aj_ruby2_keywords")
+        ActiveJob::JobArgumentFilter.filter_argument_hash(argument)
+          .without("_aj_symbol_keys", "_aj_ruby2_keywords")
           .transform_values { |v| as_renderable_argument(v) }
           .map { |k, v| "#{k}: #{v}" }
           .join(", ")

lib/active_job/job_argument_filter.rb

@@ -0,0 +1,28 @@
+class ActiveJob::JobArgumentFilter
+  FILTERED = "[FILTERED]"
+
+  class << self
+    def filter_arguments(arguments)
+      arguments.each do |argument|
+        if argument.is_a?(Hash)
+          filter_argument_hash(argument)
+        end
+      end
+    end
+
+    def filter_argument_hash(argument)
+      return argument if filters.blank?
+
+      argument.each do |key, value|
+        if filters.include?(key.to_s)
+          argument[key] = FILTERED
+        end
+      end
+    end
+
+    private
+      def filters
+        MissionControl::Jobs::Current.application&.filter_arguments
+      end
+  end
+end

lib/active_job/queue_adapters/resque_ext.rb

@@ -188,13 +188,19 @@ def deserialize_resque_job(resque_job_hash, index)
           args_hash = resque_job_hash.dig("payload", "args") || resque_job_hash.dig("args")
           ActiveJob::JobProxy.new(args_hash&.first).tap do |job|
             job.last_execution_error = execution_error_from_resque_job(resque_job_hash)
-            job.raw_data = resque_job_hash
+            job.raw_data = filter_raw_data_arguments(resque_job_hash)
             job.position = jobs_relation.offset_value + index
             job.failed_at = resque_job_hash["failed_at"]&.to_datetime&.utc
             job.status = job.failed_at.present? ? :failed : :pending
           end
         end
 
+        def filter_raw_data_arguments(raw_data)
+          args = raw_data.dig("payload", "args")
+          ActiveJob::JobArgumentFilter.filter_arguments(args.first["arguments"]) if args
+          raw_data
+        end
+
         def execution_error_from_resque_job(resque_job_hash)
           if resque_job_hash["exception"].present?
             ActiveJob::ExecutionError.new \

lib/active_job/queue_adapters/solid_queue_ext.rb

@@ -98,7 +98,7 @@ def deserialize_and_proxy_solid_queue_job(solid_queue_job, job_status = nil)
       ActiveJob::JobProxy.new(solid_queue_job.arguments).tap do |job|
         job.status = job_status
         job.last_execution_error = execution_error_from_solid_queue_job(solid_queue_job) if job_status == :failed
-        job.raw_data = solid_queue_job.as_json
+        job.raw_data = filter_raw_data_arguments(solid_queue_job.as_json)
         job.failed_at = solid_queue_job&.failed_execution&.created_at if job_status == :failed
         job.finished_at = solid_queue_job.finished_at
         job.blocked_by = solid_queue_job.concurrency_key
@@ -109,6 +109,12 @@ def deserialize_and_proxy_solid_queue_job(solid_queue_job, job_status = nil)
       end
     end
 
+    def filter_raw_data_arguments(raw_data)
+      arguments = raw_data.dig("arguments", "arguments")
+      ActiveJob::JobArgumentFilter.filter_arguments(arguments)
+      raw_data
+    end
+
     def status_from_solid_queue_job(solid_queue_job)
       SolidQueueJobs::STATUS_MAP.invert[solid_queue_job.status]
     end

lib/mission_control/jobs.rb

@@ -26,6 +26,8 @@ module Jobs
     mattr_accessor :show_console_help, default: true
     mattr_accessor :backtrace_cleaner
 
+    mattr_accessor :filter_arguments, default: []
+
     mattr_accessor :importmap, default: Importmap::Map.new
 
     mattr_accessor :http_basic_auth_user

lib/mission_control/jobs/application.rb

@@ -2,11 +2,12 @@
 class MissionControl::Jobs::Application
   include MissionControl::Jobs::IdentifiedByName
 
-  attr_reader :servers
+  attr_reader :servers, :filter_arguments
 
   def initialize(name:)
     super
     @servers = MissionControl::Jobs::IdentifiedElements.new
+    @filter_arguments = []
   end
 
   def add_servers(queue_adapters_by_name)
@@ -17,4 +18,8 @@ def add_servers(queue_adapters_by_name)
         backtrace_cleaner: cleaner, application: self)
     end
   end
+
+  def filter_arguments=(arguments)
+    @filter_arguments = Array(arguments).map(&:to_s)
+  end
 end

lib/mission_control/jobs/applications.rb

@@ -1,8 +1,9 @@
 # A container to register applications
 class MissionControl::Jobs::Applications < MissionControl::Jobs::IdentifiedElements
-  def add(name, queue_adapters_by_name = {})
+  def add(name, queue_adapters_by_name = {}, filter_arguments = [])
     self << MissionControl::Jobs::Application.new(name: name).tap do |application|
       application.add_servers(queue_adapters_by_name)
+      application.filter_arguments = filter_arguments.presence || MissionControl::Jobs.filter_arguments
     end
   end
 end

test/active_job/job_argument_filter_test.rb

@@ -0,0 +1,38 @@
+require "test_helper"
+
+class ActiveJob::JobArgumentFilterTest < ActiveSupport::TestCase
+  setup do
+    @application = MissionControl::Jobs::Application.new(name: "BC4")
+    MissionControl::Jobs::Current.application = @application
+  end
+
+  test "filter_arguments" do
+    arguments = [
+      "deliver",
+      {
+        email_address: "jorge@37signals.com",
+        profile: { name: "Jorge Manrubia" },
+        message: "Hello!"
+      }
+    ]
+    @application.filter_arguments = %i[ email_address message ]
+
+    filtered = ActiveJob::JobArgumentFilter.filter_arguments(arguments)
+
+    assert_equal "deliver", filtered[0]
+    assert_equal({ email_address: "[FILTERED]", profile: { name: "Jorge Manrubia" }, message: "[FILTERED]" }, filtered[1])
+  end
+
+  test "filter_argument_hash" do
+    argument = {
+      email_address: "jorge@37signals.com",
+      message: "Hello!"
+    }
+    filtered = ActiveJob::JobArgumentFilter.filter_argument_hash(argument)
+    assert_equal({ email_address: "jorge@37signals.com", message: "Hello!" }, filtered)
+
+    @application.filter_arguments = %i[ message ]
+    filtered = ActiveJob::JobArgumentFilter.filter_argument_hash(argument)
+    assert_equal({ email_address: "jorge@37signals.com", message: "[FILTERED]" }, filtered)
+  end
+end

test/dummy/config/initializers/mission_control_jobs.rb

@@ -26,5 +26,7 @@ def redis_connection_for(app, server)
     [ server, queue_adapter ]
   end.to_h
 
-  MissionControl::Jobs.applications.add(app, queue_adapters_by_name)
+  filter_arguments = %i[ author ]
+
+  MissionControl::Jobs.applications.add(app, queue_adapters_by_name, filter_arguments)
 end