logsinks

package
v0.5.7 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: MIT Imports: 11 Imported by: 0

Documentation

Overview

Package logsinks persists per-environment osquery log sink configurations to the database. Each row is one sink instance (Splunk, Kafka, S3, DB, stdout, …) with a JSON-encoded config blob whose shape is determined by the sink Type and validated against the Registry.

osctrl-tls reads these rows at boot and whenever it consumes a reload_log_sinks service command, building a per-environment exporter fan-out. Rows with EnvironmentID=0 are the global fallback used by any environment that has no rows of its own. Rows with EnvironmentID != 0 override the global set for that environment.

The YAML logger: section remains the seed source on first boot; once an operator edits a sink through the API (Source="db") the YAML value is ignored for that sink until the DB row is deleted.

Index

Constants

View Source 
const NoEnvironmentID uint = 0

NoEnvironmentID is the sentinel for global (non-env-scoped) sink rows. Mirrors settings.NoEnvironmentID / serviceconfig.NoEnvironmentID so callers can pass the same constant regardless of package.

View Source 
const SourceDB = "db"

SourceDB marks a row that has been edited or created through the API.

View Source 
const SourceService = "service"

SourceService marks a row seeded from the resolved service configuration — flags, environment variables, or YAML, whichever the operator used. The seed is a one-time create-if-missing bootstrap; once a row exists it is never overwritten by a later boot.

View Source 
const SourceYAML = "yaml"

SourceYAML is retained for backwards compatibility with rows seeded by older versions that used "yaml" as the source value. New seeds use SourceService. Reads treat both as non-DB-edited.

Variables

View Source 
var ErrInvalidSinkConfig = errors.New("invalid log sink configuration")

ErrInvalidSinkConfig is returned when a sink's Config JSON is invalid or fails to decode into the type's config struct.

View Source 
var ErrInvalidSinkType = errors.New("invalid log sink type")

ErrInvalidSinkType is returned when a sink type is not in the Registry.

View Source 
var ErrSinkNotFound = errors.New("log sink not found")

ErrSinkNotFound is returned when a sink row is not found.

View Source 
var ErrTargetHasSinks = errors.New("target environment already has log sinks")

ErrTargetHasSinks is returned by Clone when the target environment already has sinks and overwrite is false.

View Source 
var Registry = map[string]SinkSpec{
	config.LoggingNone: {
		Type:        config.LoggingNone,
		Description: "Discard logs. No destination.",
		HasSecret:   false,
		Decode: func(raw json.RawMessage) (any, error) {

			return struct{}{}, nil
		},
		Build: func(_ any, smgr *settings.Settings) (logging.DataExporter, error) {
			n, err := logging.CreateLoggerNone()
			if err != nil {
				return nil, err
			}
			n.Settings(smgr)
			return n, nil
		},
	},
	config.LoggingStdout: {
		Type:        config.LoggingStdout,
		Description: "Write logs to stdout.",
		HasSecret:   false,
		Decode: func(raw json.RawMessage) (any, error) {
			return struct{}{}, nil
		},
		Build: func(_ any, smgr *settings.Settings) (logging.DataExporter, error) {
			s, err := logging.CreateLoggerStdout()
			if err != nil {
				return nil, err
			}
			s.Settings(smgr)
			return s, nil
		},
	},
	config.LoggingFile: {
		Type:        config.LoggingFile,
		Description: "Rotating local file logger.",
		HasSecret:   false,
		Fields: []FieldSpec{
			{Name: "filePath", Label: "File path", Type: FieldString, Required: true, Placeholder: "/var/log/osquery.log", Help: "Absolute path to the log file."},
			{Name: "maxSize", Label: "Max size (MB)", Type: FieldInteger, Default: 100, Help: "Rotate after this size in MB."},
			{Name: "maxBackups", Label: "Max backups", Type: FieldInteger, Default: 3, Help: "Number of rotated files to keep."},
			{Name: "maxAge", Label: "Max age (days)", Type: FieldInteger, Default: 28, Help: "Days to retain rotated files."},
			{Name: "compress", Label: "Compress", Type: FieldBoolean, Default: false, Help: "Gzip rotated files."},
		},
		Decode: decodeTyped[config.LocalLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			f, err := logging.CreateLoggerFile(cfg.(*config.LocalLogger))
			if err != nil {
				return nil, err
			}
			f.Settings(smgr)
			return f, nil
		},
	},
	config.LoggingDB: {
		Type:         config.LoggingDB,
		Description:  "Persist logs to a SQL database (separate from or same as the primary).",
		HasSecret:    true,
		SecretFields: []string{"password"},
		Fields: []FieldSpec{
			{Name: "type", Label: "DB type", Type: FieldSelect, Required: true, Options: []string{"postgres", "mysql", "sqlite"}, Default: "postgres", Help: "Database backend."},
			{Name: "host", Label: "Host", Type: FieldString, Placeholder: "127.0.0.1", Help: "Database host. Ignored by sqlite."},
			{Name: "port", Label: "Port", Type: FieldInteger, Default: 5432, Help: "Database port. Defaults match PostgreSQL."},
			{Name: "name", Label: "Database name", Type: FieldString, Required: true, Placeholder: "osctrl", Help: "Database/schema name."},
			{Name: "username", Label: "Username", Type: FieldString, Placeholder: "postgres"},
			{Name: "password", Label: "Password", Type: FieldPassword, Secret: true, Placeholder: "prefer env vars/secrets in prod", Help: "Database password."},
			{Name: "sslmode", Label: "SSL mode", Type: FieldString, Placeholder: "disable", Help: "PostgreSQL SSL mode: disable, require, verify-full."},
			{Name: "maxIdleConns", Label: "Max idle conns", Type: FieldInteger, Default: 20},
			{Name: "maxOpenConns", Label: "Max open conns", Type: FieldInteger, Default: 100},
			{Name: "connMaxLifetime", Label: "Conn max lifetime (min)", Type: FieldInteger, Default: 30},
			{Name: "connRetry", Label: "Conn retry (s)", Type: FieldInteger, Default: 10, Help: "Seconds to retry at startup; 0 fails fast."},
			{Name: "filePath", Label: "SQLite file path", Type: FieldString, Placeholder: "./osctrl.db", Help: "SQLite database file path when type is sqlite."},
		},
		Decode: decodeTyped[config.YAMLConfigurationDB](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			d, err := logging.CreateLoggerDBConfig(cfg.(*config.YAMLConfigurationDB))
			if err != nil {
				return nil, err
			}
			d.Settings(smgr)
			return d, nil
		},
	},
	config.LoggingSplunk: {
		Type:         config.LoggingSplunk,
		Description:  "Splunk HTTP Event Collector (HEC).",
		HasSecret:    true,
		SecretFields: []string{"token"},
		Fields: []FieldSpec{
			{Name: "url", Label: "HEC URL", Type: FieldString, Required: true, Placeholder: "https://splunk.example.com:8088/services/collector", Help: "Splunk HTTP Event Collector endpoint."},
			{Name: "token", Label: "HEC token", Type: FieldPassword, Secret: true, Required: true, Placeholder: "prefer env vars/secrets in prod", Help: "Splunk HEC token."},
			{Name: "host", Label: "Host", Type: FieldString, Placeholder: "osctrl", Help: "Source host value attached to events."},
			{Name: "index", Label: "Index", Type: FieldString, Placeholder: "main", Help: "Splunk index name."},
		},
		Decode: decodeTyped[config.SplunkLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			s, err := logging.CreateLoggerSplunk(cfg.(*config.SplunkLogger))
			if err != nil {
				return nil, err
			}
			s.Settings(smgr)
			return s, nil
		},
	},
	config.LoggingGraylog: {
		Type:        config.LoggingGraylog,
		Description: "Graylog GELF over HTTP.",
		HasSecret:   false,
		Fields: []FieldSpec{
			{Name: "url", Label: "URL", Type: FieldString, Required: true, Placeholder: "https://graylog.example.com:12202/gelf", Help: "Graylog GELF HTTP input endpoint."},
			{Name: "host", Label: "Host", Type: FieldString, Placeholder: "osctrl", Help: "Source host value attached to messages."},
			{Name: "queries", Label: "Queries stream", Type: FieldString, Placeholder: "osquery_queries", Help: "Stream/name for distributed query logs."},
			{Name: "status", Label: "Status stream", Type: FieldString, Placeholder: "osquery_status", Help: "Stream/name for status logs."},
			{Name: "results", Label: "Results stream", Type: FieldString, Placeholder: "osquery_results", Help: "Stream/name for result logs."},
		},
		Decode: decodeTyped[config.GraylogLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			g, err := logging.CreateLoggerGraylog(cfg.(*config.GraylogLogger))
			if err != nil {
				return nil, err
			}
			g.Settings(smgr)
			return g, nil
		},
	},
	config.LoggingLogstash: {
		Type:        config.LoggingLogstash,
		Description: "Logstash over HTTP, TCP, or UDP.",
		HasSecret:   false,
		Fields: []FieldSpec{
			{Name: "host", Label: "Host", Type: FieldString, Required: true, Placeholder: "logstash.example.com"},
			{Name: "port", Label: "Port", Type: FieldString, Required: true, Placeholder: "5044", Help: "Logstash input port (string in the config)."},
			{Name: "protocol", Label: "Protocol", Type: FieldSelect, Options: []string{"http", "tcp", "udp"}, Default: "http"},
			{Name: "path", Label: "Path", Type: FieldString, Placeholder: "/osquery", Help: "Optional HTTP/TCP path depending on protocol."},
		},
		Decode: decodeTyped[config.LogstashLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			l, err := logging.CreateLoggerLogstash(cfg.(*config.LogstashLogger))
			if err != nil {
				return nil, err
			}
			l.Settings(smgr)
			return l, nil
		},
	},
	config.LoggingKinesis: {
		Type:         config.LoggingKinesis,
		Description:  "AWS Kinesis Data Streams.",
		HasSecret:    true,
		SecretFields: []string{"secretKey", "sessionToken"},
		Fields: []FieldSpec{
			{Name: "stream", Label: "Stream", Type: FieldString, Required: true, Placeholder: "osquery-logs", Help: "Kinesis stream name."},
			{Name: "region", Label: "Region", Type: FieldString, Required: true, Placeholder: "us-east-1"},
			{Name: "endpoint", Label: "Endpoint", Type: FieldString, Placeholder: "https://kinesis.us-east-1.amazonaws.com", Help: "Optional custom endpoint."},
			{Name: "accessKey", Label: "Access key ID", Type: FieldString, Placeholder: "prefer instance/task roles", Help: "Optional static access key."},
			{Name: "secretKey", Label: "Secret access key", Type: FieldPassword, Secret: true, Placeholder: "prefer instance/task roles"},
			{Name: "sessionToken", Label: "Session token", Type: FieldPassword, Secret: true, Placeholder: "optional"},
		},
		Decode: decodeTyped[config.KinesisLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			k, err := logging.CreateLoggerKinesis(cfg.(*config.KinesisLogger))
			if err != nil {
				return nil, err
			}
			k.Settings(smgr)
			return k, nil
		},
	},
	config.LoggingS3: {
		Type:         config.LoggingS3,
		Description:  "AWS S3 object storage.",
		HasSecret:    true,
		SecretFields: []string{"secretAccessKey"},
		Fields: []FieldSpec{
			{Name: "bucket", Label: "Bucket", Type: FieldString, Required: true, Placeholder: "osquery-logs", Help: "S3 bucket name for log objects."},
			{Name: "region", Label: "Region", Type: FieldString, Required: true, Placeholder: "us-east-1"},
			{Name: "accessKey", Label: "Access key ID", Type: FieldString, Placeholder: "prefer instance/task roles"},
			{Name: "secretAccessKey", Label: "Secret access key", Type: FieldPassword, Secret: true, Placeholder: "prefer instance/task roles"},
		},
		Decode: decodeTyped[config.S3Logger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			s, err := logging.CreateLoggerS3(cfg.(*config.S3Logger))
			if err != nil {
				return nil, err
			}
			s.Settings(smgr)
			return s, nil
		},
	},
	config.LoggingKafka: {
		Type:         config.LoggingKafka,
		Description:  "Apache Kafka producer.",
		HasSecret:    true,
		SecretFields: []string{"sasl.password"},
		Fields: []FieldSpec{
			{Name: "bootstrapServers", Label: "Bootstrap servers", Type: FieldString, Required: true, Placeholder: "broker1:9092,broker2:9092", Help: "Comma-separated Kafka bootstrap servers."},
			{Name: "topic", Label: "Topic", Type: FieldString, Required: true, Placeholder: "osquery-logs"},
			{Name: "sslCALocation", Label: "CA cert path", Type: FieldString, Placeholder: "/etc/ssl/certs/ca.pem", Help: "CA certificate path for TLS verification."},
			{Name: "connectionTimeout", Label: "Connection timeout", Type: FieldString, Placeholder: "5s", Help: "Go duration string, e.g. 5s, 10s, 1m."},
			{Name: "sasl.mechanism", Label: "SASL mechanism", Type: FieldSelect, Options: []string{"", "SCRAM-SHA-256", "SCRAM-SHA-512"}, Help: "SASL mechanism. Empty disables SASL."},
			{Name: "sasl.username", Label: "SASL username", Type: FieldString, Placeholder: "kafka-user"},
			{Name: "sasl.password", Label: "SASL password", Type: FieldPassword, Secret: true, Placeholder: "prefer env vars/secrets in prod"},
		},
		Decode: decodeTyped[config.KafkaLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			k, err := logging.CreateLoggerKafka(cfg.(*config.KafkaLogger))
			if err != nil {
				return nil, err
			}
			k.Settings(smgr)
			return k, nil
		},
	},
	config.LoggingElastic: {
		Type:        config.LoggingElastic,
		Description: "Elasticsearch index writer.",
		HasSecret:   false,
		Fields: []FieldSpec{
			{Name: "host", Label: "Host", Type: FieldString, Required: true, Placeholder: "elastic.example.com"},
			{Name: "port", Label: "Port", Type: FieldString, Required: true, Placeholder: "9200", Help: "Elasticsearch port (string in the config)."},
			{Name: "indexPrefix", Label: "Index prefix", Type: FieldString, Placeholder: "osquery", Help: "Prefix for generated index names."},
			{Name: "dateSeparator", Label: "Date separator", Type: FieldString, Placeholder: ".", Help: "Separator inside date suffixes, e.g. . for YYYY.MM.DD."},
			{Name: "indexSeparator", Label: "Index separator", Type: FieldString, Placeholder: "-", Help: "Separator between prefix and date suffix, e.g. - for prefix-YYYY.MM.DD."},
		},
		Decode: decodeTyped[config.ElasticLogger](),
		Build: func(cfg any, smgr *settings.Settings) (logging.DataExporter, error) {
			e, err := logging.CreateLoggerElastic(cfg.(*config.ElasticLogger))
			if err != nil {
				return nil, err
			}
			e.Settings(smgr)
			return e, nil
		},
	},
}

