FluxFlow.Components.Observability 3.0.0

FluxFlow.Components.Observability

Standalone observer nodes for FluxFlow — counter, logger, and metrics. Each depends only on FluxFlow.Nodes (and FluxFlow.Mapping for the counter's optional predicate) — no engine, registry, or runtime. You new the node and LinkTo the next one.

Nodes

Every message travels as a FlowMessage<T> envelope. Each node broadcasts its result on Output carrying the same correlation id as the input; failures surface on Errors (with the input's correlation id and a Code from ObservabilityErrorCodes) and diagnostics flow on Events. The package emits neutral contracts only — hosts decide whether those entries or snapshots become app logs, dashboards, files, telemetry, or test assertions.

Counter

await using var node = new FlowCounterNode<MyMessage>(
    new FlowCounterOptions { InputType = "message", Name = "received", Predicate = "value.Enabled" },
    expressionEngine);

node.Output.LinkTo(snapshotSink, new DataflowLinkOptions { PropagateCompletion = false });
await node.Input.SendAsync(FlowMessage.Create(message));

FlowCounterNode emits FlowCounterSnapshot values with count, rejected count, last observed timestamp, name, and input type. When a predicate is configured it is compiled once at construction from the supplied IFlowExpressionEngine; inputs the predicate rejects are not counted (but are tallied in RejectedCount). With no predicate every input is counted and no engine is required. Pass an IFlowMapContextFactory<TInput> to control the variables the predicate sees (defaults to input/value).

Logger

await using var node = new FlowLoggerNode<MyMessage>(
    new FlowLoggerOptions
    {
        InputType = "message",
        Level = "Information",
        Category = "workflow",
        MessageTemplate = "Observed {kind} item #{sequence}"
    },
    attributeSelectors: new Dictionary<string, IObservabilityValueSelector<MyMessage>>
    {
        ["kind"] = new KindSelector()
    });

FlowLoggerNode emits FlowLogEntry values. Supply IObservabilityValueSelector<TInput> selectors keyed by attribute name to enrich each entry; an attribute-selector failure is reported on Errors, the offending attribute is skipped, and the entry is still emitted. An unsupported Level throws InvalidOperationException at construction.

Metrics

await using var node = new FlowMetricsNode<MyMessage>(
    new FlowMetricsOptions { InputType = "message", Name = "received", SizeSelector = "payloadBytes" },
    sizeSelector: new PayloadSizeSelector());

FlowMetricsNode emits FlowMetricSnapshot values with total count, current rate, average rate, last observed timestamp, and optional size values. The optional size selector can return numeric values, strings, byte arrays, or collections; a size-selector failure is reported on Errors and the node keeps processing.

Runtime timing

Snapshots and log entries use the node's clock for Timestamp (default TimeProvider.System). Provide a deterministic clock for tests:

new FlowMetricsNode<MyMessage>(options, sizeSelector, clock: new FakeTimeProvider(timestamp));

A time provider also makes counter/metrics rate calculations deterministic.