DOCSEARCHbyAlgolia
Submit Docs

Gush Documentation

repository·master·Indexed 22 days ago

https://github.com/chaps-io/gush

Gush is a parallel workflow runner that uses Redis for storage and ActiveJob for scheduling and execution. It enables the definition of complex, dependency-based workflows using a Directed Acyclic Graph (DAG) model. Key features include pipelining for passing JSON-serializable data between jobs, a CLI for monitoring workflow status, and visualization tools via ImageMagick. It supports TTL strategies for Redis data management and provides a Gush::Client for programmatic workflow control.

Tokens
6.5K
Snippets
35
Records
38
Agent score
77%

What's inside Gush

  1. Manage workflow data expiration and TTL

    master

    By default, Gush and Redis keep workflow data indefinitely. To prevent Redis from growing too large, you should implement a TTL (Time To Live) strategy.

    1. Set a global TTL

    Configure a global expiration time in seconds via config.ttl. It is recommended to set this to a duration of approximately one week rather than very short intervals.

    2. Purge expired data

    Setting the TTL is not enough; you must periodically call Client#expire_workflows to actually clear the expired stored workflow data, job data, and indexes. It is recommended to call this at least once for every 1000 workflows created.

    3. Individual workflow expiration

    If you need granular control, you can call flow.expire!(ttl) on a specific workflow instance. Passing -1 will prevent the workflow from ever expiring.

    # config/initializers/gush.rb
Gush.configure do |config|
  config.redis_url = "redis://localhost:6379"
  config.concurrency = 5
  config.ttl = 3600*24*7 # Example: 1 week in seconds
end

# Periodically run this to clear expired data
Gush::Client.new.expire_workflows

  2. Define a Gush Workflow

    master

    Workflows are defined by inheriting from Gush::Workflow and implementing a configure method. You use the run method to declare jobs and their dependencies.

    Dependency Management

    • Single dependency: Use after: JobClass.
    • Multiple dependencies: Use an array after: [JobClass1, JobClass2].
    • Reverse dependency: Use before: JobClass.

    Passing Arguments

    Workflows can accept primitive arguments in their constructor, which are then passed to the configure method. You can also pass globals to all jobs in the workflow.

    class SampleWorkflow < Gush::Workflow
  def configure(url_to_fetch_from)
    run FetchJob1, params: { url: url_to_fetch_from }
    run FetchJob2, params: { some_flag: true, url: 'http://url.com' }

    run PersistJob1, after: FetchJob1
    run PersistJob2, after: FetchJob2

    run Normalize,
        after: [PersistJob1, PersistJob2],
        before: Index

    run Index
  end
end

# Creating and starting the workflow
flow = SampleWorkflow.create("http://url.com/data", globals: { creator_id: 123})
flow.start!

  3. Pass data between jobs using Pipelining

    master

    Gush allows jobs to pass data to their dependent jobs using the output method. The receiving job can access this data via the payloads array.

    Note: Data must be JSON-serializable.

    How it works

    1. The ancestor job calls output(data).
    2. The dependent job accesses payloads.first[:output].

    payloads is an array of hashes containing the id, class, and the returned output for every ancestor job.

    class DownloadVideo < Gush::Job
  def perform
    path = "/tmp/video.mp4"
    output(path)
  end
end

class EncodeVideo < Gush::Job
  def perform
    # Access the output from the ancestor
    video_path = payloads.first[:output]
  end
end

  4. Migrate to Gush 3.0

    master

    Gush 3.0 introduces indexing for faster workflow pagination and changes how workflow data is expired in Redis.

    After upgrading your gem, you must run the migration command to update the internal data structures.

    bundle exec gush migrate

  5. Execute and Monitor Workflows

    master

    To run workflows, you must first start a background worker process compatible with your ActiveJob backend (e.g., Sidekiq). Gush uses the gush queue by default.

    Execution Steps

    1. Start Worker: bundle exec sidekiq -q gush (for Sidekiq).
    2. Create Instance: flow = MyWorkflow.create(args).
    3. Start: flow.start!.

    Monitoring

    Use reload to refresh the instance state before checking status.

    • flow.reload
    • flow.status (returns :running, :finished, or :failed)
    # 1. Start worker (in terminal)
# bundle exec sidekiq -q gush

# 2. In your code
flow = PublishBookWorkflow.create("http://url.com/book.pdf", "978-0470081204")
flow.start!

# 3. Monitor
flow.reload
puts flow.status

  6. Install Gush

    master

    To use Gush, add it to your Gemfile and ensure you have a Gushfile in your project root to load your workflows and jobs.

    1. Add to Gemfile

    gem 'gush', '~> 5.0'

    2. Create a Gushfile

    For Ruby on Rails: Require the Rails environment and ensure your job/workflow directories are in the autoload_paths.

    # Gushfile