Registry maps each config.Logging* type to its SinkSpec. This is the single source of truth for what sink types exist and how each is configured. Adding a new sink type means: implement the exporter in pkg/logging, add the config struct to pkg/config, and add a SinkSpec here.

Functions

func BuildExporters 

func BuildExporters(rows []LogSink, smgr *settings.Settings) *logging.MultiExporter

BuildExporters constructs a logging.MultiExporter from a set of sink rows for one environment. Disabled sinks are skipped. Sink build errors are logged and the sink is skipped (one bad sink does not break the whole fan-out). Returns a MultiExporter containing all enabled, buildable sinks. BuildExporters constructs a logging.MultiExporter from a set of sink rows for one environment. Disabled sinks are skipped. Sink build errors are logged and the sink is skipped (one bad sink does not break the whole fan-out). Each exporter is wrapped in a CountedExporter linked to its SinkStats so the background stats writer can track bytes/count per sink.

func MergeSecrets 

func MergeSecrets(typ, prevCfgJSON, newCfgJSON string) (string, error)

MergeSecrets replaces any "***" placeholder values in newCfgJSON with the corresponding values from prevCfgJSON. Used by the API Update handler so an edit that did not touch a secret field preserves the previously stored secret instead of writing the placeholder.

func RedactedConfig 

