Help us shape the future of Go! Take the 2026 Go Developer Survey

logstore

package
v0.26.0 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2026 License: Apache-2.0 Imports: 31 Imported by: 0

Documentation

Overview

Package logstore is the log store (epic memql#4893, design record docs/superpowers/specs/2026-09-03-logs-design.md): every node's log lines persisted beside the observability rows in a dedicated log_line hypertable, thirty days of retention with an archive before the sweep, and the Go-served reads the v1:observability:logLine builtins are executed by.

Four halves, one package:

  • sink.go -- the batching Sink installed by app/database.go on every node type (the component/observe TimescaleSink precedent). It implements core/logger.Sink; the handler chain logger.New builds hands it every line at or above the store floor.
  • reader.go -- Search / Tail / Sources / Status over bun.
  • sweep.go -- the nightly retention sweep (archive, then delete, never the second without the first) and the restore.
  • client.go + plugin.go -- the OS write (logsRecordClient) and the `logs` plug-in that binds all eight builtins to their executors.

The concept declares no row tier and no row of it ever passes graph admission: access is a ROLE FLOOR enforced in each handler (design L3), the integrationConfigure precedent, because a builtin carries no @requiresRank.

Index

Constants

View Source 
const (
	MaxClientLines = 50

	// ClientClockSkew is how far a client `at` may sit from the node clock
	// and still be kept; one outside it is replaced by the node's now, which
	// keeps a browser with a wrong clock from writing into next week.
	ClientClockSkew = 5 * time.Minute

	// The per-(user, session) bucket: 120 lines, refilled two per second.
	ClientBucketCapacity = 120
	ClientBucketRefill   = 2.0

	// ReasonRateLimited is the reply's reason when the bucket refuses.
	ReasonRateLimited = "rate_limited"
)

The caps (design L9). A call past any of them is refused WHOLE, with a sentence naming the line and the cap, so the shell can see what it did rather than lose half a batch silently.

View Source 
const (
	// EnvRetentionDays is how many UTC days of lines the store keeps before
	// the nightly sweep archives and deletes them. Default 30, clamped 1..365.
	EnvRetentionDays = "MEMQL_LOGS_RETENTION_DAYS"

	// EnvLevel is the store's own floor on this node (debug / info / warn /
	// error / off), independent of the console level. Read once by
	// core/logger; reported here through LevelName.
	EnvLevel = "MEMQL_LOGS_LEVEL"

	// EnvMaxLinesPerSecond bounds the per-node token bucket in front of the
	// queue. Default 2000, clamped 10..100000. Above it lines are DROPPED and
	// counted (memql_logs_dropped_total{reason="rate"}), never queued.
	EnvMaxLinesPerSecond = "MEMQL_LOGS_MAX_LINES_PER_SECOND"

	// EnvArchiveContainer is the blob container the sweep archives expired
	// days into. Defaults to the cluster's own container,
	// MEMQL_AZURE_BLOB_CONTAINER. With neither set the sweep REFUSES to
	// delete anything (design L7: no archive, no delete).
	EnvArchiveContainer = "MEMQL_LOGS_ARCHIVE_CONTAINER"
)

The store's environment. All four are registered `component: observability`, `scope: node`, `optional: true` in scripts/secrets/manifest.yaml.

View Source 
const (
	DefaultRetentionDays = 30
	MinRetentionDays     = 1
	MaxRetentionDays     = 365

	DefaultMaxLinesPerSecond = 2000
	MinMaxLinesPerSecond     = 10
	MaxMaxLinesPerSecond     = 100000
)

The documented defaults and bounds.

View Source 
const (
	DefaultLimit = 200
	MaxLimit     = 500
)

The read bounds (design D). A page defaults to 200 rows and is capped at 500; a caller asking for more gets 500, never an unbounded scan.

View Source 
const (
	DefaultQueueSize     = 4096
	DefaultMaxBatch      = 256
	DefaultFlushInterval = time.Second

	// DefaultReportInterval is how often the sink says, in ONE warn line under
	// component `logs`, how many lines it dropped and why. That line is itself
	// stored, so the Logs app shows its own gaps.
	DefaultReportInterval = time.Minute

	// MaxMessageBytes caps a stored message; MaxAttributesBytes caps the
	// serialized attributes object. Both apply to engine lines and to OS lines
	// alike (design L9 states them for the OS write).
	MaxMessageBytes    = 4096
	MaxAttributesBytes = 8192

	// NodeTypeOS is the node type of a line written by the MemQL OS front end
	// through logsRecordClient. Such a line has no node.
	NodeTypeOS = "os"
)

The handler's bounds (design L5). Queue 4096 lines, batch 256, flush 1s. Overflow is dropped and counted by reason; a caller never waits on the database.

View Source 
const (
	// SweepMaxDaysPerRun bounds one run. A store that fell far behind its
	// retention catches up over several nights rather than in one run that
	// holds the lock for hours.
	SweepMaxDaysPerRun = 60

	// SweepPageSize is how many rows one archive page reads;
	// SweepDeleteBatch how many one DELETE removes.
	SweepPageSize    = 5000
	SweepDeleteBatch = 5000

	// RestoreBatch is how many rows one restore INSERT carries.
	RestoreBatch = 1000
)
View Source 
const ComponentName = common.ComponentName("logstore")

ComponentName is the bus-side identifier for the log store's sink. Used in lifecycle logs and dependency ordering; not surfaced to end users.

View Source 
const ConceptID = "v1:observability:logLine"

The concept every row of the store is returned as.

Variables

View Source 
var ErrNoDatabase = fmt.Errorf("logstore: this node has no database")

ErrNoDatabase is answered by every read on a node that has none.

View Source 
var ErrNoStore = errors.New("logstore: this node keeps no log lines (no database, or the store has not started)")

ErrNoStore is returned by WriteLines when this node has no sink -- no database, or not yet booted. The caller answers with a sentence rather than a count: the OS drops and counts, and a browser cannot retry into a node that keeps no lines.

Functions

func ArchiveContainer 

func ArchiveContainer() string

ArchiveContainer returns the archive container: MEMQL_LOGS_ARCHIVE_CONTAINER, else MEMQL_AZURE_BLOB_CONTAINER, else "" -- which the sweep reads as "no archive configured" and refuses to delete.

func ArchiveObjectName 

func ArchiveObjectName(day time.Time, nodeType string) string

ArchiveObjectName is logs/<YYYY-MM-DD>/<nodeType>.ndjson.gz.

func LevelName 

func LevelName() string

LevelName returns the store floor this node runs with (debug / info / warn / error / off), the one parse of MEMQL_LOGS_LEVEL that core/logger made.

func MaxLinesPerSecond 

func MaxLinesPerSecond() int

MaxLinesPerSecond returns MEMQL_LOGS_MAX_LINES_PER_SECOND clamped to 10..100000, or the default 2000 when unset or unparseable.

func ParseClientLines 

func ParseClientLines(raw []any, now time.Time) ([]logger.Line, error)

ParseClientLines validates and converts the raw `lines` argument -- the []object the builtin declares, arriving as []any of map[string]any -- into logger.Lines. Any violation refuses the WHOLE call with a sentence naming the line index and the cap.

func ParseDay 

func ParseDay(day string) (time.Time, error)

ParseDay parses a UTC day, YYYY-MM-DD.

func RetentionDays 

func RetentionDays() int

RetentionDays returns MEMQL_LOGS_RETENTION_DAYS clamped to 1..365, or the default 30 when unset or unparseable. Never unbounded: a value past the bound lands ON the bound rather than being honoured, because "keep logs for 10000 days" is a disk decision nobody made on purpose.

func ValidateSession 

func ValidateSession(session string) error

ValidateSession checks the tab session id.

Types

type ArchiveObject 

type ArchiveObject struct {
	Day      string `json:"day"`
	NodeType string `json:"nodeType"`
	Object   string `json:"object"`
	// Size is the object's bytes when the archiver reports sizes; SizeKnown
	// says whether it did, so an absent size is never read as zero.
	Size      int64 `json:"size,omitempty"`
	SizeKnown bool  `json:"sizeKnown"`
}

ArchiveObject is one row of logsArchiveList.

type Archiver 

type Archiver interface {
	Upload(ctx context.Context, container, object string, data []byte, contentType string) (string, error)
	Download(ctx context.Context, container, object string) ([]byte, error)
	ListPrefix(ctx context.Context, container, prefix string) ([]string, error)
}

Archiver is the narrow slice of object storage the sweep and the restore need. azureblob.AzureBlobUploader is the production one; tests hold an in-memory map.

type BunDBProvider 

type BunDBProvider func() *bun.DB

BunDBProvider resolves the database handle at Start time, or nil when the database is not ready yet (the component retries briefly). The same seam component/observe.SinkComponent uses, for the same reason: this package must not import the database component's graph.

type ClientLimiter 

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

ClientLimiter is the per-(user, session) bucket. In memory and per replica, like abuse.IPRateLimiter, and for the same reason: a browser reconnecting to a sibling starts a fresh bucket, and the cap is a courtesy against a storm rather than an authority (design I).

func NewClientLimiter 

func NewClientLimiter() *ClientLimiter

NewClientLimiter builds the limiter at the L9 bounds.

func (*ClientLimiter) Allow 

func (l *ClientLimiter) Allow(userId, session string, n int, now time.Time) bool

Allow asks for n lines' worth of tokens for one (user, session). A refusal consumes nothing: the whole call is refused and the shell drops it.

func (*ClientLimiter) Snapshot 

func (l *ClientLimiter) Snapshot() int

Snapshot reports the bucket count, for tests and telemetry.

type ClientReply 

type ClientReply struct {
	Accepted int    `json:"accepted"`
	Dropped  int    `json:"dropped"`
	Reason   string `json:"reason,omitempty"`
}

ClientReply is what logsRecordClient answers.

type InsertFunc 

type InsertFunc func(ctx context.Context, rows []Row) error

InsertFunc writes one batch. The production one is a bun multi-row insert; tests inject one that records, blocks, or fails.

type Integration 

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

Integration binds the store to the engine.

func NewIntegration 

func NewIntegration(log *slog.Logger, db, directDB func() *bun.DB) *Integration

NewIntegration wires the production integration. The archive client is built only when a container is configured, and its absence is a refusal at sweep time rather than a factory error: a cluster with no object storage keeps its lines, which is the design.

func (*Integration) Capabilities 

func (i *Integration) Capabilities() []memql.IntegrationCapability

Capabilities implements memql.IntegrationProvider.

func (*Integration) IntegrationName 

func (i *Integration) IntegrationName() string

IntegrationName implements memql.IntegrationProvider.

type Query 

type Query struct {
	// Window bounds: occurred_at >= WindowStart AND occurred_at < WindowEnd.
	// A zero bound is open. logsSearch requires both; logsTail has none.
	WindowStart time.Time
	WindowEnd   time.Time

	NodeTypes  []string
	Nodes      []string
	Components []string
	Levels     []string

	Apps            []string
	SubjectConcepts []string

	Subject        string
	SubjectConcept string
	Session        string
	UserId         string

	// Text is a case-insensitive substring of the message, subject, or structured attributes.
	Text string

	Limit int

	// Search cursor: rows strictly older than (BeforeAt, BeforeId).
	BeforeAt time.Time
	BeforeId string

	// Tail cursor: rows strictly newer than (AfterAt, AfterId).
	AfterAt time.Time
	AfterId string
}

Query is the facet set logsSearch and logsTail share.

THE SCOPE RULE: Apps and SubjectConcepts together form ONE predicate, ORed -- an app's slice of the store is "lines tagged with me OR about the things I own" -- and every other facet ANDs. Both empty means no scope predicate.

type Report 

type Report struct {
	Boundary     time.Time `json:"boundary"`
	DaysArchived int       `json:"daysArchived"`
	RowsArchived int64     `json:"rowsArchived"`
	RowsDeleted  int64     `json:"rowsDeleted"`
	Days         []string  `json:"days,omitempty"`
	Objects      []string  `json:"objects,omitempty"`
	Container    string    `json:"container,omitempty"`
	Skipped      string    `json:"skipped,omitempty"`
	Refused      string    `json:"refused,omitempty"`
	Truncated    bool      `json:"truncated,omitempty"` // more expired days remain
	Took         string    `json:"took"`

	// ChunksDropped is how many chunks went whole instead of being deleted
	// from; zero means every row of the run went by DELETE. A drop reports
	// chunks, not rows, so the rows inside a dropped chunk are counted in
	// RowsDeleted at the count their day archived.
	ChunksDropped int64 `json:"chunksDropped"`
}

Report is what one sweep answers.

type RestoreReport 

type RestoreReport struct {
	Day      string   `json:"day"`
	NodeType string   `json:"nodeType,omitempty"`
	Restored int64    `json:"restored"`
	Skipped  int64    `json:"skipped"`
	Objects  []string `json:"objects"`
	// Note says what a caller must know: a restored day is older than the
	// retention boundary by definition and is swept again at the next run.
	Note string `json:"note"`
}

RestoreReport is what one restore answers.

type Row 

type Row struct {
	bun.BaseModel `bun:"table:log_line,alias:ll"`

	OccurredAt     time.Time       `bun:"occurred_at,pk" json:"occurredAt"`
	ID             string          `bun:"id,pk" json:"id"`
	NodeType       string          `bun:"node_type,notnull" json:"nodeType"`
	Node           string          `bun:"node,notnull" json:"node"`
	Level          string          `bun:"level,notnull" json:"level"`
	Component      string          `bun:"component,notnull" json:"component"`
	App            string          `bun:"app,notnull" json:"app"`
	Message        string          `bun:"message,notnull" json:"message"`
	Attributes     json.RawMessage `bun:"attributes,type:jsonb,nullzero" json:"attributes,omitempty"`
	Subject        string          `bun:"subject,notnull" json:"subject"`
	SubjectConcept string          `bun:"subject_concept,notnull" json:"subjectConcept"`
	Session        string          `bun:"session,notnull" json:"session"`
	UserId         string          `bun:"user_id,notnull" json:"userId"`
}

Row is one row of log_line. The bun tags are the table; the json tags are the concept's field names, which is the shape a logLine node's payload, an archive's NDJSON record and a restore's input all share. 

func Search(ctx context.Context, db *bun.DB, q Query) ([]Row, error)

Search answers logsSearch.

func Tail 

func Tail(ctx context.Context, db *bun.DB, q Query) ([]Row, error)

Tail answers logsTail: always ascending.

type Sink 

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

Sink is the batching writer over log_line, and the process's core/logger.Sink. Write is a per-second token bucket followed by a non-blocking channel send; one goroutine drains the queue into batched inserts. Lossy under pressure BY DESIGN: a full queue drops the line and counts it, because the alternative -- blocking the caller inside a log call -- turns logging into a latency hazard on every hot path in the process.

func Current 

func Current() *Sink

Current returns the sink the SinkComponent installed on this node, or nil when the node has no database (or has not finished booting). The plug-in reads stats and writes OS lines through it.

func NewSink 

func NewSink(db *bun.DB, opts SinkOptions) *Sink

NewSink constructs the sink. It does not start the drain; call Start.

func (*Sink) MaxLinesPerSecond 

func (s *Sink) MaxLinesPerSecond() int

MaxLinesPerSecond is the bucket's rate on this node.

func (*Sink) Node 

func (s *Sink) Node() string

func (*Sink) NodeType 

func (s *Sink) NodeType() string

NodeType and Node are what this sink stamps engine lines with.

func (*Sink) ReportDropsNow 

func (s *Sink) ReportDropsNow()

ReportDropsNow forces the gap report. For tests, so they do not wait a minute for the ticker.

func (*Sink) Start 

func (s *Sink) Start()

Start begins the background drain. Idempotent.

func (*Sink) Stats 

func (s *Sink) Stats() Stats

Stats returns a snapshot of the counters.

func (*Sink) Stop 

func (s *Sink) Stop(ctx context.Context) error

Stop flushes what is queued and stops the drain. Blocks until done or ctx is cancelled. Idempotent.

func (*Sink) Write 

func (s *Sink) Write(l logger.Line)

Write implements core/logger.Sink for the engine path: the line is stamped with this node's type and id.

func (*Sink) WriteLines 

func (s *Sink) WriteLines(lines []logger.Line, st Stamp) WriteResult

WriteLines is the instance form of the package function.

func (*Sink) WriteStamped 

func (s *Sink) WriteStamped(l logger.Line, st Stamp) bool

WriteStamped enqueues one line under an explicit stamp. Non-blocking. It reports whether the line was queued; a false is already counted under its reason.

type SinkComponent 

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

SinkComponent owns the Sink's lifecycle so app/database.go can wire it in beside the observe sink on every node type, and shutdown flushes the queue.

On a node with no database the component installs a no-op sink and says so once: the pre-boot ring drains into nothing, the console keeps every line it kept, and logsRecordClient answers that this node keeps no lines.

func NewSinkComponent 

func NewSinkComponent(log *slog.Logger, provider BunDBProvider, opts SinkOptions, order int) *SinkComponent

NewSinkComponent constructs the dependency wrapper. order 0 falls back to 50, safely after the database component (around 2) has started.

func (*SinkComponent) ComponentName 

func (c *SinkComponent) ComponentName() common.ComponentName

func (*SinkComponent) IsRunning 

func (c *SinkComponent) IsRunning() bool

IsRunning, Order, ComponentName, Ready -- the standard Dependency surface. Ready closes when Start has finished regardless of whether the DB resolved.

func (*SinkComponent) Order 

func (c *SinkComponent) Order() int

func (*SinkComponent) Ready 

func (c *SinkComponent) Ready() <-chan struct{}

func (*SinkComponent) Sink 

func (c *SinkComponent) Sink() *Sink

Sink returns the running sink, or nil.

func (*SinkComponent) Start 

func (c *SinkComponent) Start(_ context.Context)

Start resolves the bun handle, builds and starts the sink, registers it as the process's logger.Sink (which drains the pre-boot ring into it) and as logstore.Current().

func (*SinkComponent) Stop 

func (c *SinkComponent) Stop(ctx context.Context)

Stop unregisters the sink first -- so lines written during the flush go to the ring rather than a closing queue -- then flushes and reports.

type SinkOptions 

type SinkOptions struct {
	QueueSize         int
	MaxBatch          int
	FlushInterval     time.Duration
	ReportInterval    time.Duration
	MaxLinesPerSecond int

	// NodeType and Node stamp every engine line. Resolved from the
	// environment when empty (MEMQL_NODE_TYPE; MEMQL_NODE_ID with the hostname
	// fallback), the way component/node resolves the node's identity.
	NodeType string
	Node     string

	// Logger is the CONSOLE logger the sink's own lines go through. Its lines
	// carry component `logs.store` and are never stored (the recursion guard);
	// the one exception, the periodic gap line, carries component `logs`.
	Logger *slog.Logger

	// Insert overrides the bun insert. Nil uses the database handed to
	// NewSink.
	Insert InsertFunc

	// Now is injectable for tests. Nil uses time.Now.
	Now func() time.Time
}

SinkOptions configures the sink. Zero values take the documented defaults.

type Source 

type Source struct {
	Kind     string `json:"kind"` // component | node | app
	Value    string `json:"value"`
	NodeType string `json:"nodeType,omitempty"`
	Count    int64  `json:"count"`
}

Source is one row of logsSources: a distinct component, a (nodeType, node) pair, or an OS app, with its line count inside the window.

func Sources 

func Sources(ctx context.Context, db *bun.DB, from, to time.Time) ([]Source, error)

Sources answers logsSources for [from, to).

type Stamp 

type Stamp struct {
	NodeType string
	Node     string
	Session  string
	UserId   string
}

Stamp is what a line carries about WHERE it came from. The engine path stamps the sink's own node type and node; the OS write path passes NodeTypeOS, a blank node, the tab session and the actor's user id.

type Stats 

type Stats struct {
	Written      uint64 `json:"written"`
	DroppedQueue uint64 `json:"droppedQueue"`
	DroppedRate  uint64 `json:"droppedRate"`
	DroppedLevel uint64 `json:"droppedLevel"`
	DroppedDB    uint64 `json:"droppedDb"`
	Flushes      uint64 `json:"flushes"`
	InsertErrors uint64 `json:"insertErrors"`
	QueueDepth   int    `json:"queueDepth"`
	QueueSize    int    `json:"queueSize"`
}

Stats is a snapshot of the sink's counters since the process started. The same numbers metrics export, kept in-process so logsStatus can answer them for the node that took the call.

func (Stats) Dropped 

func (s Stats) Dropped() uint64

Dropped is every drop reason summed.

type StatusReport 

type StatusReport struct {
	RetentionDays     int    `json:"retentionDays"`
	Level             string `json:"level"`
	MaxLinesPerSecond int    `json:"maxLinesPerSecond"`
	ArchiveConfigured bool   `json:"archiveConfigured"`
	ArchiveContainer  string `json:"archiveContainer,omitempty"`
	ArchiveClient     bool   `json:"archiveClient"`
	NodeType          string `json:"nodeType"`
	Node              string `json:"node"`
	// StoreActive is false on a node with no sink: no database, or the store
	// is off here. Counters are then absent rather than zero.
	StoreActive bool        `json:"storeActive"`
	Counters    *Stats      `json:"counters,omitempty"`
	Table       TableStatus `json:"table"`
	TableError  string      `json:"tableError,omitempty"`
}

StatusReport is what logsStatus answers: the cluster's configuration, this node's store, and the table's bounds.

type Sweeper 

type Sweeper struct {
	// DB is the pooled handle for row traffic.
	DB *bun.DB
	// LockDB is the DIRECT handle for the advisory-lock session. Nil falls
	// back to DB, which is correct on a cluster with no pooler.
	LockDB *bun.DB

	Archive   Archiver
	Container string

	// RetentionDays is the window. Zero or negative uses the configured
	// value.
	RetentionDays int
	// MaxDays bounds one run. Zero uses SweepMaxDaysPerRun.
	MaxDays int

	Now    func() time.Time
	Logger *slog.Logger
	// contains filtered or unexported fields
}

Sweeper runs the retention sweep, the restore and the archive listing.

func (*Sweeper) ListArchive 

func (s *Sweeper) ListArchive(ctx context.Context) ([]ArchiveObject, error)

ListArchive answers logsArchiveList: every logs/<day>/<nodeType>.ndjson.gz object, newest day first. Nil archiver means no archive is configured; the caller says so.

func (*Sweeper) Restore 

func (s *Sweeper) Restore(ctx context.Context, day string, nodeType string) (RestoreReport, error)

Restore brings an archived day back (design L8): list logs/<day>/, filter on nodeType when given, download, gunzip, parse, insert with ON CONFLICT DO NOTHING in batches. Idempotent on (occurred_at, id): a second restore restores nothing.

func (*Sweeper) Run 

func (s *Sweeper) Run(ctx context.Context) (Report, error)

Run sweeps once. Refusals (no container) and skips (another sweep holds the lock) are answers, not errors: they land on the report and in the one log line, and the automation step succeeds.

type TableStatus 

type TableStatus struct {
	OldestAt    *time.Time `json:"oldestAt,omitempty"`
	NewestAt    *time.Time `json:"newestAt,omitempty"`
	RowEstimate int64      `json:"rowEstimate"`
	Timescale   bool       `json:"timescale"`
}

TableStatus is what the store holds: its bounds and an estimate of its size. RowEstimate is TimescaleDB's approximate_row_count where the extension is present and pg_class.reltuples otherwise -- an estimate either way, because a count(*) over thirty days of lines is exactly the scan a status call must not run.

func Status 

func Status(ctx context.Context, db *bun.DB) (TableStatus, error)

Status answers the table half of logsStatus.

type WriteResult 

type WriteResult struct {
	Accepted int
	Dropped  int
}

WriteResult is what an explicit write answers: how many lines the queue took and how many it dropped, by the sink's own reasons.

func WriteLines 

func WriteLines(_ context.Context, lines []logger.Line, st Stamp) (WriteResult, error)

WriteLines enqueues lines carrying an explicit stamp, through the same queue and bucket the engine's own lines use. The OS write path (design E) is its caller: nodeType=os, node blank, the tab session and the actor's user id. The queue never blocks; a line the bucket or the queue refuses is counted under its reason and reported as dropped here.

The context is accepted for the signature's sake and not waited on -- a non-blocking enqueue has nothing to cancel.

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.