Class: Hookshot

Inherits:
Object
  • Object
show all
Defined in:
lib/hookshot.rb,
lib/hookshot/version.rb

Constant Summary collapse

PREFIX = 
"hookshot"
NEW_JOBS_LIST = 
"#{PREFIX}:jobs"
DELAYED_SET = 
"#{PREFIX}:delayed"
FAILURES_LIST = 
"#{PREFIX}:failures"
BLACKLIST = 
"#{PREFIX}:throttle:blacklist"
WHITELIST = 
"#{PREFIX}:throttle:whitelist"
FINAL_FAILURE =

Special sentinel value returned from ‘get_next_failure` when the job will no longer be retried.

-1
JOB_KEY_LIFETIME =

Jobs are retried for up to 48 hours. Though we delete the job info when hookshot is done with each job, it’s still a good idea to clean up any keys that have managed to stick around somehow.

86400 * 4
FailureQueueEmpty =

Raised by ‘get_next_failure` if there are no pending failures.

Class.new(StandardError)
VERSION = 
"0.3.2"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis) ⇒ Hookshot

Hookshot expects to be initialized with an instance of ‘Redis` provided by the `redis` rubygem.

Example:

require 'hookshot'
require 'redis'
hookshot = Hookshot.new(Redis.new(port: 6379, host: 'localhost'))



37
38
39
 
# File 'lib/hookshot.rb', line 37

def initialize(redis)
  @redis = redis
end

Instance Attribute Details

#redis ⇒ Object (readonly)

Returns the value of attribute redis.



27
28
29
 
# File 'lib/hookshot.rb', line 27

def redis
  @redis
end

Instance Method Details

#blacklist ⇒ Object

blacklist returns an array of currently-blacklisted domains.

Example:

hookshot.blacklist
#=> ["example.com"]



203
204
205
 
# File 'lib/hookshot.rb', line 203

def blacklist
  redis.smembers(BLACKLIST)
end

#blacklist!(domain) ⇒ Object

blacklist! adds a domain to the list of currently blacklisted domains. Any jobs submitted with a domain in this list will be automatically dropped by hookshot.

The format of the domain should be the full domain. The port should not be included if it is ‘80` or `443`. For example:

| URL | Domain | |—————————|———————| | example.com/post | example.com | | example.com/post | example.com | | example.com:8000 | example.com:8000 | | example.com:80 | example.com |

Example:

hookshot.blacklist!("example.com")



172
173
174
 
# File 'lib/hookshot.rb', line 172

def blacklist!(domain)
  redis.sadd(BLACKLIST, domain)
end

#enqueue(url:, headers:, context:, payload:, activate_at: nil) ⇒ Object

enqueue takes a URL, a hash of headers, a payload (request body), and a ‘context` value. It submits these values to hookshot for processing.

The ‘context` value can be any string, and will be returned to you via `get_next_failure` if the job can’t be successfully completed after about 48 hours of retries.

In Shopify, we pass our ‘WebhookSubscription` object ID for the `context` value, so that we can notify merchants when their webhooks are failing, and delete subscriptions that fail consistently.

‘activate_at` is an optional parameter that specifies a number of seconds to wait before making this job active in hookshot. You should normally call `enqueue_in` rather than pass `activate_at` explicitly.

Example:

hookshot.enqueue(
  url: 'http://localhost:8080/post',
  headers: {"X-My-Header" => "value"},
  context: "42",
  payload: "request body")



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/hookshot.rb', line 63

def enqueue(url:, headers:, context:, payload:, activate_at: nil)
  uuid = SecureRandom.uuid
  redis.pipelined do
    redis.hmset(
      job_key(uuid),
      "url", url,
      "headers", serialize_headers(headers),
      "context", context,
      "payload", payload,
      "failures", 0)
    redis.expire(job_key(uuid), JOB_KEY_LIFETIME)
    if activate_at
      redis.zadd(DELAYED_SET, activate_at, uuid)
    else
      redis.lpush(NEW_JOBS_LIST, uuid)
    end
  end

  uuid
end

#enqueue_in(duration, url:, headers:, context:, payload:) ⇒ Object