func RedactedConfig(typ, cfgJSON string) string

RedactedConfig returns the Config JSON with secret fields replaced by the placeholder "***", for the sink Type. Non-secret types return the raw Config unchanged. If the Config does not decode as a JSON object, the raw value is returned untouched (best-effort redaction).

func SupportedTypes 

func SupportedTypes() []string

SupportedTypes returns the Registry keys in a stable, sorted order, for API/UI discovery.

func ValidateSink 

func ValidateSink(name, typ, cfgJSON string) error

ValidateSink returns an error if the proposed sink is invalid: unknown type, empty name, or Config that does not decode against the type.

func ValidateType 

func ValidateType(typ string) bool

ValidateType reports whether the sink type is registered.

Types

type FieldSpec 

type FieldSpec struct {
	// Name is the JSON key inside the Config blob. Nested keys use dot
	// notation, e.g. "sasl.mechanism".
	Name string
	// Label is the human-facing label shown in the form.
	Label string
	// Type selects the input control.
	Type FieldType
	// Required marks the field as mandatory.
	Required bool
	// Secret marks the field as credential-bearing. The API redacts
	// secret fields in read responses unless reveal=1 is passed; the
	// SPA renders them as password inputs. Secret fields are also
	// listed in SinkSpec.SecretFields for the redaction code path.
	Secret bool
	// Placeholder is shown when the field is empty.
	Placeholder string
	// Help is a one-line description shown under the input.
	Help string
	// Options is the list of allowed values for FieldSelect.
	Options []string
	// Default is the value used when the field is absent on create.
	Default any
}

