iop/apps/node/internal/node/liveness_observability.go

package node

import (
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"

	runtime "iop/packages/go/execution"
)

// nodeLivenessObserver emits bounded, operator-queryable evidence for every
// exactly-once claimed stall on either execution path. It is process-global in
// production so repeated Node construction never re-registers metric names,
// and it is test-injectable so package tests can verify the closed label set
// and the safe log contract without touching the default prometheus registerer.
//
// The observer never changes stall detection, fence/probe ordering, terminal
// delivery, or request/session/raw prompt/response handling. Observer failure
// or disabled logging cannot suppress the terminal.
type nodeLivenessObserver struct {
	stalls   *prometheus.CounterVec
	duration *prometheus.HistogramVec
	logger   *zap.Logger
}

// productionStalls is the process-global counter registered once against the
// default Prometheus registerer. Every Node reuses this single instance.
var productionStalls *prometheus.CounterVec

// productionDuration is the process-global histogram registered once against
// the default Prometheus registerer. Every Node reuses this single instance.
var productionDuration *prometheus.HistogramVec

// init registers the production collector set exactly once with the default
// Prometheus registerer. Per-Node construction never calls promauto or
// MustRegister; test constructors supply an isolated registerer instead.
func init() {
	productionStalls = prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "iop",
		Subsystem: "node",
		Name:      "response_stalls_total",
		Help:      "Total claimed response stalls grouped by execution path, provider health, liveness classification, and attempt fence.",
	}, []string{"execution_path", "provider_health", "liveness_classification", "attempt_fence"})
	prometheus.MustRegister(productionStalls)

	productionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Namespace:                       "iop",
		Subsystem:                       "node",
		Name:                            "response_stall_duration_seconds",
		Help:                            "Idle duration in seconds for every claimed response stall.",
		Buckets:                         prometheus.ExponentialBuckets(0.05, 2, 10),
		NativeHistogramBucketFactor:     1.1,
		NativeHistogramMaxBucketNumber:  100,
		NativeHistogramMinResetDuration: 1 << 60,
	}, []string{"execution_path", "provider_health", "liveness_classification", "attempt_fence"})
	prometheus.MustRegister(productionDuration)
}

// newProductionNodeLivenessObserver returns the shared production observer.
// Tests must not call this; they call newNodeLivenessObserverForTest
// with a private prometheus.Registry to avoid polluting the default registerer.
func newProductionNodeLivenessObserver(logger *zap.Logger) *nodeLivenessObserver {
	return &nodeLivenessObserver{
		stalls:   productionStalls,
		duration: productionDuration,
		logger:   logger,
	}
}

// newNodeLivenessObserverForTest returns an observer backed by a private
// prometheus.Registry. The returned observer's Stalls and Duration fields
// expose the underlying collectors so tests can inspect gathered metrics
// without touching the process-wide default registerer.
func newNodeLivenessObserverForTest(logger *zap.Logger, reg prometheus.Registerer) *nodeLivenessObserver {
	stalls := prometheus.NewCounterVec(prometheus.CounterOpts{
		Namespace: "iop",
		Subsystem: "node",
		Name:      "response_stalls_total",
		Help:      "Total claimed response stalls grouped by execution path, provider health, liveness classification, and attempt fence.",
	}, []string{"execution_path", "provider_health", "liveness_classification", "attempt_fence"})
	reg.MustRegister(stalls)

	duration := prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Namespace:                       "iop",
		Subsystem:                       "node",
		Name:                            "response_stall_duration_seconds",
		Help:                            "Idle duration in seconds for every claimed response stall.",
		Buckets:                         prometheus.ExponentialBuckets(0.05, 2, 10),
		NativeHistogramBucketFactor:     1.1,
		NativeHistogramMaxBucketNumber:  100,
		NativeHistogramMinResetDuration: 1 << 60,
	}, []string{"execution_path", "provider_health", "liveness_classification", "attempt_fence"})
	reg.MustRegister(duration)

	return &nodeLivenessObserver{
		stalls:   stalls,
		duration: duration,
		logger:   logger,
	}
}

// executionPathAllowlist enumerates the only values the observer accepts for
// the execution_path label. Anything else is normalized to "unknown".
var executionPathAllowlist = map[string]struct{}{
	"normalized":      {},
	"provider_tunnel": {},
}

// healthAllowlist enumerates the only values the observer accepts for the
// provider_health label. Anything else is normalized to "unknown".
var healthAllowlist = map[runtime.ProviderStatus]runtime.ProviderStatus{
	runtime.ProviderStatusAvailable:   runtime.ProviderStatusAvailable,
	runtime.ProviderStatusUnavailable: runtime.ProviderStatusUnavailable,
}

// classificationAllowlist enumerates the only values the observer accepts for
// the liveness_classification label. Anything else is normalized to "health_unknown".
var classificationAllowlist = map[runtime.ProviderHealth]runtime.ProviderHealth{
	runtime.RequestStalled:    runtime.RequestStalled,
	runtime.ProviderUnhealthy: runtime.ProviderUnhealthy,
}

// fenceAllowlist enumerates the only values the observer accepts for the
// attempt_fence label. Anything else is normalized to "unknown".
var fenceAllowlist = map[string]struct{}{
	"confirmed":   {},
	"unconfirmed": {},
}

// normalizeNodeLivenessLabels returns the closed four-tuple of label values
// for the counter, histogram, and dedicated structured log. Every value is
// validated against its allowlist; anything outside is normalized to "unknown"
// so a future classification or status never leaks an unbounded cardinality
// into the metric series.
func normalizeNodeLivenessLabels(executionPath string, obs stallObservation) [4]string {
	var path string
	if _, ok := executionPathAllowlist[executionPath]; ok {
		path = executionPath
	} else {
		path = "unknown"
	}

	health := runtime.ProviderStatusUnknown
	if v, ok := healthAllowlist[obs.health.Status]; ok {
		health = v
	}

	classification := runtime.HealthUnknown
	if v, ok := classificationAllowlist[obs.health.Health]; ok {
		classification = v
	}

	var fence string
	if _, ok := fenceAllowlist[obs.fence]; ok {
		fence = obs.fence
	} else {
		fence = "unknown"
	}

	return [4]string{path, string(health), string(classification), fence}
}

// Observe emits one counter observation, one duration sample, and one
// structured log entry for the given claimed stall. It is invoked exactly
// once per claimed stall from the production watchdog seams.
//
// Observer failure never suppresses the terminal: metrics and logs are
// fire-and-forget evidence; the terminal is the delivery contract.
func (o *nodeLivenessObserver) Observe(executionPath string, obs stallObservation) {
	if o == nil {
		return
	}
	defer func() { _ = recover() }()

	labels := normalizeNodeLivenessLabels(executionPath, obs)

	o.stalls.WithLabelValues(labels[0], labels[1], labels[2], labels[3]).Inc()
	o.duration.WithLabelValues(labels[0], labels[1], labels[2], labels[3]).Observe(obs.idle.Seconds())

	if o.logger == nil {
		return
	}

	o.logger.Info(
		"node_response_stall_observation",
		zap.String("execution_path", labels[0]),
		zap.String("provider_health", labels[1]),
		zap.String("liveness_classification", labels[2]),
		zap.String("attempt_fence", labels[3]),
		zap.Int64("idle_duration_ms", obs.idle.Milliseconds()),
	)
}