added support for multiple handlers

Author: terjesannum

CHANGELOG.md

@@ -1,8 +1,12 @@
 # Changelog
 
+## [2.1.0] - 2017-05-15
+### Added
+- Support multiple handlers (@terjesannum)
+
 ## [2.0.1] - 2017-05-14
 ### Added
-- Released as Gem
+- Released as Gem (@terjesannum)
 
 ## [1.7] - 2017-04-24
 ### Added

README.md

@@ -175,6 +175,35 @@ measurement = app.downloads, tags = platform => iOS;device => iPad , value = 92,
 
 The event output tags will be merged with client and check definition tags and sent to InfluxDB as usual.
 
+# Multiple handlers
+
+If you need to have multiple handlers, eg. for different precision, proxy mode, writing to different influx databases etc, this can be done by configuring **additional_handlers**:
+
+```
+{
+  "influxdb-extension": {
+    "hostname": "influxdb",
+    "port": 8086,
+    "database": "metrics",
+    "username": "sensu",
+    "password": "sensu",
+    "buffer_size": 1000,
+    "buffer_max_age": 10,
+    "additional_handlers": ["events", "events_nano"]
+  },
+  "events": {
+    "proxy_mode": true,
+    "precision": "s"
+  },
+  "events_nano": {
+    "proxy_mode": true,
+    "precision": "n"
+  }
+}
+```
+
+Settings for the additional handlers will be merged with the **influxdb-extension** settings, so you only need to specify the settings you want to change for that handler.
+
 # Performance
 
 The extension will buffer up points until it reaches the configured **buffer_size** length or **buffer_max_age**, and then post all the points in the buffer to InfluxDB. 

influxdb-extension.json.tmpl

@@ -10,7 +10,12 @@
         "password": "<influxdb password>",
         "buffer_size": 100,
         "buffer_max_age": 10,
-        "precision": "s"
+        "precision": "s",
+        "additional_handlers": ["influxdb-proxy-nano"]
+    },
+    "influxdb-proxy-nano": {
+      "proxy_mode": true,
+      "precision": "n"
     }
 }
 

lib/sensu/extensions/influxdb.rb

@@ -18,60 +18,104 @@ def description
       "Transforms and sends metrics to InfluxDB"
     end
 
-    def post_init
-      influxdb_config = settings[@@extension_name] || Hash.new
-      validate_config(influxdb_config)
-
-      hostname         = influxdb_config[:hostname] || "127.0.0.1"
-      port             = influxdb_config[:port] || "8086"
-      database         = influxdb_config[:database]
-      ssl              = influxdb_config[:ssl] || false
-      ssl_ca_file      = influxdb_config[:ssl_ca_file]
-      ssl_verify       = if influxdb_config.key?(:ssl_verify) then influxdb_config[:ssl_verify] else true end
-      precision        = influxdb_config[:precision] || "s"
-      retention_policy = influxdb_config[:retention_policy]
+    @@default_config = {
+      :hostname       => "127.0.0.1",
+      :port           => "8086",
+      :ssl            => false,
+      :precision      => "s",
+      :protocol       => "http",
+      :buffer_size    => 100,
+      :buffer_max_age => 10,
+      :proxy_mode     => false
+    }
+
+    def create_config(name, defaults)
+      if settings[name].nil?
+        Raise ArgumentError "no configuration for #{name} provided. exiting..."
+      end
+      config = defaults.merge(settings[name])
+      @logger.debug("Config for #{name} created: #{config}")
+      validate_config(name, config)
+
+      hostname         = config[:hostname]
+      port             = config[:port]
+      database         = config[:database]
+      ssl              = config[:ssl]
+      ssl_ca_file      = config[:ssl_ca_file]
+      ssl_verify       = if config.key?(:ssl_verify) then config[:ssl_verify] else true end
+      precision        = config[:precision]
+      retention_policy = config[:retention_policy]
       rp_queryparam    = if retention_policy.nil? then "" else "&rp=#{retention_policy}" end
       protocol         = if ssl then "https" else "http" end
-      username         = influxdb_config[:username]
-      password         = influxdb_config[:password]
+      username         = config[:username]
+      password         = config[:password]
       auth_queryparam  = if username.nil? or password.nil? then "" else "&u=#{username}&p=#{password}" end
-      @BUFFER_SIZE     = if influxdb_config.key?(:buffer_size) then influxdb_config[:buffer_size].to_i else 100 end
-      @BUFFER_MAX_AGE  = if influxdb_config.key?(:buffer_max_age) then influxdb_config[:buffer_max_age].to_i else 10 end
-      @PROXY_MODE      = influxdb_config[:proxy_mode] || false
+      buffer_size      = config[:buffer_size]
+      buffer_max_age   = config[:buffer_max_age]
+      proxy_mode       = config[:proxy_mode]
 
       string = "#{protocol}://#{hostname}:#{port}/write?db=#{database}&precision=#{precision}#{rp_queryparam}#{auth_queryparam}"
