Help us shape the future of Go! Take the 2026 Go Developer Survey

topology

package
v0.6.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 21, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Overview

Package topology owns the mode-selected service-topology read contract.

Index

Constants

View Source 
const (
	// MaxHostsPerNode caps the hosts list stamped on one service-map node.
	MaxHostsPerNode = 20
	// MaxServicesPerHost caps the services list carried by one host.
	MaxServicesPerHost = 100
)

Host projection bounds (#288). The registry already bounds hosts and service-host pairs per tenant; these bound what one response carries.

View Source 
const (
	KindService = "service"
	KindHost    = "host"
)

Node kinds. A host entity (IsHostEntity) is rendered as a host by every consumer: it is never a service, never carries a CALLS edge and never enters a service count.

View Source 
const (
	GraphWindow        = 5 * time.Minute
	GraphFlushInterval = 5 * time.Second
)
View Source 
const (
	DropInvalidIdentity = "invalid_identity" // empty tenant/service or over-length value
	DropOutOfWindow     = "out_of_window"    // older than retention or more than 2m in the future
	DropUnknownEntity   = "unknown_entity"   // span for a resource that was never admitted
	DropInvalidSpan     = "invalid_span"     // ID length or duration out of range
	DropEntityCap       = "entity_cap"       // graphMaxEntities reached
	DropChildrenCap     = "children_cap"     // graphMaxChildren waiting on one parent
	DropPendingCap      = "pending_cap"      // graphMaxPending unflushed buckets
	UnmatchedExpired    = "expired"          // parent did not arrive within graphSpanTTL
	UnmatchedEvicted    = "evicted"          // span buffer full; child evicted before its parent arrived
	UnmatchedEndpoint   = "unknown_endpoint" // an endpoint entity was pruned before the join
)

Loss reasons, reported per reason in the graph metadata and on otelcontext_instance_graph_lost_total{kind,reason}.

View Source 
const (
	// RegistryMaxHostsPerTenant bounds distinct non-empty hosts per tenant.
	RegistryMaxHostsPerTenant = 1000
	// RegistryMaxEntriesPerTenant bounds service-host(-workload) entries per tenant.
	RegistryMaxEntriesPerTenant = 10000
	// RegistryIdleTTL evicts an entry no signal has touched for this long.
	RegistryIdleTTL = 24 * time.Hour
	// RegistryMaxValueBytes bounds service, host and workload values; the
	// persisted columns are 255 bytes wide on every adapter and an over-length
	// value would poison every later flush of the same batch.
	RegistryMaxValueBytes = 255
	// RegistryFlushInterval is the dirty-tick period used by Run.
	RegistryFlushInterval = 30 * time.Second
)

Resource registry bounds (#279). Constants, mirroring the SignalStore caps: a shared __other__ host would be exactly the cross-host merge the registry exists to avoid, so overflow is counted and dropped, never collapsed.

View Source 
const (
	RegistryKindHost   = "host"
	RegistryKindPair   = "pair"
	RegistryKindLength = "length"
)

Overflow kinds, also the metric label values.

View Source 
const HostPrefix = "host/"

HostPrefix is the reserved service-name namespace of host entities (#280): a resource carrying host.id or host.name and no service.name is identified as HostPrefix + host.name (host.id when there is no name). The registry schema is unchanged; the entity kind is derived from the name.

Variables

This section is empty.

Functions

func IsHostEntity 

func IsHostEntity(service string) bool

IsHostEntity reports whether service names a host entity.

func NodeKind 

func NodeKind(service string) string

NodeKind returns the kind of the named node.

Types

type AggregateProvider 

type AggregateProvider struct {
	HostReader
	// contains filtered or unexported fields
}

AggregateProvider is both the ordinary mode-selected provider and the narrow aggregate projection source GraphRAG already consumes.

func NewAggregateProvider 

func NewAggregateProvider(engine *aggregate.Engine) (*AggregateProvider, error)

func (*AggregateProvider) Identity 

func (p *AggregateProvider) Identity(context.Context) Identity

func (*AggregateProvider) PruneTopology 

func (p *AggregateProvider) PruneTopology()

func (*AggregateProvider) Snapshot 

func (p *AggregateProvider) Snapshot(ctx context.Context, q Query) (Snapshot, error)

func (*AggregateProvider) Source 

func (*AggregateProvider) Source() Source

func (*AggregateProvider) TopologyEpoch 

func (p *AggregateProvider) TopologyEpoch() uint64

The following methods implement graphrag.AggregateSource. Keeping that projection narrow avoids teaching GraphRAG about range queries.

func (*AggregateProvider) TopologyRevision 

func (p *AggregateProvider) TopologyRevision(tenant string) uint64

func (*AggregateProvider) TopologySnapshot 

func (p *AggregateProvider) TopologySnapshot(tenant string) aggregate.TopologySnapshot

func (*AggregateProvider) TopologyTenants 

func (p *AggregateProvider) TopologyTenants() []string

type Edge 

type Edge struct {
	Source       string  `json:"source"`
	Target       string  `json:"target"`
	CallCount    int64   `json:"call_count"`
	AvgLatencyMs float64 `json:"avg_latency_ms"`
	ErrorRate    float64 `json:"error_rate"`
	Status       string  `json:"status,omitempty"`
}

Edge is one directed service dependency.

type Host 

type Host struct {
	Name         string    `json:"name"`
	ServiceCount int       `json:"service_count"`
	Services     []string  `json:"services"`
	LastSeen     time.Time `json:"last_seen"`
	Signals      []string  `json:"signals"`
}

Host is one host with the services observed on it. Services is sorted, never nil, capped at MaxServicesPerHost and never lists a host entity; ServiceCount is the uncapped total.

type HostProjection 

type HostProjection struct {
	// Hosts is sorted by name and never nil.
	Hosts []Host
	// contains filtered or unexported fields
}

HostProjection is one tenant's host view derived from one registry snapshot: hosts with their services, and per service its hosts.

func ProjectHosts 

func ProjectHosts(entries []ResourceEntry) HostProjection

ProjectHosts folds registry entries of one tenant into a HostProjection. Entries without a host contribute nothing.

func (HostProjection) Host 

func (p HostProjection) Host(name string) (Host, bool)

Host returns the named host.

func (HostProjection) ServiceHosts 

func (p HostProjection) ServiceHosts(service string) (int, []string)

ServiceHosts returns how many hosts service was observed on and the first MaxHostsPerNode of them, sorted.

type HostReader 

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

HostReader is the single reader of the resource registry for every topology consumer. Both providers embed it, so legacy, shadow and aggregate answer host questions identically. Without a registry it projects nothing.

func (*HostReader) Hosts 

func (h *HostReader) Hosts(ctx context.Context) HostProjection

Hosts projects the registry for the tenant on ctx. It is built per call: the registry snapshot is already sorted and bounded per tenant.

func (*HostReader) SetRegistry 

func (h *HostReader) SetRegistry(r *Registry)

SetRegistry wires the registry read by Hosts.

type Identity 

type Identity struct {
	Epoch    string `json:"epoch,omitempty"`
	Revision uint64 `json:"revision,omitempty"`
}

Identity orders full-replacement snapshots within one process generation.

func (Identity) String 

func (i Identity) String() string

String returns the stable cache and stream identity.

type InstanceCatalog added in v0.6.4 

type InstanceCatalog struct {
	Services     []InstanceNode `json:"services"`
	Environments []string       `json:"environments"`
	ServiceNames int            `json:"service_names"`
	Instances    int            `json:"instances"`
	Unresolved   int            `json:"unresolved"`
	Hosts        int            `json:"hosts"`
	Truncated    bool           `json:"truncated"`
}

type InstanceEdge added in v0.6.4 

type InstanceEdge struct {
	ID           string  `json:"id"`
	Source       string  `json:"source"`
	Target       string  `json:"target"`
	Calls        int64   `json:"calls"`
	Errors       int64   `json:"errors"`
	Rate         float64 `json:"request_rate_rps"`
	AvgLatencyMS float64 `json:"avg_latency_ms"`
	ErrorRate    float64 `json:"error_rate"`
	// contains filtered or unexported fields
}

type InstanceGraph added in v0.6.4 

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

InstanceGraph is a best-effort, pre-sampling visualization projection. A five-second dirty tick persists summaries; raw ingestion ACKs are not changed. A crash can lose the unflushed interval. Correlation is bounded and in-memory, with no raw trace/span IDs persisted by this projection.

func NewInstanceGraph added in v0.6.4 

func NewInstanceGraph(ctx context.Context, db *gorm.DB, retention time.Duration) (*InstanceGraph, error)

func (*InstanceGraph) Catalog added in v0.6.4 

func (g *InstanceGraph) Catalog(tenant string, start, end time.Time) InstanceCatalog

Catalog reads bounded identity metadata, not raw telemetry or aggregate history. It is safe to call independently of the expensive graph read.

func (*InstanceGraph) Flush added in v0.6.4 

func (g *InstanceGraph) Flush(ctx context.Context) error

Flush snapshots dirty values without holding the ingestion mutex during I/O. A failed transaction keeps the original pending values. Successful commit subtracts only that snapshot; concurrent observations remain pending.

func (*InstanceGraph) Observe added in v0.6.4 

func (g *InstanceGraph) Observe(entity, traceID, spanID, parentID string, at time.Time, micros int64, request, failed bool)

Observe records one successfully accepted span before exemplar filtering. Both arrival orders work, including calls between copies of the same service. A parent in a different trace or tenant can never create an edge.

func (*InstanceGraph) Prune added in v0.6.4 

func (g *InstanceGraph) Prune(ctx context.Context) error

Prune removes at most 1000 rows per tick. Query visibility always clamps to retention even when physical cleanup is behind.

func (*InstanceGraph) Register added in v0.6.4 

func (g *InstanceGraph) Register(i InstanceIdentity, signal Signal, at time.Time) string

Register is called once per resource batch for all three signals. It records presence without inventing trace counts for metrics-only/logs-only resources.

func (*InstanceGraph) Run added in v0.6.4 

func (g *InstanceGraph) Run(ctx context.Context)

func (*InstanceGraph) SetMetrics added in v0.6.8 

func (g *InstanceGraph) SetMetrics(m *telemetry.Metrics)

SetMetrics wires loss counters and size gauges. Call before ingestion starts.

func (*InstanceGraph) Snapshot added in v0.6.4 

func (g *InstanceGraph) Snapshot(ctx context.Context, tenant string, q InstanceQuery) (InstanceSnapshot, error)

Snapshot materializes at most 50,000 stored rows plus the bounded pending map. Scope is applied before storage reads, with at most 400 bind variables per query. The gate excludes a concurrent flush, preventing double counts between durable rows and pending values without blocking ingestion on I/O.

type InstanceIdentity added in v0.6.4 

type InstanceIdentity struct {
	Tenant, Environment, Namespace, Service, Instance, Host, Workload string
}

InstanceIdentity never derives environment from a host name. Explicit instance IDs are scoped by tenant, environment, namespace and service. Host/workload is a fallback; host alone is explicitly ambiguous.

type InstanceMetadata added in v0.6.4 

type InstanceMetadata struct {
	Start          time.Time         `json:"start"`
	End            time.Time         `json:"end"`
	ResolutionS    int               `json:"resolution_seconds"`
	Coverage       string            `json:"coverage"`
	Note           string            `json:"note"`
	Truncated      bool              `json:"truncated"`
	AvailableNodes int               `json:"available_nodes"`
	ReadRows       int               `json:"read_rows"`
	PendingSpans   int               `json:"pending_parent_spans"`
	Dropped        uint64            `json:"process_dropped"`
	Unmatched      uint64            `json:"process_unmatched"`
	DropReasons    map[string]uint64 `json:"process_drop_reasons"`
	MissReasons    map[string]uint64 `json:"process_unmatched_reasons"`
	LastLoss       *time.Time        `json:"last_loss,omitempty"`
	LastFlush      time.Time         `json:"last_flush"`
	FlushFailed    bool              `json:"flush_failed"`
}

type InstanceNode added in v0.6.4 

type InstanceNode struct {
	ID              string   `json:"id"`
	ServiceID       string   `json:"service_id"`
	Name            string   `json:"name"`
	Namespace       string   `json:"namespace"`
	Environment     string   `json:"environment"`
	Instance        string   `json:"instance,omitempty"`
	Identity        string   `json:"identity"`
	Hosts           []string `json:"hosts"`
	Instances       []string `json:"instances"`
	InstanceCount   int      `json:"instance_count"`
	UnresolvedCount int      `json:"unresolved_count"`
	Spans           int64    `json:"spans"`
	Requests        int64    `json:"requests"`
	Errors          int64    `json:"errors"`
	Rate            float64  `json:"request_rate_rps"`
	AvgLatencyMS    float64  `json:"avg_latency_ms"`
	ErrorRate       float64  `json:"error_rate"`
	// contains filtered or unexported fields
}

type InstanceQuery added in v0.6.4 

type InstanceQuery struct {
	Start, End  time.Time
	Environment *string // nil selects all, empty selects Unknown only
	Mode        string
	ServiceID   string // opaque service group identity, never an ambiguous name
	Hops, Limit int
}

type InstanceSnapshot added in v0.6.4 

type InstanceSnapshot struct {
	Nodes []InstanceNode   `json:"nodes"`
	Edges []InstanceEdge   `json:"edges"`
	Meta  InstanceMetadata `json:"meta"`
}

type LegacyProvider 

type LegacyProvider struct {
	HostReader
	// contains filtered or unexported fields
}

LegacyProvider preserves the existing historical repository query and the current GraphRAG/graph/DB preference for live topology.

func NewLegacyProvider 

func NewLegacyProvider(repo legacyRepository, serviceGraph *graph.Graph, graphRAG *graphrag.GraphRAG) (*LegacyProvider, error)

func (*LegacyProvider) Identity 

func (*LegacyProvider) Identity(context.Context) Identity

func (*LegacyProvider) Snapshot 

func (p *LegacyProvider) Snapshot(ctx context.Context, q Query) (Snapshot, error)

func (*LegacyProvider) Source 

func (*LegacyProvider) Source() Source

type Metadata 

type Metadata struct {
	Source       Source    `json:"source,omitempty"`
	Start        time.Time `json:"start,omitempty"`
	End          time.Time `json:"end,omitempty"`
	Coverage     string    `json:"coverage,omitempty"`
	CoverageNote string    `json:"coverage_note,omitempty"`
	Epoch        string    `json:"epoch,omitempty"`
	Revision     uint64    `json:"revision,omitempty"`
	Truncated    bool      `json:"truncated,omitempty"`

	DroppedServices   uint64                     `json:"dropped_services,omitempty"`
	DroppedOperations uint64                     `json:"dropped_operations,omitempty"`
	DroppedEdges      uint64                     `json:"dropped_edges,omitempty"`
	DroppedMetrics    uint64                     `json:"dropped_metrics,omitempty"`
	IdentityCoverage  aggregate.IdentityCoverage `json:"identity_coverage"`
}

Metadata is additive ownership and completeness information.

func (Metadata) Identity 

func (m Metadata) Identity() Identity

Identity returns the ordering pair carried by this metadata.

type Node 

type Node struct {
	Name         string  `json:"name"`
	TotalTraces  int64   `json:"total_traces"`
	ErrorCount   int64   `json:"error_count"`
	AvgLatencyMs float64 `json:"avg_latency_ms"`

	RequestRateRPS    float64             `json:"request_rate_rps,omitempty"`
	ErrorRate         float64             `json:"error_rate,omitempty"`
	P99LatencyMs      float64             `json:"p99_latency_ms,omitempty"`
	LatencyProvenance *latency.Provenance `json:"latency_provenance,omitempty"`
	SpanCount         int64               `json:"span_count,omitempty"`
	HealthScore       float64             `json:"health_score,omitempty"`
	Status            string              `json:"status,omitempty"`
	Alerts            []string            `json:"alerts,omitempty"`

	// Host projection (#288), stamped by the provider from the resource
	// registry. Kind is KindService or KindHost; Hosts is sorted and capped
	// at MaxHostsPerNode while HostCount is the uncapped total.
	Kind      string   `json:"kind,omitempty"`
	HostCount int      `json:"host_count,omitempty"`
	Hosts     []string `json:"hosts,omitempty"`
}

Node is the wire-neutral service projection shared by all consumers.

type Provider 

type Provider interface {
	Source() Source
	Identity(context.Context) Identity
	Snapshot(context.Context, Query) (Snapshot, error)
	// Hosts is the tenant-scoped host projection of the resource registry.
	// Every provider answers it from the same registry, so legacy, shadow
	// and aggregate agree on hosts.
	Hosts(context.Context) HostProjection
}

Provider is selected once from AGGREGATE_MODE and injected into every live topology consumer.

type Query 

type Query struct {
	Start    time.Time
	End      time.Time
	Services []string
}

Query selects a range and an optional closed set of services. A zero range asks for the provider's current live replacement.

type Registry 

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

Registry is the bounded in-memory resource registry. Registration is one mutex acquisition and one map lookup; the already-present path allocates nothing.

func NewRegistry 

func NewRegistry(metrics *telemetry.Metrics) *Registry

NewRegistry returns an empty registry. metrics may be nil.

func (*Registry) Evict 

func (r *Registry) Evict(now time.Time) int

Evict drops entries idle past RegistryIdleTTL at now and returns how many. Their rows are deleted by the next Flush.

func (*Registry) Flush 

func (r *Registry) Flush(ctx context.Context, db *gorm.DB) error

Flush upserts dirty entries and deletes evicted rows. A failed write leaves the affected entries dirty (or evicted) for the next tick.

func (*Registry) Load 

func (r *Registry) Load(ctx context.Context, db *gorm.DB) (int, error)

Load reloads persisted entries through the same bounds as Register. Rows already idle past the TTL are loaded too; the first tick evicts them and the following flush deletes their rows.

func (*Registry) Overflow 

func (r *Registry) Overflow(tenant, kind string) uint64

Overflow returns how many registrations the bound of kind refused for tenant.

func (*Registry) Register 

func (r *Registry) Register(tenant, service, host, workload, kind string, signal Signal, now time.Time) bool

Register records that signal arrived from the resource at now. It reports false when a per-tenant bound refused a new entry.

func (*Registry) Run 

func (r *Registry) Run(ctx context.Context, db *gorm.DB, interval time.Duration)

Run evicts, publishes gauges and flushes on every tick until ctx is done. The caller flushes once more at shutdown.

func (*Registry) Snapshot 

func (r *Registry) Snapshot() []ResourceEntry

Snapshot returns every live entry ordered by key.

func (*Registry) TenantSnapshot 

func (r *Registry) TenantSnapshot(tenant string) []ResourceEntry

TenantSnapshot returns tenant's live entries ordered by key.

type ResourceEntry 

type ResourceEntry struct {
	ResourceKey
	// Kind names the slot that filled Workload (pod|container|process), or "".
	Kind     string
	Signals  Signal
	LastSeen time.Time
}

ResourceEntry is one row of a read-only registry snapshot.

type ResourceKey 

type ResourceKey struct {
	Tenant   string
	Service  string
	Host     string
	Workload string
}

ResourceKey identifies one registered resource. Host is host.id else host.name; Workload is k8s.pod.uid else container.id else process.pid. Both may be empty: a resource without a host registers the service alone.

type Signal 

type Signal uint8

Signal is one bit of ResourceEntry.Signals. 

const (
	SignalTraces Signal = 1 << iota
	SignalLogs
	SignalMetrics
)

func (Signal) Names 

func (s Signal) Names() []string

Names renders the signal bits as sorted names.

type Snapshot 

type Snapshot struct {
	Nodes []Node   `json:"nodes"`
	Edges []Edge   `json:"edges"`
	Meta  Metadata `json:"meta,omitempty"`
}

Snapshot is always a full replacement. Nodes and Edges are never nil.

type Source 

type Source string

Source identifies the producer that owns a snapshot. 

const (
	SourceLegacy    Source = "legacy"
	SourceAggregate Source = "aggregate"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.