require_relative './config/environment.rb'

    In config/application.rb:

    config.autoload_paths += ["#{Rails.root}/app/jobs", "#{Rails.root}/app/workflows"]

    For plain Ruby: Manually require your workflow and job files.

    # Gushfile
require_relative 'lib/workflows/example_workflow.rb'
require_relative 'lib/jobs/some_job.rb'
require_relative 'lib/jobs/some_other_job.rb'

  7. Manage job state and lifecycle

    master

    A Gush::Job instance tracks its own lifecycle through several state-checking methods. These are useful for monitoring or conditional logic within a workflow engine:

    • enqueued?: Returns true if the job has been enqueued.
    • started?: Returns true if the job has started running.
    • running?: Returns true if the job has started but not yet finished.
    • finished?: Returns true if the job has reached a terminal state (success or failure).
    • succeeded?: Returns true if the job finished successfully.
    • failed?: Returns true if the job failed.
    • ready_to_start?: Returns true if the job is not currently running, enqueued, finished, or failed, AND all its parent jobs (incoming) have succeeded.

  8. Configure Redis locking options to prevent RedisMutex::LockError

    master

    If you encounter RedisMutex::LockError when processing a large number of jobs, you can customize the locking_duration and polling_interval in your Gush configuration. This helps manage how long the system waits for locks to be released and how frequently it polls.

    # config/initializers/gush.rb
Gush.configure do |config|
  config.redis_url = "redis://localhost:6379"
  config.concurrency = 5
  config.locking_duration = 2 # how long you want to wait for the lock to be released, in seconds
  config.polling_interval = 0.3 # how long the polling interval should be, in seconds
end

  9. Prevent overlapping workflow executions

    master

    To avoid starting a new scheduled iteration of a workflow while a previous instance of the same class is still running, you can implement a check using Gush::Client#all_workflows.

    Since the core library does not currently provide a Workflow.find_by_class(klass) method, you can manually iterate through running workflows to check for name matches.

    GUSH_CLIENT = Gush::Client.new

# Call this method before starting a new workflow instance
def find_by_class(klass)
  GUSH_CLIENT.all_workflows.each do |flow|
    return true if flow.to_hash[:name] == klass && flow.running?
  end
  return false
end

  10. Customize Job Enqueueing and ActiveJob options

    master

    You can customize how jobs are enqueued in a workflow using specific options in the run method, or by overriding methods in the Gush::Job class.

    Workflow-level options

    When calling run, you can specify:

    • queue: The name of the queue.
    • wait: A delay (e.g., 5.seconds).

    Class-level overrides

    To pass additional options to ActiveJob.set, override worker_options. To completely change the integration, override enqueue_worker!.

    # Using options in workflow
run AdminNotificationJob, after: jobs, queue: 'admin', wait: 5.seconds

# Overriding worker_options in a Job class
class ScheduledJob < Gush::Job
  def worker_options
    super.merge(wait_until: Time.at(params[:start_at]))
  end
end

# Overriding enqueue_worker! for custom integration
class SynchronousJob < Gush::Job
  def enqueue_worker!(options = {})
    Gush::Worker.perform_now(workflow_id, name)
  end
end

  11. Configure Gush global settings

    master

    Gush uses a Configuration object to manage global settings for workflows, including Redis connectivity, concurrency, and polling behavior. You can configure these settings by passing a hash to the initializer or by setting the attributes on the configuration instance.

    Configuration Options

    OptionTypeDefaultDescription
    concurrencyInteger5The number of concurrent workers/jobs.
    namespaceString'gush'The Redis namespace used to isolate Gush data.
    redis_urlString'redis://localhost:6379'The connection URL for the Redis instance.
    gushfileString/Path'Gushfile'The path to the Gushfile that defines your workflows.
    ttlInteger-1Time-to-live for jobs.
    locking_durationInteger2How long to wait for a lock to be released, in seconds.
    polling_intervalFloat0.3The polling interval in seconds.

    Note: The polling_interval option is initialized via the key :polling_internal in the hash.

    # Example of initializing configuration with custom settings
config = Gush::Configuration.new(
  concurrency: 10,
  redis_url: 'redis://production-redis:6379',
  namespace: 'my_app_workflows'
)

  12. Configure Gush::Graph output options

    master

    When initializing Gush::Graph, you can pass an options hash to control the output file:

    • filename: The name of the file to be generated. Defaults to "graph.png".
    • path: A Pathname object specifying the exact location where the graph should be saved. If not provided, it defaults to a temporary directory using the provided filename.

    Note: The file format is inferred from the extension of the filename. If the extension is 3 characters long (e.g., .png, .svg), it is used as the Graphviz format.

    # Using a custom filename
Graph.new(workflow, filename: 'output.svg')

# Using a specific path
Graph.new(workflow, path: Pathname.new('/tmp/custom_location.png'))