Documentation
¶
Overview ¶
Package admin is Murmur's read-only control plane. It implements the AdminService defined in proto/murmur/admin/v1/admin.proto via Connect-RPC, which means the same endpoint speaks gRPC (for Go/JVM/Rust clients), gRPC-Web (for browsers without a proxy), and the Connect protocol (the browser UI's default — plain HTTP+JSON, no special transport).
The .proto is the contract. Anyone is welcome to point a different UI at the same endpoint, or generate a client in another language with `buf generate`. The Go server below is one implementation among many.
Pipelines register themselves with Server.Register; Server.Handler returns an http.Handler that serves the AdminService routes. Pair it with FullHandler if you also want the embedded UI on the same port.
Index ¶
- Variables
- func UIHandler() http.Handler
- type Authenticator
- type JWTVerifier
- type Option
- type PipelineInfo
- type QueryFn
- type Server
- func (s *Server) FullHandler() http.Handler
- func (s *Server) GetPipelineMetrics(_ context.Context, req *connect.Request[adminv1.GetPipelineMetricsRequest]) (*connect.Response[adminv1.GetPipelineMetricsResponse], error)
- func (s *Server) GetRange(_ context.Context, req *connect.Request[adminv1.GetRangeRequest]) (*connect.Response[adminv1.GetRangeResponse], error)
- func (s *Server) GetState(_ context.Context, req *connect.Request[adminv1.GetStateRequest]) (*connect.Response[adminv1.GetStateResponse], error)
- func (s *Server) GetWindow(_ context.Context, req *connect.Request[adminv1.GetWindowRequest]) (*connect.Response[adminv1.GetWindowResponse], error)
- func (s *Server) Handler() http.Handler
- func (s *Server) Health(_ context.Context, _ *connect.Request[adminv1.HealthRequest]) (*connect.Response[adminv1.HealthResponse], error)
- func (s *Server) ListPipelines(_ context.Context, _ *connect.Request[adminv1.ListPipelinesRequest]) (*connect.Response[adminv1.ListPipelinesResponse], error)
- func (s *Server) Register(info PipelineInfo, query QueryFn)
Constants ¶
This section is empty.
Variables ¶
var ErrInvalidAuth = errors.New("admin auth: invalid bearer token")
ErrInvalidAuth is reported when the supplied bearer token doesn't match any configured token or fails JWT verification.
var ErrMissingAuth = errors.New("admin auth: missing or malformed Authorization header")
ErrMissingAuth is reported when the Authorization header is absent or not of the form "Bearer <token>". The middleware translates this to 401 Unauthorized.
Functions ¶
func UIHandler ¶
UIHandler returns an http.Handler that serves the embedded Vite build (web/dist copied into pkg/admin/dist by the build pipeline). Falls back to index.html for unknown paths so client-side routing works (refresh on /pipelines/foo).
If the embedded FS is empty (build wasn't run), the handler returns 503 with a pointer to the build instructions.
Types ¶
type Authenticator ¶ added in v0.2.0
Authenticator validates a request's credentials. Implementations are free to inspect any header, but the supplied helpers all read the HTTP Authorization header for a bearer token.
The middleware calls Authenticate before the request reaches the Connect handler. Returning a non-nil error short-circuits the request with a 401 Unauthorized; a nil return passes the request through.
type JWTVerifier ¶ added in v0.2.0
JWTVerifier is the contract for OIDC / JWT validation. Wire your preferred library (golang-jwt, go-oidc, lestrrat-go/jwx, …) here. Verify receives the raw bearer token (the part after "Bearer "). A non-nil return is reported to the client as 401 with no body.
type Option ¶
type Option func(*Server)
Option configures a Server. Apply via NewServer(rec, opts...).
func WithAllowedOrigins ¶
WithAllowedOrigins sets the list of origins the admin server will respond to from cross-origin browsers. Pass exact origins ("https://dashboard.example") or the wildcard "*". Default is no CORS headers — same-origin only — which is the right default for the embedded UI in cmd/murmur-ui. Use "*" for local development against the Vite dev server (it proxies /api so cross-origin only matters when the UI is served by some other origin).
func WithAuth ¶ added in v0.2.0
func WithAuth(a Authenticator) Option
WithAuth installs an Authenticator. When set, requests that fail authentication get a 401 with no body before reaching the Connect handler. CORS preflight (OPTIONS) is always allowed through so the browser can complete the CORS negotiation before sending the credentialed request.
Auth is OFF by default; the server is then suitable only for same-origin / network-isolated deployments. Always pair public-internet exposure with WithAuth or place the admin server behind an authenticating reverse proxy.
func WithAuthToken ¶ added in v0.2.0
WithAuthToken builds an Authenticator that accepts any of the supplied bearer tokens. Empty tokens are silently ignored (so callers can wire an env-var-driven list without filtering nils). Token comparison is constant-time, so a leaked token doesn't reveal its prefix via timing side channels.
Use the multi-token form for rotation: deploy with the new token, wait for clients to roll, then drop the old.
func WithJWTVerifier ¶ added in v0.2.0
func WithJWTVerifier(v JWTVerifier) Option
WithJWTVerifier wraps a JWTVerifier as an Authenticator. Useful when admin tokens are issued by an OIDC IdP (Okta, Auth0, internal SSO) rather than hand-distributed.
type PipelineInfo ¶
type PipelineInfo struct {
Name string
MonoidKind string
Windowed bool
WindowGranSec int64
WindowRetSec int64
StoreType string
CacheType string
SourceType string
}
PipelineInfo is the Go-friendly form of murmur.admin.v1.PipelineInfo. We keep a separate struct so package consumers don't have to import generated code, and so JSON tags work for any caller that wants to log it.
type QueryFn ¶
QueryFn is the read-side closure each registered pipeline supplies. The op argument is one of "get" / "window" / "range"; params carries the per-op arguments (entity, bucket, duration_s, start, end, decode). Returns the raw stored bytes plus a `present` flag.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server is the admin control surface. Concurrent-safe.
func NewServer ¶
NewServer constructs an admin Server. The recorder is shared with whatever runtime you've installed it on (typically streaming.Run via WithMetrics).
CORS is closed by default — pass WithAllowedOrigins(…) to open it up to the origins that should be able to talk to this server.
func (*Server) FullHandler ¶
FullHandler combines the admin Connect-RPC service and the embedded UI on a single mux. Requests under "/api/" are stripped of the prefix and forwarded to the AdminService handler (which expects paths of the form "/murmur.admin.v1.AdminService/<RPC>"); everything else is served by the SPA handler.
The "/api" prefix is a UI-side convention so the same dist works under `npm run dev` (Vite proxy: /api → :8080) and in production (embedded). Other non-browser clients can speak directly to the Connect service at "/" by pointing at <host>:<port>/murmur.admin.v1.AdminService/<RPC> if the API is the only thing on the port.
func (*Server) GetPipelineMetrics ¶
func (s *Server) GetPipelineMetrics(_ context.Context, req *connect.Request[adminv1.GetPipelineMetricsRequest]) (*connect.Response[adminv1.GetPipelineMetricsResponse], error)
GetPipelineMetrics implements murmur.admin.v1.AdminService/GetPipelineMetrics. Returns the in-memory metrics snapshot for one pipeline (events / errors / latency percentiles). Returns CodeNotFound for unknown pipeline names so clients can distinguish "no metrics yet" from "wrong name."
func (*Server) GetRange ¶
func (s *Server) GetRange(_ context.Context, req *connect.Request[adminv1.GetRangeRequest]) (*connect.Response[adminv1.GetRangeResponse], error)
GetRange implements murmur.admin.v1.AdminService/GetRange. Merges every bucket whose ID falls in [start_unix, end_unix] inclusive. Same error-code conventions as GetState plus CodeInvalidArgument when end < start.
func (*Server) GetState ¶
func (s *Server) GetState(_ context.Context, req *connect.Request[adminv1.GetStateRequest]) (*connect.Response[adminv1.GetStateResponse], error)
GetState implements murmur.admin.v1.AdminService/GetState. Reads one (entity[, bucket]) tuple from the registered pipeline's store. Honors `decode=true` to populate StateValue.decoded with a monoid-typed view (see decodeForKind). Returns CodeInvalidArgument for missing entity, CodeNotFound for unknown pipeline, CodeInternal for store-side failures.
func (*Server) GetWindow ¶
func (s *Server) GetWindow(_ context.Context, req *connect.Request[adminv1.GetWindowRequest]) (*connect.Response[adminv1.GetWindowResponse], error)
GetWindow implements murmur.admin.v1.AdminService/GetWindow. Merges the N most-recent buckets covering `duration_seconds` ending at the server's now via the pipeline's monoid. Same error-code conventions as GetState plus CodeInvalidArgument for non-positive durations.
func (*Server) Handler ¶
Handler returns an http.Handler implementing the AdminService routes. Mount at "/" or under a subpath via http.StripPrefix.
Middleware order (outer-most first):
- CORS — sets Access-Control-* headers and short-circuits OPTIONS preflight before the auth middleware sees it.
- Auth — when WithAuth / WithAuthToken / WithJWTVerifier is configured, enforces the Authorization header; rejects with 401 on missing or invalid credentials.
- AdminService handler (Connect-RPC).
func (*Server) Health ¶
func (s *Server) Health(_ context.Context, _ *connect.Request[adminv1.HealthRequest]) (*connect.Response[adminv1.HealthResponse], error)
Health implements murmur.admin.v1.AdminService/Health. Always returns {status: "ok"}; the response is generated synchronously without touching pipeline state, so a slow store cannot make the health endpoint flap.
func (*Server) ListPipelines ¶
func (s *Server) ListPipelines(_ context.Context, _ *connect.Request[adminv1.ListPipelinesRequest]) (*connect.Response[adminv1.ListPipelinesResponse], error)
ListPipelines implements murmur.admin.v1.AdminService/ListPipelines. Returns every registered pipeline's metadata, sorted by name for stable client-side rendering. Cheap (RAM-only) so the dashboard polls it every few seconds without coordination.
func (*Server) Register ¶
func (s *Server) Register(info PipelineInfo, query QueryFn)
Register installs a pipeline's metadata and query closure. Idempotent — re-registering the same name overwrites.
Source Files