objectstore

package
v0.2.0-beta.18 Latest Latest
Warning

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

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

Documentation

Overview

Package objectstore is the S3-compatible storage layer shared by log archive and other features that write to a connected storage destination.

Index

Constants

View Source
const (
	EnvArchivePrefix     = "APP_LOG_ARCHIVE_PREFIX"
	EnvArchiveMaxLines   = "APP_LOG_ARCHIVE_MAX_LINES_PER_OBJECT"
	EnvArchiveMaxWindows = "APP_LOG_ARCHIVE_MAX_WINDOWS_PER_RUN"
	EnvArchiveBackfill   = "APP_LOG_ARCHIVE_MAX_BACKFILL"
	EnvArchiveTick       = "APP_LOG_ARCHIVE_TICK"
)

Env vars tuning the archiver; unset or invalid values use the defaults.

View Source
const (
	EnvBuildCachePrefix          = "APP_BUILD_CACHE_PREFIX"
	EnvBuildCacheClearMaxObjects = "APP_BUILD_CACHE_CLEAR_MAX_OBJECTS"
	EnvBuildCacheStatsMaxObjects = "APP_BUILD_CACHE_STATS_MAX_OBJECTS"
)

Env names for the build cache bounds.

View Source
const (
	PresetAWS    = "aws"
	PresetR2     = "r2"
	PresetB2     = "b2"
	PresetMinIO  = "minio"
	PresetWasabi = "wasabi"
	PresetCustom = "custom"
)

Preset identifies a well-known S3-compatible provider.

View Source
const (
	ReasonInvalidCredentials = "invalid_credentials" //nolint:gosec // reason code, not a credential
	ReasonAccessDenied       = "access_denied"
	ReasonBucketNotFound     = "bucket_not_found"
	ReasonRegionMismatch     = "region_mismatch"
	ReasonEndpointBlocked    = "endpoint_blocked"
	ReasonTLS                = "tls_error"
	ReasonUnreachable        = "unreachable"
	ReasonIntegrity          = "integrity_mismatch"
	ReasonUnknown            = "unknown"
)

Failure reason codes returned by Probe.

Variables

View Source
var ErrBuildCacheNotConfigured = errors.New("objectstore: build cache is not configured for this app")

ErrBuildCacheNotConfigured is returned when an app has no enabled build cache setting.

View Source
var ErrInvalidBuildCacheApp = errors.New("objectstore: invalid app name for build cache")

ErrInvalidBuildCacheApp is returned for an app name that cannot be a key prefix.

View Source
var ErrInvalidDump = errors.New("objectstore: invalid dump request")

ErrInvalidDump is returned for a bad dump request.

Functions

func AppPrefix

func AppPrefix(root, appName string) string

AppPrefix is the key prefix holding one app's archived logs.

func Classify

func Classify(err error) (string, string)

Classify maps an S3 or transport error to a stable reason code and a message safe to show (it never contains credentials).

func IsArchiveKey

func IsArchiveKey(root, key string) bool

IsArchiveKey reports whether key is under the archive root and free of traversal.

func ObjectKey

func ObjectKey(root, resourceID string, startNs int64) string

ObjectKey is root/<kind>/<name>/yyyy/mm/dd/hh/<start ns>.ndjson.gz. It is keyed by the chunk start only, so retrying a chunk overwrites it instead of leaving a duplicate; chunks are contiguous and sort by name.

func PresetForProvider

func PresetForProvider(provider string) string

PresetForProvider infers a preset for a backup target created without one.

func ResourceID

func ResourceID(appName string) string

ResourceID is the telemetry resource ID a log line of appName is stored under.

func ValidateEndpoint

func ValidateEndpoint(endpoint string) error

ValidateEndpoint checks scheme and host shape. Internal addresses are refused at dial time by netguard, not here, so DNS tricks cannot bypass it.

Types

type ArchiveStore

