dbrpc

package module
v0.16.7 Latest Latest
Warning

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

 Go to latest Published: Jul 23, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

Documentation

Overview

Package dbrpc is the client/server DB: it serves an embedded ClassAd log (package db) over a message transport (CEDAR in production), with out-of-order request multiplexing so a slow call never head-of-line-blocks others on the same connection. See DESIGN.md.

Index

Constants

View Source 
const DefaultTable = "ads"

DefaultTable is the table name a single-DB server serves and the client targets when a table is not named -- the historical, single-collection view.

Variables

View Source 
var ErrBucketedUnsupported = errors.New("dbrpc: server does not support bucketed aggregation")

ErrBucketedUnsupported is returned by AggregateBucketed against a server too old to implement the opcode (it rejects the request), so a caller can fall back to client-side bucketing.

View Source 
var ErrConnClosed = errors.New("dbrpc: connection closed")

ErrConnClosed reports that the dbrpc connection is no longer usable. It wraps the underlying transport cause, so errors.Is(err, ErrConnClosed) identifies the class while errors.Unwrap reaches the specific I/O error.

View Source 
var WireBatchBudget = 64 << 10

WireBatchBudget caps one wire-row batch frame's payload bytes. It bounds the per-stream buffer on BOTH sides (each holds ~one frame), amortizes the per-frame syscall/wakeup cost, and stays well under the transport's 1MB message ceiling. Measured sensitivity (2000x21KB-ad scans over TCP loopback): 16KB->64KB gains ~6%, 64KB->256KB ~4%, beyond 256KB flat -- so the default stays memory-lean at 64KB (~128KB per active stream across both sides) and a deployment can raise it for the last few percent on whole-ad scans (projected streams are insensitive: their rows are small enough that every budget batches deeply).

Functions

func OpName added in v0.16.3 

func OpName(op uint8) string

OpName maps a CallStats.Op byte to a short stable label for the write-path ops a collector times (begin/commit/batch); anything else returns "op<N>". Used to label the per-call timing metrics without exposing the op constants.

Types

type AdKV added in v0.13.0 

type AdKV struct {
	Key string
	Ad  string // old-ClassAd text
}

AdKV is one keyed ad for the batch writers.

type AdReject added in v0.13.0 

type AdReject struct {
	Index int
	Err   string
}

AdReject reports an ad a batch writer could not apply (unparseable text), by its index in the items slice, with the server's error. The other ads were applied; the caller typically logs the rejects and carries on.

type AggFunc 

type AggFunc uint8

AggFunc is a SQL aggregate function. 

const (
	AggCount AggFunc = iota // COUNT(*) or COUNT(col)
	AggSum
	AggAvg
	AggMin
	AggMax
)

type AggRow 

type AggRow struct {
	Group  []string
	Values []string
}

AggRow is one group's result: the group-by column values followed by the aggregate values, all rendered as strings (aligned with the request's group columns and aggregate specs).

type AggSpec 

type AggSpec struct {
	Func AggFunc
	Arg  string
}

AggSpec is one aggregate in a query: a function over an argument attribute. Arg "*" (only meaningful for COUNT) counts every row in the group; otherwise Arg is an attribute name evaluated per ad.

type CallStats added in v0.16.3 

type CallStats struct {
	Op        uint8         // request op byte (see OpName)
	WriteWait time.Duration // blocked acquiring the write lock (another send ahead)
	Send      time.Duration // in conn.WriteMsg -- getting the request onto the wire
	Wait      time.Duration // from send complete to response received
}

