Directory Monitoring

Contents

Watching a directory for new files and kicking off a job to process each one is one of the most common tasks in any batch processing system. Almost every team ends up writing it themselves, along with all the awkward parts: detecting when a file has finished uploading, archiving it so it is not picked up twice, retrying after failures, and securing which directories may be read.

Rocket Job ships this as a built-in feature called Dirmon (Directory Monitor). Because it is part of Rocket Job it also comes with a full management screen in Rocket Job Mission Control, the web UI, so files, schedules, and failures can be managed without writing or deploying any code.

Dirmon is driven by two pieces:

The key idea is that a DirmonEntry can populate the fields of the job it starts. A new file is not just handed to a job: the entry decides which job class runs and pre-sets its fields. That is what makes Dirmon able to launch any user-defined or built-in job, fully configured, for each incoming file.

Why per-entry job fields matter

A very common deployment looks like this: customers upload files over SFTP, each into their own account-specific directory. The files all need the same processing, but the job needs to know which account a file belongs to, and often a few other details such as an email address to notify when the output is ready for pickup.

With Dirmon this is solved by creating one DirmonEntry per account. Each entry watches that account’s directory and sets the account-specific fields on the job through its properties hash:

class CustomerImportJob < RocketJob::Job
  include RocketJob::Batch

  # Keep the job after completion so the output file can be downloaded.
  self.destroy_on_complete = false

  # Account that owns this file. Set per DirmonEntry.
  # `user_editable: true` also makes it editable in Mission Control.
  field :account_id, type: Integer, user_editable: true

  # Where to send the "your file is ready" notification.
  field :notify_email, type: String, user_editable: true

  input_category serializer: :encrypt
  output_category

  after_batch :notify_when_ready

  # Called once per record, spread across all available workers.
  def perform(row)
    Importer.new(account_id: account_id).transform(row)
  end

  private

  def notify_when_ready
    CustomerMailer.import_complete(notify_email, id).deliver_later if notify_email
  end
end

Create one entry per account, each setting the fields specific to that account:

RocketJob::DirmonEntry.create!(
  name:              "ACME customer imports",
  pattern:           "/var/sftp/acme/incoming/*.csv",
  job_class_name:    "CustomerImportJob",
  archive_directory: "/var/sftp/acme/archive",
  properties:        {
    account_id:   42,
    notify_email: "ops@acme.example.com",
    priority:     25
  }
).enable!

RocketJob::DirmonEntry.create!(
  name:              "Globex customer imports",
  pattern:           "/var/sftp/globex/incoming/*.csv",
  job_class_name:    "CustomerImportJob",
  archive_directory: "/var/sftp/globex/archive",
  properties:        {
    account_id:   77,
    notify_email: "files@globex.example.com"
  }
).enable!

Now any file dropped into /var/sftp/acme/incoming starts a CustomerImportJob with account_id 42 and the ACME notification address, while files for Globex start the same job scoped to their own account. The processing code is written once; the per-account context comes from the entry.

The properties hash can set any field on the target job that has a writer, including the built-in fields such as priority, description, and run_at, as well as the batch input_categories / output_categories. Setting a value for a field the job does not define is rejected by validation, so a typo fails fast when the entry is saved rather than silently doing nothing.

Fields that should also be editable from the Mission Control web UI must be declared with user_editable: true, as shown above. Setting properties programmatically does not require it.

How a scan works

DirmonJob does not start a job the instant a file appears, because the file may still be uploading. Instead it waits for the file to stabilize:

  1. On each run it lists the files matching every enabled entry’s pattern.
  2. A newly seen file’s size is recorded and carried forward to the next run.
  3. On the next run, if the file’s size is unchanged, it is considered complete and its job is started. If the size changed, it is recorded again and rechecked on the following run.

This means a file is normally picked up one scan interval after it finishes uploading. Object stores are a special case: on stores where partial files are not visible until the upload completes (such as Amazon S3), the file is processed on the first scan that sees it, with no stabilization wait.

When a file is ready, Dirmon archives it first and then starts the job, so the same file can never be picked up twice (see Archiving).

Creating a DirmonEntry

entry = RocketJob::DirmonEntry.create!(
  name:              "Daily price feed",
  pattern:           "/data/prices/*.csv",
  job_class_name:    "PriceFeedJob",
  archive_directory: "/data/prices/archive",
  properties:        { priority: 30 }
)

The fields of DirmonEntry:

Enabling and disabling entries

A new DirmonEntry starts in the pending state and is not scanned until it is enabled. This is deliberate: an entry can be reviewed before it goes live.

entry.enable!     # pending/disabled/failed -> enabled
entry.disable!    # enabled/failed -> disabled

The states are:

A snapshot of how many entries are in each state:

RocketJob::DirmonEntry.counts_by_state
# => { pending: 1, enabled: 37, disabled: 3, failed: 1 }

How the file reaches the job

When a file stabilizes, Dirmon archives it and enqueues a RocketJob::Jobs::UploadFileJob, which builds the target job from the entry’s properties and then hands the archived file to it. How the file is delivered depends on what the job supports:

A job that is neither a batch job nor declares one of those fields cannot be used as a Dirmon target and is rejected by validation.

Archiving

Before a job is started, the file is moved to the entry’s archive_directory. Moving first guarantees the file is out of the watched path before processing begins, so it cannot be discovered again on a later scan. The archived file name is prefixed with the downstream job’s id, which ties the stored file to the job that processed it.

Security: restricting which paths may be read

A pattern can point anywhere the process user can read, so Dirmon supports an allow-list of root paths. When any paths are registered, every resolved file must live under one of them or it is skipped and a warning is logged.

# In an initializer, not via the web UI, so it cannot be tampered with.
RocketJob::DirmonEntry.add_whitelist_path("/var/sftp")

RocketJob::DirmonEntry.get_whitelist_paths
# => ["/var/sftp"]

RocketJob::DirmonEntry.delete_whitelist_path("/var/sftp")

Notes:

Starting the directory monitor

Dirmon scanning is itself a scheduled Rocket Job, so it must be started once per installation:

RocketJob::Jobs::DirmonJob.create!

DirmonJob is a cron job that runs every 5 minutes (*/5 * * * * UTC) at priority 30. Override either when creating it:

RocketJob::Jobs::DirmonJob.create!(
  cron_schedule: "*/1 * * * * UTC",
  priority:      25
)

The schedule and priority can be changed at any time afterwards, either from Mission Control or in code:

RocketJob::Jobs::DirmonJob.first.update_attributes(
  cron_schedule: "*/5 * * * * UTC",
  priority:      20
)

Starting a second DirmonJob while one is already queued or running is rejected with a validation error, so it is safe to call create! from deploy automation guarded by a rescue, or use the non-raising create.

High availability

DirmonJob has no dedicated process and no single point of failure. After each scan completes it schedules the next instance of itself and then destroys the current one. Because Rocket Job picks any available worker for the next run, scanning continues even as individual workers and containers come and go. There is only ever one DirmonJob queued or running at a time.

If a scan raises an exception, the responsible DirmonEntry is moved to the failed state with the exception recorded, so the rest of the entries keep working and the failure can be investigated and re-enabled from Mission Control.

Managing Dirmon in the web UI

Everything above can be done from Rocket Job Mission Control: create and edit entries, set their user_editable job fields, enable, disable, and re-enable them, inspect failures, and adjust the DirmonJob schedule and priority. Because Dirmon ships with Rocket Job, this UI is available with no extra code, which is exactly what teams otherwise rebuild by hand for every project.