Class: Sensu::Transport::SNSSQS

Inherits:
Base
  • Object
show all
Defined in:
lib/sensu/transport/snssqs.rb

Constant Summary collapse

STRING_STR = 
"String".freeze
KEEPALIVES_STR = 
"keepalives".freeze
PIPE_STR = 
"pipe".freeze
TYPE_STR = 
"type".freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ SNSSQS

Returns a new instance of SNSSQS.



15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/sensu/transport/snssqs.rb', line 15

def initialize
  @connected = false
  @subscribing = false

  # as of sensu 0.23.0 we need to call succeed when we have
  # successfully connected to SQS.
  #
  # we already have our own logic to maintain the connection to
  # SQS, so we can always say we're connected.
  #
  # See:
  # https://github.com/sensu/sensu/blob/cdc25b29169ef2dcd2e056416eab0e83dbe000bb/CHANGELOG.md#0230---2016-04-04
  succeed()
end

Instance Attribute Details

#logger ⇒ Object

Returns the value of attribute logger.



8
9
10
 
# File 'lib/sensu/transport/snssqs.rb', line 8

def logger
  @logger
end

Instance Method Details

#acknowledge(info, &callback) ⇒ Object

acknowledge will delete the given message from the SQS queue.



106
107
108
109
110
111
112
113
114
115
 
# File 'lib/sensu/transport/snssqs.rb', line 106

def acknowledge(info, &callback)
  EM.defer {
    @sqs.delete_message(
      queue_url: @settings[:consuming_sqs_queue_url],
      receipt_handle: info.receipt_handle,
    )
    statsd_incr("sqs.#{@settings[:consuming_sqs_queue_url]}.message.deleted")
    callback.call(info) if callback
  }
end

#connect(settings) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/sensu/transport/snssqs.rb', line 32

def connect(settings)
  @settings = settings
  @connected = true
  @results_callback = proc {}
  @keepalives_callback = proc {}
  @sqs = Aws::SQS::Client.new(region: @settings[:region])
  @sns = Aws::SNS::Client.new(region: @settings[:region])

  # connect to statsd, if necessary
  @statsd = nil
  if @settings[:statsd_addr] != ""
    pieces = @settings[:statsd_addr].split(':')
    @statsd = Statsd.new(pieces[0], pieces[1].to_i).tap { |sd|
      sd.namespace = @settings[:statsd_namespace]
    }
    @statsd_sample_rate = @settings[:statsd_sample_rate].to_f
  end
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


30
 
# File 'lib/sensu/transport/snssqs.rb', line 30

def connected?; @connected; end

#publish(type, pipe, message, options = {}, &callback) ⇒ Object

publish publishes a message to the SNS topic.

The type, pipe, and options are transformed into SNS message attributes and included with the message.



121
122
123
124
125
126
127
128
129
130
 
# File 'lib/sensu/transport/snssqs.rb', line 121

def publish(type, pipe, message, options = {}, &callback)
  attributes = {
    TYPE_STR => str_attr(type),
    PIPE_STR => str_attr(pipe)
  }
  options.each do |k, v|
    attributes[k.to_s] = str_attr(v.to_s)
  end
  EM.defer { send_message(message, attributes, &callback) }
end

#statsd_incr(stat) ⇒ Object 



51
52
53
 
# File 'lib/sensu/transport/snssqs.rb', line 51

def statsd_incr(stat)
  @statsd.increment(stat, @statsd_sample_rate) unless @statsd.nil?
end

#statsd_time(stat) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
 
# File 'lib/sensu/transport/snssqs.rb', line 55

def statsd_time(stat)
  # always measure + run the block, but only if @statsd is set
  # do we actually report it.
  start = Time.now
  result = yield
  if !@statsd.nil?
    @statsd.timing(stat, ((Time.now - start) * 1000).round(5), @statsd_sample_rate)
  end
  result
end

#subscribe(type, pipe, funnel = nil, options = {}, &callback) ⇒ Object

subscribe will begin “subscribing” to the consuming sqs queue.

What this really means is that we will start polling for messages from the SQS queue, and, depending on the message type, it will call the appropriate callback.

This assumes that the SQS Queue is consuming “Raw” messages from SNS.

“subscribing” means that the “callback” parameter will be called when there is a message for you to consume.

“funnel” and “type” parameters are completely ignored.



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/sensu/transport/snssqs.rb', line 79

def subscribe(type, pipe, funnel = nil, options = {}, &callback)
  self.logger.info("subscribing to type=#{type}, pipe=#{pipe}, funnel=#{funnel}")

  if pipe == KEEPALIVES_STR
    @keepalives_callback = callback
  else
    @results_callback = callback
  end

  unless @subscribing
    do_all_the_time {
      EM::Iterator.new(receive_messages, 10).each do |msg, iter|
        statsd_time("sqs.#{@settings[:consuming_sqs_queue_url]}.process_timing") {
          if msg.message_attributes[PIPE_STR].string_value == KEEPALIVES_STR
            @keepalives_callback.call(msg, msg.body)
          else
            @results_callback.call(msg, msg.body)
          end
        }
        iter.next
      end
    }
    @subscribing = true
  end
end