CallStats is the per-call timing breakdown delivered to Client.Observer. The three durations partition a unary round-trip and pinpoint a stall: a large WriteWait means another sender held the single-writer lock; a large Send means conn.WriteMsg itself blocked (the socket's send buffer filled because the peer was not reading -- comms/server backpressure, not this client); a large Wait means the request left promptly but the server (or return path) was slow.

type Catalog 

type Catalog interface {
	// Table returns the named table's DB.
	Table(name string) (*db.DB, bool)
	// CreateTable creates (or returns the existing) named table.
	CreateTable(name string) (*db.DB, error)
	// CreateTableInMemory creates (or returns the existing) named table as RAM-only.
	CreateTableInMemory(name string) (*db.DB, error)
	// ConvertTableToMemory drops an existing table's on-disk backing, keeping its data in
	// RAM only. DAEMON-gated by the server (it changes a table's durability).
	ConvertTableToMemory(name string) error
	// DropTable removes the named table and its data.
	DropTable(name string) error
	// Tables lists the table names.
	Tables() []string
	// ArchiveTable / CreateArchiveTable / ArchiveTables manage append-only history
	// tables. A single-table server does not support them.
	ArchiveTable(name string) (*db.ArchiveTable, bool)
	CreateArchiveTable(name string, cfg db.ArchiveConfig) (*db.ArchiveTable, error)
	ArchiveTables() []string
	// CreateView / DropView / Views manage materialized views; ViewBacking returns a view's
	// in-memory backing so reads resolve a view name to its materialized rows.
	CreateView(name string, spec db.ViewSpec) error
	DropView(name string) error
	Views() []string
	ViewBacking(name string) (*db.DB, bool)
	// Exporter registry: passive storage of external-sink definitions and their opaque
	// resume state. The server never runs exporters; it only persists these for the
	// out-of-process exporter that reads them.
	CreateExporter(def db.ExporterDef) error
	DropExporter(name string) error
	Exporters() []db.ExporterDef
	Exporter(name string) (db.ExporterDef, bool)
	SaveExporterState(name string, state []byte) error
	LoadExporterState(name string) ([]byte, bool, error)
}

Catalog is the set of named tables the server serves. *db.Catalog implements it; a single-DB server wraps its DB as a one-table catalog.

type Client 

type Client struct {

	// Observer, if set before use, receives a timing breakdown for each completed unary
	// call. It lets an embedder (e.g. a collector's Prometheus metrics) localize where a
	// round-trip spends its time -- client-side write-lock contention, the socket write
	// (TCP backpressure when the server stops reading), or the server+return wait --
	// without this package depending on any metrics library. Must not block.
	Observer func(CallStats)
	// contains filtered or unexported fields
}

Client is one multiplexed connection to a dbrpc Server. It issues requests with monotonic ids and a single reader goroutine demuxes responses by id, so many calls (and independent transactions) proceed concurrently over the one connection with no head-of-line blocking. Safe for concurrent use. For many connections, see Pool.

func NewClient 

func NewClient(conn MsgConn) *Client

NewClient runs the mux over conn. Close (or a transport error) fails all in-flight and future calls.

func (*Client) Admin 

func (c *Client) Admin(ctx context.Context, action string, args ...string) (string, error)

Admin runs a management action (index/hot-set) on the default table. Requires a DAEMON-authorized connection (see AdminTable).

func (*Client) AdminTable 

func (c *Client) AdminTable(ctx context.Context, table, action string, args ...string) (string, error)

AdminTable runs a management action on the named table; it returns the server's human-readable result. Every admin action retunes or restructures the store, so it requires DAEMON authorization -- refused on read-only and ordinary read-write connections alike.

func (*Client) Aggregate 

func (c *Client) Aggregate(ctx context.Context, constraint string, groupBy []string, aggs []AggSpec) ([]AggRow, error)

Aggregate runs a server-side GROUP BY: the server buckets the constraint match by the group-by column tuple in a hash map and returns one AggRow per group. With no group columns it returns a single row aggregating the whole match. The aggregation happens on the server, so only the (small) grouped result crosses the wire, not every matched ad.

func (*Client) AggregateBucketed added in v0.13.0 

func (c *Client) AggregateBucketed(ctx context.Context, constraint string, groups []GroupCol, aggs []AggSpec) ([]AggRow, error)

AggregateBucketed is Aggregate with group columns that may be time-bucketed (see GroupCol), pushing the bucketing to the server so only the grouped rows cross the wire. Against a server that does not implement the opcode it returns an error that wraps ErrBucketedUnsupported.

func (*Client) AggregateBucketedTable added in v0.13.0 

func (c *Client) AggregateBucketedTable(ctx context.Context, table, constraint string, groups []GroupCol, aggs []AggSpec) ([]AggRow, error)

AggregateBucketedTable is AggregateBucketed on the named table.

func (*Client) AggregateTable 

func (c *Client) AggregateTable(ctx context.Context, table, constraint string, groupBy []string, aggs []AggSpec) ([]AggRow, error)

AggregateTable is Aggregate on the named table.

func (*Client) ArchiveAppend added in v0.7.0 

func (c *Client) ArchiveAppend(ctx context.Context, name, adText string) error

ArchiveAppend appends an ad (old-ClassAd text) to the named history table.

func (*Client) ArchiveQuery added in v0.7.0 

func (c *Client) ArchiveQuery(ctx context.Context, name, constraint string, limit int) ([]string, error)

ArchiveQuery returns up to limit (<= 0 = all) newest-first matches (old-ClassAd texts) from the named history table -- the condor_history "last K" pattern.

func (*Client) ArchiveRotate added in v0.7.0 

func (c *Client) ArchiveRotate(ctx context.Context, name string) (int, error)

ArchiveRotate enforces the named archive's retention policy now (using the server's clock for age-based rules), returning how many sealed segments were dropped.

func (*Client) ArchiveTables added in v0.7.0 

func (c *Client) ArchiveTables(ctx context.Context) ([]string, error)

ArchiveTables lists the history table names.

func (*Client) BackupKeyTable added in v0.7.0 

func (c *Client) BackupKeyTable(ctx context.Context, table string) ([]byte, error)

BackupKeyTable retrieves the named table's backup key -- the escrow key that decrypts its encrypted snapshots independently of the pool keys. DAEMON-level. Errors if encryption is not enabled.

func (*Client) Begin 

func (c *Client) Begin(ctx context.Context) (*Tx, error)

Begin starts a new independent transaction on the default table ("ads").

func (*Client) BeginTable 

func (c *Client) BeginTable(ctx context.Context, table string) (*Tx, error)

BeginTable starts a new independent transaction on the named table.

func (*Client) Close 

