Using Redis TTL to expire keys after a certain point

README.md

@@ -348,6 +348,37 @@ This requires that you have imagemagick installed on your computer:
 bundle exec gush viz <NameOfTheWorkflow>
 ```
 
+### Cleaning up afterwards
+
+Running `NotifyWorkflow.create` inserts multiple keys into Redis every time it is ran.  This data might be useful for analysis but at a certain point it can be purged via Redis TTL.  By default gush and Redis will keep keys forever.  To configure expiration you need to 2 things.  Create initializer (specify config.ttl in seconds, be different per environment).  
+
+```ruby
+# config/initializers/gush.rb
+Gush.configure do |config|
+  config.redis_url = "redis://localhost:6379"
+  config.concurrency = 5
+  config.ttl = 3600*24*7
+end
+```
+
+And you need to call `flow.expire!` (optionally passing custom TTL value overriding `config.ttl`).  This gives you control whether to expire data for specific workflow.  Best NOT to set TTL to be too short (like minutes) but about a week in length.  And you can run `Client.expire_workflow` and `Client.expire_job` passing appropriate IDs and TTL (pass -1 to NOT expire) values.  
+
+### Avoid overlapping workflows
+
+Since we do not know how long our workflow execution will take we might want to avoid starting the next scheduled workflow iteration while the current one with same class is still running.  Long term this could be moved into core library, perhaps `Workflow.find_by_class(klass)`
+
+```ruby
+# config/initializers/gush.rb
+GUSH_CLIENT = Gush::Client.new
+# call this method before NotifyWorkflow.create
+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
+```
+
 ## Contributors
 
 - [Mateusz Lenik](https://github.com/mlen)

lib/gush/cli.rb

@@ -16,6 +16,7 @@ def initialize(*)
         config.concurrency = options.fetch("concurrency", config.concurrency)
         config.redis_url   = options.fetch("redis",       config.redis_url)
         config.namespace   = options.fetch("namespace",   config.namespace)
+        config.ttl         = options.fetch("ttl",         config.ttl)
       end
       load_gushfile
     end

lib/gush/client.rb

@@ -144,6 +144,21 @@ def destroy_job(workflow_id, job)
       end
     end
 
+    def expire_workflow(workflow, ttl=nil)
+      ttl = ttl || configuration.ttl
+      connection_pool.with do |redis|
+        redis.expire("gush.workflows.#{workflow.id}", ttl)
+      end
+      workflow.jobs.each {|job| expire_job(workflow.id, job, ttl) }
+    end
+
+    def expire_job(workflow_id, job, ttl=nil)
+      ttl = ttl || configuration.ttl
+      connection_pool.with do |redis|
+        redis.expire("gush.jobs.#{workflow_id}.#{job.name}", ttl)
+      end
+    end
+
     def enqueue_job(workflow_id, job)
       job.enqueue!
       persist_job(workflow_id, job)

lib/gush/configuration.rb

@@ -1,6 +1,6 @@
 module Gush
   class Configuration
-    attr_accessor :concurrency, :namespace, :redis_url
+    attr_accessor :concurrency, :namespace, :redis_url, :ttl
 
     def self.from_json(json)
       new(Gush::JSON.decode(json, symbolize_keys: true))
@@ -11,6 +11,7 @@ def initialize(hash = {})
       self.namespace   = hash.fetch(:namespace, 'gush')
       self.redis_url   = hash.fetch(:redis_url, 'redis://localhost:6379')
       self.gushfile    = hash.fetch(:gushfile, 'Gushfile')
+      self.ttl         = hash.fetch(:ttl, -1)
     end
 
     def gushfile=(path)
@@ -25,7 +26,8 @@ def to_hash
       {
         concurrency: concurrency,
         namespace:   namespace,
-        redis_url:   redis_url
+        redis_url:   redis_url,
+        ttl:         ttl
       }
     end
 

lib/gush/workflow.rb

@@ -53,6 +53,10 @@ def persist!
       client.persist_workflow(self)
     end
 
+    def expire! (ttl=nil)
+      client.expire_workflow(self, ttl)
+    end
+
     def mark_as_persisted
       @persisted = true
     end

spec/gush/client_spec.rb

@@ -94,6 +94,16 @@
     end
   end
 
+  describe "#expire_workflow" do
+    it "sets TTL for all Redis keys related to the workflow" do
+      workflow = TestWorkflow.create
+
+      client.expire_workflow(workflow, -1)
+
+      # => TODO - I believe fakeredis does not handle TTL the same.   
+    end
+  end
+
   describe "#persist_job" do
     it "persists JSON dump of the job in Redis" do