> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockx.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Observability
> BlockX structured logging, Prometheus metrics, OpenTelemetry tracing, usage collection, and client id propagation
Observability is not a standalone process; it is a set of adapter-layer infrastructure shared by the Worker, Coordinator, Sync Invoker, Bundle clusters, and Python Executor. It consists of three Go packages: `internal/obs` (logging, metrics, tracing), `internal/usage` (billing usage collection), and `internal/common/clientid` (request attribution ID propagation). See the [Architecture overview](/en/architecture/overview) for where they sit in the system.
## Responsibilities and boundaries
* `internal/obs` is responsible for: installing the process-level `slog` handler, automatically injecting `task_id`/`call_id`/`instance_id` from ctx into logs, declaring all Prometheus metrics and providing the `Record*` functions, initializing the OpenTelemetry tracer, and starting the `/metrics` HTTP endpoint.
* `internal/usage` is responsible for: accumulating, per `client_id`, the executor's effective CPU time at the moment each task converges, and periodically freezing it into records written to the Kafka topic `chaintable-usage`.
* `internal/common/clientid` is responsible for: reading and writing `client-id` and the compatibility header `x-instance-id` at gRPC/HTTP boundaries, and carrying the full ID in ctx inside the process.
* Not responsible for: the log aggregation stack, alerting rules, the Prometheus server, or the OTLP collector. These all live on the deployment side.
* Invariant: `core/` packages (Sans-IO) do not import `internal/obs`; only adapter and `cmd/` code writes logs, records metrics, or opens spans. See [Design principles and code conventions](/en/architecture/design-principles).
* Invariant: the log `instance_id`, the chaintable-log `trace_id`, and the Prometheus `instance_id` label all use `clientid.LogKey` to strip the `instance:` prefix; usage messages and all downstream headers keep the full original value.
* Invariant: Prometheus labels are limited to closed enumerations (`status`/`phase`/`backend`/`kind`, etc.) plus the two high-cardinality dimensions `instance_id`/`function_id`, which are protected by the TTL reaper. Never use task\_id, call\_id, URLs, or error text as labels.
* Invariant: metrics are registered to the default registry via `promauto`; every vec that carries an `instance_id` or `function_id` label must also be listed in `instanceIDVecs`/`functionIDVecs` in `internal/obs/metrics_reaper.go`, otherwise `TestReaperVecListsMatchDeclarations` fails.
## Code location
| Path | Purpose |
| --------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------- |
| `internal/obs/logger.go` | `Init`, the ctx-binding helpers for `ContextHandler` (`WithTaskID`, etc.), and all `Attr*`/`LogType*` constants |
| `internal/obs/handler.go` | `ContextHandler`: injects the correlation fields from ctx into every `slog.Record` |
| `internal/obs/chlog_bridge.go` | chaintable-log Kafka bridge: forwards every slog record to the `instance-logs` topic as well |
| `internal/obs/table_dedupe.go` | Time-window deduplication of `table_read`/`table_write` logs keyed by `(instance, table, operation)` |
| `internal/obs/metrics.go` | Most `blockx_task_*`/`blockx_worker_*` metric declarations, the `Record*` functions, and `StartPrometheusEndpoint` |
| `internal/obs/metrics_reaper.go` | TTL reaper for `instance_id`/`function_id` series |
| `internal/obs/metrics_io_adaptive.go` | `blockx_io_adaptive_*` custom Collector (IO adaptive concurrency limits) |
| `internal/obs/metrics_noderpc_batch.go` | `blockx_noderpc_semantic_batch_*` |
| `internal/obs/metrics_syncinvoke.go` | `blockx_syncinvoke_*` |
| `internal/obs/tracing.go` | `InitTracing`, `Tracer`, `AddUDSEvent`, and the tracing environment variable constants |
| `internal/usage/collector.go` | `Collector`, the `Record`/`Sink` interfaces, active/pending buffering and flush logic |
| `internal/usage/kafka.go` | `KafkaSink`, `NewKafkaCollector` (probes topic partitions at startup) |
| `internal/usage/metrics.go` | The three `blockx_usage_*` publish health metrics |
| `internal/common/clientid/client_id.go` | Header constants, `FromIncomingGRPC`/`IntoOutgoingGRPC`/`IntoHTTP`/`LogKey` |
| `internal/coordinator/adapters/etcd_watcher_metrics.go` | `blockx_coordinator_registry_workers` |
| `internal/io/core/stats/` | CKMS quantile accumulators for task-level IO/call latency (feeds the `task_finished` log and `RecordTask*Converged`) |
| `internal/worker/app/app.go`, `internal/coordinator/app/app.go`, `cmd/syncinvoker/main.go` | Wiring for the three process types: `obs.Init` → `InitTracing` → `StartPrometheusEndpoint` → `InitChlog` → usage collector |
| `python/blockx_executor/logging_setup.py`, `tracing_setup.py` | The Python executor-side counterparts, sharing the same set of environment variables |
| `analyze_spans.py` (repo root) | Normalizes Go and Python span JSON and prints the span tree per trace |
| `docs/specs/logging.md`, `blockx-log-types.md`, `tracing.md`, `client-id-propagation.md`, `2026-07-13-blockx-usage-collection.md` | Corresponding specs |
## Core types and interfaces
| Identifier | File | Description |
| ---------------------------------------------------------------------------------------------------------------- | ----------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `obs.Init()` | `internal/obs/logger.go` | Reads `BLOCKX_LOG_LEVEL`/`BLOCKX_LOG_FORMAT`, installs a JSON (or text) handler wrapped in a `ContextHandler` as `slog.Default()` |
| `obs.ContextHandler` | `internal/obs/handler.go` | Implements the full `slog.Handler`; on `Handle`, pulls `task_id`/`call_id`/`executor_id`/`attempt_seq`/`instance_id` from ctx and appends them to the record |
| `obs.WithTaskID` / `WithCallID` / `WithExecutorID` / `WithAttemptSeq` / `WithInstanceID` / `WithExecutorCallIDs` | `internal/obs/logger.go` | Bind correlation fields to ctx; empty strings are not bound |
| `obs.BindProcess(key, value)` | `internal/obs/logger.go` | Appends a process-level constant to the default logger's base attrs; called once at startup |
| `obs.Attr*` / `obs.LogType*` | `internal/obs/logger.go` | Structured field names and the `type` field enumeration; call sites must use the constants |
| `obs.FailureAttrs(component, code, err, extra...)` | `internal/obs/metrics.go` | Uniform attrs for error paths; includes the goroutine stack at debug level |
| `obs.RecordTaskCallsConverged` / `RecordTaskIOConverged` / `RecordTaskPhase` | `internal/obs/metrics.go` | Write the call/IO/phase metrics once when a task converges |
| `obs.PrometheusConfig` / `obs.StartPrometheusEndpoint` / `obs.HTTPHandler` | `internal/obs/metrics.go` | Scrape endpoint configuration and startup; extra routes (pprof, health probes) hang off the same HTTP server |
| `obs.RegisterIOAdaptiveMetrics(backend, provider)` | `internal/obs/metrics_io_adaptive.go` | Registers a snapshot function once per adaptive IO backend |
| `obs.InitTracing(ctx, service)` / `obs.Tracer(name)` / `obs.TracingEnabled()` | `internal/obs/tracing.go` | Selects the exporter from environment variables; installs a noop provider when none is configured |
| `obs.ChlogConfig` / `NewChlogLogger` / `InitChlog` | `internal/obs/chlog_bridge.go` | Configuration and installation of the chaintable-log Kafka bridge |
| `usage.Recorder` / `usage.Sample` / `usage.Record` / `usage.Sink` | `internal/usage/collector.go` | Non-blocking entry point from the Worker into the collector, single-task sample, Kafka message, and publish backend |
| `usage.Collector` / `NewKafkaCollector` | `internal/usage/collector.go`, `kafka.go` | In-process aggregator and its Kafka constructor |
| `clientid.FromIncomingGRPC` / `WithContext` / `FromContext` / `IntoOutgoingGRPC` / `IntoHTTP` / `LogKey` | `internal/common/clientid/client_id.go` | All read/write rules for the client id |
Key signatures (copied from the code):
```go theme={null}
// internal/obs/metrics.go
func StartPrometheusEndpoint(ctx context.Context, service string, cfg PrometheusConfig, extra ...HTTPHandler) (func(context.Context) error, error)
// internal/obs/tracing.go
func InitTracing(ctx context.Context, service string) (func(context.Context) error, error)
// internal/usage/collector.go
type Recorder interface {
Record(clientID string, sample Sample)
}
type Sink interface {
Publish(ctx context.Context, records []Record) error
Close(ctx context.Context) error
}
```
## Data flow / execution flow
A request's attribution identifier enters from gRPC metadata and flows along ctx to four outlets: logs, spans, metrics, and usage:
```mermaid theme={null}
flowchart LR
A["gRPC request metadata: client-id / x-instance-id"] --> B["clientid.FromIncomingGRPC + WithContext"]
B --> C["ctx carries the full ID, obs.WithTaskID / WithCallID"]
C --> D["slog.*Context: ContextHandler injects instance_id (LogKey)"]
C --> E["otel span: otelgrpc / uds tracer"]
C --> F["obs.Record*: label instance_id (LogKey)"]
C --> G["usage.Recorder.Record: full client_id"]
C --> H["clientid.IntoOutgoingGRPC / IntoHTTP to downstream BlockDB / NodeRPC / Meta"]
D --> D1["stderr JSON"]
D --> D2["chlog bridge to Kafka instance-logs"]
E --> E1["OTLP gRPC or BLOCKX_OTEL_SPAN_FILE"]
F --> F1["/metrics scraped by Prometheus"]
G --> G1["Kafka chaintable-usage"]
```
`SubmitTask` in `internal/worker/adapters/grpc_server.go`, `Invoke`/`DebugInvoke` in `internal/syncinvoker/adapters/grpc_server.go`, and `internal/coordinator/adapters/grpc_server.go` all begin with `ctx = clientid.WithContext(ctx, clientid.FromIncomingGRPC(ctx))`. `client-id` takes priority, falling back to `x-instance-id` when it is missing; if both are missing the result is `unknown`.
The Worker rebinds with `clientid.WithContext(ctx, rt.instanceID)` inside the task runtime (`internal/worker/adapters/orchestrator_dispatch.go`, `orchestrator_phases.go`) so that asynchronous goroutines and IO callbacks do not lose the ID. `obs.WithInstanceID` is just an alias for `clientid.WithContext`; it does not store a second copy of the value.
`ContextHandler.Handle` writes `instance_id` after sanitizing it with `clientid.LogKey`; `metricInstanceID` applies the same sanitization to the Prometheus label and refreshes the series' last-seen timestamp. The chlog bridge uses `LogInstanceIDFromCtx` as the `trace_id`.
When a task converges, `orchestrator_phases.go` calls `o.usageRecorder.Record(taskInstanceID, taskUsage)`, passing the full ID. The `internal/sdk/blockdb`, `logicaltypes`, `noderpc`, and `meta` adapters use `IntoOutgoingGRPC`/`IntoHTTP` to double-write both headers to downstream services.
## Logging
### Format and fields
* Each log entry is one line of JSON written to **stderr** (`internal/obs/logger.go`). `time` is millisecond-precision UTC (`2006-01-02T15:04:05.000Z07:00`), aligned with the Python executor so that Go/Python logs from the same process tree can be parsed as a single interleaved stream.
* Correlation fields are injected automatically from ctx: `task_id`, `call_id`, `executor_id`, `attempt_seq` (emitted once bound, even when it is 0), and `instance_id`.
* Two rules (from `docs/specs/logging.md` §4.1): functions that have a ctx use `slog.InfoContext(ctx, ...)`; goroutine closures without a ctx must receive a `*slog.Logger` that the caller pre-bound with `slog.Default().With(obs.AttrCallID, id)`.
* Field names always go through the `obs.Attr*` constants. The error field is `err`.
* `obs.Init()` takes no arguments; `obs.BindProcess` has no call sites in `cmd/` or `app`, and `service` / `worker_addr` only appear in a few logs that pass them explicitly.
### Structured log types (the `type` field)
The `obs.LogType*` constants are the complete set of legal values for the `type` field. The most important ones from a user's perspective:
| `type` | Trigger point | Description |
| ------------------------------------------------------ | ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `task_finished` (constant name `LogTypeTaskConverged`) | `internal/worker/adapters/orchestrator_phases.go` | One summary log when a task reaches a terminal state: `phases[]`, `call_functions[]`, `call_failed_clusters[]`, `io_error_clusters[]`, `io_backends`, `block_latency_ms`, etc. |
| `task_activation_failed` | Same as above | Task activation failed |
| `table_read` / `table_write` / `table_write_debug` | `internal/obs/table_dedupe.go`, `internal/plugin/event/*` | Table read/write audit, deduplicated by `(instance, table, operation)` within a 10-minute / 1-minute window |
| `audit_reject` | `internal/worker/adapters/audit_gate.go`, `internal/syncinvoker/adapters/service.go` | Code audit rejection (enforce and dark modes share the type and are distinguished by msg) |
| `executor_wedge_reaped` | `internal/worker/adapters/orchestrator_wedge.go` | A wedged executor was hard-killed; worth alerting on |
| `function_code_refresh` | `internal/functioncode/adapters/redis/syncer.go` | Function Code View refresh |
| `usage_publish_failed` / `usage_publish_recovered` | `internal/usage/collector.go` | usage Kafka publish failed/recovered, emitted per record |
The `LogTypeTaskPhaseFinished`, `LogTypeCallFailed`, and `LogTypeIOError` constants still exist but are no longer emitted separately; their content is folded into `task_finished`.
### chaintable-log bridge
Once `CHAINTABLE_LOG_BROKERS` is set, `obs.InitChlog` replaces the inner handler of `ContextHandler` with `chlogBridgeHandler`: every record is still written to stderr and is simultaneously forwarded to Kafka (default topic `instance-logs`), mapping `type` → chlog `WithType`, the remaining attrs → the data payload, and the instance id from ctx → `trace_id`. Records without an instance id (coordinator background logs, etc.) are silently dropped by the SDK.
## Metrics
### Naming and registration
* Namespace `blockx`, snake\_case, counters end in `_total`, histograms carry a unit suffix (`_milliseconds`, `_seconds_total`, `_bytes`).
* All declarations are package-level `promauto.New*Vec(...)`, exposed to adapters through the `obs.Record*` functions; adapters do not hold `prometheus.*` objects directly. IO adaptive rate limiting is the exception: it uses a custom `prometheus.Collector` that pulls snapshots.
* Task-dimension metrics are **observed once when the task converges**: avg/p90/p99 are computed in-process by the CKMS accumulators in `internal/io/core/stats/` and observed once each with a `stat` label. Execution-count-weighted averages are obtained by dividing the two counters `*_duration_seconds_total / *_duration_samples_total`.
### Metric families and definition files
| Metric family | Definition file | Record points |
| ------------------------------------------------------------------------------------------ | ------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------- |
| `blockx_task_*` (phase, calls, function, io, cache, latency) | `internal/obs/metrics.go` | `internal/worker/adapters/orchestrator_phases.go`, `orchestrator_actor.go` |
| `blockx_worker_*` (active tasks, dispatcher, executor, stream, writer, barrier, retention) | `internal/obs/metrics.go` | `internal/worker/adapters/orchestrator.go`, `orchestrator_dispatch.go`, `executor/adapter.go`, `executor/handlers.go` |
| `blockx_function_code_*` | `internal/obs/metrics.go` | `internal/functioncode/adapters/redis/syncer.go` |
| `blockx_io_adaptive_*` | `internal/obs/metrics_io_adaptive.go` | `internal/io/adaptive/observed/backend.go` |
| `blockx_noderpc_semantic_batch_*` | `internal/obs/metrics_noderpc_batch.go` | `internal/sdk/noderpc/batchobserved/observer.go` |
| `blockx_syncinvoke_*` | `internal/obs/metrics_syncinvoke.go` | `internal/syncinvoker/adapters/service.go` |
| `blockx_usage_*` | `internal/usage/metrics.go` | `internal/usage/collector.go` |
| `blockx_coordinator_registry_workers` | `internal/coordinator/adapters/etcd_watcher_metrics.go` | Same file |
`internal/worker/adapters/call_metrics_test.go`, `executor_metrics_test.go`, and `task_barrier_metrics_test.go` are tests for these record points, not definition files. `internal/worker/adapters/resource_metrics.go` is the CPU/memory sampler used by heartbeats (`WorkerResourceSampler`); it does not export Prometheus metrics.
### Most important metrics
| Metric name | Type | Labels | Meaning |
| --------------------------------------------------------------------------------------------------------------------- | ------------------- | --------------------------------------------- | -------------------------------------------------------------------------------- |
| `blockx_task_phase_duration_milliseconds` | histogram | `instance_id, phase, status` | Wall-clock duration of each phase; `phase="all"` is the total task duration |
| `blockx_task_calls_total` | counter | `instance_id, status` | Successful/failed call counts accumulated when the task converges |
| `blockx_task_function_calls_total` | counter | `instance_id, function_id, call_kind, status` | Terminal-state logical call count per function |
| `blockx_task_function_effective_duration_seconds_total` / `_whole_duration_seconds_total` / `_duration_samples_total` | counter | `instance_id, function_id, call_kind` | Execution-count-weighted sums of CPU / CPU+IO time and the sample count |
| `blockx_task_calls_effective_duration_milliseconds` / `blockx_task_calls_whole_duration_milliseconds` | histogram | `instance_id, function_id, call_kind, stat` | avg/p90/p99 observed once each per task |
| `blockx_task_io_ops_total` | counter | `instance_id, backend, operation, status` | Number of IO operations |
| `blockx_task_io_backend_duration_milliseconds` | histogram | `instance_id, backend, operation, stat` | Per-task IO backend latency avg/p90/p99 |
| `blockx_task_io_cache_hits_total` / `blockx_task_io_singleflight_hits_total` | counter | `instance_id, backend` | Cache / singleflight hits |
| `blockx_task_block_latency_milliseconds` / `blockx_task_upstream_latency_milliseconds` | histogram | `instance_id` | Latency from the block timestamp / upstream table completion to task convergence |
| `blockx_task_errors_total` | counter | `instance_id, component, error_code` | Categorized task-level errors |
| `blockx_worker_active_tasks` | gauge | — | Number of tasks currently retained by WorkerCore |
| `blockx_worker_dispatcher_calls` | gauge | `state` | Number of ready/running/waiting calls in the dispatcher |
| `blockx_worker_executor_calls` | gauge | `state` | Internal call state counts reported by executor heartbeats |
| `blockx_worker_executor_cpu_cores` / `blockx_worker_executor_cpu_seconds_total` | gauge / counter | — | Executor process CPU usage |
| `blockx_worker_logical_calls_terminal_total` | counter | `kind, outcome` | Logical call terminal-state count |
| `blockx_worker_stream_productions_active` / `blockx_worker_stream_credit_waiters` | gauge | — | Streaming call production and backpressure |
| `blockx_io_adaptive_limit` / `blockx_io_adaptive_pressure` | gauge | `backend, kind` / `backend, state` | Current adaptive concurrency limit and pressure |
| `blockx_syncinvoke_calls_total` / `blockx_syncinvoke_call_duration_milliseconds` | counter / histogram | `status, failure_code` / `status` | Sync Invoker call volume and latency |
| `blockx_usage_publish_failures_total` / `blockx_usage_pending_records` | counter / gauge | `service, resource_type` | usage Kafka publish health; alerting on `increase(...[1m]) > 0` is recommended |
| `blockx_coordinator_registry_workers` | gauge | `format` | Number of workers the Coordinator discovers per registry format |
## Tracing
* `obs.InitTracing(ctx, service)` always sets the W3C TraceContext propagator. Exporter selection: `BLOCKX_OTEL_SPAN_FILE` takes priority (JSON-lines appended to a file, shared across processes) → `OTEL_EXPORTER_OTLP_ENDPOINT` (OTLP/gRPC, insecure) → noop when neither is set. `obs.TracingEnabled()` lets hot paths skip `tracer.Start`.
* Sampling: `AlwaysSample` by default; a value such as `OTEL_TRACES_SAMPLER_ARG=0.1` switches to `ParentBased(TraceIDRatioBased)`. Production must set this explicitly.
* The service name is passed in by the entry point: `blockx-worker`, `blockx-bundle-worker`, `blockx-coordinator`, `blockx-bundle-coordinator`, `blockx-syncinvoker` (`Profile.Service` in `cmd/*/main.go`); on the Python side it is `blockx-executor`.
* Automatic spans: Coordinator→Worker gRPC uses `otelgrpc.NewClientHandler()`/`NewServerHandler()` (`internal/coordinator/adapters/worker_rpc.go`, `internal/worker/app/app.go`).
* Manual spans: Worker↔Executor UDS uses the `udsTracer` in `internal/worker/adapters/executor/send.go` (tracer name `github.com/Chaintable/blockx/worker/uds`), with span names `uds.send.`/`uds.recv.`/`worker.io.dispatch`; trace context crosses UDS via the `MessageEnvelope.traceContext` field. The Sync Invoker uses `obs.Tracer("blockx-syncinvoker")`.
* `BLOCKX_OTEL_UDS_EVENTS=1` enables the fine-grained UDS pipeline events from `obs.AddUDSEvent`/`AddUDSEventAt`; for perf debugging only.
* `analyze_spans.py` reads the span file, normalizes the two JSON formats produced by Go's `stdouttrace` and Python's `ConsoleSpanExporter`, and prints the span tree and event offsets per trace. Typical usage:
```bash theme={null}
: > /tmp/blockx_spans.jsonl
BLOCKX_OTEL_SPAN_FILE=/tmp/blockx_spans.jsonl BLOCKX_OTEL_UDS_EVENTS=1 \
go test -v -count=1 -run TestWorkerProcess_RpcWait ./cmd/worker/...
python3 analyze_spans.py /tmp/blockx_spans.jsonl
```
See `docs/specs/tracing.md` §4 in the blockx repo for the complete span/event list.
## Usage collection
* What is collected: when each task converges, `taskRuntime.snapshotUsage()` (`internal/worker/adapters/task_runtime.go`) converts the effective execution time reported by the executor (`callEffectiveSum`, in milliseconds) into `usage.Sample{CPUTimeMicros}`. This excludes executor queueing, IO waits, and the Go-side CPU spent in the Builder/Result Handler.
* Attribution: `Collector.Record(clientID, sample)` accumulates into `active` by full `client_id`; an empty ID is recorded as `unknown`; samples with `CPUTimeMicros<=0` are discarded.
* Where it goes: by default `Flush` runs every 5s, freezing `active` into `[]usage.Record` (`id` UUID, `client_id`, `service`, `resource_type`, `usage` in milliseconds, `timestamp` at freeze time) and writing synchronously to the topic `chaintable-usage` via `KafkaSink.Publish` (key is `client_id`, `RequiredAcks=RequireAll`, `MaxAttempts=1`). On failure the records stay in `pending` and are retried in the next round with the same `id`; consumers deduplicate by `id`.
* Service enumeration: `usage.ServiceBlockXWorker = "blockx_worker"` (`cmd/worker`), `usage.ServiceBlockXBundleWorker = "blockx_bundle_worker"` (`cmd/bundle_worker`); `resource_type` is fixed to `usage.ResourceTypeCompute = "compute"`. These are bound by `Profile.UsageService` and cannot be changed through configuration. The Sync Invoker does not create a collector.
* Fail-closed: unless explicitly `Disabled`, `NewKafkaCollector` requires non-empty brokers and a topic with partitions; if the probe fails, the process refuses to start.
## Client id propagation
| Location | What it does |
| -------------------- | ------------------------------------------------------------------------------------------------------------------------ |
| Inbound gRPC handler | `clientid.WithContext(ctx, clientid.FromIncomingGRPC(ctx))`: `client-id` > `x-instance-id` > `unknown` |
| In-process | `clientid.FromContext(ctx)` returns the full value; `obs.InstanceIDFromCtx` is equivalent (returns `unknown` when empty) |
| Logs/metrics/chlog | `clientid.LogKey(id)`: `instance:` → ``; values without the prefix are returned unchanged |
| Outbound gRPC | `clientid.IntoOutgoingGRPC(ctx)`: copies the metadata and overwrites both headers |
| Outbound HTTP | `clientid.IntoHTTP(req)`: same as above, taking the value from `req.Context()` |
| Shadow submit | `internal/worker/adapters/shadow_forwarder.go` rewrites the ID to `-shadow` before calling `IntoOutgoingGRPC` |
## State and lifecycle
* **usage Collector**: `Start` launches the ticker goroutine; `Close` sets the closing barrier (after which `Record` is rejected), stops the ticker, retries the final `Flush` every 200ms within the shutdown ctx provided by the caller, and then calls `sink.Close`. An error returned from `Close` is treated as a billing-critical error and the Worker exits non-zero.
* **Prometheus series TTL**: once an `instance_id`/`function_id` label value has been idle for longer than `BLOCKX_METRICS_SERIES_TTL_MS` (default 24h), the reaper (default every 10min, `BLOCKX_METRICS_SERIES_SWEEP_INTERVAL_MS`) deletes it with `DeletePartialMatch`, at most 256 values per axis per round. An explicit non-positive value disables it. The reaper starts with `StartPrometheusEndpoint` and runs even when no scrape address is configured.
* **Table read/write log deduplication**: 10-minute window for `table_read`, 1-minute window for `table_write`, 64 shards with at most 2048 keys per shard.
* **tracing / UDS events switches**: environment variables are read once at process startup; changing them at runtime has no effect.
## Monitoring endpoints
| Endpoint | Process | Description |
| --------------------------------------------------------------------- | ------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- |
| `GET /metrics` (path can be overridden with `BLOCKX_PROMETHEUS_PATH`) | worker, bundle worker, coordinator, bundle coordinator, syncinvoker | Served by `obs.StartPrometheusEndpoint` on `BLOCKX_PROMETHEUS_LISTEN_ADDR`; does not listen when the address is empty |
| `GET /debug/pprof/` | worker, bundle worker | `workerDebugHTTPHandlers` in `internal/worker/app/app.go` mounts it on the same server; `WORKER_DEBUG_PPROF=false` disables it, enabled by default |
| `GET /readyz`, `GET /healthz` | syncinvoker | `cmd/syncinvoker/health.go`; `/readyz` returns 503 when dependencies are not ready or while draining |
Worker and Coordinator currently have no HTTP health probes and no gRPC health service.
## Configuration
Everything is configured through environment variables (the Worker also accepts the corresponding fields in its JSON config file; see `Prometheus`/`Chlog`/`Usage` in `internal/worker/app/config.go`):
| Environment variable | Default | Effect | Defined in |
| --------------------------------------------------------------------------------------- | ------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------- |
| `BLOCKX_LOG_LEVEL` | `info` | `debug`/`info`/`warn`/`error`; invalid values fall back to `info`; shared by Go and Python | `internal/obs/logger.go` |
| `BLOCKX_LOG_FORMAT` | `json` | `json`/`text` | Same as above |
| `BLOCKX_PROMETHEUS_LISTEN_ADDR` | empty (not listening) | `/metrics` listen address, e.g. `:9090` | `internal/obs/metrics.go` |
| `BLOCKX_PROMETHEUS_PATH` | `/metrics` | Scrape path | Same as above |
| `BLOCKX_METRICS_SERIES_TTL_MS` | `86400000` | Idle reclamation TTL for high-cardinality series; non-positive disables it | `internal/obs/metrics_reaper.go` |
| `BLOCKX_METRICS_SERIES_SWEEP_INTERVAL_MS` | `600000` | Sweep interval | Same as above |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | empty (noop) | OTLP/gRPC address, may carry an `http://` prefix | `internal/obs/tracing.go` |
| `BLOCKX_OTEL_SPAN_FILE` | empty | JSON-lines span file; mutually exclusive with OTLP, the file takes priority | Same as above |
| `OTEL_TRACES_SAMPLER_ARG` | empty (sample everything) | Head sampling ratio 0.0–1.0 | Same as above |
| `BLOCKX_OTEL_UDS_EVENTS` | off | `1` enables UDS pipeline events | Same as above |
| `CHAINTABLE_LOG_BROKERS` | empty (disabled) | chaintable-log Kafka brokers, comma-separated | `internal/obs/chlog_bridge.go` |
| `CHAINTABLE_LOG_TOPIC` | `instance-logs` | chaintable-log topic | Same as above |
| `BLOCKX_USAGE_DISABLED` | `false` | Explicitly disables billing; when disabled, the remaining usage validation is skipped | `internal/usage/collector.go` |
| `BLOCKX_USAGE_BROKERS` | empty (startup fails unless disabled) | usage Kafka brokers | Same as above |
| `BLOCKX_USAGE_TOPIC` | `chaintable-usage` | usage topic | Same as above |
| `BLOCKX_USAGE_FLUSH_INTERVAL_MS` | `5000` | Flush interval, valid range `(0, 5000]` | Same as above |
| `WORKER_DEBUG_PPROF` | `true` | Whether to mount `/debug/pprof/` on the metrics server | `internal/worker/app/app.go` |
| `WORKER_CPU_PROFILE` / `WORKER_TRACE` / `WORKER_MUTEX_PROFILE` / `WORKER_BLOCK_PROFILE` | empty | Write a pprof / runtime trace to a file for the lifetime of the process; the coordinator equivalents are `COORDINATOR_CPU_PROFILE` / `COORDINATOR_TRACE` | `internal/worker/app/app.go`, `internal/coordinator/app/app.go` |
For the complete deployment-side variable list, see `docs/deploy.md` §9 in the blockx repo and the [Deployment overview](/en/development/deployment) on this site.
## Performance analysis
* Methodology: `docs/specs/perf-methodology.md` in the blockx repo (the model → measure → compare → attribute → decide loop); reproduction guide in `docs/specs/perf-test-how-to-v2.md`; conclusions in `perf-test-report-v2.md`.
* Test cases: `e2e/perf/` is organized by S0–S5 stage prefixes (`s0_framework_*`, `s1_builder_*`, `s2_call_count_*`, `s3_cpu_*`, `s4_io_*`, `s5_plugin_*`), indexed in `e2e/perf/README.md`. Artifacts land in `e2e/perf/_artifacts//spans.jsonl` and `pyspy/`.
* How to run: `./test.sh -run TestPerf_XXX` runs inside a cgroup (default 4 vCPU / 8 GB); running `go test ./e2e/perf/...` directly is the unconstrained baseline. Subtests with `with_trace` set `BLOCKX_OTEL_SPAN_FILE` automatically.
* Online load testing: `go run ./cmd/perf -coordinator host:port -n 100 -c 4 -mode full|meta|reserve`, which outputs P50/P90/P99.
* Process-level attribution: write `WORKER_CPU_PROFILE`/`WORKER_TRACE` to a file, or capture `/debug/pprof/profile`.
See [Testing](/en/development/testing) for more.
## Extension points
* **Adding a log field or `type`**: add an `Attr*`/`LogType*` constant in `internal/obs/logger.go`; if the type is user-visible, update `docs/specs/blockx-log-types.md` as well.
* **Adding a Prometheus metric**: declare it with `promauto` in `internal/obs/metrics.go` (or the `metrics_*.go` of the corresponding subsystem) and add a `Record*` function; labels may only be closed enumerations plus `instance_id`/`function_id`; vecs carrying either of these two labels must be added to `instanceIDVecs`/`functionIDVecs` in `metrics_reaper.go`. `instance_id` values must go through `metricInstanceID`.
* **Adding adaptive metrics for a new IO backend**: register through `internal/io/adaptive/observed`, which calls `obs.RegisterIOAdaptiveMetrics(backend, provider)`.
* **Adding spans**: only in the adapter layer; first consider whether an event (`obs.AddUDSEvent`) can express it; across goroutines, use a child span rather than adding events to a parent span that has already ended.
* **Adding an entry point that needs billing**: pass a new `usage.Service*` constant via `Profile.UsageService` in `cmd//main.go`; `service`/`resource_type` may not be overridden from configuration.
* **Adding a downstream adapter**: call `clientid.IntoOutgoingGRPC(ctx)` or `clientid.IntoHTTP(req)` before sending; do not read the metadata yourself.
## Testing
```bash theme={null}
go test ./internal/obs/... ./internal/usage/... ./internal/common/clientid/...
```
* `internal/obs/handler_test.go`: slogtest compliance of `ContextHandler`, field injection, `WithAttrs`/`WithGroup` preserving injection, `LogKey` sanitization.
* `internal/obs/chlog_bridge_test.go`: bridge double-write, `type` kept out of data, group prefixes, `trace_id` sanitization.
* `internal/obs/metrics_test.go`, `metrics_function_test.go`, `stream_metrics_test.go`, `metrics_syncinvoke_test.go`, `metrics_io_adaptive_test.go`: the labels and values written by each `Record*` function.
* `internal/obs/metrics_reaper_test.go`, `metrics_reaper_consistency_test.go`: TTL reaper behavior; `TestReaperVecListsMatchDeclarations` parses the source to ensure every vec with `instance_id`/`function_id` is registered in the reaper lists.
* `internal/obs/table_dedupe_test.go`: deduplication windows and shard caps.
* `internal/usage/collector_test.go`, `metrics_test.go`: closing barrier, Kafka topic probing, publish failure metrics.
* `internal/common/clientid/client_id_test.go`: header priority, double-write, `LogKey`.
* Record-point tests live in `internal/worker/adapters/call_metrics_test.go`, `executor_metrics_test.go`, and `task_barrier_metrics_test.go`; process-level log schema assertions are in `cmd/worker/worker_process_logging_test.go`.
## Related docs
Specs in the blockx repo:
* `docs/specs/logging.md` — log schema, field table, Go/Python call contract.
* `docs/specs/blockx-log-types.md` — user-facing list of `type` logs, `task_finished` field definitions, how to derive metrics from logs.
* `docs/specs/tracing.md` — how to enable it, complete span/event list, `analyze_spans.py` usage, troubleshooting table.
* `docs/specs/client-id-propagation.md` — header priority, outbound double-write, log/monitoring sanitization rules.
* `docs/specs/2026-07-13-blockx-usage-collection.md` — usage message format, CPU accounting definition, flush and failure semantics.
* `docs/specs/perf-methodology.md`, `docs/specs/perf-test-how-to-v2.md` — performance testing methodology and reproduction.
* `docs/deploy.md` §6, §9 — deployment-side observability configuration and environment variables.
Pages on this site: [Worker](/en/components/worker), [Call execution subsystem](/en/components/call-execution), [IO access subsystem](/en/components/io-subsystem), [Sync Invoker](/en/components/sync-invoker), [Deployment overview](/en/development/deployment), [Testing](/en/development/testing).