Merge pull request #175 from mozilla-services/dev

Sprint Oct 16

Author: Mike Trinkala

moz_logging/CMakeLists.txt

@@ -3,6 +3,6 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 cmake_minimum_required(VERSION 3.0)
-project(moz-logging VERSION 0.0.1 LANGUAGES C)
+project(moz-logging VERSION 0.0.2 LANGUAGES C)
 set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "Mozilla Infrastructure Logging Module")
 include(sandbox_module)

moz_logging/io_modules/decoders/moz_logging/json_heka.lua

@@ -34,6 +34,7 @@ Decode and inject the resulting message
 --]]
 
 -- Imports
+local clf    = require "lpeg.common_log_format"
 local cjson  = require "cjson"
 local string = require "string"
 local util   = require "heka.util"
@@ -73,6 +74,14 @@ function decode(data, dh)
              else
                  msg.Type = dh.Type
              end
+
+             local agent = msg.Fields.agent
+             if agent then
+                 msg.Fields.user_agent_browser,
+                 msg.Fields.user_agent_version,
+                 msg.Fields.user_agent_os = clf.normalize_user_agent(agent)
+             end
+
              ok, msg = pcall(inject_message, msg)
              if not ok then
                  err_msg.Payload = msg

moz_logging/io_modules/decoders/moz_logging/line_splitter.lua

@@ -0,0 +1,97 @@
+-- This Source Code Form is subject to the terms of the Mozilla Public
+-- License, v. 2.0. If a copy of the MPL was not distributed with this
+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+--[[
+# Line Splitter Message Decoder Module
+
+Converts a single message packed with individual new line delimited messages
+into multiple messages.
+
+## Decoder Configuration Table
+
+```lua
+decoders_line_splitter = {
+  sub_decoder = Decoder module name or grammar module name
+}
+```
+
+
+## Functions
+
+### decode
+
+Decode and inject the resulting message
+
+*Arguments*
+- data (string) - JSON messages, one per line, with a Heka schema
+- default_headers (optional table) - Heka message table containing the default
+  header values to use, if they are not populated by the decoder. If 'Fields'
+  is specified it should be in the hashed based format see:
+  http://mozilla-services.github.io/lua_sandbox/heka/message.html
+
+*Return*
+- nil - throws an error on an invalid data type, JSON parse error,
+  inject_message failure etc.
+
+--]]
+
+-- Imports
+local module_name   = ...
+local module_cfg    = require "string".gsub(module_name, "%.", "_")
+local cfg = read_config(module_cfg) or {}
+assert(type(cfg.sub_decoder) == "string", "sub_decoder must be set")
+
+local string    = require "string"
+local sd_module = require(cfg.sub_decoder)
+
+local assert = assert
+local pairs  = pairs
+local pcall  = pcall
+local type   = type
+
+local inject_message = inject_message
+local read_config    = read_config
+
+local M = {}
+setfenv(1, M) -- Remove external access to contain everything in the module
+
+local sub_decoder
+if cfg.sub_decoder:match("^decoders%.") then
+    sub_decoder = sd_module.decode
+    assert(type(sub_decoder) == "function", "sub_decoders, no decode function defined")
+else
+    local grammar = sd_module.grammar or sd_module.syslog_grammar
+    assert(type(grammar) == "userdata", "sub_decoders, no grammar defined")
+    sub_decoder = function(data, dh)
+        if not dh then dh = {} end
+        local fields = grammar:match(data)
+        if not fields then return "parse failed" end
+        for k,v in pairs(fields) do
+            dh.Fields[k] = v
+        end
+        inject_message(dh)
+    end
+end
+
+local err_msg = {
+    Logger  = read_config("Logger"),
+    Type    = "error",
+    Payload = nil,
+    Fields  = {
+        data = nil
+    }
+}
+
+function decode(data, dh)
+     for line in string.gmatch(data, "([^\n]+)\n*") do
+         local err = sub_decoder(line, dh)
+         if err then
+             err_msg.Payload = err
+             err_msg.Fields.data = line
+             pcall(inject_message, err_msg)
+         end
+     end
+end
+
+return M

moz_pioneer/CMakeLists.txt

