dataexport

package
v0.6.0 Latest Latest
Warning

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

Go to latest
Published: Oct 17, 2026 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Overview

Package dataexport sends logs, audit events and metrics to destinations an operator already runs. Every pipeline is opt-in, bounded in memory and isolated from the container log stream, the API and the reconcilers.

Index

Constants

View Source
const AllowPrivateEnv = "APP_EXPORT_ALLOW_PRIVATE_NETWORKS"

AllowPrivateEnv lets export destinations resolve to private addresses, for a Loki or collector on the same host or network as the control plane.

Variables

View Source
var ErrBadSignature = errors.New("dataexport: signature mismatch")

ErrBadSignature is returned by VerifySignature.

View Source
var ErrBlockedDestination = errors.New("dataexport: destination resolves to a blocked internal address")

ErrBlockedDestination wraps a refusal to reach an internal address.

View Source
var ErrInvalid = errors.New("invalid export destination")

ErrInvalid wraps every validation failure so callers can answer 400.

Functions

func EncodeNDJSON

func EncodeNDJSON(recs []AuditRecord) ([]byte, error)

EncodeNDJSON renders records one JSON object per line.

func EncodeOTLP

func EncodeOTLP(series []Series, serviceName string, now time.Time) ([]byte, error)

EncodeOTLP renders series as an OTLP/HTTP JSON ExportMetricsServiceRequest. Counters become cumulative monotonic sums, everything else gauges.

func EncodeRemoteWrite

func EncodeRemoteWrite(series []Series, now time.Time) []byte

EncodeRemoteWrite renders series as a snappy block compressed prompb.WriteRequest. Series without a timestamp use now.

func IsVendorLogKind

func IsVendorLogKind(k store.ExportKind) bool

IsVendorLogKind reports whether k is a credentialed vendor destination.

func LokiPushURL

func LokiPushURL(endpoint string) string

LokiPushURL appends Loki's push path when the endpoint has none.

func OTLPLogsURL

func OTLPLogsURL(endpoint string) string

OTLPLogsURL appends the standard logs path when the endpoint has none.

func OTLPURL

func OTLPURL(endpoint string) string

OTLPURL appends the standard metrics path when the endpoint has none.

func ParseSyslogEndpoint

func ParseSyslogEndpoint(raw string) (scheme, hostport string, err error)

ParseSyslogEndpoint splits "tcp://host:port" or "tls://host:port".

func RemoteWriteURL

func RemoteWriteURL(endpoint string) string

RemoteWriteURL returns the endpoint unchanged: remote write receivers differ in path (/api/v1/push, /api/prom/push, /api/v1/write), so none is guessed.

func SanitizePrefix

func SanitizePrefix(name string) string

SanitizePrefix turns a brand short name into a Prometheus metric prefix.

func Sign

func Sign(key string, at time.Time, body []byte) string

Sign returns the signature header value "t=<unix>,v1=<hex>" for body: an HMAC-SHA256 over "<unix>.<body>", so a captured batch cannot be replayed with a fresh timestamp.

func ValidLevel

func ValidLevel(v string) bool

ValidLevel reports whether v is an accepted min_level filter value.

func ValidateHTTPEndpoint

func ValidateHTTPEndpoint(ctx context.Context, raw string) error

ValidateHTTPEndpoint rejects, at configuration time, an endpoint that is not an absolute http(s) URL or that points at an internal address.

func ValidateSyslogEndpoint

func ValidateSyslogEndpoint(ctx context.Context, raw string) error

ValidateSyslogEndpoint applies the same internal-address rule as ValidateHTTPEndpoint to a syslog endpoint.

func VendorEndpoint

func VendorEndpoint(kind store.ExportKind, o store.ExportOptions) (string, error)

VendorEndpoint is the ingest URL for a vendor kind and its options.

func VerifySignature

func VerifySignature(key, header string, body []byte, now time.Time, tolerance time.Duration) error

VerifySignature checks a Sign header against body, rejecting timestamps further than tolerance from now (zero disables the age check).

func WritePrometheus

func WritePrometheus(w io.Writer, series []Series) error

WritePrometheus renders series in the Prometheus text exposition format, grouped by metric name in a stable order.

Types

type AuditRecord