func (c *Client) Close() error

Close closes the underlying connection; pending and future calls fail.

func (*Client) ConvertTableToMemory added in v0.10.2 

func (c *Client) ConvertTableToMemory(ctx context.Context, name string) error

ConvertTableToMemory drops an existing table's on-disk backing, keeping its current contents in RAM only (they are gone after a server restart). Requires DAEMON authorization. Best run during low write activity: a write that races the conversion can be lost (the server takes a consistent snapshot but does not globally quiesce writers).

func (*Client) CreateArchiveTable added in v0.7.0 

func (c *Client) CreateArchiveTable(ctx context.Context, name string, cfg db.ArchiveConfig) error

CreateArchiveTable creates (or no-ops if present) an append-only history table. cfg configures indexes / zone maps / retention on first creation.

func (*Client) CreateExporter added in v0.14.0 

func (c *Client) CreateExporter(ctx context.Context, def db.ExporterDef) error

CreateExporter registers an external-sink exporter definition. DAEMON-only. The server stores the definition (and later its resume state) but never runs the exporter -- an out-of-process exporter reads these back and does the work.

func (*Client) CreateTable 

func (c *Client) CreateTable(ctx context.Context, name string) error

CreateTable creates (or no-ops if present) the named table.

func (*Client) CreateTableInMemory added in v0.10.2 

func (c *Client) CreateTableInMemory(ctx context.Context, name string) error

CreateTableInMemory creates (or no-ops if present) the named table as RAM-only: its data lives only in memory and is not recovered across a server restart. On a persistent server this avoids the disk I/O of persistence for high-churn, reconstructible data.

func (*Client) CreateView added in v0.12.0 

func (c *Client) CreateView(ctx context.Context, name string, spec db.ViewSpec) error

CreateView creates a materialized view named name from spec. The server materializes it synchronously and then maintains it live from the base table's change stream, so an error (e.g. the base table exceeds the view's cardinality limit) is reported here. A view is queried like a table (SELECT ... FROM <name>) but is read-only.

func (*Client) DeleteWhere added in v0.8.0 

func (c *Client) DeleteWhere(ctx context.Context, constraint string) (int, error)

DeleteWhere removes every ad matching constraint from the default table and returns the number removed. The delete runs server-side as a single call -- the db.DeleteWhere pushdown (batched, optimistic-retry, self-healing) -- rather than a per-key query-then-delete round trip per ad. Refused on a read-only connection.

func (*Client) DeleteWhereTable added in v0.8.0 

func (c *Client) DeleteWhereTable(ctx context.Context, table, constraint string) (int, error)

DeleteWhereTable is DeleteWhere against a named table. Express a time-based expiry sweep as the constraint (e.g. "<now> > LastHeardFrom + Lifetime").

func (*Client) Diagnostics 

func (c *Client) Diagnostics(ctx context.Context) (*Diagnostics, error)

Diagnostics fetches the default table's storage stats, hot set, indexes, and tuning suggestions.

func (*Client) DiagnosticsTable 

func (c *Client) DiagnosticsTable(ctx context.Context, table string) (*Diagnostics, error)

DiagnosticsTable fetches the named table's diagnostics.

func (*Client) DropExporter added in v0.14.0 

func (c *Client) DropExporter(ctx context.Context, name string) error

DropExporter removes an exporter's definition and resume state. DAEMON-only.

func (*Client) DropTable 

func (c *Client) DropTable(ctx context.Context, name string) error

DropTable removes the named table and its data.

func (*Client) DropView added in v0.12.0 

func (c *Client) DropView(ctx context.Context, name string) error

DropView removes a materialized view (its definition and its in-memory data).

func (*Client) Explain 

func (c *Client) Explain(ctx context.Context, constraint string) (*db.QueryExplain, error)

Explain reports how the default table would execute a constraint query.

func (*Client) ExplainTable 

func (c *Client) ExplainTable(ctx context.Context, table, constraint string) (*db.QueryExplain, error)

ExplainTable reports how the named table would execute a constraint query.

func (*Client) GetExporter added in v0.14.0 

func (c *Client) GetExporter(ctx context.Context, name string) (db.ExporterDef, bool, error)

GetExporter returns a single exporter's full definition (including Config). DAEMON-only. The bool is false when no exporter of that name exists.

func (*Client) GetExporterState added in v0.14.0 

func (c *Client) GetExporterState(ctx context.Context, name string) ([]byte, bool, error)

GetExporterState returns an exporter's last checkpointed resume-state blob. DAEMON-only. The bool is false when the exporter has never checkpointed (start from the beginning).

func (*Client) ListExporters added in v0.14.0 

func (c *Client) ListExporters(ctx context.Context) ([]ExporterInfo, error)

ListExporters returns the registered exporters' names and kinds (no config).

func (*Client) ListViews added in v0.12.0 

func (c *Client) ListViews(ctx context.Context) ([]string, error)

ListViews returns the materialized view names.

func (*Client) MatchExplain 

func (c *Client) MatchExplain(ctx context.Context, reqTable, jobSelector, resTable, targetWhere string) (*db.MatchExplain, error)

