Add schemaName/Version support to pioneer

Author: trink

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