Observability

​

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.
• 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

​

Core types and interfaces

// 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

​

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)

​

chaintable-log bridge

​

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

​

Most important metrics

​

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.<MsgType>/uds.recv.<MsgType>/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:

: > /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

​

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

​

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

​

Configuration

​

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/<TestName>/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.

​

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/<entry>/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

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

• 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.