type ArchiveStore interface {
	ListLogArchivePolicies(ctx context.Context) ([]store.LogArchivePolicy, error)
	RecordLogArchivePolicyRun(ctx context.Context, id, at string, watermarkNs int64, runErr string) error
	InsertLogArchiveRun(ctx context.Context, r store.LogArchiveRun) error
	FinishLogArchiveRun(ctx context.Context, r store.LogArchiveRun) error
	FailStaleLogArchiveRuns(ctx context.Context, at string) error
}

ArchiveStore is the persistence surface the archiver needs.

type Archiver

type Archiver struct {
	Store   ArchiveStore
	Clients *Resolver
	Logs    LogReader
	Logger  *slog.Logger
	Opts    Options
	Now     func() time.Time
	// contains filtered or unexported fields
}

Archiver ships node-local logs to a storage destination as gzip NDJSON. One run at a time, so a slow bucket applies backpressure instead of piling up goroutines.

func NewArchiver

func NewArchiver(st ArchiveStore, clients *Resolver, logs LogReader, logger *slog.Logger, opts Options) *Archiver

NewArchiver builds an Archiver.

func (*Archiver) Run

func (a *Archiver) Run(ctx context.Context)

Run checks for due policies every Opts.Tick until ctx ends. With no policies a tick is one small query, so idle cost stays near zero.

func (*Archiver) RunDue

func (a *Archiver) RunDue(ctx context.Context)

RunDue runs every enabled policy whose interval has elapsed since its last run.

func (*Archiver) RunPolicy

func (a *Archiver) RunPolicy(ctx context.Context, p store.LogArchivePolicy) error

RunPolicy archives everything a policy has not shipped yet, up to MaxWindowsPerRun hour windows, then advances its watermark. A failed window keeps the watermark in place so the next run retries it.

func (*Archiver) StartDump

func (a *Archiver) StartDump(ctx context.Context, req DumpRequest) (store.LogArchiveRun, error)

StartDump validates a manual dump, records a running row, and archives the range in the background. Poll the run row (or Wait) for the outcome.

func (*Archiver) Wait

func (a *Archiver) Wait()

Wait blocks until every in-flight manual dump has finished.

type BuildCache

type BuildCache struct {
	Store    BuildCacheStore
	Resolver BuildCacheResolver
	Options  BuildCacheOptions
	Logger   *slog.Logger
	// ResolveHost is net.DefaultResolver.LookupNetIP unless a test overrides it.
	ResolveHost func(ctx context.Context, network, host string) ([]netip.Addr, error)
}

BuildCache resolves per-app BuildKit remote cache settings against storage destinations and runs cache hygiene operations on the bucket.

func (*BuildCache) Clear

func (b *BuildCache) Clear(ctx context.Context, app string) (ClearResult, error)

Clear deletes objects under the app's cache prefix, up to ClearMaxObjects.

func (*BuildCache) Effective

func (b *BuildCache) Effective(ctx context.Context, app string) (store.BuildCacheSetting, bool, error)

Effective returns the setting an app builds with: its own row when present (an explicit disable wins), else the global default. The bool is false when no enabled setting applies.

func (*BuildCache) KeyPrefix

func (b *BuildCache) KeyPrefix(app string) (string, error)

KeyPrefix is the per-app bucket prefix the cache lives under.

func (*BuildCache) RecordBuild

func (b *BuildCache) RecordBuild(ctx context.Context, app, warning string)

RecordBuild stores the latest build outcome on the setting the app resolved to. Failures are logged, never returned: recording must not fail a build.

func (*BuildCache) S3Cache

func (b *BuildCache) S3Cache(ctx context.Context, app string) (*build.S3Cache, error)

S3Cache builds the BuildKit s3 cache config for app, or nil when the app has no enabled setting. The result carries credentials.

func (*BuildCache) Stats

func (b *BuildCache) Stats(ctx context.Context, app string) (CacheStats, error)

Stats lists the app's cache prefix, bounded by StatsMaxObjects.

type BuildCacheOptions

type BuildCacheOptions struct {
	Prefix          string
	ClearMaxObjects int
	StatsMaxObjects int
}

BuildCacheOptions bounds the build cache bucket operations.

func BuildCacheOptionsFromEnv

