logfwd

package
v0.5.0 Latest Latest
Warning

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

Go to latest
Published: Aug 3, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Overview

Package logfwd provides log forwarding from plexd mesh nodes to the control plane.

Index

Constants

View Source
const DefaultBatchSize = 200

DefaultBatchSize is the default maximum number of log entries per report batch.

View Source
const DefaultCollectInterval = 10 * time.Second

DefaultCollectInterval is the default interval between log collection cycles.

View Source
const DefaultReportInterval = 30 * time.Second

DefaultReportInterval is the default interval between reporting logs to the control plane.

View Source
const DefaultRingBufferCapacity = 1000

DefaultRingBufferCapacity is the default number of log lines retained.

View Source
const MaxLineBytes = 16384

MaxLineBytes is the maximum line length before truncation.

Variables

This section is empty.

Functions

This section is empty.

Types

type Config

type Config struct {
	// Enabled controls whether log forwarding is active.
	// Default: true (set by ApplyDefaults).
	Enabled bool `yaml:"enabled"`

	// CollectInterval is the interval between collection cycles.
	// Must be at least 5s.
	CollectInterval time.Duration `yaml:"collect_interval"`

	// ReportInterval is the interval between reporting to the control plane.
	// Must be >= CollectInterval.
	ReportInterval time.Duration `yaml:"report_interval"`

	// BatchSize is the maximum number of log entries per report batch.
	// Must be at least 1. Default: 200.
	BatchSize int `yaml:"batch_size"`

	// FilePatterns is a list of glob patterns for file-based log collection.
	// Example: ["/var/log/app/*.log", "/var/log/syslog"]
	FilePatterns []string `yaml:"file_patterns"`

	// Filter defines optional log filtering rules.
	// When non-empty, the Forwarder wraps sources with a FilteringSource.
	Filter FilterConfig `yaml:"filter"`

	// LocalEndpoint configures an optional local data-plane endpoint.
	LocalEndpoint api.LocalEndpointConfig `yaml:"local_endpoint"`
}

Config holds the configuration for log forwarding.

func (*Config) ApplyDefaults

func (c *Config) ApplyDefaults()

ApplyDefaults sets default values for zero-valued fields. On a zero-valued Config, Enabled defaults to true. To disable log forwarding, set Enabled=false before or after calling ApplyDefaults.

func (*Config) Validate

func (c *Config) Validate() error

Validate checks that configuration values are within acceptable ranges.

type FileSource

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

FileSource implements LogSource by reading from log files matching a glob pattern.

func NewFileSource

func NewFileSource(pattern, hostname string, logger *slog.Logger) *FileSource

NewFileSource creates a new FileSource. pattern is a glob expression (e.g., "/var/log/app/*.log").

func (*FileSource) Collect

func (s *FileSource) Collect(ctx context.Context) ([]api.LogEntry, error)

Collect reads new lines from all files matching the glob pattern.

type FilterConfig

type FilterConfig struct {
	// MinSeverity drops entries below this severity level.
	// Empty string means no severity filtering.
	MinSeverity string `yaml:"min_severity"`

	// IncludeUnits, if non-empty, only passes entries matching one of these unit names.
	IncludeUnits []string `yaml:"include_units"`

	// ExcludeUnits drops entries matching any of these unit names.
	ExcludeUnits []string `yaml:"exclude_units"`
}

FilterConfig defines log filtering rules.

func (*FilterConfig) IsEmpty

func (fc *FilterConfig) IsEmpty() bool

IsEmpty returns true if no filtering rules are configured.

type FilteringSource

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

FilteringSource wraps a LogSource and applies filter rules to its output.

func NewFilteringSource

func NewFilteringSource(inner LogSource, config FilterConfig) *FilteringSource

NewFilteringSource creates a filtering wrapper around a LogSource.

func (*FilteringSource) Collect

func (f *FilteringSource) Collect(ctx context.Context) ([]api.LogEntry, error)

Collect reads from the inner source and applies filters.

type Forwarder

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

Forwarder orchestrates log collection and reporting.

func NewForwarder

func NewForwarder(cfg Config, sources []LogSource, reporter LogReporter, nodeID string, hostname string, logger *slog.Logger) *Forwarder

NewForwarder creates a new Forwarder. Config defaults are applied automatically. When cfg.Filter is non-empty, all sources are wrapped with a FilteringSource.

func (*Forwarder) RegisterSource

func (f *Forwarder) RegisterSource(s LogSource)

RegisterSource adds a log source to the forwarder. Must be called before Run; it is not safe for concurrent use.

func (*Forwarder) RingBuffer

func (f *Forwarder) RingBuffer() *RingBuffer

RingBuffer returns the attached ring buffer, or nil if none is set.

func (*Forwarder) Run

func (f *Forwarder) Run(ctx context.Context) error

Run starts the collect and report loops. It blocks until ctx is cancelled.

func (*Forwarder) SetRingBuffer

func (f *Forwarder) SetRingBuffer(rb *RingBuffer)

SetRingBuffer attaches a ring buffer for log snapshot support. Must be called before Run.

func (*Forwarder) Status

func (f *Forwarder) Status() (enabled bool, bufferSize, sourceCount, errorCount int, lastReport time.Time)

Status returns a snapshot of the forwarder's operational status.

type IngestClient added in v0.2.0

type IngestClient interface {
	ReportLogs(ctx context.Context, nodeID string, lines []api.LogLine) (*api.IngestReceipt, error)
}

