telemetry

package
v0.2.0-beta.1 Latest Latest
Warning

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

 Go to latest Published: Sep 13, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package telemetry is the node-local metrics store, decided in ADR 009: a dedicated SQLite database (telemetry.db, separate from internal/store's levelrail.db) holding container resource-usage samples, queried in place rather than shipped anywhere centrally. Keeping metrics node-local instead of centralizing them is what keeps idle cost near zero as the number of managed apps grows.

Index

Constants

View Source 
const (
	// MetricDeployCount is one sample per completed deploy cutover
	// (RecordDeploy), always Value 1. A raw sample count therefore *is*
	// deploy frequency for a range: Aggregate's own Count field (how
	// many raw samples fell in a bucket) already answers "how many
	// deploys in this window" with no separate counter or rate
	// computation needed.
	MetricDeployCount = "deploy_count"
	// MetricBuildDuration is one sample per completed build
	// (RecordBuildDuration), Value the build's wall-clock duration in
	// seconds, taken directly from build.Result.Duration rather than
	// re-measured here.
	MetricBuildDuration = "build_duration_seconds"
)

Metric names for the remaining gap: deploy frequency and build duration. Unlike every other metric in this store, neither comes from a Docker stats poll (collector.go's sampleValues): a deploy or a build is a discrete event, not a continuously observable resource, so each is recorded once, at the moment it happens, by its own caller (internal/reconcile/application for a deploy cutover, internal/deploy for a completed build) rather than by Collector.

View Source 
const (
	MetricDiskUsedBytes  = "disk_used_bytes"
	MetricDiskTotalBytes = "disk_total_bytes"
)

MetricDiskUsedBytes and MetricDiskTotalBytes are the sample metric names HostDiskCollector writes, and the names internal/api and internal/alerting read back.

View Source 
const (
	MetricOSPatchesAvailable         = "os_patches_available"
	MetricOSSecurityPatchesAvailable = "os_security_patches_available"
)

MetricOSPatchesAvailable and MetricOSSecurityPatchesAvailable are the sample metric names HostPatchCollector writes, and the names internal/api reads back (node_patch_status.go, node_metrics.go).

Variables

View Source 
var ErrNoSupportedPackageManager = errors.New("telemetry: no supported package manager found")

ErrNoSupportedPackageManager means none of apt, dnf, or yum exist on this host: a real, permanent condition (e.g. a non-Linux dev machine, or a distribution this checker doesn't know), not a transient failure.

Functions

This section is empty.

Types

type AggregatedPoint 

type AggregatedPoint struct {
	// Timestamp is the bucket's start, not a sample's own timestamp.
	Timestamp time.Time
	// Value is the average of every sample that fell in this bucket.
	Value float64
	// Count is how many raw samples contributed, so a caller (or a
	// future frontend) can tell a bucket built from one sample from one
	// built from sixty, without a separate query.
	Count int
}

AggregatedPoint is one bucketed value from Aggregate.

func Aggregate 

func Aggregate(samples []Sample, from time.Time, step time.Duration) []AggregatedPoint

Aggregate buckets samples (assumed already timestamp-ascending, which Federator.QueryMetrics already guarantees) into fixed-width windows starting at from, each bucket's value the average of the samples inside it. step <= 0 means "no aggregation," one point per sample. An empty bucket is omitted, not returned as a zero: a sparse series (a container that stopped reporting for an hour) should read as a gap in the chart, not a real reading of zero.

type Collector 

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

Collector polls a caller-supplied set of targets on an interval and writes what it collects to a Store.

func NewCollector 

func NewCollector(source StatsSource, store *DB, interval time.Duration, logger *slog.Logger) *Collector

NewCollector builds a Collector. logger defaults to slog.Default() if nil.

func (*Collector) CollectOnce 

func (c *Collector) CollectOnce(ctx context.Context, targets []Target) error

CollectOnce polls every target once and writes whatever succeeded. One target's stats call failing (its container just stopped between discovery and polling, for instance) doesn't block the others': the same "one broken resource must not stop convergence of everything else" principle reconcile.Engine.ReconcileAll already applies to controllers, applied here to collection targets. Returns a joined error of every target that failed, purely for the caller to log or count; a partial failure is not treated as this tick failing overall, since the successful samples were still written.

func (*Collector) Run 

func (c *Collector) Run(ctx context.Context, targetsFunc func(context.Context) ([]Target, error)) error

Run calls CollectOnce every interval until ctx is done. targetsFunc is invoked fresh on every tick, not captured once at construction, so the polled set stays current as services are created, redeployed, or removed between ticks: the same level-triggered, re-derive-every-pass principle every reconcile.Controller already follows, applied here to "what should be collected" instead of "what should be running."

type DB 

type DB struct {
	*sql.DB
}

DB wraps a *sql.DB opened against telemetry.db: WAL mode, migrated to the latest version before Open returns. Deliberately a separate type from store.DB, not a shared one: ADR 009 keeps this on its own SQLite file precisely so its write pattern (a batch every collection tick) never contends with internal/store's own WAL.

func Open 

func Open(ctx context.Context, path string) (*DB, error)

Open opens (creating if needed) the SQLite database at path, applies pragmas, and runs every pending migration. path is a plain filesystem path, not a DSN, matching internal/store.Open's own shape.

func (*DB) LatestByMetric 

func (db *DB) LatestByMetric(ctx context.Context, metric string) ([]Sample, error)

LatestByMetric returns the most recent sample for every resource_id that has ever recorded metric, one row per resource: the "what is everything doing right now" read a dashboard-wide ranking needs (e.g. "which app is using the most CPU"), where fetching each resource's own time series one at a time would mean one query per app instead of one query total. An empty (nil) result and a nil error mean "no resource has recorded this metric yet," the same convention Query's own doc comment establishes.

func (*DB) Query 

func (db *DB) Query(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)

Query returns every sample for resourceID/metric with a timestamp in [from, to], oldest first. An empty (nil) result and a nil error both mean "no samples in this range," not an error: a resource with no activity yet, or a range before collection started, is a valid observed state, the same convention docker.InspectByName's "not found is not an error" doc comment already establishes elsewhere in this codebase.

func (*DB) QueryDeployLog 

func (db *DB) QueryDeployLog(ctx context.Context, attemptID string) ([]DeployLogEntry, error)

QueryDeployLog returns every persisted line for attemptID, oldest first: a full replay, not a windowed range query like QueryLogs, since a deploy attempt's log is a bounded, already-finished (by the time this is the code path serving it, see internal/api/deploys.go's SSE handler) event with no reason to page through it by time range. An empty (nil) result and a nil error both mean "no lines for this attempt," the same "absence is not an error" convention QueryLogs already establishes, covering both a genuinely empty log (the plain image-tag trigger path, which never calls WriteDeployLogBatch at all) and an attempt ID that was never written for any reason.

func (*DB) QueryLogs 

func (db *DB) QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)

QueryLogs returns every log entry for resourceID with a timestamp in [from, to], oldest first, optionally filtered to lines whose message matches query. An empty query returns every line in range with no text filter applied. A non-empty query runs against log_entries_fts (see migrations/0002_log_entries.sql), wrapped as a single quoted phrase (double quotes inside query are escaped by doubling, FTS5's own escape convention) rather than passed through as raw FTS5 query syntax: this keeps the query API a plain substring-ish search from a caller's point of view, not something that can fail with an FTS5 syntax error just because the search text happened to contain a character FTS5's query grammar treats specially (e.g. a bare "-" or unbalanced quote).

An empty (nil) result and a nil error both mean "no lines matched," not an error, the same convention Query already establishes for metric samples.

func (*DB) RecordBuildDuration 

func (db *DB) RecordBuildDuration(ctx context.Context, serviceName string, d time.Duration, at time.Time) error

RecordBuildDuration writes one MetricBuildDuration sample for serviceName at at.

func (*DB) RecordDeploy 

func (db *DB) RecordDeploy(ctx context.Context, serviceName string, at time.Time) error

RecordDeploy writes one MetricDeployCount sample for serviceName at at. Callers must call this only when a deploy actually happened (a new container was created/started and passed its readiness probe), not on every reconcile tick that finds nothing to do: internal/reconcile/ application.Controller's own "Deployed" vs "AlreadyRunning" distinction is exactly that signal.

func (*DB) Retain 

func (db *DB) Retain(ctx context.Context, cutoff time.Time) (deleted int64, err error)

Retain deletes every sample older than the given cutoff, returning how many rows were removed. The documented default retention (15 days) is a caller concern (the collector or a scheduled sweep decides the cutoff), not hardcoded here, per the house "no hardcoded thresholds" rule.

func (*DB) RetainDeployLogs 

func (db *DB) RetainDeployLogs(ctx context.Context, cutoff time.Time) (deleted int64, err error)

RetainDeployLogs deletes every deploy log line older than cutoff, returning how many rows were removed. Mirrors RetainLogs' shape exactly, same "caller decides the cutoff, no hardcoded threshold" house rule; see cmd/levelrail/main.go's retention sweep wiring for the default this project actually runs with, chosen separately from RetainLogs' 15-day container-log default per this table's own "bounded, terminal event" reasoning (migrations/0003_deploy_logs.sql).

func (*DB) RetainLogs 

func (db *DB) RetainLogs(ctx context.Context, cutoff time.Time) (deleted int64, err error)

RetainLogs deletes every log entry older than cutoff, returning how many rows were removed. Mirrors Retain's shape exactly (same signature pattern, same "caller decides the cutoff, no hardcoded threshold" house rule); the FTS index stays in sync automatically via log_entries_ad (migrations/0002_log_entries.sql), not because of anything this method does itself.

func (*DB) WriteDeployLogBatch 

func (db *DB) WriteDeployLogBatch(ctx context.Context, entries []DeployLogEntry) error

WriteDeployLogBatch inserts every entry in one transaction, the same atomicity and "no upsert, a retried batch can produce duplicate rows" tradeoff WriteLogBatch already documents for the identical reason: a deploy attempt's output has no natural idempotency key either. internal/deploylog.Recorder is this method's only caller, and it already batches by count (see that package's own batchMaxLines) before calling this, matching WriteLogBatch's batching discipline.

func (*DB) WriteLogBatch 

func (db *DB) WriteLogBatch(ctx context.Context, entries []LogEntry) error

WriteLogBatch inserts every entry in one transaction, the same atomicity WriteSamples already gives metric collection ticks, applied here to one log batch (see logBatchMaxLines/logBatchMaxWait). Unlike WriteSamples, there's no ON CONFLICT upsert: metric samples have a natural idempotency key (resource, metric, timestamp) a retried collection tick can safely overwrite, but log lines don't, since a container legitimately emitting the same text twice (e.g. two "OK" lines) is normal, not a duplicate to be coalesced. This is a real, documented gap: if a batch write fails partway and the caller decides to retry with the same lines, retried lines land as new rows rather than replacing anything, so a transient write failure during streaming can produce duplicate log rows in rare cases. Closing that gap would need its own idempotency key (e.g. a source-assigned sequence number per container), deliberately deferred rather than solved silently here.

func (*DB) WriteSamples 

func (db *DB) WriteSamples(ctx context.Context, samples []Sample) error

WriteSamples inserts every sample in one transaction: a collection tick writes many samples (several metrics per running container) and they should land atomically, not partially, if the process is interrupted mid-write. Writing the same (resource, metric, timestamp) twice replaces the value rather than erroring, since a collector that retries a tick after a transient failure should be safe to just run again, not required to first check what already landed.

type DeployLogEntry 

type DeployLogEntry struct {
	AttemptID string
	// Stream is "stdout" or "stderr".
	Stream    string
	Timestamp time.Time
	Message   string
}

DeployLogEntry is one line of a deploy attempt's build output.

type DrainConfig 

type DrainConfig struct {
	ResourceID string
	Type       DrainSinkType
	Target     string
	Enabled    bool
}

DrainConfig is one resource's external log-forwarding configuration, the telemetry-package-local shape a caller's configFunc (see Run) resolves from wherever it actually persists (internal/store's LogDrain, for application services).

type DrainForwarder 

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

DrainForwarder taps LogBroadcaster the same way a live SSE viewer does (Subscribe/Publish, broadcast.go), so it is purely an additional consumer of the log stream: it never touches LogCollector, WriteLogBatch, or QueryLogs, and a drain forwarding failure can never affect the node-local store or a live tail.

func NewDrainForwarder 

func NewDrainForwarder(broadcaster *LogBroadcaster, logger *slog.Logger, programTag string) *DrainForwarder

NewDrainForwarder builds a DrainForwarder reading from broadcaster. logger defaults to slog.Default() if nil. programTag is the syslog program tag a DrainSinkSyslog sink identifies itself with (see newSyslogSink); callers pass brand.Brand.ShortName, following the same "resolve brand outside this package" convention internal/reconcile/application.WithNetworkPrefix already establishes.

func (*DrainForwarder) Run 

func (f *DrainForwarder) Run(ctx context.Context, resync time.Duration, configFunc func(context.Context) ([]DrainConfig, error)) error

Run derives the desired drain set via configFunc on a resync interval and keeps exactly one forwarding goroutine alive per currently-enabled ResourceID, mirroring LogCollector.Run's own reconcile-loop shape exactly (same active-map/start/cancel pattern) so the two collectors read as one family even though they serve different consumers.

type DrainSink 

type DrainSink interface {
	Send(ctx context.Context, entries []LogEntry) error
}

DrainSink is the narrow surface a log drain forwarder needs from a concrete sink implementation (HTTPSink, syslog). Send should treat a partial failure (e.g. some lines rejected) as a whole-batch error: there's no per-line retry here, a failed batch is simply dropped and logged, the same best-effort shape LogBroadcaster.Publish's own doc comment already establishes for a slow live-tail subscriber.

type DrainSinkType 

type DrainSinkType string

DrainSinkType selects which external sink protocol a drain uses. Defined locally rather than importing internal/store's LogDrainType: this package's own doc comments elsewhere (e.g. DesiredService's NodeStatus precedent) already establish "define locally, don't import a higher-level package's vocabulary" for cross-package enums like this. 

const (
	DrainSinkHTTP   DrainSinkType = "http"
	DrainSinkSyslog DrainSinkType = "syslog"
)

The two sink protocols buildDrainSink knows how to construct.

type Federator 

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

Federator fans a query out to every configured source and merges results, per ADR 008: "the control plane fans queries out to agents and merges results at query time; query is pull, not push." Today there is exactly one source (this node's own local DB), so federation is a merge-of-one; the shape is what Phase 3 fills in with real per-node sources.

func NewFederator 

func NewFederator(metrics []MetricsSource, logs []LogsSource) *Federator

NewFederator builds a Federator over an explicit source list, for Phase 3 once real per-node sources exist alongside (or instead of) the local one.

func NewLocalFederator 

func NewLocalFederator(db *DB) *Federator

NewLocalFederator builds a Federator with exactly one source: this process's own local DB. The single-node shape every control plane runs today, until Phase 3 adds real remote agents.

func (*Federator) LatestByMetric 

func (f *Federator) LatestByMetric(ctx context.Context, metric string) ([]Sample, error)

LatestByMetric is LatestByMetric's federated equivalent: fans out to every source and keeps, per resource_id, whichever source reported the newest sample. A resource split across sources (not possible today, single-node only, but Phase 3's per-node agents could each report a same-named resource) should never happen in practice, so "newest wins" is a defensive tie-break, not a real merge strategy.

func (*Federator) QueryLogs 

func (f *Federator) QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)

QueryLogs is QueryMetrics' log-query equivalent, merged and sorted by timestamp ascending the same way.

func (*Federator) QueryMetrics 

func (f *Federator) QueryMetrics(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)

QueryMetrics fans out to every metrics source and merges results, sorted by timestamp ascending. A source erroring doesn't fail the whole query: ADR 008's Consequences require the query layer to "handle partial results gracefully... deciding what the dashboard shows when [an agent is unreachable]," so this returns whatever the healthy sources answered, plus the errors as a joined error the caller can log or surface as a partial-result warning, not use to discard everything.

type HTTPSink 

type HTTPSink struct {
	URL    string
	Client *http.Client
}

HTTPSink POSTs a JSON array of httpSinkLine to URL for every Send call, the generic-webhook log drain shape (Coolify-parity item, see this package's drain.go doc comment).

func NewHTTPSink 

func NewHTTPSink(url string, client *http.Client) *HTTPSink

NewHTTPSink builds an HTTPSink. client defaults to a fresh *http.Client with httpSinkTimeout if nil.

func (*HTTPSink) Send 

func (s *HTTPSink) Send(ctx context.Context, entries []LogEntry) error

Send implements DrainSink.

type HostDiskCollector 

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

HostDiskCollector polls one filesystem path's capacity on an interval and writes disk_total_bytes/disk_used_bytes samples under one resource ID, the host-level counterpart to Collector's per-container polling above: a real reading of the volume that actually matters operationally (where images, volumes, and the SQLite stores live), not a per-container Docker stat.

func NewHostDiskCollector 

func NewHostDiskCollector(path, resourceID string, store *DB, interval time.Duration, logger *slog.Logger) *HostDiskCollector

NewHostDiskCollector builds a HostDiskCollector. logger defaults to slog.Default() if nil.

func (*HostDiskCollector) CollectOnce 

func (c *HostDiskCollector) CollectOnce(ctx context.Context) error

CollectOnce reads path's current capacity once and writes it.

func (*HostDiskCollector) Run 

func (c *HostDiskCollector) Run(ctx context.Context) error

Run calls CollectOnce every interval until ctx is done, the same "log and keep going, one bad tick must not stop the collector" shape Collector.Run above already establishes.

type HostPatchChecker 

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

HostPatchChecker looks up how many OS package updates a host has available, via whichever supported package manager exists on PATH.

func NewHostPatchChecker 

func NewHostPatchChecker() *HostPatchChecker

NewHostPatchChecker builds a HostPatchChecker that shells out to the real package manager binaries on PATH.

func (*HostPatchChecker) Check 

func (c *HostPatchChecker) Check(ctx context.Context) (PatchCounts, error)

Check runs the first supported package manager it finds (apt, then dnf, then yum) and returns its upgrade counts, or ErrNoSupportedPackageManager if none exist. A found manager whose command itself fails returns that error wrapped, never a false zero.

type HostPatchCollector 

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

HostPatchCollector polls one host's available OS package updates on an interval and writes MetricOSPatchesAvailable/ MetricOSSecurityPatchesAvailable samples under one resource ID, the same shape HostDiskCollector already establishes for host disk space.

func NewHostPatchCollector 

func NewHostPatchCollector(checker *HostPatchChecker, resourceID string, store *DB, interval time.Duration, logger *slog.Logger) *HostPatchCollector

NewHostPatchCollector builds a HostPatchCollector. checker defaults to NewHostPatchChecker() if nil, logger to slog.Default() if nil.

func (*HostPatchCollector) CollectOnce 

func (c *HostPatchCollector) CollectOnce(ctx context.Context) error

CollectOnce checks this host's available updates once and writes them. ErrNoSupportedPackageManager is not an error from this method's own perspective: it logs at Debug and returns nil without writing a sample, so a host with no supported package manager reports "unknown, not checked" (no recent sample) rather than a crashed collector or a false zero.

func (*HostPatchCollector) Run 

func (c *HostPatchCollector) Run(ctx context.Context) error

Run calls CollectOnce every interval until ctx is done, the same "log and keep going, one bad tick must not stop the collector" shape HostDiskCollector.Run already establishes.

type LogBroadcaster 

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

LogBroadcaster fans out every log line LogCollector.StreamOne receives to whichever SSE viewers are currently watching that line's ResourceID, live, the moment it arrives. It exists for the same reason internal/deploylog.Recorder exists for build/deploy output: persistence alone (WriteLogBatch, batched up to logBatchMaxWait behind real time) is fine for a historical search but not for a view labeled "live", which needs to feel instant.

Deliberately simpler than deploylog.Recorder, in one specific way: there is no Start/Finish lifecycle here. A deploy attempt has a real bounded shape (started, eventually finishes, gets evicted), a container's ResourceID does not: a service can be redeployed, restarted, or have zero subscribers for most of its life, with no single moment that means "this resource is done, stop remembering it." So LogBroadcaster never eagerly allocates state for a resource, only when something actually calls Subscribe, and Unsubscribe cleans up empty entries rather than relying on a Finish call that has no natural trigger here. There is also no "lines so far" in-memory replay (contrast Recorder.Snapshot): that job belongs to the persisted store (QueryLogs) via internal/api's live-log handler, which backfills recent context from there before calling Subscribe, since only the store, not this in-memory broadcaster, survives a control plane restart.

One instance is shared across every LogCollector.StreamOne goroutine (the publishing side, one per currently-running container) and every live-log SSE handler request (the subscribing side), the same "exactly one shared instance, wired once at startup" shape deploylog.Recorder's own package doc comment establishes for build logs: a viewer connected through the HTTP API must see the same lines the collector is receiving directly from Docker, which only works if both sides publish to and read from the same subscriber set.

func NewLogBroadcaster 

func NewLogBroadcaster() *LogBroadcaster

NewLogBroadcaster builds an empty LogBroadcaster.

func (*LogBroadcaster) Publish 

func (b *LogBroadcaster) Publish(entry LogEntry)

Publish delivers entry to every subscriber currently watching entry.ResourceID. Non-blocking by design: a subscriber whose channel is full (see broadcastBufferSize) is skipped for this entry rather than stalling the call, since Publish's real caller (StreamOne) is on the same goroutine that is also consuming Docker's own log stream and batching writes to the store, neither of which may block on an HTTP response writer somewhere downstream that a slow or stuck client left half-read.

func (*LogBroadcaster) Subscribe 

func (b *LogBroadcaster) Subscribe(resourceID string) (ch <-chan LogEntry, unsubscribe func())

Subscribe registers a new live listener for resourceID, returning a channel that receives every entry Published for that resourceID from this point forward, and an unsubscribe func the caller must call exactly once, typically deferred, once it stops reading (e.g. the SSE client disconnected). Safe to call for a resourceID with no active producer (the container hasn't started yet, or already stopped, or this control plane never ran a collector for it): Subscribe never fails, it may simply never receive anything, which is the correct behavior for a live tail opened against a currently quiet or not-yet-running app rather than an error condition.

type LogCollector 

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

LogCollector streams logs from a caller-supplied set of targets and writes what it receives to a Store, batching writes per logBatchMaxLines/logBatchMaxWait, and, when a broadcaster is configured, fans every line out live at the same time (see StreamOne).

func NewLogCollector 

func NewLogCollector(source LogSource, store *DB, broadcaster *LogBroadcaster, logger *slog.Logger) *LogCollector

NewLogCollector builds a LogCollector. broadcaster may be nil: without one, StreamOne still batches and persists every line exactly as before, it just has nowhere to publish a live copy, so a control plane started without one simply has no live log tail (internal/api's live log route returns 501 in that case, the same "not configured" shape this package's own optional dependencies already use elsewhere). logger defaults to slog.Default() if nil.

func (*LogCollector) Run 

func (lc *LogCollector) Run(ctx context.Context, resync time.Duration, targetsFunc func(context.Context) ([]LogTarget, error)) error

Run derives targets via targetsFunc on a resync interval and keeps exactly one StreamOne goroutine alive per currently-desired container ID, starting new ones as containers appear and cancelling ones for containers no longer desired, until ctx is done. This is the streaming-collector equivalent of Collector.Run's "re-derive targets every tick" principle: the set of things being collected stays current as services are created, redeployed, or removed, without needing a separate mechanism to notice that a target disappeared mid-stream versus never existed.

func (*LogCollector) StreamOne 

func (lc *LogCollector) StreamOne(ctx context.Context, target LogTarget) error

StreamOne follows target's log stream (Follow: true, from the current point forward) until ctx is cancelled or the stream ends on its own (the container stopped), batching lines per logBatchMaxLines/ logBatchMaxWait and writing each batch to the store. Blocking: callers run it in its own goroutine per target, one long-lived subscription per running container, not a polling tick like Collector.CollectOnce: logs are a push stream from Docker, not a point-in-time snapshot, so this package's two collectors have genuinely different shapes even though they share a store and a retention pattern.

type LogEntry 

type LogEntry struct {
	ResourceID string
	// Stream is "stdout" or "stderr".
	Stream    string
	Timestamp time.Time
	Message   string
	// Structured is true when Message's full text parsed as a JSON
	// object; see classifyLine.
	Structured bool
	// FieldsJSON holds Message's parsed JSON when Structured is true,
	// stored in its own column, separately from the raw line, so a future
	// log viewer doesn't need
	// to re-parse Message to tell structured and plain lines apart or to
	// render one differently from the other. Empty when Structured is
	// false.
	FieldsJSON string
}

LogEntry is one line of container log output, landed in or read back from the store.

type LogSource 

type LogSource interface {
	Logs(ctx context.Context, containerID string, follow bool, since time.Time) (<-chan docker.LogLine, <-chan error)
}

LogSource is the narrow Docker surface a LogCollector needs: nothing about container discovery, only "give me a demultiplexed log stream for this ID." *docker.Client satisfies this structurally. Deliberately not part of docker.Runtime, the same reasoning StatsSource's doc comment already gives for the same choice: adding this to Runtime would mean updating every existing fake Runtime implementation across the codebase for a capability only this package needs.

type LogTarget 

type LogTarget struct {
	ResourceID  string
	ContainerID string
}

LogTarget is one container a LogCollector should stream logs from.

type LogsSource 

type LogsSource interface {
	QueryLogs(ctx context.Context, resourceID string, from, to time.Time, query string) ([]LogEntry, error)
}

LogsSource is the log-query equivalent of MetricsSource.

type MetricsSource 

type MetricsSource interface {
	Query(ctx context.Context, resourceID, metric string, from, to time.Time) ([]Sample, error)
	LatestByMetric(ctx context.Context, metric string) ([]Sample, error)
}

MetricsSource is what the query layer needs from one agent, local or remote, to answer a metrics query. *DB satisfies this today (the single-node, in-process "agent"); a Phase 3 gRPC client would satisfy it too without Federator's callers (internal/api) ever changing. This mirrors the same precedent already set for the reconcile agent transport: it must also work in single-node mode, communicating over an in-memory transport that implements the same interface.

type PatchCounts 

type PatchCounts struct {
	Manager  string
	Total    int
	Security int
}

PatchCounts is one point-in-time reading of a host's available OS package updates.

type Sample 

type Sample struct {
	ResourceID string
	Metric     string
	Timestamp  time.Time
	Value      float64
}

Sample is one metric reading at one point in time for one resource (e.g. resource_id "service:web", metric "cpu_percent").

func SumAcrossResources 

func SumAcrossResources(nodeResourceID, metric string, samples []Sample) []Sample

SumAcrossResources collapses samples from multiple resources (e.g. every container currently placed on one node) for a single metric into one per-timestamp total: "how much of this metric was in use, summed across everything running here, at each collection tick."

Samples are grouped by their exact Timestamp, not a time bucket: Collector.CollectOnce (collector.go) captures one `now` per tick and stamps every target polled in that tick with it, and WriteSamples/Query round-trip that value at one-second precision (store.go), so two containers collected in the same tick share the exact same Timestamp here, not merely an approximately-close one. A caller that wants coarser buckets afterward (a chart's `step` query param) runs this function's output through Aggregate, the same as a single resource's own series already does (internal/api/metrics.go's handleQueryMetrics).

Mixing metrics in one call would produce a meaningless sum (adding a percent to a byte count), so this assumes every sample passed in already shares one metric; callers query one metric per resource and concatenate before calling this, matching Federator.QueryMetrics' own "one metric per call" contract.

The returned samples' ResourceID and Metric are the caller-supplied nodeResourceID/metric, not copied from any individual input sample: the result represents a virtual, computed-on-the-fly aggregate resource, not any single container, and is never itself written back to the store.

Two known, deliberately accepted limitations, not bugs:

  • The exact-Timestamp grouping this function relies on holds only because everything collected today runs through one process's own Collector.CollectOnce, sharing one clock. The Phase 3 plan (per-node agents, each with an independent local collector) breaks this assumption: two agents' clocks are never perfectly synchronized, so their samples for "the same tick" won't share an identical Unix second. Revisit this grouping (a tolerance window, not exact equality) before this function is ever fed cross-agent samples, not after.
  • A bucket is never flagged as partially-covered: if one resource has a sample at a given timestamp and another (also placed on this node for that whole range) doesn't, the sum silently reflects only the resource that reported, indistinguishable here from "every resource genuinely totaled this." Neither zero-filled nor crashed, both worse options, but a caller cannot currently tell "real total" from "partial total" from this function's output alone.

type StatsSource 

type StatsSource interface {
	Stats(ctx context.Context, containerID string) (docker.ContainerStats, error)
}

StatsSource is the narrow Docker surface a Collector needs: nothing about container discovery, only "give me a snapshot for this ID." *docker.Client satisfies this structurally. Deliberately not part of docker.Runtime (the interface every reconcile controller depends on and every existing fake implements): adding a method there would require updating every one of those fakes for a capability only this package needs, so this stays its own minimal, consumer-defined interface instead, the same shape internal/reconcile/application's ServiceStore already establishes for the same reason.

type SyslogSink 

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

SyslogSink writes each entry to a syslog daemon via the standard library's log/syslog package, which this build tag mirrors exactly: that package itself only builds on the platforms listed above. This platform targets Linux-only nodes in production, so the unsupported set (Windows, Plan 9, WASM) is not a real gap for a managed node, only for compiling this package on an unsupported dev machine, which the sibling stub file (drain_syslog_stub.go) covers.

func (*SyslogSink) Send 

func (s *SyslogSink) Send(ctx context.Context, entries []LogEntry) error

Send implements DrainSink. log/syslog's Writer takes no context, so this only checks ctx before each line rather than mid-write, enough to stop promptly on shutdown without needing a custom transport.

type Target 

type Target struct {
	ResourceID  string
	ContainerID string
}

Target is one container a Collector should poll on each tick. ResourceID is the metrics store's own identifier for whatever this container belongs to (e.g. "service:web"), deliberately not the raw container ID: a redeploy gives a service a new container ID (per internal/reconcile/application's deterministic-name-per-image design) but its metric history should stay queryable under one stable identifier across that change.

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.