Documentation
¶
Overview ¶
Package capacity reports how full the platform's database and buckets are and whether objects and rows still agree (#1899). Running out of disk or bucket quota is one of the commonest real outages, and before it nothing reported either.
Three loops run on every replica:
- the table sampler reads each growing table's size and row estimate, each HNSW index's size and the transaction id age from the catalog (MCP_PLATFORM_CAPACITY_INTERVAL, default 15m);
- the usage flush adds the puts and deletes this replica made to the shared storage_usage rows and reads every row back (every 10s), so an upload shows within seconds on every replica;
- the full listing walks every platform prefix, resets the usage rows to what it found and counts orphaned objects and dangling rows (MCP_PLATFORM_STORAGE_SCAN_INTERVAL, default 6h). It runs on one replica at a time, under an advisory lock.
Between listings a usage row is the last listing plus the writes since. A delete removes its object from the count but not its bytes, which the listing corrects.
Index ¶
- Constants
- Variables
- func DetectBackend(ctx context.Context, endpoint string) string
- func Endpoint(toolkits map[string]any, conn string) (endpoint string, found bool)
- func ParseBudgets(v string) ([]observability.BucketBudget, error)
- func ReadStorage(ctx context.Context, db *sql.DB, backends map[string]string) (observability.StorageCapacity, error)
- func SampleTables(ctx context.Context, db *sql.DB) (observability.DatabaseCapacity, error)
- type BucketPrefix
- type Config
- type Layout
- type Recorder
- type ScanResult
- type Scanner
- type Service
- type Sources
- type Walker
Constants ¶
const ( DefaultTableInterval = 15 * time.Minute DefaultScanInterval = 6 * time.Hour DefaultFlushInterval = 10 * time.Second DefaultOrphanGrace = 15 * time.Minute )
Defaults for the loops.
const ( EnvTableInterval = "MCP_PLATFORM_CAPACITY_INTERVAL" EnvScanInterval = "MCP_PLATFORM_STORAGE_SCAN_INTERVAL" EnvOrphanGrace = "MCP_PLATFORM_STORAGE_ORPHAN_GRACE" EnvBudgets = "MCP_PLATFORM_STORAGE_BUDGETS" EnvBackend = "MCP_PLATFORM_STORAGE_BACKEND" )
The environment the service reads. Capacity reporting is an observability concern, configured beside OTEL_* and the other MCP_PLATFORM_* telemetry knobs rather than in the platform's YAML.
Variables ¶
var Tables = []string{
"audit_logs",
"call_records",
"script_runs",
"script_versions",
"script_recordings",
"notifications",
"index_jobs",
"connection_auth_events",
"memory_records",
"portal_knowledge_pages",
"portal_knowledge_page_versions",
"knowledge_changesets",
"portal_assets",
"portal_asset_versions",
"resources",
"resource_versions",
"prompt_versions",
"webhook_windows",
"webhook_rejections",
"config_changelog",
"content_producers",
"tool_embeddings",
"api_catalog_operation_embeddings",
"graphql_operation_embeddings",
"portal_knowledge_page_embedding_chunks",
"script_embedding_chunks",
"catalog_datasets",
}
Tables is every platform table that grows with use and is reported by name. A table the platform keeps small (settings, definitions, the OAuth state) is left out; a table a migration has not created yet reads as absent, never as an error. The names are the migrations' own, which is what bounds the table label.
Functions ¶
func DetectBackend ¶
DetectBackend names the object store an endpoint is: no endpoint is AWS S3, an amazonaws.com or googleapis.com host names itself, and anything else is asked once, by the Server header its root answers with (SeaweedFS sends one). A store that does not answer, or answers as something else, is other.
func Endpoint ¶
Endpoint is the endpoint of the S3 connection named conn in the toolkits configuration, or of the default connection when conn is empty. Empty when there is no such connection or it names no endpoint (AWS).
func ParseBudgets ¶
func ParseBudgets(v string) ([]observability.BucketBudget, error)
ParseBudgets reads "bucket=size,bucket=size", where size is a whole number of bytes or one with a unit (KiB, MiB, GiB, TiB, or KB, MB, GB, TB). Empty is no budgets.
func ReadStorage ¶
func ReadStorage(ctx context.Context, db *sql.DB, backends map[string]string) (observability.StorageCapacity, error)
ReadStorage reads the usage rows, the last reconcile and the last scan. backends names each bucket the deployment configures and the object store it is on, which labels its rows; a row for any other bucket (one the deployment no longer uses) is not reported. Nil reports every row, labeled other.
func SampleTables ¶
func SampleTables(ctx context.Context, db *sql.DB) (observability.DatabaseCapacity, error)
SampleTables reads the table sizes, the vector index sizes and the transaction id age. Each read is a catalog lookup; none scans a table.
Types ¶
type BucketPrefix ¶
BucketPrefix is one prefix of one bucket.
type Config ¶
type Config struct {
TableInterval time.Duration
ScanInterval time.Duration
FlushInterval time.Duration
OrphanGrace time.Duration
// Budgets is the byte budget per bucket an operator set.
Budgets []observability.BucketBudget
// Backend, when set, names the object store kind instead of asking it.
Backend string
}
Config is the service's cadence and the operator's budgets.
func ConfigFromEnv ¶
ConfigFromEnv reads the configuration. An unset variable takes its default. A value that does not parse takes its default too, and is named in the error returned beside a configuration that keeps every variable that did parse: a typo in one never drops the budgets set in another. An unset orphan grace is DefaultOrphanGrace; zero is accepted and counts an object the moment it has no row.
type Layout ¶
Layout is where the platform keeps its objects: the portal's bucket and key prefix, and the managed-resources bucket (which also holds webhook segments and, unless a deployment points maps elsewhere, basemap archives).
func (Layout) Classify ¶
Classify names the purpose of the object at key in bucket by where it sits, the one rule the full listing and every put and delete share. ok is false for an object outside the platform's prefixes.
A key says what an object is for only as far as its layout does. Inside the portal prefix a tile is told apart by its file name, a collection mosaic and a script's outputs by their second segment, and a GraphQL or API export by its directory; a Trino export is written in an asset's own shape, so it is counted with the portal's assets.
func (Layout) Prefixes ¶
func (l Layout) Prefixes() []BucketPrefix
Prefixes is every bucket and key prefix a full listing walks, each once.
type Recorder ¶
type Recorder struct {
// contains filtered or unexported fields
}
Recorder collects the puts and deletes the platform makes (through internal/objectobs.DoObject) and adds them to the shared usage rows on Flush. Collecting first keeps a burst of writes to one bucket from becoming a burst of updates to one row.
func NewRecorder ¶
NewRecorder returns a recorder that sorts keys by layout.
func (*Recorder) Flush ¶
Flush adds every pending delta to its row. A delta that fails to write is kept for the next flush.
func (*Recorder) RecordObject ¶
RecordObject notes one object put (objects 1, bytes stored) or deleted (objects -1). A key outside the platform's prefixes is ignored, and so is a tile: the thumbnail worker redraws a tile in place at the same key, so a put is not a new object, and the full listing alone counts them. Every other purpose writes a new key per version, run, segment or build.
type ScanResult ¶
type ScanResult struct {
Usage map[usageKey]usageDelta
Snapshot map[usageKey]usageDelta
Orphaned map[string]int64
Dangling map[string]int64
Objects int64
Duration time.Duration
Buckets []string
}
ScanResult is what one full listing found. Usage counts the objects that existed when it began; Snapshot is the usage rows as they stood then, so a save keeps what the rows gained while the listing ran (Save).
func (ScanResult) Save ¶
Save writes a scan's result in one transaction. Each usage row of a listed bucket becomes the listing's total plus what the row gained after the listing read it: a put made while the listing ran is kept, not overwritten, and not counted twice (the listing skipped objects modified after it began). A row for a bucket the deployment no longer lists is removed. Then the reconcile counts and the scan's own duration.
type Scanner ¶
type Scanner struct {
DB *sql.DB
Layout Layout
// Walkers maps a bucket to the client that lists it.
Walkers map[string]Walker
// OrphanGrace is how recently an object may have been written and still
// not be counted as an orphan: a writer stores the object before the row
// that references it.
OrphanGrace time.Duration
// contains filtered or unexported fields
}
Scanner lists every platform prefix and compares what it finds with the rows that reference objects.
func (*Scanner) Scan ¶
func (s *Scanner) Scan(ctx context.Context) (ScanResult, error)
Scan reads the usage rows and the references, then lists every prefix. It changes nothing. An object modified after the listing began is not counted in its usage: the write that made it moved the row while the listing ran, and Save keeps that.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service runs the loops and holds the last readings.
func Start ¶
func Start(m *observability.Metrics, src Sources, cfg Config) *Service
Start installs the usage recorder, registers the service as m's capacity sampler and starts the loops. With no database or no metrics it does nothing and returns a service whose Close is a no-op.
func (*Service) Close ¶
Close stops the loops, writes what the recorder still holds, and removes it. Nil-safe.
func (*Service) Sample ¶
func (s *Service) Sample(_ context.Context) observability.CapacitySample
Sample is the scrape-time read: the last readings, never a query.
type Sources ¶
type Sources struct {
DB *sql.DB
Layout Layout
// Walkers maps each platform bucket to the client that lists it.
Walkers map[string]Walker
// Endpoints maps each platform bucket to the S3 endpoint of the
// connection it is reached through, each asked once which object store it
// is unless Config.Backend names it (DetectBackend). The portal's and the
// managed resources' buckets may be on different stores.
Endpoints map[string]string
// Budgets, when set, reads the byte budgets an administrator set per
// bucket (#2148), which override Config.Budgets bucket by bucket. It is
// read with the table sampler.
Budgets func(ctx context.Context) (map[string]int64, error)
}
Sources is what the service reads.
type Walker ¶
type Walker interface {
Walk(ctx context.Context, bucket, prefix string, fn func(s3adapter.WalkedObject) error) error
PutObject(ctx context.Context, bucket, key string, data []byte, contentType string) error
}
Walker lists every object under a prefix of a bucket, and writes the listing's start marker. pkg/portal/s3adapter.ClientAdapter is one.
Source Files