Documentation
¶
Overview ¶
Package ingest is the mAPI-ng collector: the Connect/gRPC IngestService handler that authenticates ingest keys, resolves the tenant, enforces the timestamp policy, and writes Summary rows to the data plane. Auth and the tenant column exist from day one so the model does not churn when the real control plane lands.
Index ¶
- type Handler
- type HandshakeRecorder
- type InstanceWindowSink
- type KeyResolver
- type Option
- func WithCardinality(allow cardinalityFunc, capacity capProvider) Option
- func WithHandshakeRecorder(r HandshakeRecorder) Option
- func WithInstanceWindowSink(sink InstanceWindowSink) Option
- func WithLimiter(allow func(tenant string) bool) Option
- func WithPayloadLimit(limit func(ctx context.Context, tenant string) int64) Option
- type RowSink
- type StaticKeyResolver
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Handler ¶
type Handler struct {
// contains filtered or unexported fields
}
Handler implements mapingv1connect.IngestServiceHandler. It authenticates the ingest key, resolves the tenant, applies the timestamp policy, converts each Summary to a storage row, and enqueues it. Unknown keys are rejected with CodeUnauthenticated; per-tenant abuse is throttled with a token bucket.
func NewHandler ¶
NewHandler builds an ingest Handler. Without options it uses the fixed default per-tenant limiter and no cardinality guard.
func (*Handler) Register ¶
func (h *Handler) Register( ctx context.Context, req *connect.Request[mapingv1.Handshake], ) (*connect.Response[mapingv1.RegisterResponse], error)
Register records a handshake: it authenticates and returns accepted, driving the dashboard onboarding state (CONTEXT Handshake).
func (*Handler) Upload ¶
func (h *Handler) Upload( ctx context.Context, req *connect.Request[mapingv1.UploadRequest], ) (*connect.Response[mapingv1.UploadResponse], error)
Upload validates and stores a batch of Summaries. Out-of-band-skew summaries are dropped and counted into RejectedSummaries rather than clamped onto now.
type HandshakeRecorder ¶
type HandshakeRecorder interface {
RecordHandshake(ctx context.Context, tenant, service, instance, sdkVersion string) error
}
HandshakeRecorder persists the one-time registration ping so the dashboard's onboarding panel can show "service connected" (CONTEXT Handshake). It is an interface, not a *control.Store, so ingest stays control-agnostic (ingest must never import control — main adapts the store to this at wiring time). A nil recorder (the default) keeps today's log-only Register behavior.
type InstanceWindowSink ¶
type InstanceWindowSink interface {
EnqueueInstanceWindow(row storage.InstanceWindowRow) error
}
InstanceWindowSink is the optional write side for per-instance USE gauges. It is separate from RowSink (a different table) and wired only when a sink is provided (WithInstanceWindowSink), so the zero-option handler and existing tests ignore instance windows entirely. The storage.Writer satisfies it via EnqueueInstanceWindow.
type KeyResolver ¶
type KeyResolver interface {
// Resolve returns the tenant for key and ok=false if the key is unknown.
Resolve(ctx context.Context, key string) (tenant string, ok bool)
}
KeyResolver maps an ingest key to a tenant. The control-plane implementation queries Postgres; local dev and tests use StaticKeyResolver.
type Option ¶
type Option func(*Handler)
Option configures an ingest Handler. Options are additive: the zero-option call reproduces the fixed default limiter and no cardinality guard, so existing wiring and tests are unaffected.
func WithCardinality ¶
func WithCardinality(allow cardinalityFunc, capacity capProvider) Option
WithCardinality enables the best-effort series-cardinality cap. allow tracks per-tenant series (adapting guardrail.Cardinality.Allow); cap resolves each tenant's budget (typically from the control plane). Both must be non-nil for the guard to run.
func WithHandshakeRecorder ¶
func WithHandshakeRecorder(r HandshakeRecorder) Option
WithHandshakeRecorder persists each accepted Register handshake through r, so the dashboard onboarding panel reflects connected services. A nil r is ignored, keeping the default log-only behavior and leaving existing tests unchanged. Recording failures never fail the handshake (see Register).
func WithInstanceWindowSink ¶
func WithInstanceWindowSink(sink InstanceWindowSink) Option
WithInstanceWindowSink enables ingestion of per-instance USE gauges through sink (typically the storage.Writer). A nil sink is ignored, keeping instance windows off by default so existing wiring and tests are unaffected.
func WithLimiter ¶
WithLimiter replaces the default per-tenant token-bucket with a caller- supplied allow function (e.g. adapting guardrail.RateLimiter fed by plan_limits). A nil allow is ignored, keeping the default.
func WithPayloadLimit ¶
WithPayloadLimit enables per-tenant logical payload enforcement. cap resolves the tenant's max ingest payload in bytes (typically from the control plane); Upload measures the decoded request with proto.Size and rejects the whole request with CodeResourceExhausted when it exceeds the cap. A nil cap or a cap returning <= 0 disables the check, so the option is purely additive: absent it (dev-without-Postgres / existing tests), no per-tenant payload check runs and behavior is unchanged.
type RowSink ¶
type RowSink interface {
// Enqueue hands one converted Summary row to the data plane.
Enqueue(row storage.Row) error
}
RowSink is the write side the handler depends on, so the handler is testable with a fake sink and does not import a live ClickHouse connection. The storage.Writer satisfies this via Enqueue.
type StaticKeyResolver ¶
type StaticKeyResolver struct {
// contains filtered or unexported fields
}
StaticKeyResolver is an in-memory KeyResolver for local dev and tests. Keys are stored hashed and every candidate is compared with constant-time equality so neither a match nor a mismatch leaks key material through timing.
func NewStaticKeyResolver ¶
func NewStaticKeyResolver(keyToTenant map[string]string) *StaticKeyResolver
NewStaticKeyResolver builds a resolver from a plaintext key -> tenant map, hashing each key at construction so no plaintext is retained.
Source Files