-      @uri = URI(string)
-      @http = Net::HTTP::new(@uri.host, @uri.port)
+      uri = URI(string)
+      http = Net::HTTP::new(uri.host, uri.port)
       if ssl
-        @http.ssl_version = :TLSv1
-        @http.use_ssl = true
-        @http.verify_mode = if ssl_verify then OpenSSL::SSL::VERIFY_PEER else OpenSSL::SSL::VERIFY_NONE end
-        @http.ca_file = ssl_ca_file
+        http.ssl_version = :TLSv1
+        http.use_ssl = true
+        http.verify_mode = if ssl_verify then OpenSSL::SSL::VERIFY_PEER else OpenSSL::SSL::VERIFY_NONE end
+        http.ca_file = ssl_ca_file
       end
-      @buffer = []
-      @buffer_flushed = Time.now.to_i
 
-      @logger.info("#{@@extension_name}: successfully initialized config: hostname: #{hostname}, port: #{port}, database: #{database}, uri: #{@uri.to_s}, username: #{username}, buffer_size: #{@BUFFER_SIZE}, buffer_max_age: #{@BUFFER_MAX_AGE}")
+      @handlers ||= Hash.new
+      @handlers[name] = {
+        "http" => http,
+        "uri" => uri,
+        "buffer" => [],
+        "buffer_flushed" => Time.now.to_i,
+        "buffer_size" => buffer_size,
+        "buffer_max_age" => buffer_max_age,
+        "proxy_mode" => proxy_mode
+      }
+
+      @logger.info("#{name}: successfully initialized handler: hostname: #{hostname}, port: #{port}, database: #{database}, uri: #{uri.to_s}, username: #{username}, buffer_size: #{buffer_size}, buffer_max_age: #{buffer_max_age}")
+      return config
+    end
+
+    def post_init
+      main_config = create_config(@@extension_name, @@default_config)
+      if settings[name].key?(:additional_handlers)
+        settings[name][:additional_handlers].each {|h| create_config(h, main_config)}
+      end
     end
 
     def run(event)
       begin
+        @logger.debug("event: #{event}")
+        event = JSON.parse(event)
 
-        if buffer_too_old? or buffer_too_big?
-          flush_buffer
+        handler = @handlers[@@extension_name]
+        unless event["check"]["handlers"].nil?
+          event["check"]["handlers"].each {|x|
+            if @handlers.has_key?(x)
+              @logger.debug("found additional handler: #{x}")
+              handler = @handlers[x]
+              break
+            end
+          }
+        end
+
+        if buffer_too_old?(handler) or buffer_too_big?(handler)
+          flush_buffer(handler)
         end
 
-        event = JSON.parse(event)
         output = event["check"]["output"]
 
-        if not @PROXY_MODE
+        if not handler["proxy_mode"]
           client_tags = event["client"]["tags"] || Hash.new
           check_tags = event["check"]["tags"] || Hash.new
           tags = create_tags(client_tags.merge(check_tags))
         end
 
         output.split(/\r\n|\n/).each do |point|
-            if not @PROXY_MODE
+            if not handler["proxy_mode"]
               measurement, field_value, timestamp = point.split(/\s+/)
 
               if not is_number?(timestamp)
@@ -93,8 +137,8 @@ def run(event)
               point = "#{measurement}#{tags} value=#{field_value} #{timestamp}"
             end
 
-            @buffer.push(point)
-            @logger.debug("#{@@extension_name}: stored point in buffer (#{@buffer.length}/#{@BUFFER_SIZE})")
+            handler["buffer"].push(point)
+            @logger.debug("#{@@extension_name}: stored point in buffer (#{handler['buffer'].length}/#{handler['buffer_size']})")
         end
         yield 'ok', 0
       rescue => e
@@ -123,39 +167,39 @@ def create_tags(tags)
         end
     end
 
-    def send_to_influxdb(payload)
-      request = Net::HTTP::Post.new(@uri.request_uri)
+    def send_to_influxdb(handler)
+      payload = handler["buffer"].join("\n")
+      request = Net::HTTP::Post.new(handler['uri'].request_uri)
       request.body = payload
 
-      @logger.debug("#{@@extension_name}: writing payload #{payload} to endpoint #{@uri.to_s}")
-      response = @http.request(request)
+      @logger.debug("#{@@extension_name}: writing payload #{payload} to endpoint #{handler['uri'].to_s}")
+      response = handler["http"].request(request)
       @logger.debug("#{@@extension_name}: influxdb http response code = #{response.code}, body = #{response.body}")
     end
 
-    def flush_buffer
-      payload = @buffer.join("\n")
-      send_to_influxdb(payload)
-      @buffer = []
-      @buffer_flushed = Time.now.to_i
+    def flush_buffer(handler)
+      send_to_influxdb(handler)
+      handler["buffer"] = []
+      handler["buffer_flushed"] = Time.now.to_i
     end
 
-    def buffer_too_old?
-      buffer_age = Time.now.to_i - @buffer_flushed
-      buffer_age >= @BUFFER_MAX_AGE
+    def buffer_too_old?(handler)
+      buffer_age = Time.now.to_i - handler["buffer_flushed"]
+      buffer_age >= handler["buffer_max_age"]
     end
 