type AuditRecord struct {
	ID          string `json:"id"`
	Time        string `json:"time"`
	ActorType   string `json:"actor_type"`
	ActorID     string `json:"actor_id"`
	ActorName   string `json:"actor_name"`
	Ability     string `json:"ability"`
	Method      string `json:"method"`
	Path        string `json:"path"`
	Status      int    `json:"status"`
	RemoteAddr  string `json:"remote_addr"`
	ClientKind  string `json:"client_kind"`
	AgentName   string `json:"agent_name,omitempty"`
	AgentClient string `json:"agent_client,omitempty"`
	Action      string `json:"action,omitempty"`
}

AuditRecord is the exported, redacted shape of one audit log row.

func NewAuditRecord

func NewAuditRecord(e store.AuditEntry) AuditRecord

NewAuditRecord converts a stored row, masking token-shaped strings, URL credentials and bearer values in every free-text field.

type Config

type Config struct {
	Resync          time.Duration
	QueueLines      int
	BatchLines      int
	BatchBytes      int
	FlushInterval   time.Duration
	MaxAttempts     int
	BackoffBase     time.Duration
	BackoffMax      time.Duration
	BreakerFailures int
	BreakerCooldown time.Duration
	SendTimeout     time.Duration
	StatusInterval  time.Duration
	AuditInterval   time.Duration
	AuditBatch      int
	AuditMaxRows    int
	MetricsInterval time.Duration
}

Config holds every interval and size the exporter uses. Zero values are replaced by defaults in ConfigFromEnv.

func ConfigFromEnv

func ConfigFromEnv(lookup func(string) (string, bool)) Config

ConfigFromEnv overlays APP_EXPORT_* variables onto the defaults. A value that does not parse or is not positive is ignored.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns the built-in defaults.

type Deps

type Deps struct {
	Store    Store
	Secrets  Secrets
	Logs     *telemetry.LogBroadcaster
	Metrics  MetricSource
	AppCount func(ctx context.Context) (int, error)
	Platform PlatformSource
	Objects  ObjectPutter
	Config   Config
	// SpoolDir, when set, keeps each log destination's backlog on disk so it
	// survives a restart. Empty keeps the backlog in memory only.
	SpoolDir   string
	Brand      string
	Logger     *slog.Logger
	HTTPClient *http.Client
	Now        func() time.Time
}

Deps wires a Service. Logs, Metrics, AppCount and Objects are optional: a destination kind that needs a missing one reports a clear error.

type Input

type Input struct {
	Name     string
	Kind     store.ExportKind
	Endpoint string
	Enabled  bool
	Apps     []string
	MinLevel string
	Options  store.ExportOptions
	Headers  map[string]string
	HMACKey  string
	// APIKey is the vendor key for the vendor kinds; it is stored as the
	// header that vendor reads and is never returned.
	APIKey string
}

Input is the writable part of a destination. Headers and HMACKey are write-only: they are stored encrypted and never returned.

type Label

type Label struct{ Name, Value string }

Label is one metric label. Only app, database, node and destination are ever emitted, which keeps cardinality bounded by what the operator runs.

type LogRecord

type LogRecord struct {
	Time    time.Time
	Kind    string
	App     string
	Stream  string
	Level   string
	Message string
	Fields  json.RawMessage
}

LogRecord is one container log line prepared for export.

func NewLogRecord

func NewLogRecord(e telemetry.LogEntry) LogRecord

NewLogRecord derives the export shape of a stored log line. The level is the structured "level" or "severity" field when the line is JSON, else "error" for stderr and "info" for stdout.

type MetricSource

type MetricSource interface {
	LatestByMetric(ctx context.Context, metric string) ([]telemetry.Sample, error)
}

MetricSource is the slice of the node-local telemetry store the metric exporters read: the newest sample per resource for one metric name.

type ObjectPutter

type ObjectPutter interface {
	PutObject(ctx context.Context, targetID, key string, body []byte, contentType string) error
}

ObjectPutter writes one object to a connected storage destination.

type Patch

type Patch struct {
	Name     *string
	Endpoint *string
	Enabled  *bool
	Apps     *[]string
	MinLevel *string
	Options  *store.ExportOptions
	// Headers replaces the stored secret headers (an empty map clears them).
	Headers *map[string]string
	// HMACKey replaces the signing key (an empty string clears it).
	HMACKey *string
	// APIKey replaces a vendor destination's key.
	APIKey *string
}

Patch is a partial update; nil fields are left unchanged.

type PlatformSource