enqeueue_in calls enqueue with an ‘activate_at` parameter to delay the job’s execution.

Example:

hookshot.enqueue_in(60, # seconds
  url: 'http://localhost:8080/post',
  headers: {"X-My-Header" => "value"},
  context: "42",
  payload: "request body")



94
95
96
97
98
99
100
101
102
 
# File 'lib/hookshot.rb', line 94

def enqueue_in(duration, url:, headers:, context:, payload:)
  enqueue_time = (Time.now + duration).to_i
  enqueue(
    url: url,
    headers: headers,
    context: context,
    payload: payload,
    activate_at: enqueue_time)
end

#get_next_failure ⇒ Object

Jobs that fail many times in a row are returned back to the application. Specifically, the ‘context` value passed in via the `enqueue*` methods is returned.

‘get_next_failure` returns two values: the number of failures so far for this job, and the `context` passed in with the job. If `nfailures` is equal to `Hookshot::FINAL_FAILURE`, the job will not be retried. However, if `nfailures` is any other value, the job will still be retried; this is just an advisory because the job has been failing for at least 24 hours.

This method is non-blocking: if there is no item present in the failures queue, it will raise FailureQueueEmpty.

Example:

loop {
  nfailures, context = hookshot.get_next_failure
  if nfailures == Hookshot::FINAL_FAILURE
    delete_webhook_subscription(context)
  end
}

Raises:



125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/hookshot.rb', line 125

def get_next_failure
  # block indefinitely waiting for the next failure.
  line = redis.lpop(FAILURES_LIST)
  raise FailureQueueEmpty if line.nil?

  nfailures, failed_id = line.split('|', 2)

  if nfailures.to_i == 0 || failed_id.empty?
    raise "Invalid line from hookshot: #{line}"
  end

  [nfailures.to_i, failed_id]
end

#queue_stats ⇒ Object

Hookshot writes a lot of statistics to statsd/datadog, but to quickly check the current queue sizes, use ‘queue_stats`.

Example:

hookshot.queue_stats
# => { pending: 42, delayed: 42, failures: 42 }



146
147
148
149
150
151
152
153
 
# File 'lib/hookshot.rb', line 146

def queue_stats
  pending, delayed, failures = redis.pipelined do
    redis.llen NEW_JOBS_LIST
    redis.zcard DELAYED_SET
    redis.llen FAILURES_LIST
  end
  { pending: pending, delayed: delayed, failures: failures }
end

#remove_blacklist(domain) ⇒ Object

remove_blacklist removes a currently-blacklisted domain from the blacklist. If the domain was not blacklisted, this method has no effect.

Example:

hookshot.remove_blacklist("google.com")



223
224
225
 
# File 'lib/hookshot.rb', line 223

def remove_blacklist(domain)
  redis.srem(BLACKLIST, domain)
end

#remove_whitelist(domain) ⇒ Object

remove_whitelist removes a currently-whitelisted domain from the whitelist. If the domain was not whitelisted, this method has no effect.

Example:

hookshot.remove_whitelist("google.com")



233
234
235
 
# File 'lib/hookshot.rb', line 233

def remove_whitelist(domain)
  redis.srem(WHITELIST, domain)
end

#whitelist ⇒ Object

whitelist returns an array of currently-whitelisted domains.

Example:

hookshot.whitelist
#=> ["example.com"]



213
214
215
 
# File 'lib/hookshot.rb', line 213

def whitelist
  redis.smembers(WHITELIST)
end

#whitelist!(domain) ⇒ Object

whitelist! adds a domain to the list of currently whitelisted domains. Normally, jobs are throttled to a maximum requests per second per domain. Whitelisted domains are granted a much higher initial rate.

The format of the domain should be the full domain. The port should not be included if it is ‘80` or `443`. For example:

| URL | Domain | |—————————|———————| | example.com/post | example.com | | example.com/post | example.com | | example.com:8000 | example.com:8000 | | example.com:80 | example.com |

Example:

hookshot.whitelist!("google.com")



193
194
195
 
# File 'lib/hookshot.rb', line 193

def whitelist!(domain)
  redis.sadd(WHITELIST, domain)
end