IngestClient is the control-plane seam a PlatformReporter reports wire log lines through. It is defined here (rather than reused from the api package) so the logfwd package owns the exact shape it depends on.

type JournalEntry

type JournalEntry struct {
	Timestamp time.Time
	Message   string
	Priority  int
	Unit      string
}

JournalEntry represents a single entry read from the systemd journal.

type JournalReader

type JournalReader interface {
	ReadEntries(ctx context.Context) ([]JournalEntry, error)
}

JournalReader abstracts systemd journal access for testability.

type JournalctlReader

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

JournalctlReader implements JournalReader by running the journalctl subprocess. It tracks cursor position to avoid re-reading entries across collection cycles.

func NewJournalctlReader

func NewJournalctlReader() *JournalctlReader

NewJournalctlReader creates a new JournalctlReader.

func (*JournalctlReader) ReadEntries

func (r *JournalctlReader) ReadEntries(ctx context.Context) ([]JournalEntry, error)

ReadEntries runs journalctl and parses JSON output. On the first call, it reads entries from the last 60 seconds. Subsequent calls read entries after the stored cursor.

type JournaldSource

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

JournaldSource implements LogSource by reading from the systemd journal.

func NewJournaldSource

func NewJournaldSource(reader JournalReader, hostname string, logger *slog.Logger) *JournaldSource

NewJournaldSource creates a new JournaldSource.

func (*JournaldSource) Collect

func (s *JournaldSource) Collect(ctx context.Context) ([]api.LogEntry, error)

Collect reads journal entries and maps them to api.LogEntry values.

type LocalReporter

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

LocalReporter sends log batches to a local data-plane endpoint. It implements the LogReporter interface.

func NewLocalReporter

func NewLocalReporter(cfg api.LocalEndpointConfig, fetcher SecretFetcher, nsk []byte, nodeID string, logger *slog.Logger) *LocalReporter

NewLocalReporter creates a LocalReporter from the given configuration.

func (*LocalReporter) ReportLogs

func (r *LocalReporter) ReportLogs(ctx context.Context, nodeID string, batch api.LogBatch) error

ReportLogs sends a log batch to the local endpoint with bearer token authentication.

type LogReporter

type LogReporter interface {
	ReportLogs(ctx context.Context, nodeID string, batch api.LogBatch) error
}

LogReporter abstracts the control plane log reporting API.

type LogSource

type LogSource interface {
	Collect(ctx context.Context) ([]api.LogEntry, error)
}

LogSource collects log entries from a specific source.

type MultiReporter

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

MultiReporter dispatches log batches to both a platform and a local reporter in parallel. The platform reporter's error is returned; local reporter errors are logged but do not affect the return value.

func NewMultiReporter

func NewMultiReporter(platform, local LogReporter, logger *slog.Logger) *MultiReporter

NewMultiReporter creates a MultiReporter that sends logs to both reporters.

func (*MultiReporter) ReportLogs

func (m *MultiReporter) ReportLogs(ctx context.Context, nodeID string, batch api.LogBatch) error

ReportLogs sends the log batch to both reporters concurrently. Only the platform reporter's error is returned.

type PlatformReporter added in v0.2.0

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

PlatformReporter implements LogReporter by converting each log entry into a wire LogLine and posting them to the platform ingest endpoint through an IngestClient.

func NewPlatformReporter added in v0.2.0

func NewPlatformReporter(client IngestClient, logger *slog.Logger) *PlatformReporter

NewPlatformReporter creates a PlatformReporter that reports through client.

func (*PlatformReporter) ReportLogs added in v0.2.0

func (r *PlatformReporter) ReportLogs(ctx context.Context, nodeID string, batch api.LogBatch) error

ReportLogs converts the batch into wire log lines and posts them through the ingest client. An entry with an empty message or a zero timestamp is skipped (the contract requires both and would 400 the whole batch); a severity outside the wire set is coerced to "info". When no lines survive, it returns nil without calling the client (the ingest contract rejects an empty array). Every line the reporter discards is counted and periodically summarized, because dropping is invisible to the forwarder's own status.

type RingBuffer

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

RingBuffer is a thread-safe, fixed-capacity circular buffer for log entries. When the buffer is full, the oldest entry is overwritten.

func NewRingBuffer

func NewRingBuffer(capacity int) *RingBuffer

NewRingBuffer creates a RingBuffer with the given capacity. If capacity is <= 0, DefaultRingBufferCapacity is used.

func (*RingBuffer) Len

func (rb *RingBuffer) Len() int

Len returns the number of entries currently stored.

func (*RingBuffer) Recent

func (rb *RingBuffer) Recent(n int) []api.LogEntry

Recent returns the most recent n entries in chronological order. If n exceeds the number of stored entries, all stored entries are returned.

func (*RingBuffer) RecentLines

func (rb *RingBuffer) RecentLines(n int) []string

RecentLines returns the message field of the most recent n entries.

func (*RingBuffer) Write

func (rb *RingBuffer) Write(entry api.LogEntry)

Write adds an entry to the ring buffer, overwriting the oldest if full.

func (*RingBuffer) WriteBatch

func (rb *RingBuffer) WriteBatch(entries []api.LogEntry)

WriteBatch adds multiple entries to the ring buffer.

type SecretFetcher

type SecretFetcher interface {
	FetchSecret(ctx context.Context, nodeID, name string, version int) (*api.SecretEnvelope, error)
}

SecretFetcher abstracts fetching encrypted secrets from the control plane.

Jump to

Keyboard shortcuts

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