-    def buffer_too_big?
-      @buffer.length >= @BUFFER_SIZE
+    def buffer_too_big?(handler)
+      handler["buffer"].length >= handler["buffer_size"]
     end
 
-    def validate_config(config)
+    def validate_config(name, config)
       if config.nil?
-        raise ArgumentError, "no configuration for #{@@extension_name} provided. exiting..."
+        raise ArgumentError, "no configuration for #{name} provided. exiting..."
       end
 
       ["hostname", "database"].each do |required_setting|
-        if config[required_setting].nil?
-          raise ArgumentError, "required setting #{required_setting} not provided to extension. this should be provided as json element with key #{@@extension_name}. exiting..."
+        if config.has_key?(required_setting)
+          raise ArgumentError, "required setting #{required_setting} not provided to extension. this should be provided as json element with key #{name}. exiting..."
         end
       end
     end

sensu-extensions-influxdb.gemspec

@@ -2,7 +2,8 @@
 
 Gem::Specification.new do |spec|
   spec.name          = "sensu-extensions-influxdb"
-  spec.version       = "2.0.1"
+  spec.version       = "2.1.0"
+  spec.license       = "MIT"
   spec.authors       = ["Johnny Horvi", "Terje Sannum"]
   spec.email         = ["[email protected]", "[email protected]"]
 

spec/influxdb-extension_spec.rb

@@ -1,25 +1,29 @@
 require "sensu/extensions/influxdb"
+require "sensu/logger"
 
 describe "Sensu::Extension::InfluxDB" do
 
   before do
     @extension = Sensu::Extension::InfluxDB.new
-    @extension.settings = {
-      "influxdb-extension" => {
-        "database" => "test",
-        "hostname" => "127.0.0.69"
-      }
+    @extension.settings = Hash.new
+    @extension.settings["influxdb-extension"] = {
+        :database => "test",
+        :hostname => "nonexistinghost",
+        :additional_handlers => ["proxy"],
+        :buffer_size => 5,
+        :buffer_max_age => 1
+    }
+    @extension.settings["proxy"] = {
+        :proxy_mode => true
     }
 
-    stubbed_logger = double("logger", :info => "info", :debug => "debug", :error => "error")
-
-    @extension.instance_variable_set("@logger", stubbed_logger)
+    @extension.instance_variable_set("@logger", Sensu::Logger.get(:log_level => :fatal))
     @extension.post_init
   end
 
   it "processes minimal event" do
     @extension.run(minimal_event.to_json) do 
-      buffer = @extension.instance_variable_get("@buffer")
+      buffer = @extension.instance_variable_get("@handlers")["influxdb-extension"]["buffer"]
       expect(buffer[0]).to eq("rspec value=69 1480697845")
     end
   end
@@ -36,14 +40,13 @@
     }
 
     @extension.run(event.to_json) do 
-      buffer = @extension.instance_variable_get("@buffer")
+      buffer = @extension.instance_variable_get("@handlers")["influxdb-extension"]["buffer"]
       expect(buffer.size).to eq(0)
     end
   end
 
 
   it "flushes buffer when full" do
-    @extension.instance_variable_set("@BUFFER_SIZE", 5)
     5.times {
       @extension.run(minimal_event.to_json) do |output,status|
         expect(output).to eq("ok")
@@ -60,12 +63,11 @@
   end
 
   it "flushes buffer when timed out" do
-    @extension.instance_variable_set("@BUFFER_MAX_AGE", 1)
     @extension.run(minimal_event.to_json) do end
     sleep(1)
     @extension.run(minimal_event.to_json) do end
 
-    buffer = @extension.instance_variable_get("@buffer")
+    buffer = @extension.instance_variable_get("@handlers")["influxdb-extension"]["buffer"]
     expect(buffer.size).to eq(1)
   end
 
@@ -91,15 +93,14 @@
 
     @extension.run(event.to_json) do end
 
-    buffer = @extension.instance_variable_get("@buffer")
+    buffer = @extension.instance_variable_get("@handlers")["influxdb-extension"]["buffer"]
     expect(buffer[0]).to eq("rspec,a=1,b=1,c=1,x=1,y=1,z=1 value=69 1480697845")
   end
 
   it "does not modify input in proxy mode" do
-    @extension.instance_variable_set("@PROXY_MODE", true)
-    @extension.run(minimal_event.to_json) do end
+    @extension.run(minimal_event_proxy.to_json) do end
 
-    buffer = @extension.instance_variable_get("@buffer")
+    buffer = @extension.instance_variable_get("@handlers")["proxy"]["buffer"]
     expect(buffer[0]).to eq("rspec 69 1480697845")
   end
 
@@ -115,3 +116,15 @@ def minimal_event
       }
     }
 end
+
+def minimal_event_proxy
+    event = {
+      "client" => {
+        "name" => "rspec"
+      },
+      "check" => {
+        "handlers" => ["proxy"],
+        "output" => "rspec 69 1480697845"
+      }
+    }
+end