DEV Community: Chad Wilken

Sidekiq Request ID Tracking

Chad Wilken — Fri, 08 Jul 2022 15:47:38 +0000

Sidekiq is the go to job system for most Ruby and Rails applications. It's blazing fast, well supported, trusted by thousands, and best of all it's free! However, I wish that it included a default method for tracking the Rails request_id that caused the job to be enqueued. This is useful if we want to see which controller action kicked off the job. No worries though, Sidekiq has the concept of middleware - similar to Rack - that allows us to implement functionality on the Sidekiq client and/or server.

Let's Get Started

Before we can implement the Sidekiq middleware we need a place to store request_id for the current request. In our case we use the Current model. Rails has some magic that resets the attributes defined on Current before and after each request - this makes it a great place to store thread-isolated data. That being said, let's go ahead and add the request_id attribute to our Current model.

class Current < ActiveSupport::CurrentAttributes
  attribute :request_id
end

We can use a before_action in our ApplicationController so that every request will store it's request_id.

class ApplicationController < ActionController::Base
  before_action do
    Current.request_id = request.request_id
  end
end

Now that we have global access to the request_id we need to create some client and server middleware for Sidekiq.

Client Middleware

Client middleware runs before the job is pushed to Redis - allowing us to modify the job. Knowing this, the middleware is pretty easy.

# You can put this in lib/request_id/sidekiq/client.rb

module RequestId
  module Sidekiq
    class Client
      def call(worker_klass, job, queue, redis_pool)
        # Stash the request_id in the job's metadata
        job['request_id'] = Current.request_id
        yield
      end
    end
  end
end

Server Middleware

The server middleware runs around the execution the job, allowing us to do something before and after the job's execution. Once again it's a class that responds to call.

# You can put this in lib/request_id/sidekiq/server.rb

module RequestId
  module Sidekiq
    class Server
      def call(worker, job, queue)
        Current.request_id = job['request_id']
        yield
      ensure
        Current.reset
      end
    end
  end
end

On the server we pull the request_id from the job's metadata and set the value on Current. We'll want to use ensure and call Current.reset so that all attributes are reset when the job completes - whether it's successful or not.

Hooking it Up

Now that we have our middleware let's tell Sidekiq to use it.

# config/initializers/sidekiq.rb

require 'lib/request_id/sidekiq/client'
require 'lib/request_id/sidekiq/server'

Sidekiq.configure_server do |config|
  if Sidekiq.server?
    config.server_middleware do |chain|
      chain.add RequestId::Sidekiq::Server
    end
  end
end

Sidekiq.configure_client do |config|
  config.client_middleware do |chain|
    chain.add RequestId::Sidekiq::Client
  end
end

Now in our job we can include the request_id that triggered the job in the logs.

class ImportUsersFromCsvWorker
  include Sidekiq::Worker

  def perform(csv)
    # …do stuff
    Rails.logger.info({ message: 'Imported Users', request_id: Current.request_id })
  end
end

Summary

Knowing what action triggered a job can be quite powerful while debugging. In a few lines of code we can implement middleware for Sidekiq allowing us to use the request_id of the request that caused our job to be enqueued. We could even go a step further and add the request_id to Sidekiq's context and include it in the default logs if we wanted to.

What Sidekiq middleware have you implemented or found useful in your app?

Building a Workflow for Async Searchkick Reindexing

Chad Wilken — Wed, 29 Jun 2022 20:36:17 +0000

We lean heavily on Elasticsearch here at CompanyCam, especially to serve our highly filterable project feed. It's incredibly fast, even when you apply multiple filters to your query while searching a largish data set. Our primary interface for interacting with Elasticsearch is using the Searchkick gem. Searchkick is a powerhouse with many features out of the box, but we bump up against the edges when trying to reindex a large collection.

Mo' Projects, Mo' Problems