FieldSpec describes one configurable field of a sink type. The schema is declarative so the frontend can render a typed form without shipping a per-type component for every sink.

type FieldType 

type FieldType string

FieldType is the input kind the frontend should render for a field. 

const (
	FieldString   FieldType = "string"
	FieldInteger  FieldType = "integer"
	FieldBoolean  FieldType = "boolean"
	FieldSelect   FieldType = "select"
	FieldPassword FieldType = "password" // string input masked by the SPA
)

type LogSink 

type LogSink struct {
	gorm.Model
	Name          string `gorm:"uniqueIndex:idx_log_sinks_unique"`
	EnvironmentID uint   `gorm:"uniqueIndex:idx_log_sinks_unique"`
	Type          string `gorm:"index"`
	Enabled       bool
	Order         int
	Config        string `gorm:"type:text"`
	Source        string // "service" (seeded) or "db" (operator-edited)
	Info          string
	// BytesSent and ExportsCount are maintained by the background stats
	// writer (SinkStatsWriter), which snapshots the in-process atomic
	// counters from the live MultiExporter and flushes them here
	// periodically. They are never written on the hot path.
	BytesSent    int64 `gorm:"default:0"`
	ExportsCount int64 `gorm:"default:0"`
}

LogSink stores one osquery log destination for one environment (or global when EnvironmentID is NoEnvironmentID). The Config field is a JSON-encoded blob whose shape is determined by Type and validated against the Registry.

type LogSinksManager 

type LogSinksManager struct {
	DB *gorm.DB
}

LogSinksManager manages the log_sinks table.

func NewLogSinksManager 

func NewLogSinksManager(backend *gorm.DB) *LogSinksManager

NewLogSinksManager initializes the manager and auto-migrates the log_sinks table.

func (*LogSinksManager) AllGrouped 

func (m *LogSinksManager) AllGrouped() (map[uint][]LogSink, error)

AllGrouped returns every sink, grouped by EnvironmentID, with the global key (0) included. Used by the API list endpoint and by the TLS boot/reload builder.

func (*LogSinksManager) BuildExportersForEnvironments 

func (m *LogSinksManager) BuildExportersForEnvironments(smgr *settings.Settings) (map[uint]*logging.MultiExporter, error)

BuildExportersForEnvironments builds a map of envID -> MultiExporter for every environment that has sinks, plus the global (key 0) set. Used at TLS boot and on reload. Errors for individual sinks are logged inside BuildExporters; this function never returns an error for a buildable-but-broken sink — it only returns an error if the DB query itself fails.

func (*LogSinksManager) CloneEnvironment 

func (m *LogSinksManager) CloneEnvironment(sourceEnvID, targetEnvID uint, overwrite bool) ([]LogSink, error)

CloneEnvironment copies all sinks from sourceEnvID to targetEnvID. If overwrite is false and the target already has any sinks, ErrTargetHasSinks is returned. If overwrite is true, the target's existing sinks are hard-deleted first. New sink names are suffixed with " (clone of env N)" when the source is non-global, or " (clone of global)" otherwise, so uniqueIndex on Name is preserved.

func (*LogSinksManager) Create 

func (m *LogSinksManager) Create(name, typ string, enabled bool, order int, cfgJSON string, envID uint, info string) (LogSink, error)

Create inserts a new sink row.

func (*LogSinksManager) Delete 

func (m *LogSinksManager) Delete(id uint) error

Delete removes a sink by ID.

func (*LogSinksManager) EffectiveFor 

func (m *LogSinksManager) EffectiveFor(envID uint) ([]LogSink, error)

EffectiveFor returns the sinks that should be used for the given environment: if the environment has any rows of its own, those are returned (override-with-fallback); otherwise the global rows (EnvironmentID == NoEnvironmentID) are returned.

func (*LogSinksManager) Get 

func (m *LogSinksManager) Get(id uint) (LogSink, error)

Get retrieves one sink by ID.

