From 6adaa37c7af413d2c991982523ffa955d687f3c6 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 23 Jun 2026 11:24:27 +0000 Subject: [PATCH 1/9] feat: add lifecycle metrics support for component updates --- src/services/update/component_watch.lua | 19 +++- src/services/update/lifecycle_metrics.lua | 106 +++++++++++++++++++ src/services/update/service.lua | 119 ++++++++++++++++++++++ 3 files changed, 242 insertions(+), 2 deletions(-) create mode 100644 src/services/update/lifecycle_metrics.lua diff --git a/src/services/update/component_watch.lua b/src/services/update/component_watch.lua index 2ada84fb..adca3714 100644 --- a/src/services/update/component_watch.lua +++ b/src/services/update/component_watch.lua @@ -8,6 +8,7 @@ local fibers = require 'fibers' local scoped_work = require 'devicecode.support.scoped_work' local bus_cleanup = require 'devicecode.support.bus_cleanup' local topics = require 'services.update.topics' +local queue = require 'devicecode.support.queue' local M = {} @@ -45,9 +46,20 @@ local function open_watches(scope, conn, components, queue_len) return watches, nil end +local function report_component_fact(params, component, ev) + if not params.events_tx then return true, nil end + return queue.try_admit_required(params.events_tx, { + kind = 'component_fact_changed', + component = component, + payload = ev.payload, + origin = ev.origin, + }, 'update_component_fact_changed_admission_failed') +end + local function watch_loop(scope, params) local observer = assert(params.observer, 'component_watch observer required') - local watches, err = open_watches(scope, assert(params.conn, 'component_watch conn required'), params.components, params.queue_len) + local conn = assert(params.conn, 'component_watch conn required') + local watches, err = open_watches(scope, conn, params.components, params.queue_len) if not watches then error(err or 'component_watch_open_failed', 0) end while true do @@ -62,7 +74,9 @@ local function watch_loop(scope, params) return { role = 'update_component_watch', reason = recv_err or 'component_watch_closed' } end if ev.op == 'retain' or ev.event == 'retain' or ev.type == 'retain' or ev.kind == 'retain' then - observer:update_component(component, ev.payload, ev.origin) + local ok_update, update_err = observer:update_component(component, ev.payload, ev.origin) + if ok_update == nil then error(update_err or 'component_observer_update_failed', 0) end + report_component_fact(params, component, ev) elseif ev.op == 'unretain' or ev.event == 'unretain' or ev.type == 'unretain' or ev.kind == 'unretain' then observer:remove_component(component, 'component_unretained') end @@ -87,6 +101,7 @@ function M.start(scope, params) observer = params.observer, components = components, queue_len = params.queue_len, + events_tx = params.events_tx, }) end, report = params.report, diff --git a/src/services/update/lifecycle_metrics.lua b/src/services/update/lifecycle_metrics.lua new file mode 100644 index 00000000..c3fd7f56 --- /dev/null +++ b/src/services/update/lifecycle_metrics.lua @@ -0,0 +1,106 @@ +-- services/update/lifecycle_metrics.lua +-- +-- Small publisher helper for component update lifecycle metrics. The Update +-- service derives lifecycle records from Device component facts, then calls +-- emit() with the phase it wants to publish. + +local M = {} + +M.METRIC_NAME = 'component_update_lifecycle' + +-- Metric payload shape consumed by services.metrics: +-- { +-- value = 'started' | 'completed' | 'failed' | 'cancelled', +-- namespace = { '', 'lifecycle', '', '' }, +-- job_id = '', component = '', ... +-- } +-- The namespace is later used as a metrics topic suffix, so each segment must +-- avoid separators and whitespace. + +---Return a string safe to use as one metric namespace segment. +---This is a topic/namespace normaliser, not a security boundary: it preserves +---letters, digits, '_' and '-', replaces every other run with '_', trims edge +---underscores, and falls back to 'unknown' if nothing usable remains. +---@param value any Original component or job identifier. +---@return string token Namespace-safe token. +local function safe_token(value) + local s = tostring(value or '') + s = s:gsub('[^%w%-_]', '_') + s = s:gsub('_+', '_') + s = s:gsub('^_+', ''):gsub('_+$', '') + if s == '' then return 'unknown' end + return s +end + +---@param record table|nil Lifecycle record built from Device component facts. +---@return string|nil component Component id when present and non-empty. +local function component_of(record) + local component = type(record) == 'table' and record.component or nil + if type(component) ~= 'string' or component == '' then return nil end + return component +end + +---Build the metric payload expected by services.metrics. +---@param record table Lifecycle record with job_id/component/state/error fields. +---@param phase string Lifecycle phase to publish. +---@param extra table|nil Additional local-observability fields. +---@return table|nil payload Metric payload, or nil on invalid input. +---@return string|nil err Error code when payload is nil. +local function payload(record, phase, extra) + local component = component_of(record) + if not component then return nil, 'component_required' end + if type(phase) ~= 'string' or phase == '' then return nil, 'phase_required' end + + local out = { + value = phase, + namespace = { safe_token(component), 'lifecycle', safe_token(record.job_id), phase }, + job_id = record.job_id, + component = component, + state = record.state, + error = record.error, + } + for k, v in pairs(extra or {}) do + out[k] = v + end + return out, nil +end + +---@param conn table|nil Bus connection fallback, used when no service_base object exists. +---@param svc table|nil service_base-like object with obs_metric(). +---@param p table Metric payload from payload(). +---@return boolean|nil ok True when published. +---@return string|nil err Error code when no sink is available. +local function publish(conn, svc, p) + if svc and type(svc.obs_metric) == 'function' then + svc:obs_metric(M.METRIC_NAME, p) + return true, nil + end + + if conn and type(conn.retain) == 'function' then + conn:retain({ 'obs', 'v1', 'update', 'metric', M.METRIC_NAME }, p) + return true, nil + end + + return nil, 'metric_sink_unavailable' +end + +---@param value any +---@return string +function M.safe_token(value) + return safe_token(value) +end + +---@param conn table|nil +---@param svc table|nil +---@param record table Lifecycle record derived from a Device component fact. +---@param phase string +---@param extra table|nil +---@return boolean|nil ok +---@return string|nil err +function M.emit(conn, svc, record, phase, extra) + local p, err = payload(record, phase, extra) + if not p then return nil, err end + return publish(conn, svc, p) +end + +return M diff --git a/src/services/update/service.lua b/src/services/update/service.lua index 6192b672..0e7d793c 100644 --- a/src/services/update/service.lua +++ b/src/services/update/service.lua @@ -33,6 +33,7 @@ local component_watch = require 'services.update.component_watch' local artifact_store_bus = require 'services.update.artifacts.store_bus' local component_backend_mod = require 'services.update.backends.component' local router_backend_mod = require 'services.update.backends.router' +local lifecycle_metrics = require 'services.update.lifecycle_metrics' local M = {} @@ -135,6 +136,116 @@ local function update_model_state(self, state, reason) end end + +local function updater_from_fact(fact) + if type(fact) ~= 'table' then return nil end + return type(fact.updater) == 'table' and fact.updater + or type(fact.update) == 'table' and fact.update + or nil +end + +local function software_from_fact(fact) + return type(fact) == 'table' and type(fact.software) == 'table' and fact.software or nil +end + +local function expected_image_id(job) + if type(job) ~= 'table' then return nil end + local meta = type(job.metadata) == 'table' and job.metadata or {} + local art = type(job.artifact) == 'table' and job.artifact or nil + local stage = type(job.stage_result) == 'table' and job.stage_result + or type(job.staged) == 'table' and job.staged + or type(job.stage) == 'table' and job.stage + or nil + local preflight = type(stage) == 'table' and type(stage.preflight) == 'table' and stage.preflight or nil + local pre_commit = type(job.commit_attempt) == 'table' and type(job.commit_attempt.pre_commit) == 'table' + and job.commit_attempt.pre_commit or nil + return job.expected_image_id + or meta.expected_image_id + or meta.image_id + or (type(stage) == 'table' and (stage.expected_image_id or stage.image_id)) + or (preflight and (preflight.expected_image_id or preflight.image_id)) + or (pre_commit and pre_commit.expected_image_id) + or (art and (art.expected_image_id or art.image_id)) +end + +local function lifecycle_record(component, fact) + local upd = updater_from_fact(fact) or {} + return { + job_id = upd.job_id, + component = component or (type(fact) == 'table' and fact.component), + state = upd.state, + error = upd.last_error or upd.error, + } +end + +local function lifecycle_metric_key(record, phase) + return table.concat({ + tostring(record and record.component or 'unknown'), + tostring(record and record.job_id or 'unknown'), + tostring(phase or 'unknown'), + }, '/') +end + +local function lifecycle_metric_sent(self, record, phase) + self._component_lifecycle_sent = self._component_lifecycle_sent or {} + return self._component_lifecycle_sent[lifecycle_metric_key(record, phase)] == true +end + +local function mark_lifecycle_metric_sent(self, record, phase, sent) + self._component_lifecycle_sent = self._component_lifecycle_sent or {} + self._component_lifecycle_sent[lifecycle_metric_key(record, phase)] = sent == true or nil +end + +local function emit_component_lifecycle_metric(self, record, phase, extra) + if lifecycle_metric_sent(self, record, phase) then + return true, nil + end + mark_lifecycle_metric_sent(self, record, phase, true) + + local ok, err = lifecycle_metrics.emit(self._conn, self._svc, record, phase, extra) + if ok ~= true then + mark_lifecycle_metric_sent(self, record, phase, false) + if self._svc and type(self._svc.obs_log) == 'function' then + self._svc:obs_log('warn', { what = 'component_lifecycle_metric_emit_failed', err = tostring(err) }) + end + return true, nil + end + return true, nil +end + +local function component_fact_phase(component, fact, self) + local upd = updater_from_fact(fact) or {} + if type(upd.job_id) ~= 'string' or upd.job_id == '' then return nil end + if upd.state == 'cancelled' or upd.state == 'aborted' then return 'cancelled' end + if upd.state == 'failed' or upd.state == 'rollback_detected' or upd.last_error ~= nil or upd.error ~= nil then + return 'failed' + end + + local record = lifecycle_record(component, fact) + if lifecycle_metric_sent(self, record, 'started') then + local job = self._jobs and self._jobs:get(upd.job_id) or nil + local expected = expected_image_id(job) + local sw = software_from_fact(fact) + if type(sw) == 'table' and expected and sw.image_id == expected then + return 'completed' + end + end + + return 'started' +end + +local function emit_component_fact_lifecycle_metric(self, ev) + local component = ev and ev.component or nil + local fact = ev and ev.payload or nil + local phase = component_fact_phase(component, fact, self) + if not phase then return true, nil end + local record = lifecycle_record(component, fact) + return emit_component_lifecycle_metric(self, record, phase, { + source = 'device_component_fact', + reason = phase, + }) +end + local function apply_generation_snapshot(self, snapshot) self._model:update(function (s) s.generation = snapshot.generation or s.generation @@ -674,6 +785,11 @@ local function reduce_event(self, ev) return end + if ev.kind == 'component_fact_changed' then + emit_component_fact_lifecycle_metric(self, ev) + return + end + if ev.kind == 'active_runtime_changed' then handle_active_runtime_changed(self, ev) return @@ -1168,6 +1284,7 @@ local function ensure_component_watch(self, reason) config = self._config, queue_len = params.component_watch_queue_len, report = service_events.reporter(cw_port, 'update_component_watch_completion_report_failed'), + events_tx = self._done_tx, }) if not cwh then return nil, cwerr or 'update_component_watch_start_failed' end self._component_watch = cwh @@ -1368,6 +1485,7 @@ function M.run(scope, params) local self = setmetatable({ _scope = scope, + _conn = params.conn, _service_id = service_id, _svc = params.svc, _model = service_model, @@ -1398,6 +1516,7 @@ function M.run(scope, params) _component_observer = component_observer, _component_watch = nil, _backend = backend, + _component_lifecycle_sent = {}, _job_runtime_ready = false, _generation_done_queue_len = params.generation_done_queue_len, _manager_route_queue_len = params.manager_route_queue_len, From 540d7a700f7c8816d77c3473afbf1d751844ea52 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 23 Jun 2026 11:24:41 +0000 Subject: [PATCH 2/9] feat: add component update lifecycle metrics tests and integration --- tests/integration/devhost/metrics_spec.lua | 37 ++++ tests/run.lua | 1 + tests/unit/metrics/config_spec.lua | 20 ++ tests/unit/update/test_lifecycle_metrics.lua | 63 +++++++ tests/unit/update/test_service_phase2.lua | 187 +++++++++++++++++++ 5 files changed, 308 insertions(+) create mode 100644 tests/unit/update/test_lifecycle_metrics.lua diff --git a/tests/integration/devhost/metrics_spec.lua b/tests/integration/devhost/metrics_spec.lua index 669b97ec..bc13ce01 100644 --- a/tests/integration/devhost/metrics_spec.lua +++ b/tests/integration/devhost/metrics_spec.lua @@ -525,4 +525,41 @@ function T.per_endpoint_state_isolation() end, { timeout = 3.0 }) end + +function T.component_update_lifecycle_metric_uses_event_namespace() + runfibers.run(function(scope) + local clock = new_test_clock() + scope:finally(function() clock:restore() end) + local bus = make_bus() + local test_conn = bus:connect() + + test_conn:retain({ "cap", "fs", "credentials", "state" }, "added") + start_mock_hal(test_conn, scope) + test_conn:retain({ "state", "time", "synced" }, true) + + local result_sub = test_conn:subscribe( + { "obs", "v1", "metrics", "output", "#" }, + { queue_len = 10, full = "drop_oldest" }) + + test_conn:retain({ "cfg", "metrics" }, bus_pipeline_config("component_update_lifecycle", 0.1)) + local svc_scope = start_metrics(bus, scope) + flush_ticks() + + test_conn:publish( + { "obs", "v1", "update", "metric", "component_update_lifecycle" }, + { value = "started", namespace = { "mcu", "lifecycle", "job-1", "started" } }) + + local msg = recv_metric(clock, result_sub, 0.5) + assert(msg ~= nil, "expected lifecycle metric publish") + assert(table.concat(msg.topic, ".") == "obs.v1.metrics.output.mcu.lifecycle.job-1.started", + "unexpected topic " .. table.concat(msg.topic, ".")) + assert(msg.payload.value == "started", + "expected value=started, got " .. tostring(msg.payload.value)) + + stop_scope(svc_scope) + clock:restore() + end, { timeout = 3.0 }) +end + + return T diff --git a/tests/run.lua b/tests/run.lua index c6ee9e24..f32ec02d 100644 --- a/tests/run.lua +++ b/tests/run.lua @@ -83,6 +83,7 @@ local files = { 'unit.update.test_generation_refactor', 'unit.update.test_ingest_artifacts', 'unit.update.test_job_repository', + 'unit.update.test_lifecycle_metrics', 'unit.update.test_job_runtime', 'unit.update.test_job_store_memory', 'unit.update.test_job_store_control_store', diff --git a/tests/unit/metrics/config_spec.lua b/tests/unit/metrics/config_spec.lua index bc25bd53..83175966 100644 --- a/tests/unit/metrics/config_spec.lua +++ b/tests/unit/metrics/config_spec.lua @@ -137,4 +137,24 @@ function T.validate_config_propagates_invalid_template_to_pipeline() assert(saw_metric_invalid_protocol, "expected warning: sim inherited invalid protocol 'invalid'") end +local function read_project_file(rel) + local candidates = { rel, '../' .. rel } + for i = 1, #candidates do + local f = io.open(candidates[i], 'rb') + if f then local data = f:read('*a'); f:close(); return data end + end + return nil, 'unable to read ' .. rel +end + +function T.bigbox_config_accepts_component_update_lifecycle_pipeline() + local cjson = require 'cjson.safe' + local text = assert(read_project_file('src/configs/bigbox-v1-cm-2.json')) + local doc = assert(cjson.decode(text)) + local metrics = assert(doc.metrics) + local ok, _, err = conf.validate_config(metrics) + assert(ok == true, 'expected bigbox metrics config to validate: ' .. tostring(err)) + assert(metrics.data.pipelines.component_update_lifecycle.protocol == 'http', + 'expected component_update_lifecycle to use HTTP pipeline') +end + return T diff --git a/tests/unit/update/test_lifecycle_metrics.lua b/tests/unit/update/test_lifecycle_metrics.lua new file mode 100644 index 00000000..9f743ad1 --- /dev/null +++ b/tests/unit/update/test_lifecycle_metrics.lua @@ -0,0 +1,63 @@ +-- tests/unit/update/test_lifecycle_metrics.lua + +local metrics = require 'services.update.lifecycle_metrics' + +local tests = {} + +local function fail(msg) error(msg or 'assertion failed', 2) end +local function assert_eq(a, b, msg) + if a ~= b then + fail(msg or ('expected ' .. tostring(b) .. ', got ' .. tostring(a))) + end +end +local function assert_not_nil(v, msg) if v == nil then fail(msg or 'expected non-nil value') end end +local function assert_nil(v, msg) if v ~= nil then fail(msg or ('expected nil, got ' .. tostring(v))) end end + +function tests.test_safe_token_replaces_namespace_unsafe_characters() + assert_eq(metrics.safe_token('job/mcu.1 ready'), 'job_mcu_1_ready') + assert_eq(metrics.safe_token('job-1_ok'), 'job-1_ok') + assert_eq(metrics.safe_token('///'), 'unknown') +end + +function tests.test_emit_publishes_component_lifecycle_payload() + local seen + local svc = { + obs_metric = function (_, name, payload) + seen = { name = name, payload = payload } + end, + } + + local ok, err = metrics.emit(nil, svc, { + job_id = 'job/mcu.1', + component = 'mcu', + state = 'ready', + error = nil, + }, 'started', { source = 'device_component_fact' }) + + assert_eq(ok, true, err) + assert_not_nil(seen, 'expected obs_metric call') + assert_eq(seen.name, 'component_update_lifecycle') + assert_eq(seen.payload.value, 'started') + assert_eq(table.concat(seen.payload.namespace, '.'), 'mcu.lifecycle.job_mcu_1.started') + assert_eq(seen.payload.job_id, 'job/mcu.1') + assert_eq(seen.payload.component, 'mcu') + assert_eq(seen.payload.state, 'ready') + assert_eq(seen.payload.source, 'device_component_fact') +end + +function tests.test_emit_rejects_missing_component_or_phase() + local ok_component, err_component = metrics.emit(nil, {}, { + job_id = 'j1', + }, 'started') + assert_nil(ok_component) + assert_eq(err_component, 'component_required') + + local ok_phase, err_phase = metrics.emit(nil, {}, { + job_id = 'j1', + component = 'mcu', + }, '') + assert_nil(ok_phase) + assert_eq(err_phase, 'phase_required') +end + +return tests diff --git a/tests/unit/update/test_service_phase2.lua b/tests/unit/update/test_service_phase2.lua index b6d26c36..5921ed60 100644 --- a/tests/unit/update/test_service_phase2.lua +++ b/tests/unit/update/test_service_phase2.lua @@ -514,4 +514,191 @@ function tests.test_artifact_store_dependency_gates_generation_after_job_store_r end) end + + +local function lifecycle_metric_topic() + return { "obs", "v1", "update", "metric", "component_update_lifecycle" } +end + +local function component_config(component) + return { + schema = "devicecode.update/1", + components = { { component = component or "mcu" } }, + } +end + +local function retain_component_fact(conn, component, opts) + opts = opts or {} + conn:retain(topics.device_component(component), { + component = component, + software = { + image_id = opts.image_id, + boot_id = opts.boot_id, + }, + updater = { + state = opts.updater_state or "ready", + job_id = opts.job_id, + last_error = opts.last_error, + }, + }) +end + +local function wait_lifecycle_metric(caller, value) + local metric_view = caller:retained_view(lifecycle_metric_topic()) + local metric + local ok = probe.wait_until(function () + local msg = metric_view:get(lifecycle_metric_topic()) + metric = msg and msg.payload or nil + return metric and metric.value == value + end, { timeout = 0.8, interval = 0.01 }) + metric_view:close() + assert_true(ok, "expected " .. tostring(value) .. " lifecycle metric") + return metric +end + +function tests.test_component_device_fact_job_id_emits_lifecycle_started_metric() + fibers.run(function (root_scope) + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "job/mcu.1", + component = "mcu", + artifact_ref = "artifact-mcu", + }, { timeout = 0.5 })) + + retain_component_fact(caller, "mcu", { + job_id = "job/mcu.1", + image_id = "image-old", + boot_id = "boot-1", + }) + + local metric = wait_lifecycle_metric(caller, "started") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.job_mcu_1.started") + assert_eq(metric.job_id, "job/mcu.1") + assert_eq(metric.component, "mcu") + assert_eq(metric.source, "device_component_fact") + + child:cancel("test complete") + end) +end + +function tests.test_component_device_fact_expected_image_emits_lifecycle_completed_metric() + fibers.run(function (root_scope) + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-success", + component = "mcu", + artifact_ref = "artifact-mcu", + metadata = { expected_image_id = "image-new" }, + }, { timeout = 0.5 })) + + retain_component_fact(caller, "mcu", { + job_id = "j-success", + image_id = "image-old", + boot_id = "boot-1", + }) + wait_lifecycle_metric(caller, "started") + + retain_component_fact(caller, "mcu", { + job_id = "j-success", + image_id = "image-new", + boot_id = "boot-2", + }) + + local metric = wait_lifecycle_metric(caller, "completed") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j-success.completed") + assert_eq(metric.state, "ready") + + child:cancel("test complete") + end) +end + +function tests.test_component_device_fact_error_emits_lifecycle_failed_metric() + fibers.run(function (root_scope) + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-fail", + component = "mcu", + artifact_ref = "artifact-mcu", + }, { timeout = 0.5 })) + + retain_component_fact(caller, "mcu", { + job_id = "j-fail", + updater_state = "failed", + last_error = "wrong_image_after_reboot", + image_id = "image-old", + boot_id = "boot-1", + }) + + local metric = wait_lifecycle_metric(caller, "failed") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j-fail.failed") + assert_eq(metric.state, "failed") + assert_eq(metric.error, "wrong_image_after_reboot") + + child:cancel("test complete") + end) +end + +function tests.test_component_device_fact_cancelled_emits_lifecycle_cancelled_metric() + fibers.run(function (root_scope) + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-cancel", + component = "mcu", + artifact_ref = "artifact-mcu", + }, { timeout = 0.5 })) + + retain_component_fact(caller, "mcu", { + job_id = "j-cancel", + updater_state = "cancelled", + image_id = "image-old", + boot_id = "boot-1", + }) + + local metric = wait_lifecycle_metric(caller, "cancelled") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j-cancel.cancelled") + assert_eq(metric.state, "cancelled") + + child:cancel("test complete") + end) +end + +function tests.test_other_component_device_fact_emits_component_lifecycle_metric() + fibers.run(function (root_scope) + local child, caller = start_service(root_scope, { + config = component_config("cm5"), + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-cm5", + component = "cm5", + artifact_ref = "artifact-cm5", + }, { timeout = 0.5 })) + + retain_component_fact(caller, "cm5", { + job_id = "j-cm5", + image_id = "cm5-old", + boot_id = "boot-1", + }) + + local metric = wait_lifecycle_metric(caller, "started") + assert_eq(table.concat(metric.namespace, "."), "cm5.lifecycle.j-cm5.started") + assert_eq(metric.component, "cm5") + + child:cancel("test complete") + end) +end + + return tests From 774c280431d269f116e73c9be9e959ac050bfe56 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 23 Jun 2026 11:27:42 +0000 Subject: [PATCH 3/9] add component update lifecycle protocol to pipelines --- src/configs/bigbox-v1-cm-2.json | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/configs/bigbox-v1-cm-2.json b/src/configs/bigbox-v1-cm-2.json index c95dfe87..8e9c3b7a 100644 --- a/src/configs/bigbox-v1-cm-2.json +++ b/src/configs/bigbox-v1-cm-2.json @@ -1582,6 +1582,9 @@ } }, "pipelines": { + "component_update_lifecycle": { + "protocol": "http" + }, "wann_type": { "template": "any_change" }, From 796d4a6367489598c3ae43023e2352c2cbd0b3f5 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Mon, 13 Jul 2026 13:03:48 +0000 Subject: [PATCH 4/9] Enhance component lifecycle metrics and HTTP publishing - Added a "process" field to the component_update_lifecycle protocol in bigbox-v1-cm-2.json. - Updated main.lua to include profiling options for services. - Introduced new functions in metrics.lua for tracing component lifecycle metrics and logging HTTP payloads. - Enhanced HTTP publishing logic to handle metric chunks and added detailed logging for each chunk. - Improved component_watch.lua to trace component fact changes and report metrics. - Updated lifecycle_metrics.lua to sanitize job IDs in namespaces and log publishing actions. - Enhanced service.lua to trace component lifecycle metrics and handle job states more robustly. - Added unit tests to validate lifecycle metrics and ensure proper namespace formatting. --- src/configs/bigbox-v1-cm-2.json | 3 +- src/devicecode/main.lua | 2 + src/services/metrics.lua | 214 +++++++++++++++++-- src/services/metrics/http.lua | 22 +- src/services/update/component_watch.lua | 47 +++- src/services/update/lifecycle_metrics.lua | 38 +++- src/services/update/service.lua | 166 +++++++++++++- tests/integration/devhost/metrics_spec.lua | 4 +- tests/unit/update/test_lifecycle_metrics.lua | 22 +- tests/unit/update/test_service_phase2.lua | 151 ++++++++++++- 10 files changed, 621 insertions(+), 48 deletions(-) diff --git a/src/configs/bigbox-v1-cm-2.json b/src/configs/bigbox-v1-cm-2.json index 8e9c3b7a..f1c9e6b1 100644 --- a/src/configs/bigbox-v1-cm-2.json +++ b/src/configs/bigbox-v1-cm-2.json @@ -1583,7 +1583,8 @@ }, "pipelines": { "component_update_lifecycle": { - "protocol": "http" + "protocol": "http", + "process": [] }, "wann_type": { "template": "any_change" diff --git a/src/devicecode/main.lua b/src/devicecode/main.lua index b85b9e61..1eb1f7ea 100644 --- a/src/devicecode/main.lua +++ b/src/devicecode/main.lua @@ -80,6 +80,8 @@ local function spawn_service(child, bus, name, mod, env, extra_opts) services = extra_opts and extra_opts.services or nil, run_http = extra_opts and extra_opts.run_http or nil, verify_login = extra_opts and extra_opts.verify_login or nil, + profile = 'trace', + min_level = 'trace' }) error(('service returned unexpectedly: %s'):format(tostring(name)), 0) diff --git a/src/services/metrics.lua b/src/services/metrics.lua index fef32663..d26bab1e 100644 --- a/src/services/metrics.lua +++ b/src/services/metrics.lua @@ -44,6 +44,7 @@ local types = require 'services.metrics.types' local unpack = unpack or rawget(table, 'unpack') local NAME = 'metrics' +local HTTP_MAX_RECORDS = 25 ------------------------------------------------------------------------------- -- Topic helpers @@ -71,6 +72,13 @@ local function now() return runtime.now() end ---@return number local function now_real() return time.realtime() end +local function topic_string(topic) + if type(topic) ~= 'table' then return nil end + local parts = {} + for i = 1, #topic do parts[i] = tostring(topic[i]) end + return table.concat(parts, '/') +end + ------------------------------------------------------------------------------- -- Metric helpers ------------------------------------------------------------------------------- @@ -127,6 +135,22 @@ local State = { fs_cap = nil, } +local function trace_component_lifecycle_metric_received(metric_name, msg, payload, extra) + if metric_name ~= 'component_update_lifecycle' then return end + if not (State.svc and type(State.svc.obs_log) == 'function') then return end + local out = { + what = 'component_lifecycle_metric_received', + metric = metric_name, + topic = topic_string(msg and msg.topic or nil), + namespace = topic_string(type(payload) == 'table' and payload.namespace or nil), + value = type(payload) == 'table' and payload.value or nil, + component = type(payload) == 'table' and payload.component or nil, + job_id = type(payload) == 'table' and payload.job_id or nil, + } + for k, v in pairs(extra or {}) do out[k] = v end + State.svc:obs_log('trace', out) +end + ------------------------------------------------------------------------------- -- Config warnings (pure: no service state) ------------------------------------------------------------------------------- @@ -241,17 +265,113 @@ local function log_publish(data) end end ----@param data table -local function http_publish(data) - local senml_list, encode_err = senml.encode_r('', data) - if encode_err then - State.svc:obs_log('error', { what = 'senml_encode_failed', err = tostring(encode_err) }) - return +local function log_component_lifecycle_http_payload(data, senml_list, body) + if not (State.svc and type(State.svc.obs_log) == 'function') then return end + local endpoints = {} + for endpoint_str, metric in pairs(data or {}) do + if type(endpoint_str) == 'string' and endpoint_str:find('%.lifecycle%.', 1, false) then + endpoints[#endpoints + 1] = string.format('%s=%s', endpoint_str, tostring(metric and metric.value)) + end + end + if #endpoints == 0 then return end + + local encoded = {} + for _, rec in ipairs(senml_list or {}) do + if type(rec) == 'table' and type(rec.n) == 'string' and rec.n:find('%.lifecycle%.', 1, false) then + local field = rec.v ~= nil and 'v' or rec.vs ~= nil and 'vs' or rec.vb ~= nil and 'vb' or 'unknown' + encoded[#encoded + 1] = string.format('%s:%s=%s@%s', rec.n, field, tostring(rec[field]), tostring(rec.t)) + end + end + + State.svc:obs_log('trace', { + what = 'component_lifecycle_metric_http_payload', + endpoints = table.concat(endpoints, ','), + encoded = table.concat(encoded, ','), + body_bytes = type(body) == 'string' and #body or nil, + }) +end + +local function component_lifecycle_http_debug(data, senml_list, channel_id, body) + local endpoints = {} + for endpoint_str, metric in pairs(data or {}) do + if type(endpoint_str) == 'string' and endpoint_str:find('%.lifecycle%.', 1, false) then + endpoints[#endpoints + 1] = string.format('%s=%s', endpoint_str, tostring(metric and metric.value)) + end + end + if #endpoints == 0 then return nil end + + local encoded = {} + for _, rec in ipairs(senml_list or {}) do + if type(rec) == 'table' and type(rec.n) == 'string' and rec.n:find('%.lifecycle%.', 1, false) then + local field = rec.v ~= nil and 'v' or rec.vs ~= nil and 'vs' or rec.vb ~= nil and 'vb' or 'unknown' + encoded[#encoded + 1] = string.format('%s:%s=%s@%s', rec.n, field, tostring(rec[field]), tostring(rec.t)) + end + end + + return { + lifecycle_endpoints = table.concat(endpoints, ','), + lifecycle_encoded = table.concat(encoded, ','), + channel_id = channel_id, + body_bytes = type(body) == 'string' and #body or nil, + } +end + +local function sorted_metric_keys(data) + local keys = {} + for endpoint_str in pairs(data or {}) do keys[#keys + 1] = endpoint_str end + table.sort(keys) + return keys +end + +local function metric_chunks(data, max_records) + local keys = sorted_metric_keys(data) + local chunks = {} + if #keys == 0 then return chunks end + + local limit = tonumber(max_records) or #keys + if limit < 1 then limit = #keys end + + for i = 1, #keys, limit do + local chunk = {} + for j = i, math.min(i + limit - 1, #keys) do + local key = keys[j] + chunk[key] = data[key] + end + chunks[#chunks + 1] = chunk end - if #senml_list == 0 then return end + return chunks +end - local body = json.encode(senml_list) +local function metric_time_window(data) + local min_t, max_t + for _, metric in pairs(data or {}) do + local t = type(metric) == 'table' and metric.time or nil + if type(t) == 'number' then + if min_t == nil or t < min_t then min_t = t end + if max_t == nil or t > max_t then max_t = t end + end + end + return min_t, max_t +end + +local function log_http_chunk(chunk_index, chunk_count, record_count, body, debug, data) + if not (State.svc and type(State.svc.obs_log) == 'function') then return end + local min_t, max_t = metric_time_window(data) + local out = { + what = 'http_publish_chunk', + chunk_index = chunk_index, + chunk_count = chunk_count, + records = record_count, + body_bytes = type(body) == 'string' and #body or nil, + time_min_ms = min_t, + time_max_ms = max_t, + } + for k, v in pairs(debug or {}) do out[k] = v end + State.svc:obs_log('trace', out) +end +---@param data table +local function http_publish(data) local valid, config_err = conf.validate_http_config(State.cloud_config) if not valid then State.svc:obs_log('error', { what = 'http_publish_skipped', err = tostring(config_err) }) @@ -274,12 +394,40 @@ local function http_publish(data) State.cloud_config.url, channel_id) local auth = 'Thing ' .. State.cloud_config.thing_key - -- Non-blocking enqueue: drop and log if the channel is at capacity. - local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body }) - :or_else(function() return true end)) - - if full then - State.svc:obs_log('error', { what = 'http_queue_full', err = 'dropping publish payload' }) + local chunks = metric_chunks(data, HTTP_MAX_RECORDS) + for chunk_index, chunk in ipairs(chunks) do + local senml_list, encode_err = senml.encode_r('', chunk) + if encode_err then + State.svc:obs_log('error', { + what = 'senml_encode_failed', + err = tostring(encode_err), + chunk_index = chunk_index, + chunk_count = #chunks, + }) + elseif #senml_list > 0 then + local body = json.encode(senml_list) + log_component_lifecycle_http_payload(chunk, senml_list, body) + local debug = component_lifecycle_http_debug(chunk, senml_list, channel_id, body) or {} + debug.channel_id = channel_id + debug.body_bytes = type(body) == 'string' and #body or nil + debug.chunk_index = chunk_index + debug.chunk_count = #chunks + debug.records = #senml_list + log_http_chunk(chunk_index, #chunks, #senml_list, body, debug, chunk) + + -- Non-blocking enqueue: drop and log if the channel is at capacity. + local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body, debug = debug }) + :or_else(function() return true end)) + + if full then + State.svc:obs_log('error', { + what = 'http_queue_full', + err = 'dropping publish payload', + chunk_index = chunk_index, + chunk_count = #chunks, + }) + end + end end end @@ -305,6 +453,29 @@ end local publish_fns = { bus = bus_publish, log = log_publish, http = http_publish } +local function metric_sample_summary(metric) + if type(metric) ~= 'table' then return tostring(metric) end + return tostring(metric.value) +end + +local function log_publish_batch(protocol, data) + if not (State.svc and type(State.svc.obs_log) == 'function') then return end + local items = {} + for endpoint_str, metric in pairs(data or {}) do + items[#items + 1] = tostring(endpoint_str) .. '=' .. metric_sample_summary(metric) + end + table.sort(items) + local min_t, max_t = metric_time_window(data) + State.svc:obs_log('trace', { + what = 'metrics_publish_batch', + protocol = protocol, + count = #items, + time_min_ms = min_t, + time_max_ms = max_t, + endpoints = table.concat(items, ','), + }) +end + ---@param values table> local function publish_all(values) for protocol, pv in pairs(values) do @@ -320,6 +491,7 @@ local function publish_all(values) end pv = set_timestamps_realtime_millis(State.base_time, pv) + log_publish_batch(protocol, pv) local fn = publish_fns[protocol] if fn == nil then @@ -343,9 +515,10 @@ local function handle_metric(msg) if not metric_name then return end local pipe_cfg = State.pipelines_map[metric_name] + local payload = msg.payload + trace_component_lifecycle_metric_received(metric_name, msg, payload, { has_pipeline = pipe_cfg ~= nil }) if not pipe_cfg then return end -- no matching pipeline, drop silently - local payload = msg.payload if type(payload) ~= 'table' then return end local value = payload.value @@ -354,11 +527,22 @@ local function handle_metric(msg) -- Optional namespace overrides the topic used as the SenML name and state key. local topic = payload.namespace or msg.topic if not validate_topic(topic) then + trace_component_lifecycle_metric_received(metric_name, msg, payload, { + what = 'component_lifecycle_metric_rejected', + has_pipeline = true, + valid_namespace = false, + }) State.svc:obs_log('warn', { what = 'metric_invalid_topic', metric = metric_name }) return end local endpoint_str = table.concat(topic, '.') + trace_component_lifecycle_metric_received(metric_name, msg, payload, { + what = 'component_lifecycle_metric_accepted', + has_pipeline = true, + valid_namespace = true, + endpoint = endpoint_str, + }) -- Get-or-create per-endpoint processing state. if not State.metric_states[endpoint_str] then diff --git a/src/services/metrics/http.lua b/src/services/metrics/http.lua index 36d032f1..a3b515e3 100644 --- a/src/services/metrics/http.lua +++ b/src/services/metrics/http.lua @@ -27,6 +27,15 @@ local blob_source = require 'devicecode.blob_source' local QUEUE_SIZE = 10 local HTTP_TIMEOUT = 60 +local function with_debug(payload, debug) + local out = {} + for k, v in pairs(payload or {}) do out[k] = v end + if type(debug) == 'table' then + for k, v in pairs(debug) do out[k] = v end + end + return out +end + --- Send a single HTTP payload to the cloud via the HTTP capability service, --- retrying with exponential backoff on failure. Returns only when the send --- succeeds or the enclosing scope is cancelled. @@ -38,12 +47,13 @@ local function send_http(http_ref, data, log_fn) local uri = data.uri local body = data.body local auth = data.auth + local debug = data.debug local sleep_duration = 1 local reply while not reply do - log_fn('trace', { what = 'http_publish_attempt', status = 'started' }) + log_fn('trace', with_debug({ what = 'http_publish_attempt', status = 'started' }, debug)) local which, result, err = fibers.perform(op.named_choice({ response = http_ref:exchange_op({ method = 'POST', @@ -56,11 +66,11 @@ local function send_http(http_ref, data, log_fn) }), timeout = sleep.sleep_op(HTTP_TIMEOUT), })) - log_fn('trace', { what = 'http_publish_attempt', status = which }) + log_fn('trace', with_debug({ what = 'http_publish_attempt', status = which }, debug)) if which == 'timeout' or not result then local err_msg = which == 'timeout' and 'timeout' or tostring(err) - log_fn('debug', { what = 'http_retry', retry_in_s = sleep_duration, err = err_msg }) + log_fn('debug', with_debug({ what = 'http_retry', retry_in_s = sleep_duration, err = err_msg }, debug)) sleep.sleep(sleep_duration) sleep_duration = math.min(sleep_duration * 2, 60) else @@ -74,13 +84,13 @@ local function send_http(http_ref, data, log_fn) for k, v in pairs(reply.result and reply.result.headers or {}) do table.insert(parts, string.format('\t%s: %s', k, v)) end - log_fn('warn', { + log_fn('warn', with_debug({ what = 'http_publish_failed', status = tostring(status), headers = table.concat(parts, '\n'), - }) + }, debug)) else - log_fn('trace', { what = 'http_publish_ok', status = status }) + log_fn('trace', with_debug({ what = 'http_publish_ok', status = status }, debug)) end end diff --git a/src/services/update/component_watch.lua b/src/services/update/component_watch.lua index adca3714..90478650 100644 --- a/src/services/update/component_watch.lua +++ b/src/services/update/component_watch.lua @@ -12,6 +12,26 @@ local queue = require 'devicecode.support.queue' local M = {} +local function trace(params, what, payload) + if type(params.trace) ~= 'function' then return end + payload = payload or {} + payload.what = what + pcall(params.trace, what, payload) +end + +local function updater_from_payload(payload) + if type(payload) ~= 'table' then return nil end + return type(payload.updater) == 'table' and payload.updater + or type(payload.update) == 'table' and payload.update + or nil +end + +local function updater_job_id(upd) + local job_id = type(upd) == 'table' and upd.job_id or nil + if type(job_id) ~= 'string' or job_id == '' then return nil end + return job_id +end + local function component_list(config, explicit) local seen, out = {}, {} local function add(c) @@ -47,13 +67,29 @@ local function open_watches(scope, conn, components, queue_len) end local function report_component_fact(params, component, ev) - if not params.events_tx then return true, nil end - return queue.try_admit_required(params.events_tx, { + local upd = updater_from_payload(ev and ev.payload) + if not params.events_tx then + trace(params, 'component_fact_changed_not_reported', { + component = component, + reason = 'events_tx_unavailable', + job_id = updater_job_id(upd), + updater_state = upd and upd.state or nil, + }) + return true, nil + end + local ok, err = queue.try_admit_required(params.events_tx, { kind = 'component_fact_changed', component = component, payload = ev.payload, origin = ev.origin, }, 'update_component_fact_changed_admission_failed') + trace(params, ok == true and 'component_fact_changed_reported' or 'component_fact_changed_report_failed', { + component = component, + job_id = updater_job_id(upd), + updater_state = upd and upd.state or nil, + err = err, + }) + return ok, err end local function watch_loop(scope, params) @@ -76,6 +112,12 @@ local function watch_loop(scope, params) if ev.op == 'retain' or ev.event == 'retain' or ev.type == 'retain' or ev.kind == 'retain' then local ok_update, update_err = observer:update_component(component, ev.payload, ev.origin) if ok_update == nil then error(update_err or 'component_observer_update_failed', 0) end + local upd = updater_from_payload(ev.payload) + trace(params, 'component_fact_retained', { + component = component, + job_id = updater_job_id(upd), + updater_state = upd and upd.state or nil, + }) report_component_fact(params, component, ev) elseif ev.op == 'unretain' or ev.event == 'unretain' or ev.type == 'unretain' or ev.kind == 'unretain' then observer:remove_component(component, 'component_unretained') @@ -102,6 +144,7 @@ function M.start(scope, params) components = components, queue_len = params.queue_len, events_tx = params.events_tx, + trace = params.trace, }) end, report = params.report, diff --git a/src/services/update/lifecycle_metrics.lua b/src/services/update/lifecycle_metrics.lua index c3fd7f56..a8ea2555 100644 --- a/src/services/update/lifecycle_metrics.lua +++ b/src/services/update/lifecycle_metrics.lua @@ -8,24 +8,26 @@ local M = {} M.METRIC_NAME = 'component_update_lifecycle' +local METRIC_TOPIC = { 'obs', 'v1', 'update', 'metric', M.METRIC_NAME } + -- Metric payload shape consumed by services.metrics: -- { --- value = 'started' | 'completed' | 'failed' | 'cancelled', +-- value = 'started' | 'staged' | 'completed' | 'failed' | 'cancelled', -- namespace = { '', 'lifecycle', '', '' }, -- job_id = '', component = '', ... -- } -- The namespace is later used as a metrics topic suffix, so each segment must --- avoid separators and whitespace. +-- avoid separators, whitespace, and dashes. ---Return a string safe to use as one metric namespace segment. ---This is a topic/namespace normaliser, not a security boundary: it preserves ----letters, digits, '_' and '-', replaces every other run with '_', trims edge +---letters, digits, and '_', replaces every other run with '_', trims edge ---underscores, and falls back to 'unknown' if nothing usable remains. ---@param value any Original component or job identifier. ---@return string token Namespace-safe token. local function safe_token(value) local s = tostring(value or '') - s = s:gsub('[^%w%-_]', '_') + s = s:gsub('[^%w_]', '_') s = s:gsub('_+', '_') s = s:gsub('^_+', ''):gsub('_+$', '') if s == '' then return 'unknown' end @@ -40,6 +42,28 @@ local function component_of(record) return component end +local function topic_string(topic) + if type(topic) ~= 'table' then return nil end + local parts = {} + for i = 1, #topic do parts[i] = tostring(topic[i]) end + return table.concat(parts, '/') +end + +local function log_bus_publish(svc, what, p, extra) + if not (svc and type(svc.obs_log) == 'function') then return end + local out = { + what = what, + metric = M.METRIC_NAME, + topic = topic_string(METRIC_TOPIC), + namespace = topic_string(type(p) == 'table' and p.namespace or nil), + value = type(p) == 'table' and p.value or nil, + component = type(p) == 'table' and p.component or nil, + job_id = type(p) == 'table' and p.job_id or nil, + } + for k, v in pairs(extra or {}) do out[k] = v end + svc:obs_log('trace', out) +end + ---Build the metric payload expected by services.metrics. ---@param record table Lifecycle record with job_id/component/state/error fields. ---@param phase string Lifecycle phase to publish. @@ -53,7 +77,7 @@ local function payload(record, phase, extra) local out = { value = phase, - namespace = { safe_token(component), 'lifecycle', safe_token(record.job_id), phase }, + namespace = { safe_token(component), 'lifecycle', safe_token(record.job_id), safe_token(phase) }, job_id = record.job_id, component = component, state = record.state, @@ -72,12 +96,14 @@ end ---@return string|nil err Error code when no sink is available. local function publish(conn, svc, p) if svc and type(svc.obs_metric) == 'function' then + log_bus_publish(svc, 'component_lifecycle_metric_bus_publish', p, { sink = 'service_base' }) svc:obs_metric(M.METRIC_NAME, p) return true, nil end if conn and type(conn.retain) == 'function' then - conn:retain({ 'obs', 'v1', 'update', 'metric', M.METRIC_NAME }, p) + log_bus_publish(svc, 'component_lifecycle_metric_bus_publish', p, { sink = 'conn_retain' }) + conn:retain(METRIC_TOPIC, p) return true, nil end diff --git a/src/services/update/service.lua b/src/services/update/service.lua index 0e7d793c..f4cea1c3 100644 --- a/src/services/update/service.lua +++ b/src/services/update/service.lua @@ -148,6 +148,24 @@ local function software_from_fact(fact) return type(fact) == 'table' and type(fact.software) == 'table' and fact.software or nil end +local function meaningful_fact_value(v) + if v == nil then return nil end + if type(v) == 'userdata' then return nil end + if type(v) == 'string' and v == '' then return nil end + return v +end + +local function updater_job_id(upd) + local job_id = type(upd) == 'table' and upd.job_id or nil + if type(job_id) ~= 'string' or job_id == '' then return nil end + return job_id +end + +local function updater_error(upd) + if type(upd) ~= 'table' then return nil end + return meaningful_fact_value(upd.last_error) or meaningful_fact_value(upd.error) +end + local function expected_image_id(job) if type(job) ~= 'table' then return nil end local meta = type(job.metadata) == 'table' and job.metadata or {} @@ -171,10 +189,20 @@ end local function lifecycle_record(component, fact) local upd = updater_from_fact(fact) or {} return { - job_id = upd.job_id, + job_id = updater_job_id(upd), component = component or (type(fact) == 'table' and fact.component), state = upd.state, - error = upd.last_error or upd.error, + error = updater_error(upd), + } +end + +local function lifecycle_record_from_job(job) + if type(job) ~= 'table' then return nil end + return { + job_id = job.job_id, + component = job.component, + state = job.state, + error = job.error, } end @@ -196,48 +224,159 @@ local function mark_lifecycle_metric_sent(self, record, phase, sent) self._component_lifecycle_sent[lifecycle_metric_key(record, phase)] = sent == true or nil end +local function trace_component_lifecycle(self, what, payload) + if not (self and self._svc and type(self._svc.obs_log) == 'function') then return end + local out = { what = what } + for k, v in pairs(payload or {}) do out[k] = v end + self._svc:obs_log('trace', out) +end + local function emit_component_lifecycle_metric(self, record, phase, extra) + local key = lifecycle_metric_key(record, phase) if lifecycle_metric_sent(self, record, phase) then + trace_component_lifecycle(self, 'component_lifecycle_metric_skipped_duplicate', { + key = key, + component = record and record.component or nil, + job_id = record and record.job_id or nil, + phase = phase, + }) return true, nil end mark_lifecycle_metric_sent(self, record, phase, true) + trace_component_lifecycle(self, 'component_lifecycle_metric_emit_begin', { + key = key, + component = record and record.component or nil, + job_id = record and record.job_id or nil, + phase = phase, + state = record and record.state or nil, + error = record and record.error or nil, + }) local ok, err = lifecycle_metrics.emit(self._conn, self._svc, record, phase, extra) if ok ~= true then mark_lifecycle_metric_sent(self, record, phase, false) if self._svc and type(self._svc.obs_log) == 'function' then - self._svc:obs_log('warn', { what = 'component_lifecycle_metric_emit_failed', err = tostring(err) }) + self._svc:obs_log('warn', { + what = 'component_lifecycle_metric_emit_failed', + key = key, + component = record and record.component or nil, + job_id = record and record.job_id or nil, + phase = phase, + err = tostring(err), + }) end return true, nil end + trace_component_lifecycle(self, 'component_lifecycle_metric_emit_ok', { + key = key, + component = record and record.component or nil, + job_id = record and record.job_id or nil, + phase = phase, + }) + return true, nil +end + +local function job_lifecycle_phase(job) + if type(job) ~= 'table' then return nil, 'job_missing' end + if type(job.job_id) ~= 'string' or job.job_id == '' then return nil, 'job_id_missing' end + if type(job.component) ~= 'string' or job.component == '' then return nil, 'component_missing' end + + local state = job.state + if state == 'created' then return nil, 'job_not_started' end + if state == 'succeeded' then return 'completed', 'job_succeeded' end + if state == 'failed' or state == 'timed_out' then return 'failed', 'job_failed' end + if state == 'cancelled' or state == 'discarded' or state == 'superseded' then + return 'cancelled', 'job_cancelled' + end + if state == 'awaiting_commit' then return 'staged', 'job_staged' end + if state == 'staging' or state == 'committing' or state == 'awaiting_return' then + return 'started', 'job_active' + end + return nil, 'job_state_unmapped' +end + +local TERMINAL_LIFECYCLE_PHASE = { + completed = true, + failed = true, + cancelled = true, +} + +local function emit_job_lifecycle_metric(self, job, reason) + local phase, phase_reason = job_lifecycle_phase(job) + local record = lifecycle_record_from_job(job) + trace_component_lifecycle(self, 'job_lifecycle_evaluated', { + component = record and record.component or nil, + job_id = record and record.job_id or nil, + state = record and record.state or nil, + error = record and record.error or nil, + phase = phase, + reason = phase_reason, + source_reason = reason, + }) + if not phase then return true, nil end + + if TERMINAL_LIFECYCLE_PHASE[phase] and not lifecycle_metric_sent(self, record, 'started') then + local ok_started, serr = emit_component_lifecycle_metric(self, record, 'started', { + source = 'update_job_state', + reason = 'terminal_without_started', + source_reason = reason, + }) + if ok_started ~= true then return ok_started, serr end + end + + return emit_component_lifecycle_metric(self, record, phase, { + source = 'update_job_state', + reason = phase_reason, + source_reason = reason, + }) +end + +local function emit_job_lifecycle_metrics_from_snapshot(self, snapshot, reason) + local jobs = snapshot and snapshot.jobs and snapshot.jobs.by_id or nil + if type(jobs) ~= 'table' then return true, nil end + for _, job in pairs(jobs) do + local ok, err = emit_job_lifecycle_metric(self, job, reason) + if ok ~= true then return ok, err end + end return true, nil end local function component_fact_phase(component, fact, self) local upd = updater_from_fact(fact) or {} - if type(upd.job_id) ~= 'string' or upd.job_id == '' then return nil end - if upd.state == 'cancelled' or upd.state == 'aborted' then return 'cancelled' end - if upd.state == 'failed' or upd.state == 'rollback_detected' or upd.last_error ~= nil or upd.error ~= nil then - return 'failed' + local job_id = updater_job_id(upd) + if not job_id then return nil, 'job_id_missing' end + if upd.state == 'cancelled' or upd.state == 'aborted' then return 'cancelled', 'updater_cancelled' end + if upd.state == 'failed' or upd.state == 'rollback_detected' or updater_error(upd) ~= nil then + return 'failed', 'updater_failed' end local record = lifecycle_record(component, fact) if lifecycle_metric_sent(self, record, 'started') then - local job = self._jobs and self._jobs:get(upd.job_id) or nil + local job = self._jobs and self._jobs:get(job_id) or nil local expected = expected_image_id(job) local sw = software_from_fact(fact) if type(sw) == 'table' and expected and sw.image_id == expected then - return 'completed' + return 'completed', 'expected_image_seen' end end - return 'started' + return 'started', 'job_observed' end local function emit_component_fact_lifecycle_metric(self, ev) local component = ev and ev.component or nil local fact = ev and ev.payload or nil - local phase = component_fact_phase(component, fact, self) + local upd = updater_from_fact(fact) or {} + local sw = software_from_fact(fact) or {} + local phase, phase_reason = component_fact_phase(component, fact, self) + trace_component_lifecycle(self, 'component_fact_lifecycle_evaluated', { + component = component, + job_id = updater_job_id(upd), + updater_state = upd.state, + software_image_id = sw.image_id, + phase = phase, + reason = phase_reason, + }) if not phase then return true, nil end local record = lifecycle_record(component, fact) return emit_component_lifecycle_metric(self, record, phase, { @@ -667,6 +806,7 @@ local function handle_job_runtime_changed(self, ev) if self._jobs and self._jobs:ready() and not self._job_runtime_ready then self._job_runtime_ready = true update_service_jobs_projection(self) + emit_job_lifecycle_metrics_from_snapshot(self, ev.snapshot, 'job_runtime_ready') local ok, err = reconcile_runtime_components(self, 'job_runtime_ready') if ok ~= true then update_model_state(self, 'failed', err or 'runtime_dependents_start_failed') @@ -674,6 +814,7 @@ local function handle_job_runtime_changed(self, ev) end else update_service_jobs_projection(self) + emit_job_lifecycle_metrics_from_snapshot(self, ev.snapshot, 'job_runtime_changed') end if self._current_generation then apply_generation_snapshot(self, self._current_generation.last_snapshot or { @@ -1285,6 +1426,9 @@ local function ensure_component_watch(self, reason) queue_len = params.component_watch_queue_len, report = service_events.reporter(cw_port, 'update_component_watch_completion_report_failed'), events_tx = self._done_tx, + trace = function (_, payload) + trace_component_lifecycle(self, payload and payload.what or 'component_watch_trace', payload) + end, }) if not cwh then return nil, cwerr or 'update_component_watch_start_failed' end self._component_watch = cwh diff --git a/tests/integration/devhost/metrics_spec.lua b/tests/integration/devhost/metrics_spec.lua index bc13ce01..1d882a96 100644 --- a/tests/integration/devhost/metrics_spec.lua +++ b/tests/integration/devhost/metrics_spec.lua @@ -547,11 +547,11 @@ function T.component_update_lifecycle_metric_uses_event_namespace() test_conn:publish( { "obs", "v1", "update", "metric", "component_update_lifecycle" }, - { value = "started", namespace = { "mcu", "lifecycle", "job-1", "started" } }) + { value = "started", namespace = { "mcu", "lifecycle", "job_1", "started" } }) local msg = recv_metric(clock, result_sub, 0.5) assert(msg ~= nil, "expected lifecycle metric publish") - assert(table.concat(msg.topic, ".") == "obs.v1.metrics.output.mcu.lifecycle.job-1.started", + assert(table.concat(msg.topic, ".") == "obs.v1.metrics.output.mcu.lifecycle.job_1.started", "unexpected topic " .. table.concat(msg.topic, ".")) assert(msg.payload.value == "started", "expected value=started, got " .. tostring(msg.payload.value)) diff --git a/tests/unit/update/test_lifecycle_metrics.lua b/tests/unit/update/test_lifecycle_metrics.lua index 9f743ad1..df3d4ba8 100644 --- a/tests/unit/update/test_lifecycle_metrics.lua +++ b/tests/unit/update/test_lifecycle_metrics.lua @@ -15,7 +15,7 @@ local function assert_nil(v, msg) if v ~= nil then fail(msg or ('expected nil, g function tests.test_safe_token_replaces_namespace_unsafe_characters() assert_eq(metrics.safe_token('job/mcu.1 ready'), 'job_mcu_1_ready') - assert_eq(metrics.safe_token('job-1_ok'), 'job-1_ok') + assert_eq(metrics.safe_token('job-1_ok'), 'job_1_ok') assert_eq(metrics.safe_token('///'), 'unknown') end @@ -45,6 +45,26 @@ function tests.test_emit_publishes_component_lifecycle_payload() assert_eq(seen.payload.source, 'device_component_fact') end +function tests.test_emit_sanitizes_full_uuid_job_id_in_namespace() + local seen + local svc = { + obs_metric = function (_, name, payload) + seen = { name = name, payload = payload } + end, + } + + local job_id = '08be26e5-e0b6-4f11-9c64-d1053e983820' + local ok, err = metrics.emit(nil, svc, { + job_id = job_id, + component = 'mcu', + state = 'ready', + }, 'started') + + assert_eq(ok, true, err) + assert_eq(seen.payload.job_id, job_id) + assert_eq(table.concat(seen.payload.namespace, '.'), 'mcu.lifecycle.08be26e5_e0b6_4f11_9c64_d1053e983820.started') +end + function tests.test_emit_rejects_missing_component_or_phase() local ok_component, err_component = metrics.emit(nil, {}, { job_id = 'j1', diff --git a/tests/unit/update/test_service_phase2.lua b/tests/unit/update/test_service_phase2.lua index 5921ed60..dbb9fd0b 100644 --- a/tests/unit/update/test_service_phase2.lua +++ b/tests/unit/update/test_service_phase2.lua @@ -584,6 +584,35 @@ function tests.test_component_device_fact_job_id_emits_lifecycle_started_metric( end) end +function tests.test_component_device_fact_json_null_error_still_emits_lifecycle_started_metric() + fibers.run(function (root_scope) + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "job/mcu.null-error", + component = "mcu", + artifact_ref = "artifact-mcu", + }, { timeout = 0.5 })) + + retain_component_fact(caller, "mcu", { + job_id = "job/mcu.null-error", + updater_state = "ready", + last_error = newproxy(false), + image_id = "image-old", + boot_id = "boot-1", + }) + + local metric = wait_lifecycle_metric(caller, "started") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.job_mcu_null_error.started") + assert_eq(metric.job_id, "job/mcu.null-error") + assert_eq(metric.error, nil) + + child:cancel("test complete") + end) +end + function tests.test_component_device_fact_expected_image_emits_lifecycle_completed_metric() fibers.run(function (root_scope) local child, caller = start_service(root_scope, { @@ -611,7 +640,7 @@ function tests.test_component_device_fact_expected_image_emits_lifecycle_complet }) local metric = wait_lifecycle_metric(caller, "completed") - assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j-success.completed") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_success.completed") assert_eq(metric.state, "ready") child:cancel("test complete") @@ -639,7 +668,7 @@ function tests.test_component_device_fact_error_emits_lifecycle_failed_metric() }) local metric = wait_lifecycle_metric(caller, "failed") - assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j-fail.failed") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_fail.failed") assert_eq(metric.state, "failed") assert_eq(metric.error, "wrong_image_after_reboot") @@ -667,7 +696,7 @@ function tests.test_component_device_fact_cancelled_emits_lifecycle_cancelled_me }) local metric = wait_lifecycle_metric(caller, "cancelled") - assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j-cancel.cancelled") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_cancel.cancelled") assert_eq(metric.state, "cancelled") child:cancel("test complete") @@ -693,12 +722,126 @@ function tests.test_other_component_device_fact_emits_component_lifecycle_metric }) local metric = wait_lifecycle_metric(caller, "started") - assert_eq(table.concat(metric.namespace, "."), "cm5.lifecycle.j-cm5.started") + assert_eq(table.concat(metric.namespace, "."), "cm5.lifecycle.j_cm5.started") assert_eq(metric.component, "cm5") child:cancel("test complete") end) end +function tests.test_job_state_awaiting_commit_emits_lifecycle_staged_metric_without_component_fact_job_id() + fibers.run(function (root_scope) + local backend = {} + function backend:stage_op(job) + return op.always({ job_id = job.job_id, expected_image_id = "image-new" }, nil) + end + + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + backend = backend, + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-state-staged", + component = "mcu", + artifact_ref = "artifact-mcu", + expected_image_id = "image-new", + }, { timeout = 0.5 })) + assert(caller:call(topics.update_manager_rpc("start-job"), { + job_id = "j-state-staged", + }, { timeout = 0.5 })) + + local metric = wait_lifecycle_metric(caller, "staged") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_state_staged.staged") + assert_eq(metric.job_id, "j-state-staged") + assert_eq(metric.source, "update_job_state") + + child:cancel("test complete") + end) +end + +function tests.test_job_state_failure_emits_lifecycle_failed_metric_without_component_fact_job_id() + fibers.run(function (root_scope) + local backend = {} + function backend:stage_op(_job) + return op.always(nil, "no_session") + end + + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + backend = backend, + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-state-fail", + component = "mcu", + artifact_ref = "artifact-mcu", + expected_image_id = "image-new", + }, { timeout = 0.5 })) + assert(caller:call(topics.update_manager_rpc("start-job"), { + job_id = "j-state-fail", + }, { timeout = 0.5 })) + + local metric = wait_lifecycle_metric(caller, "failed") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_state_fail.failed") + assert_eq(metric.job_id, "j-state-fail") + assert_eq(metric.component, "mcu") + assert_eq(metric.source, "update_job_state") + assert_eq(metric.error, "no_session") + + child:cancel("test complete") + end) +end + +function tests.test_job_state_reconcile_success_emits_lifecycle_completed_metric() + fibers.run(function (root_scope) + local backend = {} + function backend:stage_op(job) + return op.always({ job_id = job.job_id, expected_image_id = "image-new" }, nil) + end + function backend:commit_capabilities() + return { policy = "idempotent_by_token" } + end + function backend:commit_op(job, ctx) + return op.always({ accepted = true, token = ctx.commit_token, job_id = job.job_id }, nil) + end + function backend:evaluate_reconcile() + return { done = true, tag = "reconciled_success", observed = { ok = true } } + end + + local child, caller = start_service(root_scope, { + config = component_config("mcu"), + backend = backend, + }) + + assert(caller:call(topics.update_manager_rpc("create-job"), { + job_id = "j-state-complete", + component = "mcu", + artifact_ref = "artifact-mcu", + expected_image_id = "image-new", + }, { timeout = 0.5 })) + assert(caller:call(topics.update_manager_rpc("start-job"), { + job_id = "j-state-complete", + }, { timeout = 0.5 })) + assert_true(probe.wait_until(function () + local status = caller:call(topics.update_manager_rpc("status"), {}, { timeout = 0.05 }) + return status and status.snapshot.jobs.by_id["j-state-complete"] + and status.snapshot.jobs.by_id["j-state-complete"].state == "awaiting_commit" + end, { timeout = 0.8, interval = 0.01 }), "expected awaiting_commit") + + assert(caller:call(topics.update_manager_rpc("commit-job"), { + job_id = "j-state-complete", + }, { timeout = 0.5 })) + + local metric = wait_lifecycle_metric(caller, "completed") + assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_state_complete.completed") + assert_eq(metric.job_id, "j-state-complete") + assert_eq(metric.component, "mcu") + assert_eq(metric.source, "update_job_state") + + child:cancel("test complete") + end) +end + return tests From 95b351e3d7f3273e726d443c12e47feefee21e74 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Mon, 13 Jul 2026 18:44:20 +0000 Subject: [PATCH 5/9] refactor: remove unused lifecycle metric logging functions and simplify HTTP payload handling --- src/services/metrics.lua | 148 +--------------------- src/services/metrics/http.lua | 22 +--- src/services/update/lifecycle_metrics.lua | 24 ---- src/services/update/service.lua | 77 +---------- 4 files changed, 8 insertions(+), 263 deletions(-) diff --git a/src/services/metrics.lua b/src/services/metrics.lua index d26bab1e..5330585b 100644 --- a/src/services/metrics.lua +++ b/src/services/metrics.lua @@ -72,13 +72,6 @@ local function now() return runtime.now() end ---@return number local function now_real() return time.realtime() end -local function topic_string(topic) - if type(topic) ~= 'table' then return nil end - local parts = {} - for i = 1, #topic do parts[i] = tostring(topic[i]) end - return table.concat(parts, '/') -end - ------------------------------------------------------------------------------- -- Metric helpers ------------------------------------------------------------------------------- @@ -135,22 +128,6 @@ local State = { fs_cap = nil, } -local function trace_component_lifecycle_metric_received(metric_name, msg, payload, extra) - if metric_name ~= 'component_update_lifecycle' then return end - if not (State.svc and type(State.svc.obs_log) == 'function') then return end - local out = { - what = 'component_lifecycle_metric_received', - metric = metric_name, - topic = topic_string(msg and msg.topic or nil), - namespace = topic_string(type(payload) == 'table' and payload.namespace or nil), - value = type(payload) == 'table' and payload.value or nil, - component = type(payload) == 'table' and payload.component or nil, - job_id = type(payload) == 'table' and payload.job_id or nil, - } - for k, v in pairs(extra or {}) do out[k] = v end - State.svc:obs_log('trace', out) -end - ------------------------------------------------------------------------------- -- Config warnings (pure: no service state) ------------------------------------------------------------------------------- @@ -265,57 +242,6 @@ local function log_publish(data) end end -local function log_component_lifecycle_http_payload(data, senml_list, body) - if not (State.svc and type(State.svc.obs_log) == 'function') then return end - local endpoints = {} - for endpoint_str, metric in pairs(data or {}) do - if type(endpoint_str) == 'string' and endpoint_str:find('%.lifecycle%.', 1, false) then - endpoints[#endpoints + 1] = string.format('%s=%s', endpoint_str, tostring(metric and metric.value)) - end - end - if #endpoints == 0 then return end - - local encoded = {} - for _, rec in ipairs(senml_list or {}) do - if type(rec) == 'table' and type(rec.n) == 'string' and rec.n:find('%.lifecycle%.', 1, false) then - local field = rec.v ~= nil and 'v' or rec.vs ~= nil and 'vs' or rec.vb ~= nil and 'vb' or 'unknown' - encoded[#encoded + 1] = string.format('%s:%s=%s@%s', rec.n, field, tostring(rec[field]), tostring(rec.t)) - end - end - - State.svc:obs_log('trace', { - what = 'component_lifecycle_metric_http_payload', - endpoints = table.concat(endpoints, ','), - encoded = table.concat(encoded, ','), - body_bytes = type(body) == 'string' and #body or nil, - }) -end - -local function component_lifecycle_http_debug(data, senml_list, channel_id, body) - local endpoints = {} - for endpoint_str, metric in pairs(data or {}) do - if type(endpoint_str) == 'string' and endpoint_str:find('%.lifecycle%.', 1, false) then - endpoints[#endpoints + 1] = string.format('%s=%s', endpoint_str, tostring(metric and metric.value)) - end - end - if #endpoints == 0 then return nil end - - local encoded = {} - for _, rec in ipairs(senml_list or {}) do - if type(rec) == 'table' and type(rec.n) == 'string' and rec.n:find('%.lifecycle%.', 1, false) then - local field = rec.v ~= nil and 'v' or rec.vs ~= nil and 'vs' or rec.vb ~= nil and 'vb' or 'unknown' - encoded[#encoded + 1] = string.format('%s:%s=%s@%s', rec.n, field, tostring(rec[field]), tostring(rec.t)) - end - end - - return { - lifecycle_endpoints = table.concat(endpoints, ','), - lifecycle_encoded = table.concat(encoded, ','), - channel_id = channel_id, - body_bytes = type(body) == 'string' and #body or nil, - } -end - local function sorted_metric_keys(data) local keys = {} for endpoint_str in pairs(data or {}) do keys[#keys + 1] = endpoint_str end @@ -342,34 +268,6 @@ local function metric_chunks(data, max_records) return chunks end -local function metric_time_window(data) - local min_t, max_t - for _, metric in pairs(data or {}) do - local t = type(metric) == 'table' and metric.time or nil - if type(t) == 'number' then - if min_t == nil or t < min_t then min_t = t end - if max_t == nil or t > max_t then max_t = t end - end - end - return min_t, max_t -end - -local function log_http_chunk(chunk_index, chunk_count, record_count, body, debug, data) - if not (State.svc and type(State.svc.obs_log) == 'function') then return end - local min_t, max_t = metric_time_window(data) - local out = { - what = 'http_publish_chunk', - chunk_index = chunk_index, - chunk_count = chunk_count, - records = record_count, - body_bytes = type(body) == 'string' and #body or nil, - time_min_ms = min_t, - time_max_ms = max_t, - } - for k, v in pairs(debug or {}) do out[k] = v end - State.svc:obs_log('trace', out) -end - ---@param data table local function http_publish(data) local valid, config_err = conf.validate_http_config(State.cloud_config) @@ -406,17 +304,9 @@ local function http_publish(data) }) elseif #senml_list > 0 then local body = json.encode(senml_list) - log_component_lifecycle_http_payload(chunk, senml_list, body) - local debug = component_lifecycle_http_debug(chunk, senml_list, channel_id, body) or {} - debug.channel_id = channel_id - debug.body_bytes = type(body) == 'string' and #body or nil - debug.chunk_index = chunk_index - debug.chunk_count = #chunks - debug.records = #senml_list - log_http_chunk(chunk_index, #chunks, #senml_list, body, debug, chunk) -- Non-blocking enqueue: drop and log if the channel is at capacity. - local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body, debug = debug }) + local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body }) :or_else(function() return true end)) if full then @@ -453,29 +343,6 @@ end local publish_fns = { bus = bus_publish, log = log_publish, http = http_publish } -local function metric_sample_summary(metric) - if type(metric) ~= 'table' then return tostring(metric) end - return tostring(metric.value) -end - -local function log_publish_batch(protocol, data) - if not (State.svc and type(State.svc.obs_log) == 'function') then return end - local items = {} - for endpoint_str, metric in pairs(data or {}) do - items[#items + 1] = tostring(endpoint_str) .. '=' .. metric_sample_summary(metric) - end - table.sort(items) - local min_t, max_t = metric_time_window(data) - State.svc:obs_log('trace', { - what = 'metrics_publish_batch', - protocol = protocol, - count = #items, - time_min_ms = min_t, - time_max_ms = max_t, - endpoints = table.concat(items, ','), - }) -end - ---@param values table> local function publish_all(values) for protocol, pv in pairs(values) do @@ -491,7 +358,6 @@ local function publish_all(values) end pv = set_timestamps_realtime_millis(State.base_time, pv) - log_publish_batch(protocol, pv) local fn = publish_fns[protocol] if fn == nil then @@ -516,7 +382,6 @@ local function handle_metric(msg) local pipe_cfg = State.pipelines_map[metric_name] local payload = msg.payload - trace_component_lifecycle_metric_received(metric_name, msg, payload, { has_pipeline = pipe_cfg ~= nil }) if not pipe_cfg then return end -- no matching pipeline, drop silently if type(payload) ~= 'table' then return end @@ -527,22 +392,11 @@ local function handle_metric(msg) -- Optional namespace overrides the topic used as the SenML name and state key. local topic = payload.namespace or msg.topic if not validate_topic(topic) then - trace_component_lifecycle_metric_received(metric_name, msg, payload, { - what = 'component_lifecycle_metric_rejected', - has_pipeline = true, - valid_namespace = false, - }) State.svc:obs_log('warn', { what = 'metric_invalid_topic', metric = metric_name }) return end local endpoint_str = table.concat(topic, '.') - trace_component_lifecycle_metric_received(metric_name, msg, payload, { - what = 'component_lifecycle_metric_accepted', - has_pipeline = true, - valid_namespace = true, - endpoint = endpoint_str, - }) -- Get-or-create per-endpoint processing state. if not State.metric_states[endpoint_str] then diff --git a/src/services/metrics/http.lua b/src/services/metrics/http.lua index a3b515e3..36d032f1 100644 --- a/src/services/metrics/http.lua +++ b/src/services/metrics/http.lua @@ -27,15 +27,6 @@ local blob_source = require 'devicecode.blob_source' local QUEUE_SIZE = 10 local HTTP_TIMEOUT = 60 -local function with_debug(payload, debug) - local out = {} - for k, v in pairs(payload or {}) do out[k] = v end - if type(debug) == 'table' then - for k, v in pairs(debug) do out[k] = v end - end - return out -end - --- Send a single HTTP payload to the cloud via the HTTP capability service, --- retrying with exponential backoff on failure. Returns only when the send --- succeeds or the enclosing scope is cancelled. @@ -47,13 +38,12 @@ local function send_http(http_ref, data, log_fn) local uri = data.uri local body = data.body local auth = data.auth - local debug = data.debug local sleep_duration = 1 local reply while not reply do - log_fn('trace', with_debug({ what = 'http_publish_attempt', status = 'started' }, debug)) + log_fn('trace', { what = 'http_publish_attempt', status = 'started' }) local which, result, err = fibers.perform(op.named_choice({ response = http_ref:exchange_op({ method = 'POST', @@ -66,11 +56,11 @@ local function send_http(http_ref, data, log_fn) }), timeout = sleep.sleep_op(HTTP_TIMEOUT), })) - log_fn('trace', with_debug({ what = 'http_publish_attempt', status = which }, debug)) + log_fn('trace', { what = 'http_publish_attempt', status = which }) if which == 'timeout' or not result then local err_msg = which == 'timeout' and 'timeout' or tostring(err) - log_fn('debug', with_debug({ what = 'http_retry', retry_in_s = sleep_duration, err = err_msg }, debug)) + log_fn('debug', { what = 'http_retry', retry_in_s = sleep_duration, err = err_msg }) sleep.sleep(sleep_duration) sleep_duration = math.min(sleep_duration * 2, 60) else @@ -84,13 +74,13 @@ local function send_http(http_ref, data, log_fn) for k, v in pairs(reply.result and reply.result.headers or {}) do table.insert(parts, string.format('\t%s: %s', k, v)) end - log_fn('warn', with_debug({ + log_fn('warn', { what = 'http_publish_failed', status = tostring(status), headers = table.concat(parts, '\n'), - }, debug)) + }) else - log_fn('trace', with_debug({ what = 'http_publish_ok', status = status }, debug)) + log_fn('trace', { what = 'http_publish_ok', status = status }) end end diff --git a/src/services/update/lifecycle_metrics.lua b/src/services/update/lifecycle_metrics.lua index a8ea2555..5e1ef295 100644 --- a/src/services/update/lifecycle_metrics.lua +++ b/src/services/update/lifecycle_metrics.lua @@ -42,28 +42,6 @@ local function component_of(record) return component end -local function topic_string(topic) - if type(topic) ~= 'table' then return nil end - local parts = {} - for i = 1, #topic do parts[i] = tostring(topic[i]) end - return table.concat(parts, '/') -end - -local function log_bus_publish(svc, what, p, extra) - if not (svc and type(svc.obs_log) == 'function') then return end - local out = { - what = what, - metric = M.METRIC_NAME, - topic = topic_string(METRIC_TOPIC), - namespace = topic_string(type(p) == 'table' and p.namespace or nil), - value = type(p) == 'table' and p.value or nil, - component = type(p) == 'table' and p.component or nil, - job_id = type(p) == 'table' and p.job_id or nil, - } - for k, v in pairs(extra or {}) do out[k] = v end - svc:obs_log('trace', out) -end - ---Build the metric payload expected by services.metrics. ---@param record table Lifecycle record with job_id/component/state/error fields. ---@param phase string Lifecycle phase to publish. @@ -96,13 +74,11 @@ end ---@return string|nil err Error code when no sink is available. local function publish(conn, svc, p) if svc and type(svc.obs_metric) == 'function' then - log_bus_publish(svc, 'component_lifecycle_metric_bus_publish', p, { sink = 'service_base' }) svc:obs_metric(M.METRIC_NAME, p) return true, nil end if conn and type(conn.retain) == 'function' then - log_bus_publish(svc, 'component_lifecycle_metric_bus_publish', p, { sink = 'conn_retain' }) conn:retain(METRIC_TOPIC, p) return true, nil end diff --git a/src/services/update/service.lua b/src/services/update/service.lua index f4cea1c3..c4d2a29e 100644 --- a/src/services/update/service.lua +++ b/src/services/update/service.lua @@ -224,55 +224,17 @@ local function mark_lifecycle_metric_sent(self, record, phase, sent) self._component_lifecycle_sent[lifecycle_metric_key(record, phase)] = sent == true or nil end -local function trace_component_lifecycle(self, what, payload) - if not (self and self._svc and type(self._svc.obs_log) == 'function') then return end - local out = { what = what } - for k, v in pairs(payload or {}) do out[k] = v end - self._svc:obs_log('trace', out) -end - local function emit_component_lifecycle_metric(self, record, phase, extra) - local key = lifecycle_metric_key(record, phase) if lifecycle_metric_sent(self, record, phase) then - trace_component_lifecycle(self, 'component_lifecycle_metric_skipped_duplicate', { - key = key, - component = record and record.component or nil, - job_id = record and record.job_id or nil, - phase = phase, - }) return true, nil end mark_lifecycle_metric_sent(self, record, phase, true) - trace_component_lifecycle(self, 'component_lifecycle_metric_emit_begin', { - key = key, - component = record and record.component or nil, - job_id = record and record.job_id or nil, - phase = phase, - state = record and record.state or nil, - error = record and record.error or nil, - }) local ok, err = lifecycle_metrics.emit(self._conn, self._svc, record, phase, extra) if ok ~= true then mark_lifecycle_metric_sent(self, record, phase, false) - if self._svc and type(self._svc.obs_log) == 'function' then - self._svc:obs_log('warn', { - what = 'component_lifecycle_metric_emit_failed', - key = key, - component = record and record.component or nil, - job_id = record and record.job_id or nil, - phase = phase, - err = tostring(err), - }) - end return true, nil end - trace_component_lifecycle(self, 'component_lifecycle_metric_emit_ok', { - key = key, - component = record and record.component or nil, - job_id = record and record.job_id or nil, - phase = phase, - }) return true, nil end @@ -295,35 +257,11 @@ local function job_lifecycle_phase(job) return nil, 'job_state_unmapped' end -local TERMINAL_LIFECYCLE_PHASE = { - completed = true, - failed = true, - cancelled = true, -} - local function emit_job_lifecycle_metric(self, job, reason) local phase, phase_reason = job_lifecycle_phase(job) local record = lifecycle_record_from_job(job) - trace_component_lifecycle(self, 'job_lifecycle_evaluated', { - component = record and record.component or nil, - job_id = record and record.job_id or nil, - state = record and record.state or nil, - error = record and record.error or nil, - phase = phase, - reason = phase_reason, - source_reason = reason, - }) if not phase then return true, nil end - if TERMINAL_LIFECYCLE_PHASE[phase] and not lifecycle_metric_sent(self, record, 'started') then - local ok_started, serr = emit_component_lifecycle_metric(self, record, 'started', { - source = 'update_job_state', - reason = 'terminal_without_started', - source_reason = reason, - }) - if ok_started ~= true then return ok_started, serr end - end - return emit_component_lifecycle_metric(self, record, phase, { source = 'update_job_state', reason = phase_reason, @@ -366,17 +304,7 @@ end local function emit_component_fact_lifecycle_metric(self, ev) local component = ev and ev.component or nil local fact = ev and ev.payload or nil - local upd = updater_from_fact(fact) or {} - local sw = software_from_fact(fact) or {} - local phase, phase_reason = component_fact_phase(component, fact, self) - trace_component_lifecycle(self, 'component_fact_lifecycle_evaluated', { - component = component, - job_id = updater_job_id(upd), - updater_state = upd.state, - software_image_id = sw.image_id, - phase = phase, - reason = phase_reason, - }) + local phase = component_fact_phase(component, fact, self) if not phase then return true, nil end local record = lifecycle_record(component, fact) return emit_component_lifecycle_metric(self, record, phase, { @@ -1426,9 +1354,6 @@ local function ensure_component_watch(self, reason) queue_len = params.component_watch_queue_len, report = service_events.reporter(cw_port, 'update_component_watch_completion_report_failed'), events_tx = self._done_tx, - trace = function (_, payload) - trace_component_lifecycle(self, payload and payload.what or 'component_watch_trace', payload) - end, }) if not cwh then return nil, cwerr or 'update_component_watch_start_failed' end self._component_watch = cwh From 52da1c2e145162bbe78c8f6e3748bb4d936a5083 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 14 Jul 2026 13:59:00 +0000 Subject: [PATCH 6/9] refactor: remove redundant logging configuration in spawn_service function --- src/devicecode/main.lua | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/devicecode/main.lua b/src/devicecode/main.lua index 1eb1f7ea..c56b7bb6 100644 --- a/src/devicecode/main.lua +++ b/src/devicecode/main.lua @@ -79,9 +79,7 @@ local function spawn_service(child, bus, name, mod, env, extra_opts) connect = connect_as, services = extra_opts and extra_opts.services or nil, run_http = extra_opts and extra_opts.run_http or nil, - verify_login = extra_opts and extra_opts.verify_login or nil, - profile = 'trace', - min_level = 'trace' + verify_login = extra_opts and extra_opts.verify_login or nil }) error(('service returned unexpectedly: %s'):format(tostring(name)), 0) From c840e25826ad4c81e54c6d8920794a171d7d10e7 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 14 Jul 2026 14:03:15 +0000 Subject: [PATCH 7/9] refactor: simplify HTTP publishing by removing unnecessary chunking logic --- src/services/metrics.lua | 64 ++++++++-------------------------------- 1 file changed, 12 insertions(+), 52 deletions(-) diff --git a/src/services/metrics.lua b/src/services/metrics.lua index 5330585b..db3203ee 100644 --- a/src/services/metrics.lua +++ b/src/services/metrics.lua @@ -44,7 +44,6 @@ local types = require 'services.metrics.types' local unpack = unpack or rawget(table, 'unpack') local NAME = 'metrics' -local HTTP_MAX_RECORDS = 25 ------------------------------------------------------------------------------- -- Topic helpers @@ -242,32 +241,6 @@ local function log_publish(data) end end -local function sorted_metric_keys(data) - local keys = {} - for endpoint_str in pairs(data or {}) do keys[#keys + 1] = endpoint_str end - table.sort(keys) - return keys -end - -local function metric_chunks(data, max_records) - local keys = sorted_metric_keys(data) - local chunks = {} - if #keys == 0 then return chunks end - - local limit = tonumber(max_records) or #keys - if limit < 1 then limit = #keys end - - for i = 1, #keys, limit do - local chunk = {} - for j = i, math.min(i + limit - 1, #keys) do - local key = keys[j] - chunk[key] = data[key] - end - chunks[#chunks + 1] = chunk - end - return chunks -end - ---@param data table local function http_publish(data) local valid, config_err = conf.validate_http_config(State.cloud_config) @@ -292,31 +265,18 @@ local function http_publish(data) State.cloud_config.url, channel_id) local auth = 'Thing ' .. State.cloud_config.thing_key - local chunks = metric_chunks(data, HTTP_MAX_RECORDS) - for chunk_index, chunk in ipairs(chunks) do - local senml_list, encode_err = senml.encode_r('', chunk) - if encode_err then - State.svc:obs_log('error', { - what = 'senml_encode_failed', - err = tostring(encode_err), - chunk_index = chunk_index, - chunk_count = #chunks, - }) - elseif #senml_list > 0 then - local body = json.encode(senml_list) - - -- Non-blocking enqueue: drop and log if the channel is at capacity. - local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body }) - :or_else(function() return true end)) - - if full then - State.svc:obs_log('error', { - what = 'http_queue_full', - err = 'dropping publish payload', - chunk_index = chunk_index, - chunk_count = #chunks, - }) - end + local senml_list, encode_err = senml.encode_r('', data) + if encode_err then + State.svc:obs_log('error', { what = 'senml_encode_failed', err = tostring(encode_err) }) + elseif #senml_list > 0 then + local body = json.encode(senml_list) + + -- Non-blocking enqueue: drop and log if the channel is at capacity. + local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body }) + :or_else(function() return true end)) + + if full then + State.svc:obs_log('error', { what = 'http_queue_full', err = 'dropping publish payload' }) end end end From 3df3b772ff1400c412bc43c3e187bb08abb64397 Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 14 Jul 2026 16:58:40 +0000 Subject: [PATCH 8/9] reverted some code to original form --- src/devicecode/main.lua | 2 +- src/services/metrics.lua | 26 ++++++++------- src/services/update/component_watch.lua | 43 +------------------------ 3 files changed, 16 insertions(+), 55 deletions(-) diff --git a/src/devicecode/main.lua b/src/devicecode/main.lua index c56b7bb6..b85b9e61 100644 --- a/src/devicecode/main.lua +++ b/src/devicecode/main.lua @@ -79,7 +79,7 @@ local function spawn_service(child, bus, name, mod, env, extra_opts) connect = connect_as, services = extra_opts and extra_opts.services or nil, run_http = extra_opts and extra_opts.run_http or nil, - verify_login = extra_opts and extra_opts.verify_login or nil + verify_login = extra_opts and extra_opts.verify_login or nil, }) error(('service returned unexpectedly: %s'):format(tostring(name)), 0) diff --git a/src/services/metrics.lua b/src/services/metrics.lua index db3203ee..60ef6889 100644 --- a/src/services/metrics.lua +++ b/src/services/metrics.lua @@ -243,6 +243,15 @@ end ---@param data table local function http_publish(data) + local senml_list, encode_err = senml.encode_r('', data) + if encode_err then + State.svc:obs_log('error', { what = 'senml_encode_failed', err = tostring(encode_err) }) + return + end + if #senml_list == 0 then return end + + local body = json.encode(senml_list) + local valid, config_err = conf.validate_http_config(State.cloud_config) if not valid then State.svc:obs_log('error', { what = 'http_publish_skipped', err = tostring(config_err) }) @@ -265,19 +274,12 @@ local function http_publish(data) State.cloud_config.url, channel_id) local auth = 'Thing ' .. State.cloud_config.thing_key - local senml_list, encode_err = senml.encode_r('', data) - if encode_err then - State.svc:obs_log('error', { what = 'senml_encode_failed', err = tostring(encode_err) }) - elseif #senml_list > 0 then - local body = json.encode(senml_list) - - -- Non-blocking enqueue: drop and log if the channel is at capacity. - local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body }) - :or_else(function() return true end)) + -- Non-blocking enqueue: drop and log if the channel is at capacity. + local full = perform(State.http_send_ch:put_op({ uri = uri, auth = auth, body = body }) + :or_else(function() return true end)) - if full then - State.svc:obs_log('error', { what = 'http_queue_full', err = 'dropping publish payload' }) - end + if full then + State.svc:obs_log('error', { what = 'http_queue_full', err = 'dropping publish payload' }) end end diff --git a/src/services/update/component_watch.lua b/src/services/update/component_watch.lua index 90478650..18833333 100644 --- a/src/services/update/component_watch.lua +++ b/src/services/update/component_watch.lua @@ -12,26 +12,6 @@ local queue = require 'devicecode.support.queue' local M = {} -local function trace(params, what, payload) - if type(params.trace) ~= 'function' then return end - payload = payload or {} - payload.what = what - pcall(params.trace, what, payload) -end - -local function updater_from_payload(payload) - if type(payload) ~= 'table' then return nil end - return type(payload.updater) == 'table' and payload.updater - or type(payload.update) == 'table' and payload.update - or nil -end - -local function updater_job_id(upd) - local job_id = type(upd) == 'table' and upd.job_id or nil - if type(job_id) ~= 'string' or job_id == '' then return nil end - return job_id -end - local function component_list(config, explicit) local seen, out = {}, {} local function add(c) @@ -67,29 +47,15 @@ local function open_watches(scope, conn, components, queue_len) end local function report_component_fact(params, component, ev) - local upd = updater_from_payload(ev and ev.payload) if not params.events_tx then - trace(params, 'component_fact_changed_not_reported', { - component = component, - reason = 'events_tx_unavailable', - job_id = updater_job_id(upd), - updater_state = upd and upd.state or nil, - }) return true, nil end - local ok, err = queue.try_admit_required(params.events_tx, { + return queue.try_admit_required(params.events_tx, { kind = 'component_fact_changed', component = component, payload = ev.payload, origin = ev.origin, }, 'update_component_fact_changed_admission_failed') - trace(params, ok == true and 'component_fact_changed_reported' or 'component_fact_changed_report_failed', { - component = component, - job_id = updater_job_id(upd), - updater_state = upd and upd.state or nil, - err = err, - }) - return ok, err end local function watch_loop(scope, params) @@ -112,12 +78,6 @@ local function watch_loop(scope, params) if ev.op == 'retain' or ev.event == 'retain' or ev.type == 'retain' or ev.kind == 'retain' then local ok_update, update_err = observer:update_component(component, ev.payload, ev.origin) if ok_update == nil then error(update_err or 'component_observer_update_failed', 0) end - local upd = updater_from_payload(ev.payload) - trace(params, 'component_fact_retained', { - component = component, - job_id = updater_job_id(upd), - updater_state = upd and upd.state or nil, - }) report_component_fact(params, component, ev) elseif ev.op == 'unretain' or ev.event == 'unretain' or ev.type == 'unretain' or ev.kind == 'unretain' then observer:remove_component(component, 'component_unretained') @@ -144,7 +104,6 @@ function M.start(scope, params) components = components, queue_len = params.queue_len, events_tx = params.events_tx, - trace = params.trace, }) end, report = params.report, From 8f372ab97ada8e1533ecad609fbbc454bb5106ce Mon Sep 17 00:00:00 2001 From: Ryan Name Date: Tue, 14 Jul 2026 17:15:26 +0000 Subject: [PATCH 9/9] refactor: update job component references and remove redundant job creation assertions --- tests/unit/update/test_service_phase2.lua | 72 +++++++++-------------- 1 file changed, 27 insertions(+), 45 deletions(-) diff --git a/tests/unit/update/test_service_phase2.lua b/tests/unit/update/test_service_phase2.lua index dbb9fd0b..97c566c6 100644 --- a/tests/unit/update/test_service_phase2.lua +++ b/tests/unit/update/test_service_phase2.lua @@ -562,12 +562,6 @@ function tests.test_component_device_fact_job_id_emits_lifecycle_started_metric( config = component_config("mcu"), }) - assert(caller:call(topics.update_manager_rpc("create-job"), { - job_id = "job/mcu.1", - component = "mcu", - artifact_ref = "artifact-mcu", - }, { timeout = 0.5 })) - retain_component_fact(caller, "mcu", { job_id = "job/mcu.1", image_id = "image-old", @@ -590,12 +584,6 @@ function tests.test_component_device_fact_json_null_error_still_emits_lifecycle_ config = component_config("mcu"), }) - assert(caller:call(topics.update_manager_rpc("create-job"), { - job_id = "job/mcu.null-error", - component = "mcu", - artifact_ref = "artifact-mcu", - }, { timeout = 0.5 })) - retain_component_fact(caller, "mcu", { job_id = "job/mcu.null-error", updater_state = "ready", @@ -617,15 +605,21 @@ function tests.test_component_device_fact_expected_image_emits_lifecycle_complet fibers.run(function (root_scope) local child, caller = start_service(root_scope, { config = component_config("mcu"), + initial_jobs = { + jobs = { + ["j-success"] = { + job_id = "j-success", + component = "mcu", + artifact_ref = "artifact-mcu", + expected_image_id = "image-new", + state = "staging", + created_seq = 1, + updated_seq = 1, + }, + }, + }, }) - assert(caller:call(topics.update_manager_rpc("create-job"), { - job_id = "j-success", - component = "mcu", - artifact_ref = "artifact-mcu", - metadata = { expected_image_id = "image-new" }, - }, { timeout = 0.5 })) - retain_component_fact(caller, "mcu", { job_id = "j-success", image_id = "image-old", @@ -653,12 +647,6 @@ function tests.test_component_device_fact_error_emits_lifecycle_failed_metric() config = component_config("mcu"), }) - assert(caller:call(topics.update_manager_rpc("create-job"), { - job_id = "j-fail", - component = "mcu", - artifact_ref = "artifact-mcu", - }, { timeout = 0.5 })) - retain_component_fact(caller, "mcu", { job_id = "j-fail", updater_state = "failed", @@ -682,12 +670,6 @@ function tests.test_component_device_fact_cancelled_emits_lifecycle_cancelled_me config = component_config("mcu"), }) - assert(caller:call(topics.update_manager_rpc("create-job"), { - job_id = "j-cancel", - component = "mcu", - artifact_ref = "artifact-mcu", - }, { timeout = 0.5 })) - retain_component_fact(caller, "mcu", { job_id = "j-cancel", updater_state = "cancelled", @@ -737,14 +719,14 @@ function tests.test_job_state_awaiting_commit_emits_lifecycle_staged_metric_with end local child, caller = start_service(root_scope, { - config = component_config("mcu"), + config = component_config("cm5"), backend = backend, }) assert(caller:call(topics.update_manager_rpc("create-job"), { job_id = "j-state-staged", - component = "mcu", - artifact_ref = "artifact-mcu", + component = "cm5", + artifact_ref = "artifact-cm5", expected_image_id = "image-new", }, { timeout = 0.5 })) assert(caller:call(topics.update_manager_rpc("start-job"), { @@ -752,7 +734,7 @@ function tests.test_job_state_awaiting_commit_emits_lifecycle_staged_metric_with }, { timeout = 0.5 })) local metric = wait_lifecycle_metric(caller, "staged") - assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_state_staged.staged") + assert_eq(table.concat(metric.namespace, "."), "cm5.lifecycle.j_state_staged.staged") assert_eq(metric.job_id, "j-state-staged") assert_eq(metric.source, "update_job_state") @@ -768,14 +750,14 @@ function tests.test_job_state_failure_emits_lifecycle_failed_metric_without_comp end local child, caller = start_service(root_scope, { - config = component_config("mcu"), + config = component_config("cm5"), backend = backend, }) assert(caller:call(topics.update_manager_rpc("create-job"), { job_id = "j-state-fail", - component = "mcu", - artifact_ref = "artifact-mcu", + component = "cm5", + artifact_ref = "artifact-cm5", expected_image_id = "image-new", }, { timeout = 0.5 })) assert(caller:call(topics.update_manager_rpc("start-job"), { @@ -783,9 +765,9 @@ function tests.test_job_state_failure_emits_lifecycle_failed_metric_without_comp }, { timeout = 0.5 })) local metric = wait_lifecycle_metric(caller, "failed") - assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_state_fail.failed") + assert_eq(table.concat(metric.namespace, "."), "cm5.lifecycle.j_state_fail.failed") assert_eq(metric.job_id, "j-state-fail") - assert_eq(metric.component, "mcu") + assert_eq(metric.component, "cm5") assert_eq(metric.source, "update_job_state") assert_eq(metric.error, "no_session") @@ -810,14 +792,14 @@ function tests.test_job_state_reconcile_success_emits_lifecycle_completed_metric end local child, caller = start_service(root_scope, { - config = component_config("mcu"), + config = component_config("cm5"), backend = backend, }) assert(caller:call(topics.update_manager_rpc("create-job"), { job_id = "j-state-complete", - component = "mcu", - artifact_ref = "artifact-mcu", + component = "cm5", + artifact_ref = "artifact-cm5", expected_image_id = "image-new", }, { timeout = 0.5 })) assert(caller:call(topics.update_manager_rpc("start-job"), { @@ -834,9 +816,9 @@ function tests.test_job_state_reconcile_success_emits_lifecycle_completed_metric }, { timeout = 0.5 })) local metric = wait_lifecycle_metric(caller, "completed") - assert_eq(table.concat(metric.namespace, "."), "mcu.lifecycle.j_state_complete.completed") + assert_eq(table.concat(metric.namespace, "."), "cm5.lifecycle.j_state_complete.completed") assert_eq(metric.job_id, "j-state-complete") - assert_eq(metric.component, "mcu") + assert_eq(metric.component, "cm5") assert_eq(metric.source, "update_job_state") child:cancel("test complete")