@@ -3,7 +3,7 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 cmake_minimum_required(VERSION 3.0)
-project(moz-pioneer VERSION 0.0.2 LANGUAGES C)
+project(moz-pioneer VERSION 0.0.3 LANGUAGES C)
 set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "Mozilla Firefox Pioneer Data Processing")
 set(CPACK_DEBIAN_PACKAGE_DEPENDS "${PACKAGE_PREFIX}-moz-ingest (>= 0.0.1), ${PACKAGE_PREFIX}-moz-telemetry (>= 1.2.13), ${PACKAGE_PREFIX}-jose (>= 0.0.1), ${PACKAGE_PREFIX}-lpeg (>= 1.0.0), ${PACKAGE_PREFIX}-rjson (>= 1.1.0)")
 string(REGEX REPLACE "[()]" "" CPACK_RPM_PACKAGE_REQUIRES ${CPACK_DEBIAN_PACKAGE_DEPENDS})

moz_pioneer/io_modules/decoders/moz_ingest/pioneer.lua

@@ -66,6 +66,7 @@ local string        = require "string"
 local module_cfg    = string.gsub(module_name, "%.", "_")
 
 local io     = require "io"
+local lfs    = require "lfs"
 local jose   = require "jose"
 local rjson  = require "rjson"
 local dt     = require "lpeg.date_time"
@@ -84,6 +85,24 @@ local inject_message       = inject_message
 local M = {}
 setfenv(1, M) -- Remove external access to contain everything in the module
 
+
+function load_json_schemas_dir(path)
+    local schemas = {}
+    for fn in lfs.dir(path) do
+        local name = fn:match("(.+%.%d+)%.schema%.json$")
+        if name then
+            local fh = assert(io.input(string.format("%s/%s", path, fn)))
+            local schema = fh:read("*a")
+            fh:close()
+            local ok, rjs = pcall(rjson.parse_schema, schema)
+            if not ok then error(string.format("%s: %s", fn, rjs)) end
+            schemas[name] = rjs
+        end
+    end
+    return schemas
+end
+
+
 local function load_cfg()
     local cfg = read_config(module_cfg)
     assert(type(cfg) == "table", module_cfg .. " must be a table")
@@ -96,7 +115,7 @@ local function load_cfg()
     if not ok then error(string.format("%s: %s", cfg.envelope_schema_file, envelope_schema)) end
 
     assert(type(cfg.schema_path) == "string", "schema_path must be set")
-    local schemas = miu.load_json_schemas(cfg.schema_path)
+    local schemas = load_json_schemas_dir(cfg.schema_path)
 
     local cnt = 0
     local jose_keys = {}
@@ -173,9 +192,10 @@ function transform_message(hsr, msg)
     msg.Fields.normalizedChannel    = mtn.channel(msg.Fields.appUpdateChannel)
 
     local pay = env:find("payload")
-    msg.Fields.studyVersion = env:value(env:find(pay, "studyVersion"))
-    msg.Fields.studyName    = env:value(env:find(pay, "studyName"))
-    msg.Fields.pioneerId    = env:value(env:find(pay, "pioneerId"))
+    msg.Fields.schemaName    = env:value(env:find(pay, "schemaName"))
+    msg.Fields.schemaVersion = env:value(env:find(pay, "schemaVersion"))
+    msg.Fields.studyName     = env:value(env:find(pay, "studyName"))
+    msg.Fields.pioneerId     = env:value(env:find(pay, "pioneerId"))
 
     -- verify the decryption and validation metadata
     local ekey  = env:value(env:find(pay, "encryptionKeyId"))
@@ -184,12 +204,10 @@ function transform_message(hsr, msg)
         error("jose\tno encryptionKeyId: " .. ekey, 0)
     end
 
-    local schema = schemas[msg.Fields.studyName]
-    if schema then schema = schema[msg.Fields.studyVersion] end
-
+    local sn = string.format("%s.%d", msg.Fields.schemaName, msg.Fields.schemaVersion)
+    local schema = schemas[sn]
     if not schema then
-        error(string.format("schema\tno study schema: %s ver: %d",
-                            msg.Fields.studyName, msg.Fields.studyVersion), 0)
+        error(string.format("schema\tno schema: %s study: %s", sn, msg.Fields.studyName), 0)
     end
 
     -- decrypt and validate the study data

moz_pioneer/tests/hindsight/input.hpb