type PlatformSource interface {
	ReadPlatformStats(ctx context.Context, now time.Time, window time.Duration) (store.PlatformStats, error)
}

PlatformSource reads control plane health numbers; *store.DB satisfies it.

type ResourceFilter

type ResourceFilter func(kind, name string) bool

ResourceFilter decides whether a resource may appear in the output, so a restricted caller of the scrape endpoint only sees what it may read.

type Secrets

type Secrets interface {
	SetValue(ctx context.Context, serviceName, envKey, plaintext string) error
	Resolve(ctx context.Context, serviceName, envKey string) (string, error)
	DeleteAll(ctx context.Context, serviceName string) error
}

Secrets stores destination credentials in envelope encryption.

type Series

type Series struct {
	Name    string
	Help    string
	Counter bool
	Labels  []Label
	Value   float64
	Time    time.Time
}

Series is one exported sample.

func CollectResourceSeries

func CollectResourceSeries(ctx context.Context, src MetricSource, prefix string, allow ResourceFilter) ([]Series, error)

CollectResourceSeries returns the latest per-resource container metrics labelled by app, database or node. Resources of other kinds are skipped.

func PlatformSeries

func PlatformSeries(prefix string, st store.PlatformStats, at time.Time) []Series

PlatformSeries turns a stats read into the fixed, unlabelled-by-resource self metric set. The only label is status on the deploy outcome gauge.

type Service

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

Service owns destination CRUD and the background workers.

func NewService

func NewService(d Deps) *Service

NewService builds a Service. Call Run to start delivery.

func (*Service) Create

func (s *Service) Create(ctx context.Context, in Input) (store.ExportDestination, error)

Create validates and stores a destination, then nudges the workers.

func (*Service) Delete

func (s *Service) Delete(ctx context.Context, id string) error

Delete removes a destination, its status and its stored secrets.

func (*Service) Get

Get returns one destination.

func (*Service) List

func (s *Service) List(ctx context.Context) ([]store.ExportDestination, error)

List returns every destination.

func (*Service) MetricPrefix

func (s *Service) MetricPrefix() string

MetricPrefix is the prefix every exported metric name carries.

func (*Service) Run

func (s *Service) Run(ctx context.Context)

Run reconciles the running workers against the stored destinations until ctx ends. A destination that fails to start is retried every resync.

func (*Service) Series

func (s *Service) Series(ctx context.Context, allow ResourceFilter) ([]Series, error)

Series returns the bounded metric set for scrape and OTLP push: per resource container metrics, platform gauges and exporter self metrics.

func (*Service) Status

func (s *Service) Status(ctx context.Context, id string) (store.ExportStatus, error)

Status returns the live delivery state of a destination, falling back to the persisted one when no worker is running.

func (*Service) StreamAudit

func (s *Service) StreamAudit(ctx context.Context, from, to time.Time, fn func(AuditRecord) error) error

StreamAudit calls fn for every redacted audit row in [from, to), oldest first, up to the configured row cap.

func (*Service) Test

func (s *Service) Test(ctx context.Context, id string) error

Test sends one probe through the destination without touching its queue, counters or cursor, and returns the delivery error if any.

func (*Service) Update

func (s *Service) Update(ctx context.Context, id string, p Patch) (store.ExportDestination, error)

Update applies a Patch, re-validating the merged result.

type Snapshot

type Snapshot struct {
	Sent        int64
	Dropped     int64
	Failures    int64
	QueueDepth  int64
	CircuitOpen bool
	LastSuccess time.Time
	LastError   string
	LastErrorAt time.Time
	CursorAt    string
	CursorID    string
}

Snapshot is a point-in-time copy of a destination's delivery counters.

type Store

type Store interface {
	SaveExportDestination(ctx context.Context, d store.ExportDestination) error
	UpdateExportDestination(ctx context.Context, d store.ExportDestination) error
	GetExportDestination(ctx context.Context, id string) (store.ExportDestination, error)
	ListExportDestinations(ctx context.Context) ([]store.ExportDestination, error)
	DeleteExportDestination(ctx context.Context, id string) error
	SaveExportStatus(ctx context.Context, s store.ExportStatus) error
	GetExportStatus(ctx context.Context, id string) (store.ExportStatus, error)
	ListAuditEntriesAfter(ctx context.Context, afterAt, afterID string, to *time.Time, limit int) ([]store.AuditEntry, error)
}

Store is the persistence surface the service needs; *store.DB satisfies it.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL