Documentation
¶
Overview ¶
Package sinks provides production audit.Sink implementations (syslog, webhook, file) plus a Fanout multiplexer and config wiring. All implementations are stdlib-only: no third-party logging or HTTP clients.
Constraint: sinks must never block the caller's goroutine beyond the time required for a buffered channel send (webhook, syslog) or a local syscall (file). Drop counters are incremented and logged on overflow; events are never silently discarded.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ParseSinks ¶
ParseSinks parses cfgJSON and constructs the enabled sinks, returning them as a slice suitable for wrapping in a Fanout. The caller is responsible for calling Close() on any Closer sinks (SyslogSink, FileSink) on shutdown.
ParseSinks never partially constructs: if any enabled sink fails to initialise it returns all previously-initialised sinks alongside the error so the caller can close them.
Types ¶
type Config ¶
type Config struct {
Syslog *SyslogConfig `json:"syslog,omitempty"`
Webhook *WebhookConfig `json:"webhook,omitempty"`
File *FileConfig `json:"file,omitempty"`
}
Config is the top-level JSON configuration block for the sinks subsystem. Example:
{
"syslog": {"network": "udp", "addr": "localhost:514"},
"webhook": {"url": "https://siem.example.com/ingest", "bearer_token": "...", "batch_size": 50},
"file": {"path": "/var/log/wardyn/audit.log", "max_bytes": 52428800, "keep": 3}
}
Omitting a top-level key disables that sink entirely.
type Fanout ¶
type Fanout struct {
// contains filtered or unexported fields
}
Fanout multiplexes a single audit event stream to multiple audit.Sink children. Per-child failures are isolated: an error from one child is logged (and counted) but never propagates to sibling sinks or to the caller.
Emit returns nil unless ALL children fail, in which case it returns the last observed error. This preserves the recorder's ability to detect a total fan-out failure while satisfying the isolation requirement.
func NewFanout ¶
NewFanout creates a Fanout over the supplied sinks. Each sink in children is wrapped in its own childState with an independent drop counter.
func (*Fanout) Close ¶
Close closes every child sink that implements io.Closer, returning the first error encountered (after attempting to close all of them). Buffering sinks (webhook, syslog) block in Close until their final batch has been flushed, so calling Fanout.Close on graceful shutdown ensures the last events are drained and awaited rather than abandoned (finding: sinks were never Closed on shutdown, so the webhook drain goroutine was never awaited).
func (*Fanout) Drops ¶
Drops returns the total drop count for the named child sink. Returns -1 if no child with that name is found.
The total aggregates two independent sources of loss:
- the fanout-local counter, incremented when the child's Emit returns an error (synchronous failure), and
- the child's own Drops() counter, if it implements dropper — buffering sinks (webhook, syslog) drop asynchronously and report nil from Emit, so without this their losses would be structurally invisible (Drops was always 0 for them).
func (*Fanout) Emit ¶
Emit delivers ev to every child sink concurrently (one goroutine per child). Per-child panics are recovered; errors are logged and counted. Emit blocks until every child has returned.
If every child returns an error Emit returns the last error seen; if at least one child succeeds Emit returns nil.
type FileConfig ¶
type FileConfig struct {
// Path is the log file path (required). Rotated files are named
// <path>.1, <path>.2, … <path>.N.
Path string `json:"path"`
// MaxBytes is the maximum size of the active log file before rotation
// (default 100 MiB). 0 means no rotation.
MaxBytes int64 `json:"max_bytes,omitempty"`
// Keep is the number of rotated files to retain (default 5). The active
// file is not counted; older files beyond Keep are deleted.
Keep int `json:"keep,omitempty"`
}
FileConfig holds the configuration for a FileSink.
type FileSink ¶
type FileSink struct {
// contains filtered or unexported fields
}
FileSink appends JSON-lines audit events to a file, rotating when the file exceeds MaxBytes. Up to Keep rotated files are retained; older ones are deleted. All operations are serialised by a mutex — the sink is safe for concurrent Emit calls.
func NewFileSink ¶
func NewFileSink(cfg FileConfig) (*FileSink, error)
NewFileSink opens (or creates) the log file at cfg.Path. Returns an error if the file cannot be opened.
type SyslogConfig ¶
type SyslogConfig struct {
// Network is the transport: "tcp", "udp", or "" for local socket.
Network string `json:"network,omitempty"`
// Addr is the remote endpoint, e.g. "host:514". Empty = local socket.
Addr string `json:"addr,omitempty"`
}
SyslogConfig is the JSON-serialisable counterpart of SyslogSink fields.
type SyslogSink ¶
type SyslogSink struct {
// Network is the syslog transport: "tcp", "udp", or "" for local socket.
Network string
// Addr is the syslog endpoint, e.g. "host:514". Ignored when Network is "".
Addr string
// contains filtered or unexported fields
}
SyslogSink emits audit events to the system syslog daemon as RFC 5424-ish messages (using Go's log/syslog, which writes RFC 3164 to local sockets and RFC 5424-ish via the "wardyn" tag). The JSON-serialised AuditEvent is the message body.
EVERY transport — the local socket (/dev/log on Linux, /var/run/syslog on macOS; Network=="") and a remote "tcp"/"udp" collector — routes through a bounded async buffer drained by a single background writer goroutine, so a wedged local syslog daemon, a full /dev/log datagram peer buffer, or a hung remote collector can never block the calling request handler (Fanout.Emit waits for every child synchronously from the request path).
func NewSyslogSink ¶
func NewSyslogSink(network, addr string) (*SyslogSink, error)
NewSyslogSink constructs and dials the syslog connection. Returns an error if the connection cannot be established.
A single background writer goroutine is started for every transport so that Emit stays non-blocking even if the daemon/collector stalls; it is shut down by Close.
func (*SyslogSink) Close ¶
func (s *SyslogSink) Close() error
Close signals the background writer to drain and stop, then closes the underlying syslog connection.
func (*SyslogSink) Drops ¶
func (s *SyslogSink) Drops() int64
Drops returns the number of events dropped due to buffer overflow or write timeout. Exposed so the never-drop-silently invariant can be observed, matching WebhookSink.Drops.
func (*SyslogSink) Emit ¶
func (s *SyslogSink) Emit(ctx context.Context, ev types.AuditEvent) error
Emit serialises ev to JSON and hands it to the background writer as an INFO syslog entry. A cancelled context causes the write to be skipped without error (the recorder is shutting down).
Emit NEVER blocks: it performs only a non-blocking enqueue onto a bounded buffer for every transport. If the buffer is full (the daemon/collector is hung and the background writer is stalled) the event is dropped and counted rather than blocking the caller's request goroutine. This prevents a hung remote collector OR a wedged local syslog daemon from stalling the synchronous request path via Fanout.Emit.
type WebhookConfig ¶
type WebhookConfig struct {
// URL is the HTTP endpoint that receives JSON-lines batches (required).
URL string `json:"url"`
// BearerToken is included as "Authorization: Bearer <token>" when non-empty.
// Setting it requires an https:// URL (see NewWebhookSink).
BearerToken string `json:"bearer_token,omitempty"`
// BatchSize is the maximum number of events per HTTP POST (default 100).
BatchSize int `json:"batch_size,omitempty"`
// FlushInterval is how long to wait before flushing a partial batch
// (default 5s). Parsed as a duration string, e.g. "5s".
FlushInterval string `json:"flush_interval,omitempty"`
// BufferSize is the capacity of the in-process event queue (default 4096).
// Events that would overflow the buffer are dropped and counted.
BufferSize int `json:"buffer_size,omitempty"`
// MaxRetries is the number of delivery attempts per batch (default 3).
MaxRetries int `json:"max_retries,omitempty"`
// RetryBaseDelay is the initial backoff delay (default 200ms).
RetryBaseDelay string `json:"retry_base_delay,omitempty"`
}
WebhookConfig holds the configuration for a WebhookSink.
type WebhookSink ¶
type WebhookSink struct {
// contains filtered or unexported fields
}
WebhookSink buffers audit events and delivers them in JSON-lines batches via HTTP POST. It never blocks the Emit caller beyond a non-blocking channel send; overflow increments the drop counter which is logged periodically.
Start the background flusher with Run(ctx); cancel the context to drain and stop cleanly. Alternatively call Close() to signal the drain and block until the final batch has been flushed (used on graceful shutdown so the last batch is awaited rather than abandoned).
func NewWebhookSink ¶
func NewWebhookSink(cfg WebhookConfig) (*WebhookSink, error)
NewWebhookSink creates a WebhookSink from cfg. Returns an error if cfg.URL is empty, durations cannot be parsed, or a bearer_token is configured on a non-https URL.
func (*WebhookSink) Close ¶
func (w *WebhookSink) Close() error
Close signals the background flusher to drain and stop, then blocks until the final batch has been flushed. It must only be called after Run has been started (the fanout always starts Run for this sink); otherwise the wait on done would block forever. It is safe to call Close multiple times and concurrently with a ctx-driven Run shutdown.
func (*WebhookSink) Drops ¶
func (w *WebhookSink) Drops() int64
Drops returns the number of events dropped due to buffer overflow.
func (*WebhookSink) Emit ¶
func (w *WebhookSink) Emit(_ context.Context, ev types.AuditEvent) error
Emit enqueues ev for delivery. If the buffer is full the event is dropped and the drop counter is incremented. Emit never blocks.
func (*WebhookSink) Run ¶
func (w *WebhookSink) Run(ctx context.Context)
Run starts the background flush loop. It returns when ctx is cancelled or Close() is called, after flushing any events remaining in the buffer (best-effort). Run must be called exactly once per WebhookSink.
Source Files