Update service heartbeat filter for InfluxDB 0.9

Change-Id: Ieb9cb70b2d65faf2e106d9e4625954798767d3a9
Implements: blueprint upgrade-influxdb-grafana

deployment_scripts/puppet/manifests/controller.pp

@@ -157,8 +157,14 @@ if $lma_collector['influxdb_mode'] != 'disabled' {
   class { 'lma_collector::mod_status': }
 
   # Enable service heartbeat metrics
-  class { 'lma_collector::metrics::service_heartbeat':
-    services => ['mysql', 'rabbitmq', 'haproxy', 'memcached', 'apache']
+  if $lma_collector['influxdb_legacy'] {
+    class { 'lma_collector::metrics::service_heartbeat_legacy':
+      services => ['mysql', 'rabbitmq', 'haproxy', 'memcached', 'apache']
+    }
+  } else {
+    class { 'lma_collector::metrics::service_heartbeat':
+      services => ['mysql', 'rabbitmq', 'haproxy', 'memcached', 'apache']
+    }
   }
 
   # Service status metrics and annotations

deployment_scripts/puppet/modules/lma_collector/files/plugins/filters/service_heartbeat.lua

@@ -11,49 +11,39 @@
 -- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 -- See the License for the specific language governing permissions and
 -- limitations under the License.
-require 'cjson'
 require 'string'
-require 'math'
+local utils = require 'lma_utils'
 
-local timeout = read_config("timeout") or 30
+local timeout = (read_config("timeout") or 30) * 1e9 -- in ns
+local hostname
 
 services = {}
-local floor = math.floor
 
 function process_message ()
-    local ts = floor(read_message("Timestamp")/1e6) -- in ms
-    local service_name = string.match(read_message("Fields[name]"), '^[^._]+')
-    local hostname = read_message("Fields[hostname]")
-    local key = string.format('%s.%s', hostname, service_name)
-
-    local service = services[key]
-    if service then
-        service.last_seen = ts
-    else
-        service = {last_seen = ts, status = 1, host = hostname, name = service_name}
-        services[key] = service
+    local ts = read_message('Timestamp')
+    local service = string.match(read_message("Fields[name]"), '^[^_]+')
+    if not hostname then
+        hostname = read_message("Fields[hostname]")
     end
+
+    local entry = services[service]
+    if entry then
+        entry.last_seen = ts
+    else
+        services[service] = {last_seen = ts}
+    end
+
     return 0
 end
 
 function timer_event(ns)
-    local current_time = floor(ns / 1e6) -- in ms
-    local datapoints = {}
-
-    for k, service in pairs(services) do
-        if current_time - service.last_seen > timeout * 1000 then
-            service.status = 0
-        else
-            service.status = 1
+    for service, data in pairs(services) do
+        local status = 1
+        if ns - data.last_seen > timeout then
+            status = 0
         end
-        datapoints[#datapoints+1] = {
-            name = string.format('%s.%s.status', service.host, service.name),
-            columns = {"time", "value"},
-            points = {{service.last_seen, service.status}}
-        }
+        utils.add_to_bulk_metric(service .. '_status', status)
     end
 
-    if #datapoints > 0 then
-        inject_payload("json", "influxdb", cjson.encode(datapoints))
-    end
+    utils.inject_bulk_metric(ns, hostname, 'service_heartbeat')
 end

deployment_scripts/puppet/modules/lma_collector/files/plugins/filters/service_heartbeat_legacy.lua

@@ -0,0 +1,59 @@
+-- Copyright 2015 Mirantis, Inc.
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+require 'cjson'
+require 'string'
+require 'math'
+
+local timeout = read_config("timeout") or 30
+
+services = {}
+local floor = math.floor
+
+function process_message ()
+    local ts = floor(read_message("Timestamp")/1e6) -- in ms
+    local service_name = string.match(read_message("Fields[name]"), '^[^._]+')
+    local hostname = read_message("Fields[hostname]")
+    local key = string.format('%s.%s', hostname, service_name)
+
+    local service = services[key]
+    if service then
+        service.last_seen = ts
+    else
+        service = {last_seen = ts, status = 1, host = hostname, name = service_name}
+        services[key] = service
+    end
+    return 0
+end
+
+function timer_event(ns)
+    local current_time = floor(ns / 1e6) -- in ms
+    local datapoints = {}
+
+    for k, service in pairs(services) do
+        if current_time - service.last_seen > timeout * 1000 then
+            service.status = 0
+        else
+            service.status = 1
+        end
+        datapoints[#datapoints+1] = {
+            name = string.format('%s.%s.status', service.host, service.name),
+            columns = {"time", "value"},
+            points = {{service.last_seen, service.status}}
+        }
+    end
+
+    if #datapoints > 0 then
+        inject_payload("json", "influxdb", cjson.encode(datapoints))
+    end
+end

deployment_scripts/puppet/modules/lma_collector/manifests/metrics/service_heartbeat.pp

@@ -21,8 +21,6 @@ class lma_collector::metrics::service_heartbeat (
   validate_array($services)
 
   if (size($services) > 0) {
-    $regexp = join(sort($services), '|')
-
     heka::filter::sandbox { 'service_heartbeat':
       config_dir      => $lma_collector::params::config_dir,
       filename        => "${lma_collector::params::plugins_dir}/filters/service_heartbeat.lua",

deployment_scripts/puppet/modules/lma_collector/manifests/metrics/service_heartbeat_legacy.pp

@@ -0,0 +1,37 @@
+#    Copyright 2015 Mirantis, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+#
+class lma_collector::metrics::service_heartbeat_legacy (
+  $services,
+  $timeout  = $lma_collector::params::heartbeat_timeout,
+) inherits lma_collector::params {
+  include lma_collector::service
+
+  validate_array($services)
+
+  if (size($services) > 0) {
+    $regexp = join(sort($services), '|')
+
+    heka::filter::sandbox { 'service_heartbeat':
+      config_dir      => $lma_collector::params::config_dir,
+      filename        => "${lma_collector::params::plugins_dir}/filters/service_heartbeat_legacy.lua",
+      message_matcher => join(['Fields[name] =~ /^', join(sort($services), '|'), '/'], ''),
+      ticker_interval => 10,
+      config          => {
+        timeout => $timeout,
+      },
+      notify          => Class['lma_collector::service'],
+    }
+  }
+}