MatchExplain reports how matchmaking the first request in reqTable matching jobSelector against resTable would execute: the job's Requirements rewritten over the slot (job constants baked in) and which resulting probes prune via a resource index. jobSelector is a constraint (e.g. `Key == "1.0"`) identifying the request.

func (*Client) MatchSorted 

func (c *Client) MatchSorted(ctx context.Context, jobText string, limit int) ([]string, error)

MatchSorted returns job's matches in the default table, ranked best-first, at most limit (<=0 = all).

func (*Client) MatchSortedTable 

func (c *Client) MatchSortedTable(ctx context.Context, table, jobText string, limit int) ([]string, error)

MatchSortedTable returns job's ranked matches in the named table.

func (*Client) MatchTables 

func (c *Client) MatchTables(ctx context.Context, reqTable, resTable, keyAttr, reqWhere, targetWhere string, limit int, significantAttrs []string) ([]MatchRow, error)

MatchTables performs cross-table matchmaking as a greedy assignment: it walks the requests in reqTable matching reqWhere (in table order) and gives each one the single best-ranked resource in resTable — bilaterally matching, passing the resource-side filter targetWhere, and not already assigned to an earlier request — then removes that resource from the pool. limit bounds the number of *requests* assigned (the first `limit` in order), not resources per request. The result is one row per assigned request (Resource is "" when it could not be placed). keyAttr is the ad attribute holding each row's key.

significantAttrs, if non-empty, enables autoclustering: requests whose significant attributes are textually identical share one ranked-candidate computation (the matchmaking runs once per distinct signature and is reused; assignment still consumes resources per request), mirroring HTCondor's significant-attribute autoclusters. It must list every attribute the match depends on (the request's Requirements and Rank, and the request attributes the resources' Requirements reference).

func (*Client) Ordered 

func (c *Client) Ordered(ctx context.Context, index int, partition string) ([]OrderedRow, error)

Ordered streams one partition of the index-th configured ordered index in sort order (the negotiator resource-request path). One-shot: the whole partition is returned (the server-side resume cursor is not carried over the wire).

func (*Client) OrderedTable 

func (c *Client) OrderedTable(ctx context.Context, table string, index int, partition string) ([]OrderedRow, error)

OrderedTable streams one partition of an ordered index in the named table.

func (*Client) PutExporterState added in v0.14.0 

func (c *Client) PutExporterState(ctx context.Context, name string, state []byte) error

PutExporterState durably stores an exporter's opaque resume-state blob. DAEMON-only. The exporter calls this after downstream has accepted its data (the at-least-once boundary).

func (*Client) Query 

func (c *Client) Query(ctx context.Context, constraint string) ([]string, error)

Query returns the committed ads (old-ClassAd texts) in the default table ("ads") matching a constraint. The server streams results; a slow scan does not block other calls.

func (*Client) QueryAsOfTable added in v0.12.0 

func (c *Client) QueryAsOfTable(ctx context.Context, table, constraint string, limit int, asOf time.Time) ([]string, error)

QueryAsOfTable is QueryTable for a point-in-time ("AS OF") query: it returns the ads in the named table matching constraint as they were at asOf. The server errors if time travel is not enabled on the table or asOf is older than the retained window.

func (*Client) QueryLimit 

func (c *Client) QueryLimit(ctx context.Context, constraint string, limit int) ([]string, error)

QueryLimit is Query with a row cap (<= 0 = all) on the default table.

func (*Client) QueryRaw added in v0.8.0 

func (c *Client) QueryRaw(ctx context.Context, constraint string) ([]string, error)

QueryRaw is Query but returns each ad as old-ClassAd wire text (the AST-free relay form for the caller): the server renders straight from the stored representation via the db QueryRaw pushdown, and the caller can forward the text without building a ClassAd. Private attributes are stripped unless the connection is privileged.

func (*Client) QueryRawProject added in v0.10.0 

func (c *Client) QueryRawProject(ctx context.Context, table, constraint string, attrs []string, limit int) ([]string, error)

QueryRawProject is QueryRawTable with a server-side projection: each returned ad carries only the attributes in attrs (matched case-insensitively) plus MyType/TargetType, so a query for a few attributes does not pull every attribute of every ad across the wire. An empty attrs behaves like QueryRawTable (all attributes). Private attributes are stripped unless the connection is privileged.

func (*Client) QueryRawProjectStream added in v0.13.0 

func (c *Client) QueryRawProjectStream(ctx context.Context, table, constraint string, attrs []string, limit int, yield func(row string) bool) error

QueryRawProjectStream is QueryRawProject (server-side projection) with the streaming delivery of QueryRawTableStream.

func (*Client) QueryRawTable added in v0.8.0 

func (c *Client) QueryRawTable(ctx context.Context, table, constraint string, limit int) ([]string, error)

QueryRawTable is QueryRaw against a named table with an optional limit.

func (*Client) QueryRawTableStream added in v0.13.0 

func (c *Client) QueryRawTableStream(ctx context.Context, table, constraint string, limit int, yield func(row string) bool) error

