correlator

Documentation

Overview

Package correlator builds and refreshes a service-to-neighbors adjacency map from sampled trace topology. The incident auto-attach path consults this map so a downstream-failure incident on payment-service and an upstream-saturation incident on api-gateway end up grouped instead of creating two separate incidents that page the oncall twice for the same outage.

Design:

• One background goroutine refreshes the map every 5 min from the topology_edges_5m MV. The map is bounded by however many service→service edges the MV holds (LIMIT 10000), so memory is small.
• Lookups are read-locked and return copies so callers don't race the next refresh.
• When the source returns nothing (cold start, no traffic yet), the previous graph is preserved — never replace with empty, since that would break correlation during a quiet window.

v0.8.67 (Faz 5) — the graph is now DIRECTED and WEIGHTED. Through Faz 4 the correlator kept a single symmetric set (svc → neighbour set): enough to answer "are A and B topologically close?" but not "which of payment-service's downstream deps carries the error traffic?". Now two maps are kept — `out` (caller → downstream callees) and `in` (callee → upstream callers) — each edge tagged with calls/errors/duration. The legacy Neighbors / AreNeighbors API is preserved exactly (union of in ∪ out), so the incident auto-attach path and the fusion evidence bundle are unchanged.

Index

• func Synthesize(anchorKind, anchorID, service string, computedAtNs int64, in SynthesisInput) chstore.RootCauseHypothesis
• type Correlator
  • func New(store *chstore.Store) *Correlator
  • func (c *Correlator) AreNeighbors(a, b string) bool
  • func (c *Correlator) Downstream(svc string) []WeightedEdge
  • func (c *Correlator) Edge(a, b string) (EdgeStat, bool)
  • func (c *Correlator) Neighbors(svc string) []string
  • func (c *Correlator) RootCauseRank(svc string) []ScoredCause
  • func (c *Correlator) Start(ctx context.Context)
  • func (c *Correlator) UpdatedAt() time.Time
  • func (c *Correlator) Upstream(svc string) []WeightedEdge
• type EdgeStat
  • func (e EdgeStat) ErrorRate() float64
• type ScoredCause
  • func RankRootCausesFromEdges(edges []chstore.ServiceEdgePair, trigger string) []ScoredCause
• type SignalEvidence
• type SynthesisInput
• type WeightedEdge

Functions

func Synthesize

func Synthesize(
	anchorKind, anchorID, service string,
	computedAtNs int64,
	in SynthesisInput,
) chstore.RootCauseHypothesis

Synthesize fuses the evidence into ONE ranked, confidence-weighted hypothesis. anchorKind/anchorID/service stamp the anchor; computedAtNs stamps when (the worker passes a single now() so a batch tick shares one timestamp). Returns a chstore.RootCauseHypothesis ready to UpsertHypothesis — the worker does no further shaping.

Determinism: candidates are built tier-by-tier (deploy, then propagation in the order RankRootCausesFromEdges already total-orders, then co-firing sorted by name), then a STABLE sort by score desc with a Service-name tie-break, so identical evidence always yields byte-identical output (the table-driven test asserts the tie-break).

Types

type Correlator

type Correlator struct {
	// contains filtered or unexported fields
}

Correlator surfaces service topology relationships — 1-hop only (direct caller or callee). Two-hop would catch transitive dependencies but is deferred to Faz 6 (decayed 2-hop); at 1 hop the recall is fine and false-positives are the bigger risk.

func New

func New(store *chstore.Store) *Correlator

func (*Correlator) AreNeighbors

func (c *Correlator) AreNeighbors(a, b string) bool

AreNeighbors reports whether a and b are within 1 hop (in either direction) in the last refreshed graph. Self-pair returns true so the incident-attach call site can use this as the single "consider these problems related" predicate.

func (*Correlator) Downstream

func (c *Correlator) Downstream(svc string) []WeightedEdge

Downstream returns svc's direct callees (the deps it calls), each with its edge weight, sorted by error-carrying volume: errors desc, then calls desc, then name asc for a stable order. This is the "which downstream dep most likely caused my failure" ranking the root-cause panel (and Faz 6's conditional probability) build on.

func (*Correlator) Edge

func (c *Correlator) Edge(a, b string) (EdgeStat, bool)

Edge returns the weight of the directed edge a → b (a calls b) and whether such an edge exists in the last refresh.

func (*Correlator) Neighbors

func (c *Correlator) Neighbors(svc string) []string

Neighbors returns a copy of the 1-hop neighbour set for svc (callers ∪ callees), or nil when svc is unknown / a leaf. Order is undefined (map iteration). Preserved verbatim from Faz 4 so the incident auto-attach path is unchanged by the directed refactor.

func (*Correlator) RootCauseRank

func (c *Correlator) RootCauseRank(svc string) []ScoredCause

RootCauseRank scores the live graph's downstream candidates for svc, best first. Read-locked snapshot — safe to call concurrently with the refresh loop.

func (*Correlator) Start

func (c *Correlator) Start(ctx context.Context)

Start launches the refresh loop. Runs an immediate refresh so the first incident attach after boot has data, then re-runs every 5 minutes. Returns when ctx is cancelled.

func (*Correlator) UpdatedAt

func (c *Correlator) UpdatedAt() time.Time

UpdatedAt reports the last successful refresh — surfaced on /api/admin/system-stats so the operator can see correlation is live.

func (*Correlator) Upstream

func (c *Correlator) Upstream(svc string) []WeightedEdge

Upstream returns svc's direct callers (the deps that call it), each with its edge weight, sorted by the same error-first key. This is the "which upstream am I taking down" / blast-radius direction.

type EdgeStat

type EdgeStat struct {
	Calls         uint64
	Errors        uint64
	SumDurationNs uint64
}

EdgeStat is the weight carried by one directed edge over the refresh window.

func (EdgeStat) ErrorRate

func (e EdgeStat) ErrorRate() float64

ErrorRate returns the fraction of calls on this edge that failed, in [0,1]. Zero calls → 0 (no traffic means no observed failure signal, not a divide-by-zero).

type ScoredCause

type ScoredCause struct {
	Service string
	Score   float64
	Hops    int
	Path    []string
}

ScoredCause is one root-cause candidate for a triggering service: the suspect dependency, its propagation score in [0,1], the hop distance, and the path it was reached by (trigger … candidate).

func RankRootCausesFromEdges

func RankRootCausesFromEdges(edges []chstore.ServiceEdgePair, trigger string) []ScoredCause

RankRootCausesFromEdges scores a one-off weighted edge list without a live Correlator — the anomaly fusion path uses this so its evidence bundle stays a pure, store-free assembly (it already holds the edges). Reuses buildGraph so the directed-edge construction has one home.

type SignalEvidence

type SignalEvidence struct {
	Kind    string  // "log_pattern" | "trace_op"
	Pattern string  // pattern name (logs) or operation name (trace ops)
	Ratio   float64 // max(current, peak) over baseline; 0 for new-template events
}

SignalEvidence is one active anomaly signal on the anchor service.

type SynthesisInput

type SynthesisInput struct {
	// Deploy — a same-service deploy that landed in the lookback window before
	// onset, or nil. FreshnessFrac ∈ [0,1] says how close to onset it landed
	// (1 = right at onset, 0 = at the far edge of the lookback window); the
	// worker computes it from (onset - deployTime) / lookback.
	Deploy        *chstore.RecentDeploy
	FreshnessFrac float64
	// Neighbours — the propagation-ranked downstream suspects (best first),
	// straight from RankRootCausesFromEdges. Score ∈ [0,1] is the error-share,
	// hop-decayed. Empty when nothing downstream carries error volume.
	Neighbours []ScoredCause
	// CoFiringServices — distinct OTHER services with an open problem co-firing
	// on the SAME anchor service's incident. In practice these are same-service
	// (so usually the anchor service itself) — the worker passes the service
	// label per co-firing problem so the candidate carries a name. Deduped +
	// sorted by the fuser for determinism.
	CoFiringServices []string
	// Signals — active log_pattern / trace_op anomalies on the anchor's own
	// service (v0.8.571). Collected into EvidenceBundle since v0.8.3xx but
	// never threaded here — the hypothesis was blind to the one evidence
	// type that carries an error SIGNATURE. Minimal shape on purpose: the
	// pure fuser takes only what it scores.
	Signals []SignalEvidence
}

SynthesisInput is the evidence the fuser ranks. The worker fills this by destructuring its anomaly.EvidenceBundle (deploy + neighbours + co-firing) for the anchor's service — Synthesize itself touches no store.

type WeightedEdge

type WeightedEdge struct {
	Service string
	EdgeStat
}

WeightedEdge is a neighbour service paired with the edge weight to it. Direction is implied by the accessor that produced it (Downstream → the service is a callee; Upstream → a caller).