@@ -10,11 +10,12 @@ require "string"
 
 local messages = {
     {Logger = "telemetry", Type = "telemetry", Fields = {
-        studyVersion = 1,
+        schemaName = "event",
+        schemaVersion = 1,
         appVersion = "45.0",
-        studyName = "example",
+        studyName = "test-study",
         pioneerId = "11111111-1111-1111-1111-111111111111",
-        submission = '{"exampleString":"foobar"}',
+        submission = '{"eventId":"enrolled"}',
         appUpdateChannel = "beta",
         normalizedChannel = "beta",
         creationTimestamp = 1.446686743312e+18,
@@ -28,9 +29,10 @@ local messages = {
         }
     },
     {Logger = "telemetry", Type = "telemetry.metadata", Fields = {
-        studyVersion = 1,
+        schemaName = "event",
+        schemaVersion = 1,
         appVersion = "45.0",
-        studyName = "example",
+        studyName = "test-study",
         pioneerId = "11111111-1111-1111-1111-111111111111",
         appUpdateChannel = "beta",
         normalizedChannel = "beta",
@@ -72,11 +74,12 @@ local messages = {
         "xpcomAbi": "x86_64-gcc3"
       },
       "payload" : {
-        "encryptedData": "eyJlbmMiOiAiQTI1NkdDTSIsICJhbGciOiAiUlNBLU9BRVAiLCAiemlwIjogIkRFRiIsICJraWQiOiAiMjk0MDkyMWUtMzY0Ni00NTFjLTg1MTAtOTcxNTUyNzU0ZTc0In0.hnXLQf6VoLkjhikf2jeNUtWlwXuCj8nKUjcrOpsWvpLpeDxOJKrxs5Iqm5myxV8UIOrHCGnowmUvvG4bu-KLdC-QzXTmMGovwE8_VxVFO6gCVmKlgLBDgcfjbKajy7JGtozomuSz-lhP5BIMzVfMF6XoVSKq2YDskzVTMGZdVnPvzFjYJpK8lEKxA_MVZF-NJ_JJKn6jtbo3y5SvuXPOKVu7JBCwvrb6VHEm0tMwhCzeH3YemUluOXTeoCMXnKx0bRJhEoeb8Bh-oNLSEYKVI1jFK-Bzc3VZ1cBMTQ_nKWorObgewdYFZsxM4bChT3dtKxpb_udG2KtaUxyxfAoZ2r0GvphSnC2lITHoHk49C54-VttMZcMyHHvFEXO_gnBsVOoP0Z2b3LXAyL6dhxg8doc4CWTj3oAA2MnzQl7ZoaiFZc4NYuav2MYpSomyGmXSLs9qHXYSMhpq3q-qSyPEm28x2iEb05HsK9e0jUyRcy1JimjaC5x4tBi4xNMRpRUNWrYkqLDsi5EneUTCmVEhKStNk6fHnqOTWoM5A_55KAxTQWknsSVaGWXQtk2O9dsDa6f7Rm3Zssb_bQVXc_EjHimXd02fSnv6rThZ2qa7-wDlKYD88vrmD358oxdwziVBpfU4DRjdMefdJybxKz_l8fwVtp7dZBNC8p469s_a2y8.ZmoNDzcRaVqrnwgO.2iyWrGdfGuAtp1Tv.9yVPdJMA63ZuNpmAKLPnTw",
+        "encryptedData": "eyJlbmMiOiAiQTI1NkdDTSIsICJhbGciOiAiUlNBLU9BRVAiLCAiemlwIjogIkRFRiIsICJraWQiOiAiMjk0MDkyMWUtMzY0Ni00NTFjLTg1MTAtOTcxNTUyNzU0ZTc0In0.ds7LT3vtshHRibx3twJQSKb8n_W6EtvpD7597KjMBsA4NySKTF0cgGE4m7MvMDXzYUeJq5K1sAnuYNAFdUML2my06rXWB3Q8gP6PRjF2hYzj84NKYVBwBr0XdgSfGqx_ja3XZX0f8LKkCUppRDo_9YuK-7kkw4_NDMLen-f3o9ta87w9Nn9lbw1m62yhkR8S2jiK3W4jpCnbxIeyZMyo-u-iCdEN4gdtH3ledcpeSZXn6b-L6d_4iQx2Y98Y5xvkSakXkipowsDd9FI7yE_gprV1pJDV29lDmH7Km_9ZSGA6NQTZs0fkOcJIhZprW4Bq_aP2tlPBU343dC-6lrVitQLxGgYgUDZduR4E0T3XJ4LJaPNIJKsTx-s9UXGur0U0qCIBT-bJKD3bBeVJmSA7ZMcuOGHktCQsx0Fr84IOInFaOCZSulPS_H0IThB23Z-Z9e-dX-c1s-YfjSUvMiyuG_mciDLo27AsN2FMOPQD-tKkOgnz231Ri3GTK777OYpsYTr8Q0vBRdVLJIi_-dIMTSuRHX3RCwduVpR_EnPdvMh_O7949W45gLoyz_Z96BarU6WDssPrCLGPkHHVDQqCO7nHz8RB5l5Jpq0Y2D_2Aaq14qDZBfldgkZJC6QufefoVtwqWWdd_R_PE_gqazJmGTsesHtgTnvT8VydnmxeyIo.LM9z4RRfWVynWd0u.SQehWUthukk7EUX2.L8zyxPzzWbDalhMdLlCPhg",
         "encryptionKeyId": "pioneer-20170901",
         "pioneerId": "11111111-1111-1111-1111-111111111111",
-        "studyName": "example",
-        "studyVersion": 1
+        "studyName": "test-study",
+        "schemaName": "event",
+        "schemaVersion": 1
       }
 }
 }]],