QueryRawTableStream is QueryRawTable that hands each matching ad's old-ClassAd wire text to yield as it arrives, instead of collecting the whole result into a slice -- so a relay (e.g. the collector) can forward each ad to its own client without buffering the entire result set. yield returns false to stop early. See streamEach for the error contract (a failure can arrive after some rows have been yielded).

func (*Client) QueryRawWireStream added in v0.16.4 

func (c *Client) QueryRawWireStream(ctx context.Context, table, constraint string, attrs []string, limit int, redact bool, yield func(row []byte) bool) error

QueryRawWireStream streams matching ads as wire-form rows (self-contained inline-names subset ads -- render them with collections.RenderRawAdInline), batched many rows per frame (WireBatchBudget on the server side). The row slice passed to yield aliases the frame buffer and is valid only until yield returns; redact requests source-side private-attribute stripping even on a privileged connection. Requires a server with opQueryRawWire; an older server answers respBad and the error surfaces here (callers fall back to the text row stream).

func (*Client) QueryTable 

func (c *Client) QueryTable(ctx context.Context, table, constraint string, limit int) ([]string, error)

QueryTable returns the committed ads in the named table matching constraint, stopping after at most limit matches (<= 0 = all). The limit is pushed to the server, which stops the scan early.

func (*Client) QueryTableStream added in v0.13.0 

func (c *Client) QueryTableStream(ctx context.Context, table, constraint string, limit int, yield func(row string) bool) error

QueryTableStream is QueryTable that hands each matching ad's text to yield as it arrives instead of collecting the whole result first (see streamEach). yield returns false to stop early.

func (*Client) Restore added in v0.7.0 

func (c *Client) Restore(ctx context.Context, r io.Reader) error

func (*Client) RestoreTable added in v0.7.0 

func (c *Client) RestoreTable(ctx context.Context, table string, r io.Reader) error

RestoreTable replaces the named table with the snapshot read from r. DAEMON-level and destructive: the server spools the upload and restores under the DB-wide lock. The upload streams in chunks, so the client does not buffer the whole snapshot.

func (*Client) SetEncryptedAttrs added in v0.7.0 

func (c *Client) SetEncryptedAttrs(ctx context.Context, table string, attrs ...string) (string, error)

SetEncryptedAttrs sets the explicit attributes encrypted at rest on the named table (private attributes are always encrypted). It is a DAEMON-level action: the server refuses it unless the connection is privileged. Passing no attributes clears the explicit set. Returns the server's human-readable result.

func (*Client) Snapshot added in v0.7.0 

func (c *Client) Snapshot(ctx context.Context, w io.Writer) error

Snapshot / Restore operate on the default table.

func (*Client) SnapshotTable added in v0.7.0 

func (c *Client) SnapshotTable(ctx context.Context, table string, w io.Writer) error

SnapshotTable streams a consistent backup of the named table to w. DAEMON-level: a snapshot carries every attribute, including private ones. The server streams it in chunks, so neither side buffers the whole database.

func (*Client) Tables 

func (c *Client) Tables(ctx context.Context) ([]string, error)

Tables lists the table names.

func (*Client) TruncateTable added in v0.7.0 

func (c *Client) TruncateTable(ctx context.Context, table string) (string, error)

TruncateTable removes every ad from the named table. It is a DAEMON-level action (destructive, DB-wide locked): the server refuses it unless the connection is privileged. Returns the server's human-readable result.

func (*Client) Watch 

func (c *Client) Watch(ctx context.Context, cursor []byte) (<-chan WatchEvent, func(), error)

Watch streams changes committed after cursor (nil = from now) on the channel, which closes when the returned stop is called, the connection fails, or the server ends the stream. A full replay leads with a reset event.

func (*Client) WatchHead added in v0.7.0 

func (c *Client) WatchHead(ctx context.Context, table string) ([]byte, error)

WatchHead returns an opaque cursor at the current head of table's change log, so a following WatchTable(table, cursor) streams only subsequent changes (no replay of current contents) -- the "tail from now" path.

func (*Client) WatchTable 

func (c *Client) WatchTable(ctx context.Context, table string, cursor []byte) (<-chan WatchEvent, func(), error)

WatchTable streams changes to the named table.

type Diagnostics 

type Diagnostics struct {
	Stats              db.Stats             `json:"stats"`
	OpStats            db.OpStats           `json:"opStats"`
	Hot                []string             `json:"hot"`
	CategoricalIndexes []string             `json:"categoricalIndexes"`
	ValueIndexes       []string             `json:"valueIndexes"`
	IndexSizes         db.IndexSizes        `json:"indexSizes"`
	Codec              db.CodecStats        `json:"codec"`
	Suggestions        []db.IndexSuggestion `json:"suggestions"`
	DropSuggestions    []db.DropSuggestion  `json:"dropSuggestions"`
	// EncryptionEnabled reports whether encryption at rest is active; EncryptedAttrs is
	// the explicit encrypted-attribute set (private attributes are always encrypted and
	// are not listed here).
	EncryptionEnabled bool     `json:"encryptionEnabled"`
	EncryptedAttrs    []string `json:"encryptedAttrs,omitempty"`
}

