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.



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

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.



10
11
12
 
# File 'lib/sensu/transport/snssqs.rb', line 10

def logger
  @logger
end

Instance Method Details

#acknowledge(info, &callback) ⇒ Object

acknowledge will delete the given message from the SQS queue.



124
125
126
127
128
129
130
131
132
133
 
# File 'lib/sensu/transport/snssqs.rb', line 124

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 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/sensu/transport/snssqs.rb', line 34

def connect(settings)
  @settings = settings
  @connected = true
  @results_callback = proc {}
  @keepalives_callback = proc {}
  # Sensu Windows install does not include a valid cert bundle for AWS
  Aws.use_bundled_cert! if Gem.win_platform?
  aws_client_settings = {region: @settings[:region]}
  unless @settings[:access_key_id].nil?
    aws_client_settings[:access_key_id] = @settings[:access_key_id]
    aws_client_settings[:secret_access_key] = @settings[:secret_access_key]
  end
  @sqs = Aws::SQS::Client.new aws_client_settings
  @sns = Aws::SNS::Client.new aws_client_settings

  # connect to statsd, if necessary
  @statsd = nil
  if !@settings[:statsd_addr].nil? and @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) 


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

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.



139
140
141
142
143
144
145
146
147
148
 
# File 'lib/sensu/transport/snssqs.rb', line 139

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 



60
61
62
 
# File 'lib/sensu/transport/snssqs.rb', line 60

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

#statsd_time(stat) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
 
# File 'lib/sensu/transport/snssqs.rb', line 64

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.

This method is intended for use by the Sensu server; fanout subscriptions initiated by the Sensu client process are treated as a no-op.

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.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/sensu/transport/snssqs.rb', line 92

def subscribe(type, pipe, funnel = nil, options = {}, &callback)
  if type == :fanout
    self.logger.debug("skipping unsupported fanout subscription type=#{type}, pipe=#{pipe}, funnel=#{funnel}")
    return
  end

  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