Add ovs logs collecting support for heka

Since ovs logging to stderr too and it's not OS service, I have to
create a multidecoder for dockerlogs. Using it, I can try different
decoders on each log string to find the right one.

It's the only option available for us right now, since Dockerlogs
plugin doesnt support docker containers filtering.

Change-Id: I53ab2beb49e5847c3e17e443a7838b0448cb066f

docker/heka/plugins/decoders/os_ovs.lua

@@ -0,0 +1,82 @@
+-- Copyright 2015-2016 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 "string"
+local l      = require 'lpeg'
+l.locale(l)
+local dt     = require "date_time"
+
+local patt   = require 'os_patterns'
+local utils  = require 'os_utils'
+
+local msg = {
+    Timestamp   = nil,
+    Type        = 'log',
+    Hostname    = nil,
+    Payload     = nil,
+    Fields      = {},
+    Severity    = nil,
+}
+
+-- ovs logs looks like this:
+-- 2016-08-10T09:27:41Z|00038|connmgr|INFO|br-ex<->tcp:127.0.0.1:6633: 2 flow_mods 10 s ago (2 adds)
+
+-- Different pieces of pattern
+local sp = patt.sp
+local colon = patt.colon
+local pipe = patt.pipe
+local dash = patt.dash
+
+local p_timestamp = l.Cg(l.Ct(dt.rfc3339_full_date * (sp + l.P"T") * dt.rfc3339_partial_time * (dt.rfc3339_time_offset + dt.timezone_offset)^-1), "Timestamp")
+local p_id = l.Cg(l.digit^-5, "Message_ID")
+local p_module = l.Cg(l.R("az")^0, "Module")
+local p_severity_label = l.Cg(l.R("AZ")^0, "SeverityLabel")
+local p_message = l.Cg(patt.Message, "Message")
+
+local ovs_grammar = l.Ct(p_timestamp * pipe * p_id * pipe * p_module * pipe * p_severity_label * pipe * p_message)
+local pattern = read_config("heka_service_pattern") or "^k8s_(.-)%..*"
+
+
+function process_message ()
+    local cont_name = read_message("Fields[ContainerName]")
+    local program = string.match(cont_name, pattern)
+    local service = nil
+
+    if program == nil then
+        program = "unknown_program"
+    else
+        service = string.match(program, '(.-)%-.*')
+    end
+
+    --- If service is still nil, it means we fail to match current service
+    --- using both patterns, so we set fallback one.
+    if service == nil then
+        service = "unknown_service"
+    end
+
+    local log = read_message("Payload")
+
+    local m = ovs_grammar:match(log)
+    if not m then return -1 end
+
+    msg.Timestamp = m.Timestamp
+    msg.Logger = service
+    msg.Payload = m.Message
+    msg.Severity = utils.label_to_severity_map[m.SeverityLabel] or 7
+    msg.Fields.module = m.Module
+    msg.Fields.message_id = m.Message_ID
+    msg.Fields.programname = program
+    msg.Fields.severity_label = m.SeverityLabel
+
+    return utils.safe_inject_message(msg)
+end

docker/heka/plugins/modules/os_patterns.lua

@@ -36,6 +36,7 @@ colon = l.P":"
 dash = l.P"-"
 dot = l.P'.'
 quote = l.P'"'
+pipe = l.P'|'
 
 local x4digit = l.xdigit * l.xdigit * l.xdigit * l.xdigit
 local uuid_dash = l.C(x4digit * x4digit * dash * x4digit * dash * x4digit * dash * x4digit * dash * x4digit * x4digit * x4digit)

service/files/heka-dockerlogs.toml

@@ -0,0 +1,10 @@
+[docker_logs_decoder]
+type = "MultiDecoder"
+subs = ['openstack_log_decoder', 'ovs_log_decoder']
+cascade_strategy = "first-wins"
+log_sub_errors = false
+
+[docker_log_input]
+type = "DockerLogInput"
+decoder = "docker_logs_decoder"
+log_decode_failures = false

service/files/heka-openstack.toml.j2

@@ -4,8 +4,3 @@ filename = "lua_decoders/os_openstack_log.lua"
 
 [openstack_log_decoder.config]
 heka_service_pattern = "{{ heka_service_pattern }}"
-
-[docker_log_input]
-type = "DockerLogInput"
-decoder = "openstack_log_decoder"
-log_decode_failures = false

service/files/heka-ovs.toml.j2

@@ -0,0 +1,6 @@
+[ovs_log_decoder]
+type = "SandboxDecoder"
+filename = "lua_decoders/os_ovs.lua"
+
+[ovs_log_decoder.config]
+heka_service_pattern = "{{ heka_service_pattern }}"

service/heka.yaml

@@ -29,6 +29,8 @@ service:
           - heka-mariadb.toml
           - heka-openstack.toml
           - heka-rabbitmq.toml
+          - heka-ovs.toml
+          - heka-dockerlogs.toml
 files:
   heka-global.toml:
     path: /etc/heka/heka-global.toml
@@ -45,3 +47,9 @@ files:
   heka-rabbitmq.toml:
     path: /etc/heka/heka-rabbitmq.toml
     content: heka-rabbitmq.toml.j2
+  heka-ovs.toml:
+    path: /etc/heka/heka-ovs.toml
+    content: heka-ovs.toml.j2
+  heka-dockerlogs.toml:
+    path: /etc/heka/heka-dockerlogs.toml
+    content: heka-dockerlogs.toml