Diagnostics is a snapshot of the store's storage, hot set, indexes, and index tuning advice -- the payload of the diagnostic ".stats"/".indexes"/".hot" commands.

type ExporterInfo added in v0.14.0 

type ExporterInfo struct {
	Name string
	Kind string
}

ExporterInfo is the name+kind pair returned by ListExporters. It deliberately omits the exporter's Config, which may hold credentials and is only returned by GetExporter to a DAEMON-authorized client.

type GroupCol added in v0.13.0 

type GroupCol struct {
	Attr        string
	BucketWidth int64
}

GroupCol is one GROUP BY column for a (possibly bucketed) aggregate: the attribute Attr, optionally floored into fixed-width buckets. BucketWidth == 0 groups by the raw attribute value; BucketWidth > 0 (seconds) groups by the epoch-aligned bucket floor(number(Attr)/BucketWidth)*BucketWidth, and a row whose Attr is not a finite number drops out of the result. This is the shared shape a bucketed aggregate (here) and a bucketed materialized view can both group by.

type MatchRow 

type MatchRow struct {
	Request  string
	Resource string
	Rank     string
}

MatchRow is one matchmaking result: a request key, the resource it matched, and the request's Rank of that resource (Rank is "" when the request has no numeric Rank for it).

type MsgConn 

type MsgConn interface {
	WriteMsg(b []byte) error
	ReadMsg() ([]byte, error)
	Close() error
}

MsgConn is a reliable, ordered, bidirectional message transport: each WriteMsg / ReadMsg carries one discrete frame. CEDAR's stream.Stream satisfies it via an adapter (see cedaradapter); tests use a length-prefixed net.Pipe. WriteMsg may be called concurrently with ReadMsg, but not concurrently with itself -- the mux serializes writers.

func NewCedarConn 

func NewCedarConn(ctx context.Context, s *stream.Stream) MsgConn