func BuildCacheOptionsFromEnv(lookup func(string) (string, bool)) BuildCacheOptions

BuildCacheOptionsFromEnv applies env overrides to the defaults.

func DefaultBuildCacheOptions

func DefaultBuildCacheOptions() BuildCacheOptions

DefaultBuildCacheOptions returns the defaults.

type BuildCacheResolver

type BuildCacheResolver interface {
	Config(ctx context.Context, targetID string) (Config, store.BackupTarget, error)
	Client(ctx context.Context, targetID string) (*Client, store.BackupTarget, error)
}

BuildCacheResolver turns a storage destination id into credentials and a client.

type BuildCacheStore

type BuildCacheStore interface {
	GetBuildCacheSetting(ctx context.Context, appName string) (store.BuildCacheSetting, error)
	RecordBuildCacheResult(ctx context.Context, scope, at, result, warning string) error
	RecordBuildCacheCleared(ctx context.Context, scope, at string) error
}

BuildCacheStore is the persistence surface the build cache service needs.

type CacheStats

type CacheStats struct {
	Prefix       string    `json:"prefix"`
	Objects      int       `json:"objects"`
	Bytes        int64     `json:"bytes"`
	LastModified time.Time `json:"last_modified,omitzero"`
	// Truncated means the listing hit the bound, so counts are lower bounds.
	Truncated bool `json:"truncated"`
}

CacheStats summarises the objects under an app's cache prefix.

type ClearResult

type ClearResult struct {
	Deleted int `json:"deleted"`
	// More means the ClearMaxObjects bound was hit; run the clear again.
	More bool `json:"more"`
}

ClearResult reports a cache clear.

type Client

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

Client is a bucket-scoped S3 client.

func New

func New(cfg Config) (*Client, error)

New builds a Client from static credentials.

func (*Client) Delete

func (c *Client) Delete(ctx context.Context, key string) error

Delete removes an object; deleting a missing key is not an error on S3.

func (*Client) Get

func (c *Client) Get(ctx context.Context, key string) (io.ReadCloser, int64, error)

Get opens an object. The caller closes the returned reader.

func (*Client) List

func (c *Client) List(ctx context.Context, prefix, token string, limit int32) (ListPage, error)

List returns up to limit keys under prefix, resuming from a previous page's Next token.

func (*Client) Put

func (c *Client) Put(ctx context.Context, key string, body io.ReadSeeker, contentType, contentEncoding string) error

Put uploads body under key. The body must be seekable so the SDK can sign and retry it.

type Config

type Config struct {
	Endpoint        string
	Region          string
	Bucket          string
	AccessKeyID     string
	SecretAccessKey string
	PathStyle       bool
	// MaxAttempts bounds SDK retries per request; zero means 3.
	MaxAttempts int
	// HTTPClient defaults to netguard.NewClient(), which refuses internal addresses.
	HTTPClient *http.Client
}

Config is everything needed to talk to one bucket.

type DumpRequest

type DumpRequest struct {
	TargetID string
	AppName  string
	From     time.Time
	To       time.Time
}

DumpRequest is a manual export of a time range.

type HealthSource

type HealthSource struct {
	Store interface {
		ListLogArchivePolicies(ctx context.Context) ([]store.LogArchivePolicy, error)
	}
}

HealthSource adapts stored policies to the alerting engine.

func (HealthSource) LogArchiveHealth

func (h HealthSource) LogArchiveHealth(ctx context.Context) ([]alerting.LogArchiveHealth, error)

LogArchiveHealth implements alerting.LogArchiveSource.

type ListPage

type ListPage struct {
	Objects []Object `json:"objects"`
	Next    string   `json:"next,omitempty"`
}

ListPage is one page of a prefix listing.

type LogReader

type LogReader interface {
	DistinctLogResources(ctx context.Context, fromNs, toNs int64) ([]string, error)
	StreamLogs(ctx context.Context, resourceID string, fromNs, toNs int64, maxLines int, fn func(telemetry.LogEntry) error) (telemetry.LogStreamResult, error)
}

LogReader is the node-local log store surface the archiver reads.