func (*LogSinksManager) List 

func (m *LogSinksManager) List(envID *uint) ([]LogSink, error)

List returns sinks across all environments when envID is nil, or for a single environment when envID is non-nil.

func (*LogSinksManager) ListByEnvironment 

func (m *LogSinksManager) ListByEnvironment(envID uint) ([]LogSink, error)

ListByEnvironment returns all sinks for one environment (EnvironmentID == envID), ordered by Order then CreatedAt.

func (*LogSinksManager) RevertToService 

func (m *LogSinksManager) RevertToService(id uint) error

RevertToService flips a sink row's Source from "db" back to "service", so the next Seed (at TLS boot or during hot-reload) re-syncs the config from the current service configuration (flags, env vars, or YAML). The config itself is NOT changed here — the sync happens in the TLS process during the reload, which has access to the resolved service parameters. The operator clicks "Revert", then "Apply" to trigger the reload; the re-seed overwrites the config with the service-config values before the exporters are rebuilt.

If the row's Source is already "service" (or the legacy "yaml"), this is a no-op. Rows that do not exist return ErrSinkNotFound.

func (*LogSinksManager) Seed 

func (m *LogSinksManager) Seed(params *config.ServiceParameters, envID uint) error

Seed translates the resolved service configuration (logger section and primary DB connection) into LogSink rows using create-if-missing semantics. The configuration may have been provided via flags, environment variables, or a YAML file — by the time it reaches here viper has merged them into a single *config.ServiceParameters, so the seed source is "the service configuration", not any one of those inputs in isolation.

On first boot it writes one row per non-empty entry in logger.types (or the legacy logger.type). Existing rows are never overwritten — operator DB edits win.

logger.alwaysLog is honored: if true and no db sink is among the configured types, a synthetic db sink named __always_log_db__ is seeded using the primary DB connection (params.DB).

func (*LogSinksManager) Update 

func (m *LogSinksManager) Update(id uint, name, typ string, enabled bool, order int, cfgJSON string, info string) (LogSink, error)

Update replaces an existing sink row's mutable fields. Config is re-validated against the existing Type (Type can be changed in the same call as long as the new Config decodes against the new Type).

func (*LogSinksManager) UpdateSinkStats 

func (m *LogSinksManager) UpdateSinkStats(sinkID uint, bytesSent, exportCount int64) error

UpdateSinkStats writes the current bytes/count values for one sink. Exposed for tests that want to verify the flush without running the background goroutine.

type SinkSpec 

type SinkSpec struct {
	Type        string
	Description string
	// HasSecret indicates the Config JSON contains credential fields.
	// When true, the API redacts SecretFields in read responses unless
	// the caller passes reveal=true.
	HasSecret bool
	// SecretFields lists the JSON keys inside Config that hold secrets.
	SecretFields []string
	// Fields is the typed form schema the SPA renders. May be empty
	// (e.g. none, stdout) — the SPA then shows no config fields.
	Fields []FieldSpec
	// Decode unmarshals a raw JSON Config into the typed config struct
	// the sink implementation expects (e.g. *config.SplunkLogger).
	Decode func(json.RawMessage) (any, error)
	// Build instantiates a DataExporter from a decoded config struct.
	Build func(any, *settings.Settings) (logging.DataExporter, error)
}

SinkSpec describes one supported sink type in the Registry.

type SinkStatsWriter 

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

SinkStatsWriter is a background goroutine that periodically snapshots the in-process atomic counters from the live MultiExporter map and flushes them to the log_sinks table. It is modeled on the TLS batchWriter pattern: a ticker fires every `interval`, the writer collects all SinkStats from every MultiExporter in the map, and issues one bulk UPDATE per sink. The hot path (Export) never touches the DB — it only does atomic Int64 adds.

The exporter map is read via the LoggerTLS.AllExporters method, which returns the current map under a read lock. This means the stats writer automatically picks up new sinks after a hot reload without needing its own reference.

func NewSinkStatsWriter 

func NewSinkStatsWriter(mgr *LogSinksManager, logTLS *logging.LoggerTLS, interval time.Duration) *SinkStatsWriter

NewSinkStatsWriter creates (but does not start) a stats writer. Call Start() to launch the background goroutine and Stop() to shut it down.

func (*SinkStatsWriter) Start 

func (w *SinkStatsWriter) Start()

func (*SinkStatsWriter) Stop 

func (w *SinkStatsWriter) Stop()

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.