Documentation
¶
Overview ¶
Package logfwd provides log forwarding from plexd mesh nodes to the control plane.
Index ¶
- Constants
- type Config
- type FileSource
- type FilterConfig
- type FilteringSource
- type Forwarder
- func (f *Forwarder) RegisterSource(s LogSource)
- func (f *Forwarder) RingBuffer() *RingBuffer
- func (f *Forwarder) Run(ctx context.Context) error
- func (f *Forwarder) SetRingBuffer(rb *RingBuffer)
- func (f *Forwarder) Status() (enabled bool, bufferSize, sourceCount, errorCount int, lastReport time.Time)
- type IngestClient
- type JournalEntry
- type JournalReader
- type JournalctlReader
- type JournaldSource
- type LocalReporter
- type LogReporter
- type LogSource
- type MultiReporter
- type PlatformReporter
- type RingBuffer
- type SecretFetcher
Constants ¶
const DefaultBatchSize = 200
DefaultBatchSize is the default maximum number of log entries per report batch.
const DefaultCollectInterval = 10 * time.Second
DefaultCollectInterval is the default interval between log collection cycles.
const DefaultReportInterval = 30 * time.Second
DefaultReportInterval is the default interval between reporting logs to the control plane.
const DefaultRingBufferCapacity = 1000
DefaultRingBufferCapacity is the default number of log lines retained.
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.
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").
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.
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 ¶
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) SetRingBuffer ¶
func (f *Forwarder) SetRingBuffer(rb *RingBuffer)
SetRingBuffer attaches a ring buffer for log snapshot support. Must be called before Run.
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 ¶
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.
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 ¶
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 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 ¶
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
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.