Documentation
¶
Overview ¶
Package sink defines the Sink interface and built-in sink implementations. Sinks are fire-and-forget output targets — separate from the structured store. Errors on sinks are logged, never fatal.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DecodeArchive ¶
DecodeArchive decodes one archive object: JSONL as FileSink and StdoutSink write it, or gzip-compressed JSONL as S3Sink writes it. The compression is detected from the content, not a file name.
Numbers in metadata decode as json.Number, not float64. A digest covers the metadata's JSON encoding, and json.Number re-encodes as exactly the literal that was archived, where a float64 loses integers above 2^53.
A malformed record stops the decode with an error naming it. Skipping it could never make a backfill accept something it should not, but it would quietly drop evidence, and an operator should see that.
Types ¶
type ArchiveReader ¶
type ArchiveReader interface {
// Name identifies the archive. It is written into the chain on any record
// recovered from it, so it should say where the archive lives.
Name() string
// ReadEvents calls fn for every event in the archive. loc says where the
// event was found, for reports. Returning an error from fn stops the read
// and returns that error.
ReadEvents(ctx context.Context, fn func(e *audit.Event, loc string) error) error
}
ArchiveReader reads back the events an archive sink wrote.
Sinks are write-only, and for everything but retention that is all they need to be. Reading exists for one job: recovering the evidence for purges that happened before retention records did (see Chronicle.BackfillRetention). A reader hands back what the archive holds and vouches for none of it. Anyone who can write to the archive can put anything there, so every event must be verified before it is believed.
type FileArchive ¶
type FileArchive struct {
// contains filtered or unexported fields
}
FileArchive reads what a FileSink wrote: every prefix-*.jsonl file in a directory, oldest name first. Files compressed after rotation, named prefix-*.jsonl.gz, are read too.
func NewFileArchive ¶
func NewFileArchive(dir, prefix string) *FileArchive
NewFileArchive reads the archive a NewFileSink(dir, prefix) wrote.
func (*FileArchive) Name ¶
func (a *FileArchive) Name() string
Name returns "file:" and the archive's directory and prefix.
func (*FileArchive) ReadEvents ¶
func (a *FileArchive) ReadEvents(ctx context.Context, fn func(e *audit.Event, loc string) error) error
ReadEvents decodes every archive file in name order.
type FileOption ¶
type FileOption func(*FileSink)
FileOption configures a FileSink.
func WithMaxSize ¶
func WithMaxSize(bytes int64) FileOption
WithMaxSize sets the maximum file size in bytes before rotation.
type FileSink ¶
type FileSink struct {
// contains filtered or unexported fields
}
FileSink writes events as JSONL to a file with optional rotation.
func NewFileSink ¶
func NewFileSink(dir, prefix string, opts ...FileOption) *FileSink
NewFileSink creates a FileSink writing JSONL to dir/prefix-*.jsonl.
type MultiSink ¶
type MultiSink struct {
// contains filtered or unexported fields
}
MultiSink fans out events to multiple sinks.
A failing sink does not stop the others: every sink is always attempted. The combined failure is then returned, so a caller that depends on the write having landed can tell. Returning nil here made MultiSink unsafe as a retention archive target, because the enforcer checks the archive write's error before purging and would delete events that were never archived.
func NewMultiSink ¶
NewMultiSink creates a MultiSink that fans out to the given sinks.
type S3Archive ¶
type S3Archive struct {
// contains filtered or unexported fields
}
S3Archive reads what an S3Sink wrote: every events-*.jsonl.gz object under its prefix, in key order.
func NewS3Archive ¶
NewS3Archive reads the archive a NewS3Sink(writer, bucket, prefix) wrote.
type S3Reader ¶
type S3Reader interface {
// ListObjects returns every key in bucket that starts with prefix.
ListObjects(ctx context.Context, bucket, prefix string) ([]string, error)
// GetObject opens the object at key.
GetObject(ctx context.Context, bucket, key string) (io.ReadCloser, error)
}
S3Reader is the read half of what an S3Sink needs, kept to two calls for the same reason S3Writer is: no AWS SDK dependency.
type S3Sink ¶
type S3Sink struct {
// contains filtered or unexported fields
}
S3Sink archives events to S3 as gzip-compressed JSONL files.
Key format: {prefix}/{category}/{year}/{month}/{day}/events-{digest}.jsonl.gz
The digest is over the batch's compressed bytes. A fixed name per category+day meant a second flush on the same day replaced the first archive, and since the retention enforcer purges the source rows once the archive write returns, the first batch was lost from both S3 and the database. Deriving the suffix from content also makes a retried flush idempotent: the same batch resolves to the same key rather than accumulating duplicates.
func (*S3Sink) Flush ¶
Flush compresses buffered events as gzip JSONL and uploads to S3. Events are partitioned by category and date.
The buffer is only cleared once every partition has uploaded. A failed flush leaves the events in place so the caller can retry, rather than dropping the batch on the floor.
type S3Writer ¶
type S3Writer interface {
// PutObject uploads data to S3 at the given key.
PutObject(ctx context.Context, bucket, key string, body io.Reader) error
}
S3Writer defines the minimal interface for writing objects to S3. This abstracts away the specific AWS SDK to keep the dependency optional.
type Sink ¶
type Sink interface {
Name() string
Write(ctx context.Context, events []*audit.Event) error
Flush(ctx context.Context) error
Close() error
}
Sink writes events to an output destination. Unlike the store, sinks are fire-and-forget — errors are logged, not fatal.
type StdoutSink ¶
type StdoutSink struct {
// contains filtered or unexported fields
}
StdoutSink writes events as JSON to an io.Writer (defaults to os.Stdout).
func NewStdoutSink ¶
func NewStdoutSink(w io.Writer) *StdoutSink
NewStdoutSink creates a StdoutSink writing to the given writer.
func (*StdoutSink) Close ¶
func (s *StdoutSink) Close() error
func (*StdoutSink) Name ¶
func (s *StdoutSink) Name() string
Source Files