type Object

type Object struct {
	Key          string    `json:"key"`
	Size         int64     `json:"size"`
	LastModified time.Time `json:"last_modified"`
}

Object is one listed key.

type Options

type Options struct {
	Prefix            string
	MaxLinesPerObject int
	MaxWindowsPerRun  int
	Lag               time.Duration
	MaxBackfill       time.Duration
	MaxDumpRange      time.Duration
	Tick              time.Duration
}

Options tune archive runs.

func DefaultOptions

func DefaultOptions() Options

DefaultOptions returns the built-in archive tuning.

func OptionsFromEnv

func OptionsFromEnv(lookup func(string) (string, bool)) Options

OptionsFromEnv applies env overrides to DefaultOptions.

type Preset

type Preset struct {
	ID    string `json:"id"`
	Label string `json:"label"`
	// EndpointTemplate may contain {account_id} and {region}. Empty means the
	// SDK resolves the endpoint (AWS) or the caller supplies one.
	EndpointTemplate string `json:"endpoint_template,omitempty"`
	DefaultRegion    string `json:"default_region,omitempty"`
	PathStyle        bool   `json:"path_style"`
	NeedsAccountID   bool   `json:"needs_account_id"`
	NeedsRegion      bool   `json:"needs_region"`
	NeedsEndpoint    bool   `json:"needs_endpoint"`
	// Provider is the backup target provider this preset is stored as.
	Provider string `json:"provider"`
}

Preset describes how a provider's endpoint, region and addressing work.

func PresetByID

func PresetByID(id string) (Preset, bool)

PresetByID looks up a preset.

func Presets

func Presets() []Preset

Presets returns the provider presets in display order.

type ProbeResult

type ProbeResult struct {
	OK      bool        `json:"ok"`
	Reason  string      `json:"reason,omitempty"`
	Message string      `json:"message,omitempty"`
	Steps   []ProbeStep `json:"steps"`
}

ProbeResult reports a put, get and delete round trip on a throwaway key.

func Probe

func Probe(ctx context.Context, c *Client) ProbeResult

Probe writes, reads back and deletes a random object to prove the credentials can do everything log archive needs.

type ProbeStep

type ProbeStep struct {
	Name  string `json:"name"`
	OK    bool   `json:"ok"`
	Error string `json:"error,omitempty"`
}

ProbeStep is the outcome of one probe operation.

type Resolved

type Resolved struct {
	Endpoint string
	Region   string
}

Resolved is a preset applied to user input.

func Resolve

func Resolve(presetID, endpoint, region, accountID string) (Resolved, error)

Resolve applies a preset to the user's endpoint, region and account ID.

type Resolver

type Resolver struct {
	Store   TargetStore
	Secrets SecretsResolver
	// HTTPClient overrides the SSRF-guarded default, for tests.
	HTTPClient *http.Client
	// MaxAttempts bounds SDK retries per request; zero means the default.
	MaxAttempts int
}

Resolver builds a Client for a stored storage destination.

func (*Resolver) Client

func (r *Resolver) Client(ctx context.Context, targetID string) (*Client, store.BackupTarget, error)

Client resolves targetID's settings and credentials into a Client.

func (*Resolver) Config

func (r *Resolver) Config(ctx context.Context, targetID string) (Config, store.BackupTarget, error)

Config resolves targetID's settings and decrypted credentials. The result carries secrets: hand it to a client, never to a log or an API response.

type SecretsResolver

type SecretsResolver interface {
	Resolve(ctx context.Context, serviceName, envKey string) (string, error)
}

SecretsResolver reads a stored credential.

type TargetStore

type TargetStore interface {
	GetBackupTarget(ctx context.Context, id string) (store.BackupTarget, error)
	GetStorageOptions(ctx context.Context, targetID string) (store.StorageOptions, bool, error)
}

TargetStore is the storage-destination lookup surface the resolver needs.

Directories

Path Synopsis
Package objectstoretest is an in-memory, path-style S3 server for tests.
Package objectstoretest is an in-memory, path-style S3 server for tests.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL