feat(executor): end-to-end latency instrumentation with correlation ids - #2361
feat(executor): end-to-end latency instrumentation with correlation ids#2361thesithunyein wants to merge 7 commits into
Conversation
…ds (KeeperHub#2289) Mint a correlation id at SQS receive and thread it through dispatch to the runner pod and the in-process engine, recording per-stage timestamps: received -> started -> dispatched -> broadcast -> completed - latency.ts: ExecutionLatency stage tracker (idempotent marks, derived durations, JSON-safe log fields) + CSPRNG correlation ids - processMessage/processExecutorMessage: correlation id minted at the earliest receipt point; dispatch emission of the receive->dispatch histogram and structured [Executor:Latency] summary line - executeInProcess: started/completed marks around the engine call, receive->started + receive->completed histograms, correlation id on existing completed/fatal logs - createWorkflowJob: KH_CORRELATION_ID env + pod label so runner logs join the same trace key - metrics types: executor.dispatch.latency_ms / executor.execution.latency_ms histograms + correlation_id / dispatch_target labels Tests: 13 new unit tests for id/order/idempotency/duration/serialization and the metric constants. Executor suite: 138 passing.
About the
|
…tracker -> SQS -> executor -> runner) Adds the event-tracker leg of the latency instrumentation, so the correlation id now spans the full pipeline the issue describes: - event-tracker: generateCorrelationId helper; mint the id + observedAt at the moment an event is first observed (EventListener.onLog) and carry both on the SQS message (workflow-sqs.ts, omitted for legacy callers) - executor: reuse the tracker-minted id and observed stage when the message carries one (falls back to minting at receive); event schema accepts the optional fields while legacy messages still validate - runner: emit KH_CORRELATION_ID on start, completion and fatal logs so the pod joins the same trace key - latency: new 'observed' stage (tracker observation -> receive queue leg) Tests: +2 tracker unit (payload carry/legacy omission), +5 executor unit (observed stage, queue leg, summary ordering, schema acceptance/rejection). Executor suite 143 passing; event-tracker unit suite 222 passing; tsc clean.
joelorzet
left a comment
joelorzet
left a comment
There was a problem hiding this comment.
The metrics half of this is dead. It will not be merged in this state.
| piece | state |
|---|---|
executor.dispatch.latency_ms |
name only, no histogram, nothing recorded |
executor.execution.latency_ms |
name only, no histogram, nothing recorded |
broadcast stage |
declared, never marked |
Every sample is discarded and logs Unknown latency metric. A declaration nothing implements is not a smaller version of the feature. It is a feature that does not exist while looking like it does.
Please finish all three here. We are not landing the shape now and the implementation later.
The correlation id half is done and worth keeping. The tracker mints the id and puts correlationId and observedAt on the SQS message, the executor reads them and mints its own when they are absent, and k8s-job.ts injects KH_CORRELATION_ID plus a correlation-id label so the runner pod joins on the same key. One run is traceable across three services, and each stage is recorded once so a retry cannot overwrite the first observation.
Four things inline.
Smaller: keeperhub-events/event-tracker/lib/correlation.ts has no trailing newline, and it duplicates generateCorrelationId. If the package boundary forces the copy, say so in the comment the way in-flight.ts does.
| // and the full receive -> terminal lifetime. Split by trigger + dispatch | ||
| // target so a slow producer, a slow queue, or a slow runner is visible | ||
| // independently. | ||
| EXECUTOR_DISPATCH_LATENCY: "executor.dispatch.latency_ms", |
There was a problem hiding this comment.
Adding a name here does not create a metric. Neither of these reaches Prometheus.
recordLatency resolves the name against a fixed map and drops anything it does not know:
const histogram = histogramMap[name];
if (histogram) {
histogram.observe(sanitizeLabels(labels), durationMs);
} else {
logWarn(`[Prometheus] Unknown latency metric: ${name}`);
}histogramMap in lib/metrics/collectors/prometheus.ts holds four latency histograms and neither of these is among them. This PR does not touch that file, and the executor imports the same collector, so every sample is discarded and each execution writes a warning instead.
Register both there with a help string and buckets, and add a test asserting a sample lands. Nothing here would have caught this.
One thing to address while you are in that file. It carries a note saying workflow execution and step metrics deliberately moved to DB-sourced gauges, and dbSourcedMetrics holds workflow.execution.duration_ms and workflow.step.duration_ms. Your two measure queue time and dispatch hand-off, which the database has no timestamps for, so runtime histograms are the right choice. Say that in the comment, otherwise the next reader sees you going against a decision recorded a few lines above.
| MODE: "mode", | ||
| CLAIM_RESULT: "claim_result", | ||
| // Latency instrumentation (issue #2289) | ||
| CORRELATION_ID: "correlation_id", |
There was a problem hiding this comment.
Keep this out of the histogram labels when you register them.
A correlation id is a fresh value per execution, so it creates one time series per execution. #2289 rules this out by name: "per-workflow labels on a latency histogram are a metrics-cost problem", and this is finer-grained than per-workflow. The comment in latency.ts calling the id "short enough for labels" points the wrong way.
The id belongs in the structured log, where it already does its job. Label the histograms with trigger, dispatch_target and stage.
| | "received" | ||
| | "started" | ||
| | "dispatched" | ||
| | "broadcast" |
There was a problem hiding this comment.
broadcast is declared here and in STAGE_ORDER, and nothing marks it. Across the whole diff the marks are observed, received, started, dispatched and completed. The only edit inside the pod that broadcasts, in workflow-runner.ts, appends the correlation id to two log lines and records no timestamp.
That is the measurement the issue exists for. #2289 lists "Transaction broadcast" as a required stage, and asks for the distribution of time from event observed to transaction broadcast, and which stage dominated it. observed is recorded and broadcast is not, so that interval cannot be computed and the per-stage attribution stops one hop short.
Mark it where the transaction actually goes to the chain.
| * Single-line structured summary matching the executor's JSON log shape | ||
| * (`[Component] key=value ...`). Parsable with a plain key=value splitter. | ||
| */ | ||
| summaryLine(params: { |
There was a problem hiding this comment.
Use logInfo for these lines rather than building the summary by hand.
lib/logging.ts exports logInfo(message, labels), which emits the canonical structured line with the labels already in the shape the log pipeline parses. This builds a second, parallel format to carry the same fields, and getting the correlation id into the logs is the point of the PR.
Scope it to the new latency lines only. The surrounding console.log calls in the executor are the existing convention there and are not yours to change.
One thing to watch: lib/logging.ts imports @sentry/nextjs. You already import lib/metrics, so most of the chain is present, but if that import breaks the executor bundle, say so and keep console.log with a note explaining why.
joelorzet
added
the
changes-requested
Addresses all four review comments on this PR: 1. Register the executor latency histograms in histogramMap. The two declared names never reached Prometheus because recordLatency resolves against a fixed map; every sample was discarded with 'Unknown latency metric'. Now registered in apiRegistry with help strings and buckets, plus a comment explaining why these are runtime histograms despite the DB-sourced-gauges note (queue/hand-off intervals leave no DB timestamps). A regression test asserts a sample lands in the exposition. 2. Remove the correlation id from histogram labels. A fresh id per execution is one time series per run (KeeperHub#2289 rules out even per-workflow labels). LabelKeys.CORRELATION_ID is gone; histograms are labeled trigger_type / dispatch_target / stage; the wrong-way comment in latency.ts is rewritten. 3. Mark the broadcast stage where the transaction actually goes to the chain: both EVM adapter broadcast points (sign-once failover and legacy signer.sendTransaction), the Solana submit point, and the Turnkey sponsored path. The write paths cannot reach the ExecutionLatency instance, so the timestamp travels via a /tmp sidecar plus a process-local counter that ships with the counter deltas. In-process runs read the marker back after the engine returns and record the observed -> broadcast distribution (executor.broadcast.latency_ms); k8s-job runs log the marker joined on correlation id, since ephemeral pods do not ship histogram observations. 4. Replace the hand-built summary lines with logInfo(message, labels) from lib/logging, scoped to the latency lines only. The executor already imports lib/logging transitively, so the Sentry chain adds nothing new. emitLog lives on ExecutionLatency; tests assert the canonical structured shape. Also: logInfo import verified bundle-safe for the executor; correlation.ts trailing newline + duplication note added per review; METRICS_REFERENCE rows added for all three histograms and the broadcast counter. New/updated tests: 152 executor + 7 web3/metrics, tsc clean.
|
Thanks for the precise review — all four points were correct, and all four are fixed in edc6193. The metrics half is real now. 1. Histograms registered (
|
…central histograms Closes the one gap the PR left open: executor.broadcast.latency_ms only filled for in-process runs, but real web3 writes dispatch to k8s Jobs, whose pods cannot write the executor's histograms (histogram observations cannot be merged across pods without losing bucket fidelity). Point observations can. The pod ships one duration per stage interval over the existing metrics ingest; the executor folds each sample into the originating run's timeline and the central histograms: - runner collects observed->broadcast (sidecar marker + KH_OBSERVED_AT) and received->completed (KH_RECEIVED_AT, injected via k8s-job.ts) - ship-metrics posts them as an optional observations array on the ingest payload; empty deltas + pending observations still post - correlation-map keeps the run's ExecutionLatency reachable by correlation id (bounded, oldest-evicted) until its observations land - observation-applier marks the timeline (broadcast epoch reconstructed from observed + pod duration), records executor.broadcast.latency_ms for every Job broadcast, and records executor.execution.latency_ms stage=completed anchored on the executor's own received stamp (a skewed pod clock cannot distort the queue leg); unknown correlation ids lose nothing for self-contained intervals and skip only start-anchored ones - ingest route returns obsApplied/obsSkipped; a bad observation is skipped, never rejected Tests: 168 executor (16 new: collection guards, map bounds, applier anchoring/skips, observations-on-the-wire), tsc clean.
|
Follow-up: the door I left open in the reply above is now closed in 14c4390. Point latency observations from runner pods.
16 new tests (168 executor total, tsc clean). Happy to trim this if the PR has grown past what you want in one change — the observation modules are self-contained ( |
|
Hi @joelorzet — friendly ping ahead of the hackathon submission deadline (Sep 18). All four review points are fixed (same-day, with tests — the executor suite is now at 168), and the follow-up in |
- export executorBroadcastsTotal (metrics-shipping destructures it; the unexported binding only passed ad-hoc tsc runs, not type-check:tsc:executor) - normalize triggerType to "manual" at collection so PendingObservation matches the strict LatencyObservation wire contract pnpm type-check:tsc:executor, pnpm type-check:tsc clean; 168/168 executor tests.
|
Small hardening commit (
Both |
|
Following up on the three metric pieces rather than reopening the review. Two are genuinely fixed; the third is still dead in production, for a different reason than before.
The
So there is no Next.js And there is a second dead metric, which is the same defect class. One more that would keep the Job half empty even after the above is fixed. And the fixed marker path has a concurrency assumption the executor breaks. Smaller, and only worth doing while you are in there: The correlation-id half is untouched by all of this and still stands on its own: the tracker mints the id, |
Second-pass review found the broadcast half of the KeeperHub#2289 instrumentation dead in production dispatch. All four defects are fixed: 1. ALS never registered in the executor/runner processes (no instrumentation.ts in either Docker stage) -> currentExecutionId() was always undefined, so markBroadcast() dropped every sidecar write and executor.broadcast.latency_ms stayed empty for in-process runs. New side-effect bootstrap module registers the storage the same way instrumentation.ts register() does in the Next runtime; imported by keeperhub-executor/index.ts and workflow-runner.ts. 2. keeperhub_executor_broadcasts_total was registered and shipped but never incremented (getBroadcastCount() had no readers outside tests). markBroadcast() now increments the registered counter via lazy import, buffering marks that land before the collector resolves so no sample is lost, and staying a no-op where the stack is unavailable. 3. takeBroadcastMarker() was called twice per run (runner log line, then the observation collector), so the second call deterministically got undefined and the Job-pod broadcast histogram stayed empty. The log line now peeks (non-destructive) and the collector takes. 4. The single fixed marker filename raced under the executor's Promise.allSettled(maxMessages: 10) in-process dispatch: run B could overwrite run A's marker before A read it back, silently losing the sample. The registry is now one file per execution id (KH_BROADCAST_MARKER_DIR), and take is by explicit execution id, so a concurrent run can only ever consume its own marker. Smaller points, same pass: dropped two unused imports from index.ts (peekLatency/takeLatency moved to their sole reader in the applier; unused LatencyObservation type), added the missing trailing newline to latency.ts, and updated the stale prometheus.ts comment that still said a central histogram for Jobs would need point-sample ingestion first - the follow-up commit in this branch added exactly that. Tests: 171 executor tests pass (3 new for the ALS bootstrap, 1 for the registered-counter flush, 1 for take-consumes semantics; broadcast-marker suite rewritten for the per-execution registry). tsc clean via the repo's own type-check:tsc:executor and type-check:tsc.
|
Thank you for the second pass - every finding checked out against the code, and all four defects are fixed in 39f978e. 1. The broadcast stage is now marked in production dispatchRoot cause as you diagnosed: no Next
2. The counter now increments
3. The double-take is goneThe runner's log line now uses a new non-destructive 4. Concurrency: the registry is now per-executionThe sidecar moved from one fixed filename to a directory ( Smaller pointsBoth unused imports dropped from VerificationYour own scripts, clean: On the landing shape: with the broadcast half now live in production dispatch, the correlation/metrics split you sketched is no longer needed - the whole PR is coherent. The offer stands though: if a smaller landing is easier to review, |
suisuss
left a comment
suisuss
left a comment
There was a problem hiding this comment.
All seven land, and the tests exercise the real call paths rather than the shape. keeperhub-executor/lib/workflow-error-context-bootstrap.ts:29 registers the AsyncLocalStorage as a side-effect import from index.ts:88 and workflow-runner.ts:48, which closes the chain - executor.workflow.ts:2150-2157 already enters the context at run start and step-handler.ts:262 wraps every step, so currentExecutionId() now resolves in both non-Next processes and the test asserts the real path instead of the broken condition. The counter is incremented at broadcast-marker.ts:76 through a lazy import with pre-resolution marks buffered, and the test reads the registered counter rather than the module-local count. peekBroadcastMarker at :120-132 leaves collectLatencyObservations as the sole consumer, with a regression test that a second collect does not re-ship. One file per execution under MARKER_DIR makes the cross-run overwrite structurally impossible. Duplicate imports, the trailing newline and the stale prometheus.ts comment are all fixed.
I fact-checked the comment: every mechanical claim holds. One is looser than stated - in the executor, loadShippableCounters only runs when an ingest POST arrives, so the first in-process broadcast may well precede resolution; the buffer covers it, so the conclusion stands but the reasoning does not.
Blocking
keeperhub-executor/lib/broadcast-marker.ts:39within-process.ts:110- the per-execution registry has no sweep.MARKER_DIRappears only at:39,:48and:81,rmSynconly at:137on a single resolved path, and there is noreaddirSync, no TTL and no startup clean.recordInProcessLatencyis the only in-process consumer and it sits inside thetry, aftermark("completed"); thecatchat:133deliberately does not call it. -> An in-process run that broadcasts and then throws leaves/tmp/kh-broadcast-markers/<executionId>.jsonbehind forever, in a pod that runs for weeks. Each file is tiny, so inodes rather than bytes are the bound, in an emptyDir with a size limit. The module doc at:23-25says these are "cleaned up by the runner's take", which is true only on the success path. The fix for the fixed-filename race converted a leak bounded at exactly one stale file into an unbounded one. -> Sweep on executor startup, or take in thecatch.
Mechanical - actionable as-is
-
broadcast-marker.ts:47-49-join(MARKER_DIR, \${executionId}.json`)with no sanitisation, and that value now drives both a path and anrmSync`. Ids are DB-generated and SQS messages are HMAC-signed, so this is hardening rather than a live hole, but the value became load-bearing for a delete in this commit. -
latency-observations.ts:79dropped themarker.executionId === executionIdcross-check whilein-process.ts:187kept it, andparseMarkervalidates shape without comparing the parsed id to the file it came from. Harmless whilemarkBroadcastis the only writer; the inconsistency between the two consumers is what will confuse the next reader. -
broadcast-marker.test.ts:88-101readsbefore, awaits onesetImmediate, marks twice, awaits one more, then assertsafter >= before + 2. One macrotask tick does not guarantee a dynamic ESM import ofprometheus.tshas settled; it passes today only because three earlier tests already triggered that import. Awaiting the module's own promise makes it deterministic rather than ordering-dependent. -
executor.workflow.ts:2150usesenterWithrather thanrun(), which mutates the current async resource's store rather than scoping a callback, under ten-way in-process concurrency. It works because the runs are on distinct async resources by the timeexecuteWorkflowruns, and the step-levelrunWithWorkflowErrorContextis a properrun()and is the path web3 writes actually take. Worth knowing that the weaker of the two mechanisms is the one the module doc credits.
With the team
- Whether
keeperhub-executor/lib/broadcast-marker.tsis safe inside the Workflow DevKit bundle.lib/web3/chain-adapter/evm.ts:6imports it statically; it already carriednode:fsandnode:path, and it now also carries a dynamic import oflib/metrics/collectors/prometheus, which begins withimport "server-only".error-context.ts's own header warns that anything reachable fromworkflow-executor.workflow.tsmust hold zero Node builtins. If the chain adapter is reachable from the workflow bundle rather than only from"use step"Node-runtime code, the DevKit build breaks - and neithertype-check:tscnortest:executorwould catch it. I flagged this last round and it has grown rather than shrunk. A fullpnpm buildsettles it in one run; I am getting that answer rather than asking you to guess.
Verdict
Changes requested on the unswept marker directory - all seven items are genuinely fixed, and the fix for the concurrency one introduced a leak that a weeks-long pod will accumulate.
The correlation-id half is still untouched and still independently shippable, and workflow-error-context-bootstrap.ts should stay whichever way the rest goes - without it the executor and runner lose workflow and org attribution on every error log, not just on markBroadcast. That is a wider fix than this PR needed to make.
suisuss
added
the
decision-needed
…reads
The per-execution marker registry had no bounded cleanup path: the success
take removed a run's file only when executeWorkflow returned normally, so
an in-process run that broadcast and then threw left its marker in
/tmp/kh-broadcast-markers forever, in a pod that runs for weeks.
Three removal paths now cover every failure mode:
- the in-process catch discards the run's own marker (best-effort, cleanup
only - no latency stage is recorded on the failure path, so histograms
keep counting only runs that reached a terminal state)
- the executor sweeps the whole registry at startup, before the consumer
can start any run; covers a process killed mid-run. Runner pods mount
their own emptyDir and are untouched
- the module doc now lists all three paths instead of crediting only the
success take
Also from review:
- executionId is allowlisted ([A-Za-z0-9_-]{1,128}, covers nanoid and
UUID) before it drives a path or an rmSync; unsafe ids still count as
broadcasts but write no file
- peek/take reject markers whose content id does not match the requested
id, so no consumer can receive a mismatched marker;
collectLatencyObservations keeps an explicit symmetric check with
in-process.ts
- the counter test awaits the lazy import's own promise via
waitForBroadcastCounterForTests instead of one setImmediate tick, and
asserts exactly rather than a range
Executor suite: 177 passing (13 marker/cleanup tests, including an
integration test that broadcasts then throws through the real
executeInProcess catch). type-check:tsc:executor and type-check:tsc clean.
|
Thanks. Every finding checked out, including the correction to my comment's reasoning. You're right that Blocking: the registry is now bounded on three paths.
An integration test pins this through the real Mechanical, all four:
Executor suite: 177 passing, up from 168. |
3 participants
What
End-to-end execution latency instrumentation for the full pipeline the issue
describes — event-tracker → SQS → executor → runner/broadcast. Every event
trigger now gets a correlation id minted at the moment it is first observed
and carried through every stage, with per-stage timestamps and latency
histograms:
Why
Latency today is only observable per-workflow from inside the engine
(
workflow.execution.duration_ms). Nothing distinguishes producer → queue →executor delay from executor → runner delay, and there is no key that joins
an event's tracker-stage logs to its SQS/executor/runner-stage logs. This
implements #2289's ask directly: a correlation id and stage timestamps from the
event-tracker through the executor, plus histograms, so a slow producer, a slow
queue and a slow runner are each visible independently — and a single run can be
traced across three systems on one key.
What changed (2 commits, 13 files, +614/−13)
Commit 1 — executor stage (
1d53470):keeperhub-executor/latency.ts(new)ExecutionLatencystage tracker — idempotent marks (first wins), derived durations, JSON-safe log fields, CSPRNGgenerateCorrelationId()(16 hex, no new deps)keeperhub-executor/index.tsdispatchExecutionmarksdispatched, emits the receive→dispatch histogram and a structured[Executor:Latency]summary line (skipped for in-process, which records its own full-timeline line)keeperhub-executor/in-process.tsstarted/completedaround the engine call; receive→started + receive→completed histograms; correlation id on Completed/Fatal logskeeperhub-executor/k8s-job.tsKH_CORRELATION_IDenv var +correlation-idpod labellib/metrics/types.tsexecutor.dispatch.latency_ms,executor.execution.latency_ms;correlation_id,dispatch_targetlabelsCommit 2 — tracker + runner legs (
4e6330e):keeperhub-events/event-tracker/lib/correlation.ts(new)generateCorrelationId()(same format as the executor's).../lib/workflow-sqs.tscorrelationId/observedAtcarried on the SQS message; absent for legacy callers (undefinedis dropped byJSON.stringify).../src/listener/event-listener.tsobservedAtminted at the moment the event is first observed;observed <tx> correlationId=…log linekeeperhub-executor/types.ts+message-schema.tscorrelationId/observedAton event messages; legacy messages without them still validate (drift guard kept)keeperhub-executor/workflow-runner.tsKH_CORRELATION_IDon start, completion and fatal logs so the pod joins the same trace keyDesign notes
first observation, keeping the histograms honest.
completedis only marked when aterminal status lands; a crash shows as a missing series, not a fast fake
reading. Error paths still carry the correlation id.
omits them for legacy callers, so older producers/messages behave exactly as
before (the executor falls back to minting its own id).
node:cryptoonly.timeline; handed-off targets (k8s-job/api) record the receive→dispatch
handoff. No double counts.
Testing
stage/queue leg/ordering, 2 tracker SQS payload carry/legacy omission)
tsc --noEmitclean for both packages (touched files)Follow-up (deliberately out of scope)
executor already handles it whenever a message carries the fields.
Verification for reviewers
Sample emitted line (tracker):
Sample emitted line (executor, in-process run):