NewCedarConn wraps an established CEDAR stream (from a client dial, or a server HandlerFunc's Conn.Stream) as a dbrpc transport. ctx bounds the connection's I/O.

func NewStreamConn 

func NewStreamConn(rw io.ReadWriteCloser) MsgConn

NewStreamConn frames messages over an ordered byte stream. Exposed so a caller can run dbrpc over any io.ReadWriteCloser (e.g. a plain TCP conn) without CEDAR.

type OrderedRow 

type OrderedRow struct {
	Signature uint64
	AdText    string
}

OrderedRow is one ad from an ordered scan (old-ClassAd text) with its cluster signature -- run-length-fold equal signatures into a resource-request list.

type Pool 

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

Pool is a set of multiplexed connections to one Server, load-balanced round-robin. It borrows the database/sql idea of a connection pool, but adapts it: an SQL driver needs a connection per in-flight query (connections do not mux), so its pool serializes checkout; here each Client already multiplexes many concurrent calls, so the pool exists for throughput across connections and fault isolation, not mutual exclusion. A transaction begun through the pool pins the connection it started on (its ops go over that one Client, in order), while that connection keeps serving everyone else.

func NewPool 

func NewPool(dial func() (MsgConn, error), n int) (*Pool, error)

NewPool dials n connections with dial and multiplexes each. On any dial error it closes the connections already opened and returns the error.

func (*Pool) Begin 

func (p *Pool) Begin(ctx context.Context) (*Tx, error)

Begin starts a transaction on a pool connection; the returned Tx's operations all run over that connection (server-side transaction state + op ordering).

func (*Pool) Close 

func (p *Pool) Close() error

Close closes every connection. The first close error is returned.

func (*Pool) MatchSorted 

func (p *Pool) MatchSorted(ctx context.Context, jobText string, limit int) ([]string, error)

MatchSorted runs a ranked match on a pool connection.

func (*Pool) Ordered 

func (p *Pool) Ordered(ctx context.Context, index int, partition string) ([]OrderedRow, error)

Ordered streams a partition of an ordered index on a pool connection.

func (*Pool) Query 

func (p *Pool) Query(ctx context.Context, constraint string) ([]string, error)

Query runs a constraint query on a pool connection, streaming the results.

func (*Pool) Watch 

func (p *Pool) Watch(ctx context.Context, cursor []byte) (<-chan WatchEvent, func(), error)

Watch starts a watch on a pool connection.

type ProposeFunc added in v0.7.0 

type ProposeFunc func(table string, ops []WriteOp) error

ProposeFunc applies a committing transaction's ops to the replicated store via consensus. table is the transaction's table; ops preserve issue order. It returns nil on a quorum-committed apply, or an error (e.g. not-the-leader) that is surfaced to the client.

type QueryLog added in v0.10.0 

type QueryLog struct {
	Op         string        // the query opcode: "Query", "QueryRaw", ...
	Table      string        // the queried table
	Constraint string        // the constraint expression ("" / "true" = match-all)
	Limit      int           // the client's LIMIT (0 = unlimited)
	Rows       int           // rows streamed back
	Duration   time.Duration // wall-clock time to run + stream the query
}

QueryLog is one observed query, passed to ServeOptions.QueryLog.

type ServeOptions 

type ServeOptions struct {
	// ReadOnly rejects the mutating operations (NewClassAd, DestroyClassAd,
	// SetAttribute, DeleteAttribute) with an error; reads, snapshots (Begin),
	// and queries still work. A read-only peer may still Begin/Abort a
	// transaction to get a stable snapshot for reads.
	ReadOnly bool

	// IncludePrivate renders returned ads with their private (secret)
	// attributes intact (classad.StringWithPrivate). When false (the default),
	// private attributes are stripped from every ad this connection sees, so an
	// under-privileged peer never learns claim ids and other secrets.
	IncludePrivate bool

	// Privileged admits the DAEMON-level administrative actions -- ALL of the admin
	// table (index/hot/compact/rewrite/codec.retrain, plus the security/durability ones
	// like the encryption toggle and truncate). These retune or restructure the store,
	// so an ordinary WRITE-level session (which may read and write ads) is refused them;
	// only a DAEMON peer sets this true. Read-only diagnostics are a separate opcode and
	// are not gated here.
	Privileged bool

	// QueryLog, if set, is called once per streamed query with a summary of what
	// the client asked for and what it cost. It is an opt-in query log for
	// operators: it makes visible, for example, a client that fetches every
	// attribute of every ad instead of projecting server-side (a slow, chatty
	// query pattern). Called from the connection's read-loop goroutine, so it must
	// not block.
	QueryLog func(QueryLog)
}

ServeOptions scopes what a single served connection may do. The zero value is full read/write access with private attributes excluded from returned ads (the historical ServeConn behavior). A privilege-scoped front end (e.g. an HTCondor daemon serving READ vs WRITE peers) sets these per connection.

type Server 

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

Server serves an embedded ClassAd log (*db.DB) to remote clients. It is embeddable: create it around a DB, optionally start its managed maintenance goroutines, and hand it each accepted connection via ServeConn. Concurrent connections and concurrent in-flight requests per connection are supported; requests are dispatched as they arrive and responses carry the request id, so a slow call never head-of-line-blocks others.

func NewServer 

func NewServer(d *db.DB) *Server

NewServer returns a single-table server over d, served as table "ads". The caller owns d's lifetime. Table create/drop are unsupported on this server.

func NewServerCatalog 

func NewServerCatalog(cat Catalog) *Server

NewServerCatalog returns a multi-table server over cat.

func (*Server) Close 

func (s *Server) Close()

Close stops the server's managed goroutines. It does not close the DB.

func (*Server) ServeConn 

func (s *Server) ServeConn(conn MsgConn) error

ServeConn runs the request loop on one connection until it errors or the peer closes. Blocking; run one per accepted connection. Watches started on the connection are cancelled when it returns. Equivalent to ServeConnOpts with the zero ServeOptions (full read/write, private attributes excluded).

func (*Server) ServeConnOpts 

func (s *Server) ServeConnOpts(conn MsgConn, opts ServeOptions) error

ServeConnOpts is ServeConn scoped by opts: a read-only and/or private-stripping view of the same DB. Use it to serve a privilege-scoped peer (e.g. an HTCondor READ-level client) from the same Server.

func (*Server) SetProposeHook added in v0.7.0 

func (s *Server) SetProposeHook(fn ProposeFunc)

SetProposeHook routes committing writes through fn (raft) instead of the local store. Call once before serving. Passing nil restores direct local commits (the default). When set, the server is otherwise read-only against its local store: the hook is the sole writer.

func (*Server) StartIdemReaper added in v0.9.0 

func (s *Server) StartIdemReaper(interval, maxAge time.Duration) (stop func())

StartIdemReaper runs reapIdemMarkers every interval, deleting idempotency markers older than maxAge, until the returned stop is called. maxAge should exceed the longest window over which a client might replay a unit of work (its retry budget), so a marker is never reaped while a late replay could still arrive. Servers that never see CommitIdempotent need not start it (there is nothing to reap).

func (*Server) StartMaintenance 

func (s *Server) StartMaintenance(interval time.Duration, opts db.MaintainOptions)

StartMaintenance starts server-managed background maintenance: a single goroutine that, each tick, re-enumerates the catalog's tables and runs one Maintain pass (index auto-tune, hot-set refresh, and -- if opts.Retrain -- dictionary retrain) on each. Re-enumerating each tick means tables created after startup are maintained too (the old per-table start missed them).

The cadence is adaptive: interval is a floor, but the wait before the next pass is raised to keep maintenance under maintenanceMaxDutyCycle of wall-clock time -- so on a large store where a pass (dominated by dictionary retrain's recompaction) takes minutes, passes automatically space out rather than consuming the daemon. Stopped by Close.

func (*Server) StartTxnReaper added in v0.9.0 

func (s *Server) StartTxnReaper(interval, maxIdle time.Duration) (stop func())

StartTxnReaper runs reapIdleTxns every interval, aborting transactions idle longer than maxIdle, until the returned stop is called. Optional: disconnect cleanup is automatic; a server that wants to bound transactions abandoned on still-open connections starts this.

type ServerError added in v0.9.0 

type ServerError struct{ Msg string }

ServerError is a logical error the server returned for a request (wire status stErr): a bad constraint, an unknown table, a malformed op. It is deterministic -- replaying the request fails the same way -- so callers should surface it rather than retry.

func (*ServerError) Error added in v0.9.0 

func (e *ServerError) Error() string

type Tx 

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

Tx is a client-side handle to a server transaction. Its operations are issued in order over the client (the transaction is server-side state); different Tx (even on one Client) proceed concurrently.

func (*Tx) Abort 

func (t *Tx) Abort(ctx context.Context) error

Abort discards the transaction.

func (*Tx) Commit 

func (t *Tx) Commit(ctx context.Context) error

Commit applies the transaction, returning *db.ConflictError with the conflicted keys if any lost a write-write race (the rest committed), or nil.

func (*Tx) CommitIdempotent added in v0.9.0 

func (t *Tx) CommitIdempotent(ctx context.Context, idemKey string) error

CommitIdempotent is Commit with exactly-once semantics across retries: the server records a durable marker under idemKey, committed atomically with this transaction's writes, so replaying the SAME unit of work (same idemKey) after an ambiguous failure -- a reconnect where the original commit's fate is unknown -- applies it at most once. The caller must generate idemKey ONCE per unit of work and reuse it verbatim on every replay. Use this for non-idempotent transactions (e.g. relative updates, appends); an already-idempotent-by-key workload can use the cheaper plain Commit. Returns the same results as Commit (nil, or *db.ConflictError on a genuine write-write conflict).

func (*Tx) DeleteAttribute 

func (t *Tx) DeleteAttribute(ctx context.Context, key, name string) error

DeleteAttribute removes key's attribute name.

func (*Tx) DestroyClassAd 

func (t *Tx) DestroyClassAd(ctx context.Context, key string) error

DestroyClassAd removes key.

func (*Tx) LookupAttr 

func (t *Tx) LookupAttr(ctx context.Context, key, name string) (string, bool, error)

LookupAttr returns key's attribute name (unparsed expression) as the transaction sees it, or ("", false).

func (*Tx) LookupClassAd 

func (t *Tx) LookupClassAd(ctx context.Context, key string) (string, bool, error)

LookupClassAd returns key's ad (old-ClassAd text) as the transaction sees it.

func (*Tx) NewClassAd 

func (t *Tx) NewClassAd(ctx context.Context, key, adText string) error

NewClassAd stores the ad (old-ClassAd text) under key.

func (*Tx) NewClassAdBatch added in v0.13.0 

func (t *Tx) NewClassAdBatch(ctx context.Context, items []AdKV) ([]AdReject, error)

NewClassAdBatch stores many ads under one transaction in a single round-trip, instead of one request/ack per ad. It returns the ads the server rejected (bad text) so the caller can log them; the rest are applied. The whole request must fit the stream's max message size, so for a large batch prefer NewClassAdBatchPipelined, which chunks and pipelines automatically.

func (*Tx) NewClassAdBatchPipelined added in v0.13.0 

func (t *Tx) NewClassAdBatchPipelined(ctx context.Context, items []AdKV, chunkSize int) ([]AdReject, error)

NewClassAdBatchPipelined stores many ads under one transaction with the round-trips pipelined: it splits items into chunks of chunkSize ads and sends EVERY chunk frame back-to-back without waiting for any chunk's ack, then collects all the acks. So the chunks flow independently of the acks and the wire cost is ~one round-trip regardless of ad count -- not one per ad, nor one serialized round-trip per chunk -- while each message stays small and the server applies a whole chunk under a single transaction- lock acquisition (chunkSize goroutines/locks for the batch, not one per ad).

It returns the ads the server rejected (unparseable), indexed into items. A whole- chunk failure -- a dropped connection, or the transaction gone ("no such transaction") -- is returned as an error so the caller aborts and replays the batch. chunkSize <= 0 defaults to 64.

func (*Tx) SetAttribute 

func (t *Tx) SetAttribute(ctx context.Context, key, name, expr string) error

SetAttribute sets key's attribute name to expr.

type WatchEvent 

type WatchEvent struct {
	Kind   uint8 // 0 upsert, 1 delete, 2 reset (see db.WatchKind)
	Key    string
	AdText string
	Cursor []byte
}

WatchEvent is one change delivered over a Watch. AdText is the ad's old-ClassAd text (empty for a delete/reset). Cursor resumes the watch just after this event.

type WriteKind added in v0.7.0 

type WriteKind uint8

WriteKind identifies a buffered mutation. The values match the semantics of the qmgmt / classad-log operations, so the hook can translate a batch 1:1 into a raft mutation batch. 

const (
	WriteNewClassAd      WriteKind = iota // store Value (old-ClassAd text) under Key
	WriteDestroyClassAd                   // remove Key
	WriteSetAttribute                     // set Key's attribute Name to expression Value
	WriteDeleteAttribute                  // remove Key's attribute Name
)

type WriteOp added in v0.7.0 

type WriteOp struct {
	Kind        WriteKind
	Key         string
	Name, Value string
}

WriteOp is one buffered mutation from a transaction, ready to be proposed to consensus.

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.