@@ -86,9 +89,10 @@ local messages = {
         uri = "/submit/telemetry/0055FAC4-8A1A-4FCA-B380-EBFDC8571A02/pioneer-study/Firefox/45.0/release/20151103030248",
         DecodeErrorType = "json",
         DecodeError = "invalid study: failed to parse offset:1 Invalid value.",
-        studyVersion = 1,
+        schemaName = "event",
+        schemaVersion = 1,
         appVersion = "45.0",
-        studyName = "example",
+        studyName = "test-study",
         pioneerId = "11111111-1111-1111-1111-111111111111",
         appUpdateChannel = "beta",
         normalizedChannel = "beta",
@@ -104,7 +108,7 @@ local messages = {
     },
     {Logger = "telemetry", Type = "telemetry.error", Fields = {
         DecodeErrorType = "json",
-        DecodeError = "study validation: SchemaURI: #/properties/exampleString Keyword: type DocumentURI: #/exampleString"
+        DecodeError = "study validation: SchemaURI: #/properties/eventId Keyword: type DocumentURI: #/eventId"
         }
     },
     {Logger = "telemetry", Type = "telemetry.error", Fields = {
@@ -114,12 +118,12 @@ local messages = {
     },
     {Logger = "telemetry", Type = "telemetry.error", Fields = {
         DecodeErrorType = "schema",
-        DecodeError = "no study schema: bogus ver: 1"
+        DecodeError = "no schema: bogus.1 study: test-study"
         }
     },
     {Logger = "telemetry", Type = "telemetry.error", Fields = {
         DecodeErrorType = "schema",
-        DecodeError = "no study schema: example ver: 2"
+        DecodeError = "no schema: event.2 study: test-study"
         }
     },
     {Logger = "telemetry", Type = "telemetry.error", Fields = {

moz_pioneer/tests/hindsight/run/analysis/verify_decoder.lua

@@ -44,19 +44,20 @@ local envelope = [[
         "encryptionKeyId": "%s",
         "pioneerId": "11111111-1111-1111-1111-111111111111",
         "studyName": "%s",
-        "studyVersion": %d
+        "schemaName": "%s",
+        "schemaVersion": %d
       }
 }
 }]]
 
 
 local submissions = {
-    {'{ "exampleString" : "foobar"}', 'pioneer-20170901', "example", 1}, -- valid
-    {'text', 'pioneer-20170901', "example", 1}, -- parse failure
-    {'{ "exampleString" : 1}', 'pioneer-20170901', "example", 1}, -- study schema validation error
-    {'{ "exampleString" : "foobar"}', 'pioneer-20200901', "example", 1}, -- no encryption key
-    {'{ "exampleString" : "foobar"}', 'pioneer-20170901', "bogus", 1}, -- no schema
-    {'{ "exampleString" : "foobar"}', 'pioneer-20170901', "example", 2}, -- no version
+    {'{ "eventId" : "enrolled"}', 'pioneer-20170901', "test-study", "event", 1}, -- valid
+    {'text', 'pioneer-20170901', "test-study", "event", 1}, -- parse failure
+    {'{ "eventId" : 1}', 'pioneer-20170901', "test-study", "event", 1}, -- study schema validation error
+    {'{ "eventId" : "foobar"}', 'pioneer-20200901', "test-study", "event", 1}, -- no encryption key
+    {'{ "eventId" : "foobar"}', 'pioneer-20170901', "test-study", "bogus", 1}, -- no schema
+    {'{ "eventId" : "foobar"}', 'pioneer-20170901', "test-study", "event", 2}, -- no version
 }
 
 local msg = {
@@ -81,7 +82,7 @@ function process_message()
     for i,v in ipairs(submissions) do
         msg.Fields.uri = string.format(uri_template, i)
         local jwe = jose.jwe_encrypt(jwk, v[1], hdr)
-        msg.Fields.content = string.format(envelope, jwe:export(), v[2], v[3], v[4])
+        msg.Fields.content = string.format(envelope, jwe:export(), v[2], v[3], v[4], v[5])
         inject_message(msg)
         if i == 1 then inject_message(msg) end -- test duplicate
         cnt = cnt + 1
@@ -103,14 +104,14 @@ function process_message()
     -- invalid import
     cnt = cnt + 1
     msg.Fields.uri = string.format(uri_template, cnt)
-    msg.Fields.content = string.format(envelope, "xxxxxxxxxxxxxxxx", v[2], v[3], v[4])
+    msg.Fields.content = string.format(envelope, "xxxxxxxxxxxxxxxx", v[2], v[3], v[4], v[5])
     inject_message(msg)
 
     -- invalid encryption
     cnt = cnt + 1
     msg.Fields.uri = string.format(uri_template, cnt)
     local jwe = jose.jwe_encrypt(jwk, submissions[1][1], hdr)
-    msg.Fields.content = string.format(envelope, string.gsub(jwe:export(), ".$", "X"), v[2], v[3], v[4])
+    msg.Fields.content = string.format(envelope, string.gsub(jwe:export(), ".$", "X"), v[2], v[3], v[4], v[5])
     inject_message(msg)
 
     return 0

moz_pioneer/tests/hindsight/run/input/generate_data.lua

@@ -3,7 +3,7 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 cmake_minimum_required(VERSION 3.0)
-project(syslog VERSION 1.0.7 LANGUAGES C)
+project(syslog VERSION 1.0.8 LANGUAGES C)
 set(CPACK_PACKAGE_DESCRIPTION_SUMMARY "Syslog parsers and collectors")
 set(CPACK_DEBIAN_PACKAGE_DEPENDS "${PACKAGE_PREFIX}-lpeg (>= 1.0.5), ${PACKAGE_PREFIX}-socket (>= 3.0)")
 string(REGEX REPLACE "[()]" "" CPACK_RPM_PACKAGE_REQUIRES ${CPACK_DEBIAN_PACKAGE_DEPENDS})

syslog/CMakeLists.txt

@@ -60,6 +60,17 @@ syslog_grammar = l.Ct(
           * l.Cg((l.P(1)-l.S" ")^0, "remote_user")
           * l.P(-1)
         )
+    + (
+            l.P"Connection from "
+            * l.Cg(ipv46, "remote_addr")
+            * l.P" port "
+            * l.Cg(l.digit^1 / tonumber, "remote_port")
+            * l.P" on "
+            * l.Cg(ipv46, "local_addr")
+            * l.P" port "
+            * l.Cg(l.digit^1 / tonumber, "local_port")
+          * l.P(-1)
+        )
     + (
             l.P"Connection closed by "
           * l.Cg(ipv46, "remote_addr")
@@ -106,6 +117,9 @@ syslog_grammar = l.Ct(
           * l.P": "
           * l.Cg(l.P(1)^1, "sshd_error")
         )
+    + (
+            l.Cg(l.P"Set " * l.P(1)^1, "set")
+        )
     )
 
 return M

syslog/modules/lpeg/linux/sshd.lua

@@ -28,3 +28,13 @@ assert(fields.remote_addr.value == '10.2.3.4', fields.remote_addr)
 assert(fields.disconnect_reason == 11, fields.disconnect_reason)
 assert(fields.disconnect_msg == 'The user disconnected the application [preauth]', fields.disconnect_msg)
 
+log = "Connection from 121.18.238.123 port 60512 on 172.31.41.219 port 22"
+fields = grammar:match(log)
+assert(fields.remote_addr.value == '121.18.238.123', fields.remote_addr)
+assert(fields.remote_port == 60512, fields.remote_port)
+assert(fields.local_addr.value == '172.31.41.219', fields.local_addr)
+assert(fields.local_port == 22, fields.local_port)
+
+log = "Set /proc/self/oom_score_adj to 0"
+fields = grammar:match(log)
+assert(fields.set == log, fields.set)