CompanyCam houses just under 21 million projects with tens of thousands of new projects added daily. On occasion we change the fields that we want to be available for filtering. In order to make the new field(s) available, we have to reindex the entire collection of records. Reindexing a collection of this size - where each record additionally pulls in values from associated records - can be quite slow. If we run the reindex synchronously it takes about 10 hours, and that is with eager loading of the associations and other optimizations. Never fear though, Searchkick accounts for this and has the ability to use ActiveJob to reindex asynchronously. The one thing that isn't accounted for is how to promote that index once the indexing is complete. You can run the task like reindex(async: { wait: true }) which will run the indexing operation async and do a periodic pull waiting for indexing to complete and then promote the index. This almost works but can still take hours and I hate to sit on a server instance waiting for this to complete - what if I get disconnected or the instance terminates due to a deploy? We decided that it was time to build a small workflow around indexing large collections asynchronously that automatically promotes itself upon completion.

Enter Tooling

My first goal was to start a collection indexing operation from our internal admin tool, Dash, and monitor the progress. With those two simple goals in mind, this is what I came up with.

I needed two jobs - one job to enqueue the indexing operation and another to monitor the operation until completion. Once those are built, I can use Rails basics to wrap this in a minimalistic UI.

The Jobs

The first job needed to:

Accept a class name and start the reindex for the given class
Store the pending index name for later usage, optimally in Redis
Enqueue another job to monitor the progress of the indexing operation

This is what I came up with:

module Searchkick
  class PerformAsyncReindexWorker
    include Sidekiq::Worker

    sidekiq_options retry: 0

    def perform(klass)
      result = klass.constantize.reindex(async: true)
      index_name = result[:index_name]
      Searchkick::AsyncReindexStatus.new.currently_reindexing << index_name
      Searchkick::MonitorAsyncReindexWorker.perform_in(5.seconds, klass, index_name)
    end
  end
end

The second job, as you may have guessed, monitors the indexing progress until completion. The basic functionality needs to check the operation and:

If incomplete, re-enqueue itself to check again in 10 seconds
If complete, promote the index for the given class and removes the index name from the collection in Redis.

module Searchkick
  class MonitorAsyncReindexWorker
    include Sidekiq::Worker

    def perform(klass, index_name)
      status = Searchkick.reindex_status(index_name)
      if status[:completed]
        klass.constantize.search_index.promote(index_name)
        Searchkick::AsyncReindexStatus.new.currently_reindexing.delete(index_name)
      else
        self.class.perform_in(10.seconds, klass, index_name)
      end
    end
  end
end

By default the Searchkick::BulkReindexJob uses the same queue as regular async reindexing, blocking user generated content from being indexed while performing a full reindex. So I also patched the Searchkick::BulkReindexJob to use a custom queue we have just for performing full collection indexing operations. In an initializer I simply did:

class Searchkick::BulkReindexJob
  queue_as { 'searchkick_full_reindex' }
end

The Status Object

You may be wondering what Searchkick::AsyncReindexStatus is. It is a simple class that includes the Redis::Objects library so that we can store a list of currently reindexing collections. It looks like this:

module Searchkick
  class AsyncReindexStatus
    include Redis::Objects

    def id
      'searchkick-async-reindex-status'
    end

    list :currently_reindexing
  end
end

Note: I opted to use Redis::Objects since it was already in our codebase and it is a bit simpler than interacting with Redis directly using Searchkick.redis.

How to Kick off the Job

An indexing operation can be kicked off in one of two ways. You can start it via the command-line if you have access such as Searchkick::PerformAsyncReindexWorker.perform_async(model_class). Instead, we built a crude interface into our internal admin tool. The UI allows us to select a model and start the indexing operation and then track it's status until completion.

The Code

For the full code that we use you can look at this gist. Always happy to hear improvements that could be made as well!

Recap

Searchkick saves us serious time and energy developing features. By taking one of its existing features like async reindexing and wrapping it in a bit of workflow, we can get even more out of it. In the end, we were able to scratch our own itch for truly async indexing operations.