api

package
v0.1.0-alpha.8 Latest Latest
Warning

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

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

Documentation

Overview

Package api hosts the public control-plane API. One TLS listener serves both gRPC (HTTP/2, application/grpc) and the grpc-gateway REST mux on the same port; content-type routing splits the two. The listener requests — but does not require — client certs, so token-bearing CLIs and mTLS nodes share it.

Services are wired incrementally through Options: a nil service is simply not registered, letting later tasks (T-04 auth, T-05 projects, …) land one at a time.

Index

Constants

View Source 
const MeshsockLabel = "zattera.dev/meshsock"

MeshsockLabel marks a node that runs the meshsock datapath (T-57): such nodes are paired directly by the peer builder regardless of public reachability, because meshsock handles NAT traversal (punch + relay).

View Source 
const TokenPrefix = "zpat_"

TokenPrefix marks a Zattera personal/session token string.

Variables

This section is empty.

Functions

func HashToken 

func HashToken(token string) string

HashToken returns the hex SHA-256 of a full token string. The whole string (prefix included) is hashed, so callers must not strip the prefix first.

func LooksLikeToken 

func LooksLikeToken(s string) bool

LooksLikeToken reports whether s carries the token prefix.

func MintToken 

func MintToken() (token string, secretHash string, err error)

MintToken returns a fresh token string (zpat_<base62>) and the hex SHA-256 of the whole string, which is what gets persisted in Token.SecretHash. The plaintext is shown to the user exactly once and never stored.

func NewGitHubWebhook 

func NewGitHubWebhook(store *state.Store, raft Applier, vault *secrets.Vault, clk clock.Clock, domain string, log *slog.Logger) (http.Handler, *github.Previews)

NewGitHubWebhook builds the POST /v1/github/webhook handler backed by cluster state. The webhook secret and App private key are read (sealed) from the KV store; a sealed vault or missing key degrades gracefully (deploys are skipped). The returned *github.Previews is the preview-environment manager; the daemon runs its SweepExpired in a leader-gated janitor loop (T-75).

func ObjectStoreFor 

func ObjectStoreFor(cfg *zatterav1.BackupConfig, sealer secrets.Sealer) (volumes.ObjectStore, error)

ObjectStoreFor builds the S3 client for a backup destination, unsealing its credentials. Shared by the backup path and the audit/event archiver (T-92), which deliberately reuse one destination and one set of credentials.

func SourceBlobHandler 

func SourceBlobHandler(uploadsDir string) http.Handler

SourceBlobHandler serves spooled source tarballs by digest to builder nodes. It is mounted on the node-mTLS API surface (the client cert is the auth); the builder fetches source_url = "<control>/internal/blobs/<digest>".

func ValidateMethodTable 

func ValidateMethodTable(info map[string]grpc.ServiceInfo) error

ValidateMethodTable asserts that every method registered on the gRPC server (excluding health) has a policy entry. A missing entry is a fail-closed hole waiting to open — surface it at startup, not at request time.

Types

type ActivatorServer 

type ActivatorServer struct {
	clusterv1.UnimplementedActivatorServiceServer
	// contains filtered or unexported fields
}

ActivatorServer wakes a scaled-to-zero environment on demand (T-70). Proxies call Activate when a request arrives for an env with no healthy endpoints; it sets effective_replicas back to max(1, min) so the scheduler places a replica. Idempotent and singleflighted per env.

func NewActivatorServer 

func NewActivatorServer(store *state.Store, raft Applier, clk clock.Clock, log *slog.Logger) *ActivatorServer

NewActivatorServer builds the activator.

func (*ActivatorServer) Activate 

func (s *ActivatorServer) Activate(ctx context.Context, req *clusterv1.ActivateRequest) (*clusterv1.ActivateResponse, error)

Activate brings a scaled-to-zero env back up to its warm replica count. It is idempotent: an already-warm env, or one woken moments ago, returns accepted without a redundant write.

type AgentExecStream 

type AgentExecStream interface {
	Send(*clusterv1.AgentExecInput) error
	Recv() (*clusterv1.AgentExecOutput, error)
	CloseSend() error
}

AgentExecStream is the control→agent Exec bidi stream (implemented by the generated gRPC client; faked in tests).

type AgentProxyStream 

type AgentProxyStream interface {
	Send(*clusterv1.TCPChunk) error
	Recv() (*clusterv1.TCPChunk, error)
	CloseSend() error
}

AgentProxyStream is the control→agent ProxyTCP bidi stream.

type AlertServer 

type AlertServer struct {
	zatterav1.UnimplementedAlertServiceServer
	// contains filtered or unexported fields
}

AlertServer implements AlertService: CRUD for alert rules and notification channels (T-74). The notify engine evaluates the rules on the leader. Channel secrets are sealed here; list responses redact them.

func NewAlertServer 

func NewAlertServer(store *state.Store, raft Applier, vault *secrets.Vault, clk clock.Clock) *AlertServer

NewAlertServer builds the alert service. sealer may be nil on a sealed node (channel writes that carry secrets then fail with FailedPrecondition).

func (*AlertServer) DeleteChannel 

func (s *AlertServer) DeleteChannel(ctx context.Context, req *zatterav1.DeleteChannelRequest) (*emptypb.Empty, error)

DeleteChannel removes a notification channel.

func (*AlertServer) DeleteRule 

func (s *AlertServer) DeleteRule(ctx context.Context, req *zatterav1.DeleteRuleRequest) (*emptypb.Empty, error)

DeleteRule removes an alert rule.

func (*AlertServer) ListChannels 

func (s *AlertServer) ListChannels(_ context.Context, _ *emptypb.Empty) (*zatterav1.ListChannelsResponse, error)

ListChannels returns channels with secrets redacted.

func (*AlertServer) ListRules 

func (s *AlertServer) ListRules(_ context.Context, _ *emptypb.Empty) (*zatterav1.ListRulesResponse, error)

ListRules returns all alert rules.

func (*AlertServer) PutChannel 

func (s *AlertServer) PutChannel(ctx context.Context, req *zatterav1.PutChannelRequest) (*zatterav1.NotificationChannel, error)

PutChannel creates or updates a notification channel, sealing any plaintext secrets supplied in the request.

func (*AlertServer) PutRule 

func (s *AlertServer) PutRule(ctx context.Context, req *zatterav1.PutRuleRequest) (*zatterav1.AlertRule, error)

PutRule creates or updates an alert rule.

type AppServer 

type AppServer struct {
	zatterav1.UnimplementedAppServiceServer
	// contains filtered or unexported fields
}

AppServer implements zatterav1.AppServiceServer.

func NewAppServer 

func NewAppServer(store *state.Store, raft Applier, clk clock.Clock, vault *secrets.Vault) *AppServer

NewAppServer builds the app service. sealer may be nil; env-var mutations then return FailedPrecondition.

func (*AppServer) ApplyAppConfig 

func (s *AppServer) ApplyAppConfig(ctx context.Context, req *zatterav1.ApplyAppConfigRequest) (*zatterav1.GetAppResponse, error)

ApplyAppConfig upserts build config and per-env ServiceSpecs. Environments absent from the request are left untouched; new ones are created.

func (*AppServer) CreateApp 

func (s *AppServer) CreateApp(ctx context.Context, req *zatterav1.CreateAppRequest) (*zatterav1.App, error)

CreateApp creates an app and its default production + staging environments.

func (*AppServer) DeleteApp 

func (s *AppServer) DeleteApp(ctx context.Context, req *zatterav1.DeleteAppRequest) (*emptypb.Empty, error)

DeleteApp deletes an app and its environments (+ env vars).

func (*AppServer) GetApp 

func (s *AppServer) GetApp(_ context.Context, req *zatterav1.GetAppRequest) (*zatterav1.GetAppResponse, error)

GetApp returns an app (by id or name) with its environments.

func (*AppServer) GetEnvVars 

func (s *AppServer) GetEnvVars(_ context.Context, req *zatterav1.GetEnvVarsRequest) (*zatterav1.GetEnvVarsResponse, error)

GetEnvVars returns env var keys; values only when reveal is set (the RBAC tier already restricts this method to DEVELOPER+).

func (*AppServer) ListApps 

func (s *AppServer) ListApps(_ context.Context, req *zatterav1.ListAppsRequest) (*zatterav1.ListAppsResponse, error)

ListApps lists a project's apps.

func (*AppServer) SetEnvVars 

func (s *AppServer) SetEnvVars(ctx context.Context, req *zatterav1.SetEnvVarsRequest) (*emptypb.Empty, error)

SetEnvVars seals each value and applies the set/unset batch. v1 semantics: changing env vars does NOT hot-restart running instances; the change folds into the next release's config hash (via DeployServer.envVarVersion) and takes effect on the next deploy or rollback.

func (*AppServer) SetReplicas 

func (s *AppServer) SetReplicas(ctx context.Context, req *zatterav1.SetReplicasRequest) (*zatterav1.Environment, error)

SetReplicas updates an environment's replica range.

type Applier 

type Applier interface {
	Apply(ctx context.Context, cmd *clusterv1.Command) error
	IsLeader() bool
}

Applier proposes commands through raft and reports leadership. *raftstore.Store implements it; T-08 wraps it with leader-forwarding.

type Auditor 

type Auditor struct {
	zatterav1.UnimplementedAuditServiceServer
	// contains filtered or unexported fields
}

Auditor records mutating API calls. The interceptor enqueues entries non-blocking; a background batcher flushes them through AppendAudit on the leader.

func NewAuditor 

func NewAuditor(store *state.Store, raft Applier, log *slog.Logger, interval time.Duration) *Auditor

NewAuditor builds the audit middleware. interval<=0 uses the 2s default.

func (*Auditor) ListEvents 

func (a *Auditor) ListEvents(ctx context.Context, req *zatterav1.ListEventsRequest) (*zatterav1.ListEventsResponse, error)

ListEvents implements zatterav1.AuditServiceServer over state.QueryEvents.

Unlike QueryAudit this is open to any authenticated user, so it scopes the result itself: org owners/admins see everything, everyone else must name a project they belong to. The tier table cannot express that, and the RBAC interceptor's project rewrite is not usable here because "" legitimately means cluster-wide for an admin (T-76).

func (*Auditor) QueryAudit 

func (a *Auditor) QueryAudit(ctx context.Context, req *zatterav1.QueryAuditRequest) (*zatterav1.QueryAuditResponse, error)

QueryAudit implements zatterav1.AuditServiceServer over state.QueryAudit.

func (*Auditor) Run 

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

Run flushes queued entries every interval (or when a batch fills). Blocks until ctx is done, then drains once more.

func (*Auditor) SetArchive 

func (a *Auditor) SetArchive(fn func() (*archive.Reader, bool))

SetArchive wires the audit/event archive reader. Called by the daemon once the cluster key is unsealed; without it include_archive returns ring data only.

func (*Auditor) UnaryInterceptor 

func (a *Auditor) UnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error)

UnaryInterceptor records the outcome of authorized mutating unary calls. It runs after auth+rbac, so only authorized calls are recorded (a denied caller cannot flood the audit log).

type AuthServer 

type AuthServer struct {
	zatterav1.UnimplementedAuthServiceServer
	// contains filtered or unexported fields
}

AuthServer implements zatterav1.AuthServiceServer.

func NewAuthServer 

func NewAuthServer(store *state.Store, raft Applier, clk clock.Clock, clusterDomain string, vault *secrets.Vault) *AuthServer

NewAuthServer builds the auth service. clusterDomain is cfg.Domain; vault is this node's cluster key holder, which Unseal installs into.

func (*AuthServer) CreateToken 

func (s *AuthServer) CreateToken(ctx context.Context, req *zatterav1.CreateTokenRequest) (*zatterav1.CreateTokenResponse, error)

CreateToken issues a personal access token for the caller.

func (*AuthServer) CreateUser 

func (s *AuthServer) CreateUser(ctx context.Context, req *zatterav1.CreateUserRequest) (*zatterav1.User, error)

CreateUser adds a user to the org (admin only; tier enforced by the policy).

func (*AuthServer) ListTokens 

func (s *AuthServer) ListTokens(ctx context.Context, _ *emptypb.Empty) (*zatterav1.ListTokensResponse, error)

ListTokens lists the caller's tokens (secret hashes redacted).

func (*AuthServer) ListUsers 

func (s *AuthServer) ListUsers(_ context.Context, _ *emptypb.Empty) (*zatterav1.ListUsersResponse, error)

ListUsers lists all org users (admin only).

func (*AuthServer) Login 

func (s *AuthServer) Login(ctx context.Context, req *zatterav1.LoginRequest) (*zatterav1.LoginResponse, error)

Login exchanges email+password for a short-lived session token.

func (*AuthServer) RevokeToken 

func (s *AuthServer) RevokeToken(ctx context.Context, req *zatterav1.RevokeTokenRequest) (*emptypb.Empty, error)

RevokeToken deletes one of the caller's tokens (admins may revoke any).

func (*AuthServer) SetUnsealHook 

func (s *AuthServer) SetUnsealHook(fn func())

SetUnsealHook registers a callback run after a successful Unseal.

func (*AuthServer) Unseal 

func (s *AuthServer) Unseal(_ context.Context, req *zatterav1.UnsealRequest) (*zatterav1.UnsealResponse, error)

Unseal installs the cluster data key on this node from the recovery passphrase (T-111). It is per-node: each sealed node must be unsealed, since the key lives in process memory and is never replicated.

func (*AuthServer) WhoAmI 

func (s *AuthServer) WhoAmI(ctx context.Context, _ *emptypb.Empty) (*zatterav1.WhoAmIResponse, error)

WhoAmI returns the caller's user and project memberships.

type Authenticator 

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

Authenticator resolves identity and enforces the method policy table. It also batches token last_used_at updates.

func NewAuthenticator 

func NewAuthenticator(store *state.Store, raft Applier, clk clock.Clock) *Authenticator

NewAuthenticator builds the auth interceptor.

func (*Authenticator) RunTokenFlusher 

func (a *Authenticator) RunTokenFlusher(ctx context.Context)

RunTokenFlusher periodically flushes accumulated last_used_at through one TouchTokens apply (leader only). Blocks until ctx is done.

func (*Authenticator) StreamInterceptor 

func (a *Authenticator) StreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error

StreamInterceptor authenticates and authorizes streaming calls.

func (*Authenticator) UnaryInterceptor 

func (a *Authenticator) UnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error)

UnaryInterceptor authenticates and authorizes unary calls.

type BackupServer 

type BackupServer struct {
	zatterav1.UnimplementedBackupServiceServer
	// contains filtered or unexported fields
}

BackupServer implements BackupService: configure the S3 destination, run a full backup on demand, and list past backups. Available only on an unsealed control node (it needs the data key to seal/unseal credentials + state).

func NewBackupServer 

func NewBackupServer(store *state.Store, raft Applier, vault *secrets.Vault, ca CAMaterial, clk clock.Clock) *BackupServer

NewBackupServer builds the backup service. sealer/ca may be nil on a sealed node; all methods then return FailedPrecondition.

func (*BackupServer) ListBackups 

func (s *BackupServer) ListBackups(_ context.Context, _ *emptypb.Empty) (*zatterav1.ListBackupsResponse, error)

ListBackups returns past backup records and the current (credential-redacted) destination config.

func (*BackupServer) SetBackupConfig 

func (s *BackupServer) SetBackupConfig(ctx context.Context, req *zatterav1.SetBackupConfigRequest) (*emptypb.Empty, error)

SetBackupConfig stores the S3 destination, sealing the plaintext credentials with the cluster data key.

func (*BackupServer) TriggerBackup 

func (s *BackupServer) TriggerBackup(ctx context.Context, _ *zatterav1.TriggerBackupRequest) (*zatterav1.BackupRecord, error)

TriggerBackup runs a full backup now (state + CA + key material + volume snapshot refs) to the configured destination and records it.

type CAMaterial 

type CAMaterial interface {
	CABundlePEM() []byte
	PrivateKeyPEM() ([]byte, error)
}

CAMaterial exposes the cluster CA cert + private key so a backup can include them (restored nodes keep the same CA, so their certs stay valid). *ca.CA satisfies it.

type DeployServer 

type DeployServer struct {
	zatterav1.UnimplementedDeployServiceServer

	// Platforms resolves an image ref's supported platforms at deploy time
	// (T-88). Optional: nil leaves releases unconstrained. Set by the daemon
	// once the embedded registry is up.
	Platforms PlatformResolver
	// contains filtered or unexported fields
}

DeployServer implements DeployService: it turns image refs (or completed builds) into Releases + red/green Deployments. The orchestrator (T-26) drives the deployment through its phases; this service only creates work.

func NewDeployServer 

func NewDeployServer(store *state.Store, raft Applier, clk clock.Clock, uploadsDir string) *DeployServer

NewDeployServer builds the deploy service. uploadsDir is where UploadSource spools source tarballs (typically <data-dir>/uploads).

func (*DeployServer) Deploy 

func (s *DeployServer) Deploy(ctx context.Context, req *zatterav1.DeployRequest) (*zatterav1.Deployment, error)

Deploy creates a release from an image ref (or completed build) and a PENDING deployment for it.

func (*DeployServer) GetDeployment 

func (s *DeployServer) GetDeployment(_ context.Context, req *zatterav1.GetDeploymentRequest) (*zatterav1.Deployment, error)

GetDeployment returns one deployment by id.

func (*DeployServer) ListDeployments 

func (s *DeployServer) ListDeployments(_ context.Context, req *zatterav1.ListDeploymentsRequest) (*zatterav1.ListDeploymentsResponse, error)

ListDeployments lists an environment's deployments.

func (*DeployServer) ListInstances 

func (s *DeployServer) ListInstances(_ context.Context, req *zatterav1.ListInstancesRequest) (*zatterav1.ListInstancesResponse, error)

ListInstances lists assignments, optionally filtered by env and app.

func (*DeployServer) ListReleases 

func (s *DeployServer) ListReleases(_ context.Context, req *zatterav1.ListReleasesRequest) (*zatterav1.ListReleasesResponse, error)

ListReleases lists an environment's releases (newest version first).

func (*DeployServer) Rollback 

func (s *DeployServer) Rollback(ctx context.Context, req *zatterav1.RollbackRequest) (*zatterav1.Deployment, error)

Rollback creates a deployment that re-promotes a prior release (default: the release before the current active one). No new release is minted.

func (*DeployServer) UploadSource 

func (s *DeployServer) UploadSource(stream grpc.ClientStreamingServer[zatterav1.UploadSourceChunk, zatterav1.UploadSourceResponse]) error

UploadSource streams a tar.gz of the source directory to the control node, spools it under <uploads>/<digest>, and creates the QUEUED Build plus a PENDING Deployment that gates on it (T-35). The build dispatcher picks the build up and the orchestrator advances the deployment once it succeeds.

func (*DeployServer) WatchDeployment 

func (s *DeployServer) WatchDeployment(req *zatterav1.GetDeploymentRequest, stream zatterav1.DeployService_WatchDeploymentServer) error

WatchDeployment streams a deployment, resending on every phase change.

type DomainServer 

type DomainServer struct {
	zatterav1.UnimplementedDomainServiceServer
	// contains filtered or unexported fields
}

DomainServer implements DomainService: custom hostnames mapped to an app's environment, with per-domain middleware. Implicit cluster subdomains (<app>-<env>.<cluster-domain>) are emitted by the route builder (T-39); this service rejects manual hostnames that collide with that namespace.

func NewDomainServer 

func NewDomainServer(store *state.Store, raft Applier, clk clock.Clock, clusterDomain string) *DomainServer

NewDomainServer builds the domain service. clusterDomain is cfg.Domain (the reserved auto-subdomain suffix; empty disables the collision check).

func (*DomainServer) AddDomain 

func (s *DomainServer) AddDomain(ctx context.Context, req *zatterav1.AddDomainRequest) (*zatterav1.Domain, error)

AddDomain validates and registers a hostname for an environment.

func (*DomainServer) ListDomains 

func (s *DomainServer) ListDomains(_ context.Context, req *zatterav1.ListDomainsRequest) (*zatterav1.ListDomainsResponse, error)

ListDomains lists a project's domains.

func (*DomainServer) RemoveDomain 

func (s *DomainServer) RemoveDomain(ctx context.Context, req *zatterav1.RemoveDomainRequest) (*emptypb.Empty, error)

RemoveDomain deletes a domain.

func (*DomainServer) SetCertStatus 

func (s *DomainServer) SetCertStatus(ctx context.Context, hostname string, cs zatterav1.CertStatus)

SetCertStatus updates a hostname's certificate status (best-effort callback from the TLS manager: PENDING → ISSUED/FAILED). Silently no-ops for unknown or implicit hostnames.

One certificate covers a hostname regardless of how many path routes share it, so the status is written to EVERY domain of that hostname — otherwise `domains ls` would show one route issued and its siblings stuck pending forever (T-104).

func (*DomainServer) SetMiddleware 

func (s *DomainServer) SetMiddleware(ctx context.Context, req *zatterav1.SetMiddlewareRequest) (*zatterav1.Domain, error)

SetMiddleware replaces a domain's middleware.

type ExecDialer 

type ExecDialer interface {
	Exec(ctx context.Context, node *zatterav1.Node) (AgentExecStream, error)
	ProxyTCP(ctx context.Context, node *zatterav1.Node) (AgentProxyStream, error)
	Top(ctx context.Context, node *zatterav1.Node, containerID string) (*clusterv1.AgentTopResponse, error)
}

ExecDialer opens AgentLocalService streams to a node over the mesh. The production implementation dials node mTLS; tests inject a fake.

type ExecServer 

type ExecServer struct {
	zatterav1.UnimplementedExecServiceServer
	// contains filtered or unexported fields
}

ExecServer implements ExecService: it resolves an instance to its node and relays the user's interactive stream (exec / port-forward / top) to that node's AgentLocalService over the mesh (T-49). The relay is a pure byte pump with a goroutine per direction; stream close propagates both ways.

func NewExecServer 

func NewExecServer(store *state.Store, dial ExecDialer, log *slog.Logger) *ExecServer

NewExecServer builds the exec service.

func (*ExecServer) Exec 

func (s *ExecServer) Exec(stream grpc.BidiStreamingServer[zatterav1.ExecInput, zatterav1.ExecOutput]) error

Exec relays a bidirectional exec/attach session to the instance's node.

func (*ExecServer) PortForward 

func (s *ExecServer) PortForward(stream grpc.BidiStreamingServer[zatterav1.PortForwardInput, zatterav1.PortForwardOutput]) error

PortForward tunnels one TCP connection to a healthy replica of the app.

func (*ExecServer) Top 

func (s *ExecServer) Top(ctx context.Context, req *zatterav1.TopRequest) (*zatterav1.TopResponse, error)

Top returns a process listing from inside the instance's container.

type GRPCExecDialer 

type GRPCExecDialer struct {
	Connect func(ctx context.Context, node *zatterav1.Node) (*grpc.ClientConn, error)
}

GRPCExecDialer is the production ExecDialer: it dials a node's AgentLocalService over the mesh with node mTLS. Connect supplies the per-node client connection; the daemon owns the transport details.

func (GRPCExecDialer) Exec 

func (g GRPCExecDialer) Exec(ctx context.Context, node *zatterav1.Node) (AgentExecStream, error)

Exec opens a bidi Exec stream, closing the connection when it ends.

func (GRPCExecDialer) ProxyTCP 

func (g GRPCExecDialer) ProxyTCP(ctx context.Context, node *zatterav1.Node) (AgentProxyStream, error)

ProxyTCP opens a bidi ProxyTCP stream, closing the connection when it ends.

func (GRPCExecDialer) Top 

func (g GRPCExecDialer) Top(ctx context.Context, node *zatterav1.Node, containerID string) (*clusterv1.AgentTopResponse, error)

Top is a unary call to the node's AgentLocalService.

type GRPCLogDialer 

type GRPCLogDialer struct {
	Connect func(ctx context.Context, node *zatterav1.Node) (*grpc.ClientConn, error)
}

GRPCLogDialer is the production LogDialer: it dials a node's AgentLocalService over the mesh with node mTLS. Connect supplies the per-node client connection; the daemon owns the transport details.

func (GRPCLogDialer) QueryLogs 

func (g GRPCLogDialer) QueryLogs(ctx context.Context, node *zatterav1.Node, q *zatterav1.LogQuery) (LogStream, error)

QueryLogs opens the stream, keeping the connection alive for its lifetime.

type GRPCStatsDialer 

type GRPCStatsDialer struct {
	Connect func(ctx context.Context, node *zatterav1.Node) (*grpc.ClientConn, error)
}

GRPCStatsDialer is the production StatsDialer: it dials a node's AgentLocalService over the mesh with node mTLS via Connect.

func (GRPCStatsDialer) Stats 

func (g GRPCStatsDialer) Stats(ctx context.Context, node *zatterav1.Node, q *zatterav1.StatsQuery) (*zatterav1.StatsResponse, error)

Stats runs a unary Stats RPC against the node, closing the connection after.

type GRPCUpgradeDialer 

type GRPCUpgradeDialer struct {
	Connect func(ctx context.Context, node *zatterav1.Node) (*grpc.ClientConn, error)
}

GRPCUpgradeDialer is the production UpgradeDialer.

func (GRPCUpgradeDialer) UpgradeBinary 

func (g GRPCUpgradeDialer) UpgradeBinary(ctx context.Context, node *zatterav1.Node, req *clusterv1.AgentUpgradeRequest) (*clusterv1.AgentUpgradeResponse, error)

type GRPCVolumeAgentDialer 

type GRPCVolumeAgentDialer struct {
	Connect func(ctx context.Context, node *zatterav1.Node) (*grpc.ClientConn, error)
}

GRPCVolumeAgentDialer dials a node's AgentLocalService to remove a docker volume. Connect supplies the per-node mTLS connection.

func (GRPCVolumeAgentDialer) RemoveVolume 

func (g GRPCVolumeAgentDialer) RemoveVolume(ctx context.Context, node *zatterav1.Node, envID, volumeName string) error

RemoveVolume runs the RemoveVolume RPC against the node, closing the connection after.

type GRPCVolumeFileDialer 

type GRPCVolumeFileDialer struct {
	Connect func(ctx context.Context, node *zatterav1.Node) (*grpc.ClientConn, error)
}

GRPCVolumeFileDialer is the production VolumeFileDialer.

func (GRPCVolumeFileDialer) ListFiles 

func (g GRPCVolumeFileDialer) ListFiles(ctx context.Context, node *zatterav1.Node, envID, volName, path string) (*clusterv1.AgentListFilesResponse, error)

func (GRPCVolumeFileDialer) ReadFile 

func (g GRPCVolumeFileDialer) ReadFile(ctx context.Context, node *zatterav1.Node, envID, volName, path string, emit func([]byte) error) error

type GossipSource 

type GossipSource interface {
	Snapshot() map[string]nodehealth.GossipLiveness
}

GossipSource is the gossip failure detector's view consumed by the liveness monitor. *mesh.Gossip implements it (via nodehealth). Nil disables the gossip signal (the monitor falls back to heartbeats only — single-node/dev).

type Identity 

type Identity struct {
	UserID  string
	NodeID  string
	OrgRole zatterav1.Role
}

Identity is the authenticated caller, resolved by the auth interceptor and stored in the request context. Exactly one of UserID / NodeID is set.

func IdentityFrom 

func IdentityFrom(ctx context.Context) (Identity, bool)

IdentityFrom returns the caller identity placed by the auth interceptor.

func (Identity) Actor 

func (i Identity) Actor() string

Actor returns a stable actor string for audit/command attribution.

func (Identity) IsNode 

func (i Identity) IsNode() bool

IsNode reports whether the caller is a cluster node (mTLS).

type JobServer 

type JobServer struct {
	zatterav1.UnimplementedJobServiceServer
	// contains filtered or unexported fields
}

JobServer implements JobService: it enqueues one-shot jobs (T-53). The scheduler (reconcileJobs) owns placement, retries and terminal transitions; this service only creates work and reports it.

func NewJobServer 

func NewJobServer(store *state.Store, raft Applier, clk clock.Clock) *JobServer

NewJobServer builds the job service.

func (*JobServer) CancelJob 

func (s *JobServer) CancelJob(ctx context.Context, req *zatterav1.CancelJobRequest) (*zatterav1.Job, error)

CancelJob marks a non-terminal job CANCELED and stops its assignment; the scheduler reaps the assignment. Terminal jobs are returned unchanged.

func (*JobServer) GetJob 

func (s *JobServer) GetJob(_ context.Context, req *zatterav1.GetJobRequest) (*zatterav1.Job, error)

GetJob returns one job by id.

func (*JobServer) ListJobs 

func (s *JobServer) ListJobs(_ context.Context, req *zatterav1.ListJobsRequest) (*zatterav1.ListJobsResponse, error)

ListJobs lists jobs for a project, optionally filtered by env/cron.

func (*JobServer) RunJob 

func (s *JobServer) RunJob(ctx context.Context, req *zatterav1.RunJobRequest) (*zatterav1.Job, error)

RunJob enqueues a one-shot job against an environment's active release.

type JoinConfig 

type JoinConfig struct {
	MeshEnabled bool
	// ControlGRPCAddr is where the joined node opens its AgentSync stream (the
	// control node's mesh IP:port, or 127.0.0.1:port with mesh disabled).
	ControlGRPCAddr string
	// RegistryAddr is the embedded registry endpoint for image pulls.
	RegistryAddr string
	// RaftPort is the port control nodes bind for the raft transport (e.g.
	// "7480"). A joining control node's raft address is meshIP:RaftPort. Empty
	// disables control-node raft handover.
	RaftPort string
}

JoinConfig carries the control-plane facts a joining node needs to come up.

type JoinServer 

type JoinServer struct {
	clusterv1.UnimplementedJoinServiceServer
	// contains filtered or unexported fields
}

JoinServer implements JoinService: token-authenticated node enrollment. It is reachable over TLS without mTLS — the join token IS the credential, verified in the handler.

func NewJoinServer 

func NewJoinServer(store *state.Store, raft Applier, clk clock.Clock, authority *ca.CA, vault *secrets.Vault, cfg JoinConfig, log *slog.Logger) *JoinServer

NewJoinServer builds the join service. vault holds the cluster data key (it is sealed on a node that never bootstrapped and has not been unsealed); the key is handed to joining control nodes so they come up already unsealed (T-55). Reading it through the vault rather than a captured keyring means a node unsealed after startup can serve joins without a restart (T-111).

func (*JoinServer) Join 

func (s *JoinServer) Join(ctx context.Context, req *clusterv1.JoinRequest) (*clusterv1.JoinResponse, error)

Join enrolls a node: it verifies + consumes the token, allocates a mesh IP, signs the node's CSR, registers the node, and issues a registry credential.

type KeyServer 

type KeyServer struct {
	clusterv1.UnimplementedKeyServiceServer
	// contains filtered or unexported fields
}

KeyServer implements KeyService: it hands the cluster data key to a control node that restarted sealed (T-112), so a reboot does not leave a node unable to touch secrets until an operator intervenes.

Trust model: the transport already proves the caller holds a cluster-signed node certificate (reqNode in the policy table). That alone is not enough — every worker has one too — so this additionally requires the calling node to carry the control role in replicated state. Handing the key only to nodes that were already entitled to it at join time keeps the blast radius of a stolen worker certificate unchanged.

func NewKeyServer 

func NewKeyServer(store *state.Store, vault *secrets.Vault, log *slog.Logger) *KeyServer

NewKeyServer builds the key handover service.

func (*KeyServer) FetchDataKey 

func (s *KeyServer) FetchDataKey(ctx context.Context, _ *clusterv1.FetchDataKeyRequest) (*clusterv1.FetchDataKeyResponse, error)

FetchDataKey returns the cluster data key to an enrolled control node.

type LeaderForwarder 

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

LeaderForwarder transparently proxies mutating unary calls that land on a follower to the current leader's API, preserving the caller's deadline and authorization metadata (the leader re-runs auth/rbac/audit on the forwarded call). It is the OUTERMOST interceptor.

Streams are not forwarded — callers redial the leader (documented in pkg/apiclient).

func NewLeaderForwarder 

func NewLeaderForwarder(isLeader func() bool, resolve func() (string, error), dialOpts []grpc.DialOption, log *slog.Logger) *LeaderForwarder

NewLeaderForwarder builds the forwarder. dialOpts must establish TLS trusting the cluster CA (no client cert: the forwarded bearer token authenticates the caller on the leader).

func (*LeaderForwarder) Invalidate 

func (f *LeaderForwarder) Invalidate()

Invalidate drops the cached leader connection. Wire it to raft's LeaderCh so a leadership change forces a fresh dial.

func (*LeaderForwarder) UnaryInterceptor 

func (f *LeaderForwarder) UnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error)

UnaryInterceptor forwards mutating calls to the leader when this node is a follower.

type LivenessMonitor 

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

LivenessMonitor is a leader-only loop that turns livestate heartbeats into durable node status: nodes with a stale (or missing) heartbeat go DOWN, a fresh heartbeat brings a DOWN node back ALIVE. Transitions apply only on change; last_heartbeat_at is batched. A newly elected leader gets a grace window before demoting nodes it simply hasn't heard from yet.

func NewLivenessMonitor 

func NewLivenessMonitor(store *state.Store, raft Applier, live *livestate.Registry, clk clock.Clock, localNodeID string, log *slog.Logger) *LivenessMonitor

NewLivenessMonitor builds the monitor.

func (*LivenessMonitor) Run 

func (m *LivenessMonitor) Run(ctx context.Context)

Run evaluates liveness every 5s until ctx is canceled. The grace window is anchored at Run start (leadership acquisition).

func (*LivenessMonitor) WithGossip 

func (m *LivenessMonitor) WithGossip(g GossipSource) *LivenessMonitor

WithGossip attaches a gossip failure detector so a node's death is caught within seconds instead of the full heartbeat deadline. Returns the monitor for chaining. Safe to pass nil (no-op).

type LogDialer 

type LogDialer interface {
	QueryLogs(ctx context.Context, node *zatterav1.Node, q *zatterav1.LogQuery) (LogStream, error)
}

LogDialer opens a QueryLogs stream to a node's AgentLocalService (over the mesh in production; a fake in tests).

type LogServer 

type LogServer struct {
	zatterav1.UnimplementedLogServiceServer
	// contains filtered or unexported fields
}

LogServer implements LogService.Query: it resolves the selector to the nodes running matching instances/builds, fans QueryLogs out to each, and merges the results by timestamp (enriching with app/env names from state).

func NewLogServer 

func NewLogServer(store *state.Store, dial LogDialer, clk clock.Clock, log *slog.Logger) *LogServer

NewLogServer builds the log service.

func (*LogServer) Query 

func (s *LogServer) Query(q *zatterav1.LogQuery, stream grpc.ServerStreamingServer[zatterav1.LogLine]) error

Query streams merged log lines for a selector.

type LogStream 

type LogStream interface {
	Recv() (*zatterav1.LogLine, error)
}

LogStream is the receive side of an AgentLocalService.QueryLogs stream.

type MeshServer 

type MeshServer struct {
	clusterv1.UnimplementedMeshServiceServer
	// contains filtered or unexported fields
}

MeshServer implements MeshService: WatchPeers streams each node its full WireGuard peer set (hub-and-spoke Phase A + worker↔worker Phase B); ReportObservedEndpoint records disco-observed reflexive endpoints (T-20).

func NewMeshServer 

func NewMeshServer(store *state.Store, raft Applier, clk clock.Clock, log *slog.Logger) *MeshServer

NewMeshServer builds the mesh peer-distribution service. raft may be nil when only the peer builder is exercised (no endpoint folding).

func (*MeshServer) PunchStream 

func (s *MeshServer) PunchStream(req *clusterv1.PunchStreamRequest, stream clusterv1.MeshService_PunchStreamServer) error

PunchStream keeps a node registered as punch-capable and forwards PunchCommands to it until the stream closes. (T-57)

func (*MeshServer) ReportObservedEndpoint 

func (s *MeshServer) ReportObservedEndpoint(ctx context.Context, req *clusterv1.ReportObservedEndpointRequest) (*clusterv1.ReportObservedEndpointResponse, error)

ReportObservedEndpoint records a node's own disco-observed reflexive endpoint. Only self-reports are accepted (the mTLS identity must match the claimed node).

func (*MeshServer) RequestPunch 

func (s *MeshServer) RequestPunch(ctx context.Context, req *clusterv1.RequestPunchRequest) (*clusterv1.RequestPunchResponse, error)

RequestPunch coordinates a simultaneous open between the caller and a target: it pushes a PunchCommand to the target (over its PunchStream) and returns the target's endpoints + a shared punch time to the caller. coordinated=false when the target is not punch-capable (no live stream). (T-57)

func (*MeshServer) RunEndpointFolder 

func (s *MeshServer) RunEndpointFolder(ctx context.Context)

RunEndpointFolder periodically folds fresh disco-observed endpoints into the nodes' durable public_endpoints (leader only), so the peer builder can hand out direct worker↔worker paths. Blocks until ctx is done.

func (*MeshServer) WatchPeers 

func (s *MeshServer) WatchPeers(req *clusterv1.WatchPeersRequest, stream clusterv1.MeshService_WatchPeersServer) error

WatchPeers streams the calling node its full PeerSet, resending (debounced) on every node change.

type MetricsServer 

type MetricsServer struct {
	zatterav1.UnimplementedMetricsServiceServer
	// contains filtered or unexported fields
}

MetricsServer implements MetricsService.Stats. v1 (T-51) serves ONLY current values from livestate — one Point per Series, stamped now — so the CLI/API shape matches the historical TSDB that lands in T-59/T-60.

func NewMetricsServer 

func NewMetricsServer(store *state.Store, live *livestate.Registry, dial StatsDialer, clk clock.Clock, log *slog.Logger) *MetricsServer

NewMetricsServer builds the metrics service. dial enables the historical (TSDB) path — a query with a `since` bound fans out to the nodes' local ring TSDBs (T-60); a nil dial keeps only the live path. clk/log default when nil.

func (*MetricsServer) Stats 

func (s *MetricsServer) Stats(ctx context.Context, q *zatterav1.StatsQuery) (*zatterav1.StatsResponse, error)

Stats returns single-point series scoped to a node, environment, app, or the whole cluster (empty scope → all nodes).

type NodeServer 

type NodeServer struct {
	zatterav1.UnimplementedNodeServiceServer
	// contains filtered or unexported fields
}

NodeServer implements zatterav1.NodeServiceServer.

func NewNodeServer 

func NewNodeServer(store *state.Store, raft Applier, clk clock.Clock, authority *ca.CA) *NodeServer

NewNodeServer builds the node service.

func (*NodeServer) CordonNode 

func (s *NodeServer) CordonNode(ctx context.Context, req *zatterav1.CordonNodeRequest) (*zatterav1.Node, error)

CordonNode makes a node unschedulable without draining it: running containers stay up, only new placements stop. This is what a rolling binary upgrade needs — DrainNode hard-stops stateful workloads, because node-pinned volumes cannot move, which would turn a zero-downtime binary swap into a database outage on every stateful node (T-94).

func (*NodeServer) CreateJoinToken 

func (s *NodeServer) CreateJoinToken(ctx context.Context, req *zatterav1.CreateJoinTokenRequest) (*zatterav1.CreateJoinTokenResponse, error)

CreateJoinToken mints a single-use (by default) join token pinned to the cluster CA. The plaintext is returned once; only its hash is stored.

func (*NodeServer) DrainNode 

func (s *NodeServer) DrainNode(ctx context.Context, req *zatterav1.DrainNodeRequest) (*zatterav1.Node, error)

DrainNode marks a node DRAINING and unschedulable; the scheduler then migrates its instances away and marks it DRAINED (T-29).

func (*NodeServer) GetNode 

func (s *NodeServer) GetNode(_ context.Context, req *zatterav1.GetNodeRequest) (*zatterav1.Node, error)

GetNode returns one node by id.

func (*NodeServer) ListNodes 

func (s *NodeServer) ListNodes(_ context.Context, _ *emptypb.Empty) (*zatterav1.ListNodesResponse, error)

ListNodes returns all registered nodes.

func (*NodeServer) RemoveNode 

func (s *NodeServer) RemoveNode(ctx context.Context, req *zatterav1.RemoveNodeRequest) (*emptypb.Empty, error)

RemoveNode deletes a drained node (or any node with force), reaping its assignments and, for a control node, removing it from the raft cluster. It refuses to remove the last control node without force.

func (*NodeServer) SetNodeLabels 

func (s *NodeServer) SetNodeLabels(ctx context.Context, req *zatterav1.SetNodeLabelsRequest) (*zatterav1.Node, error)

SetNodeLabels updates a node's labels and schedulable flag.

func (*NodeServer) SetUpgrader 

func (s *NodeServer) SetUpgrader(r upgrade.Resolver, d UpgradeDialer)

SetUpgrader wires cluster upgrade (T-95). Without it UpgradePlan and UpgradeNode report Unimplemented.

func (*NodeServer) UncordonNode 

func (s *NodeServer) UncordonNode(ctx context.Context, req *zatterav1.UncordonNodeRequest) (*zatterav1.Node, error)

UncordonNode returns a cordoned or drained node to service. Without it a drained node only comes back by restarting its daemon (which re-registers it as ALIVE) — liveness deliberately never reverts DRAINING/DRAINED.

func (*NodeServer) UpgradeNode 

func (s *NodeServer) UpgradeNode(ctx context.Context, req *zatterav1.UpgradeNodeRequest) (*zatterav1.UpgradeNodeResponse, error)

UpgradeNode swaps one node's binary. The CLI drives the rollout; this is one step of it, so a failure stops the run with the rest of the cluster untouched.

func (*NodeServer) UpgradePlan 

func (s *NodeServer) UpgradePlan(ctx context.Context, req *zatterav1.UpgradePlanRequest) (*zatterav1.UpgradePlanResponse, error)

UpgradePlan resolves the target release and returns the node-by-node order.

Ordering is a correctness constraint, not a preference: the FSM is additive-only and an unknown mutation is surfaced as an error *without* halting the node, so a new leader proposing a mutation an old follower cannot apply diverges that follower's state silently. An old leader only ever proposes mutations newer followers understand, so the leader goes last.

type Options 

type Options struct {
	CA     *ca.CA
	Listen string // e.g. ":8443"
	Logger *slog.Logger

	// Server certificate SANs. Must include 127.0.0.1 (the gateway dials the
	// public port over loopback) and localhost.
	DNSNames []string
	IPs      []net.IP

	// Public services (nil = not registered).
	AuthService    zatterav1.AuthServiceServer
	ProjectService zatterav1.ProjectServiceServer
	AppService     zatterav1.AppServiceServer
	DeployService  zatterav1.DeployServiceServer
	NodeService    zatterav1.NodeServiceServer
	StateService   zatterav1.StateServiceServer
	AuditService   zatterav1.AuditServiceServer
	LogService     zatterav1.LogServiceServer
	DomainService  zatterav1.DomainServiceServer
	ExecService    zatterav1.ExecServiceServer
	MetricsService zatterav1.MetricsServiceServer
	JobService     zatterav1.JobServiceServer
	VolumeService  zatterav1.VolumeServiceServer
	BackupService  zatterav1.BackupServiceServer
	AlertService   zatterav1.AlertServiceServer

	// Node↔control services (mTLS node identity, no REST gateway).
	AgentSyncService clusterv1.AgentSyncServiceServer
	// JoinService is token-authenticated (no mTLS), no REST gateway.
	JoinService clusterv1.JoinServiceServer
	// KeyService hands the cluster data key to an enrolled control node that
	// restarted sealed (T-112). mTLS node-authenticated, no REST gateway.
	KeyService clusterv1.KeyServiceServer
	// MeshService distributes WireGuard peer sets (mTLS node identity).
	MeshService clusterv1.MeshServiceServer
	// RouteService streams route snapshots to node proxies (mTLS node identity).
	RouteService clusterv1.RouteServiceServer
	// ActivatorService wakes scaled-to-zero envs on a proxy's request (mTLS node
	// identity).
	ActivatorService clusterv1.ActivatorServiceServer

	// GitHubWebhook, if set, is mounted as a raw HTTP handler at
	// /v1/github/webhook (signature-authenticated, not part of the gRPC policy).
	GitHubWebhook http.Handler

	// PublicHostname + PublicCertificate serve a publicly-trusted (ACME) cert
	// for the API's public hostname so CLIs need no cluster CA (T-90). All other
	// SNIs (loopback, mesh IP, node identities) keep the CA server cert. When
	// PublicCertificate is nil the CA cert is always used.
	PublicHostname    string
	PublicCertificate func(*tls.ClientHelloInfo) (*tls.Certificate, error)

	// SourceBlobHandler, if set, serves build source tarballs by digest under
	// /internal/blobs/ to builder nodes (mTLS-gated, T-54).
	SourceBlobHandler http.Handler

	// Interceptors run in the given order (auth → rbac → audit → leader-forward
	// per later tasks). Health checks bypass them via a method skip inside each.
	UnaryInterceptors  []grpc.UnaryServerInterceptor
	StreamInterceptors []grpc.StreamServerInterceptor
}

Options configures the API server. A nil service field is not registered.

type PlatformResolver 

type PlatformResolver func(ctx context.Context, imageRef string) []string

PlatformResolver reports the OCI platforms an image reference can run on (empty = unknown/unconstrained). Implementations must be best-effort: they log their own failures and return nil rather than an error a deploy would trip over.

type ProjectServer 

type ProjectServer struct {
	zatterav1.UnimplementedProjectServiceServer
	// contains filtered or unexported fields
}

ProjectServer implements zatterav1.ProjectServiceServer.

func NewProjectServer 

func NewProjectServer(store *state.Store, raft Applier, clk clock.Clock, rbac *RBAC) *ProjectServer

NewProjectServer builds the project service.

func (*ProjectServer) AddMember 

func (s *ProjectServer) AddMember(ctx context.Context, req *zatterav1.AddMemberRequest) (*zatterav1.ProjectMember, error)

AddMember resolves user_email → user and grants a project role.

func (*ProjectServer) CreateProject 

func (s *ProjectServer) CreateProject(ctx context.Context, req *zatterav1.CreateProjectRequest) (*zatterav1.Project, error)

CreateProject creates a project; the creator becomes its OWNER member.

func (*ProjectServer) DeleteProject 

func (s *ProjectServer) DeleteProject(ctx context.Context, req *zatterav1.DeleteProjectRequest) (*emptypb.Empty, error)

DeleteProject cascades: apps, environments (+ env vars), domains, volumes, then the project itself.

func (*ProjectServer) GetProject 

func (s *ProjectServer) GetProject(_ context.Context, req *zatterav1.GetProjectRequest) (*zatterav1.Project, error)

GetProject returns one project (project_id resolved by RBAC).

func (*ProjectServer) ListMembers 

func (s *ProjectServer) ListMembers(_ context.Context, req *zatterav1.ListMembersRequest) (*zatterav1.ListMembersResponse, error)

ListMembers lists a project's members.

func (*ProjectServer) ListProjects 

func (s *ProjectServer) ListProjects(ctx context.Context, _ *emptypb.Empty) (*zatterav1.ListProjectsResponse, error)

ListProjects lists the caller's projects (all, for org admins).

func (*ProjectServer) RemoveMember 

func (s *ProjectServer) RemoveMember(ctx context.Context, req *zatterav1.RemoveMemberRequest) (*emptypb.Empty, error)

RemoveMember revokes a member's project access.

type RBAC 

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

RBAC enforces per-project roles. It resolves the request's project_id field (accepting a project NAME and rewriting it to the canonical id) and compares the caller's effective role to minRole.

func NewRBAC 

func NewRBAC(store *state.Store) *RBAC

NewRBAC builds the RBAC interceptor.

func (*RBAC) UnaryInterceptor 

func (r *RBAC) UnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error)

UnaryInterceptor authorizes project-scoped unary calls.

type Requirement 

type Requirement int

Requirement is the auth tier a method demands.

type RouteProvider 

type RouteProvider interface {
	Current() *clusterv1.RouteSnapshot
	Subscribe() (int, <-chan *clusterv1.RouteSnapshot)
	Unsubscribe(id int)
}

RouteProvider is the route-table source the RouteService streams from. The scheduler's RouteBuilder implements it.

type RouteServer 

type RouteServer struct {
	clusterv1.UnimplementedRouteServiceServer
	// contains filtered or unexported fields
}

RouteServer implements RouteService.WatchRoutes: it streams the current route snapshot on connect (skipping the resend when the client already has that version) and every rebuild thereafter. Node-mTLS only (policy table).

func NewRouteServer 

func NewRouteServer(routes RouteProvider) *RouteServer

NewRouteServer wraps a route provider.

func (*RouteServer) WatchRoutes 

func (s *RouteServer) WatchRoutes(req *clusterv1.WatchRoutesRequest, stream grpc.ServerStreamingServer[clusterv1.RouteSnapshot]) error

WatchRoutes streams route snapshots to a node's proxy/DNS.

type Server 

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

Server owns the TLS listener, the gRPC server and the REST gateway.

func New 

func New(opts Options) (*Server, error)

New binds the listener and wires the gRPC server, health service and REST gateway. Call Serve to start accepting.

func (*Server) Addr 

func (s *Server) Addr() net.Addr

Addr returns the bound listen address.

func (*Server) Endpoint 

func (s *Server) Endpoint() string

Endpoint returns the loopback dial target (127.0.0.1:port).

func (*Server) Health 

func (s *Server) Health() *health.Server

Health exposes the grpc health server (liveness wiring can flip statuses).

func (*Server) Serve 

func (s *Server) Serve(ctx context.Context) error

Serve blocks serving TLS until ctx is canceled, then shuts down gracefully.

type SnapshotDispatcher 

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

SnapshotDispatcher runs volume snapshot/restore/prune by dialing the volume's node over AgentLocalService with an S3 target built from the (unsealed) backup config + cluster data key. It also satisfies scheduler.SnapshotDispatcher.

func NewSnapshotDispatcher 

func NewSnapshotDispatcher(store *state.Store, raft Applier, vault *secrets.Vault, connect func(context.Context, *zatterav1.Node) (*grpc.ClientConn, error), clk clock.Clock, log *slog.Logger) *SnapshotDispatcher

NewSnapshotDispatcher builds the dispatcher. The vault supplies both the sealer and the data key, so a node unsealed after startup can snapshot without a restart. Previously the key was captured as bytes at construction time, which made that impossible (T-111). Original note: it comes from the unsealed keyring; connect dials a node's AgentLocalService.

func (*SnapshotDispatcher) Prune 

func (d *SnapshotDispatcher) Prune(ctx context.Context, _ *zatterav1.Volume, deadSnapshotIDs []string) error

Prune (scheduler-facing) deletes the dead snapshots' manifest objects and garbage-collects orphaned chunks against S3.

func (*SnapshotDispatcher) Restore 

func (d *SnapshotDispatcher) Restore(ctx context.Context, vol *zatterav1.Volume, manifestKey string) error

Restore streams a restore of the snapshot into the volume's node.

func (*SnapshotDispatcher) RunSnapshot 

func (d *SnapshotDispatcher) RunSnapshot(ctx context.Context, vol *zatterav1.Volume) (*zatterav1.VolumeSnapshot, error)

RunSnapshot performs one snapshot synchronously: it records a RUNNING snapshot, streams the node-side op, and finalizes the record COMPLETE/FAILED.

func (*SnapshotDispatcher) Snapshot 

func (d *SnapshotDispatcher) Snapshot(_ context.Context, vol *zatterav1.Volume) error

Snapshot (scheduler-facing) fires a snapshot in the background so the leader loop never blocks on the upload.

type StateServer 

type StateServer struct {
	zatterav1.UnimplementedStateServiceServer
	// contains filtered or unexported fields
}

StateServer implements zatterav1.StateServiceServer: a human-readable YAML export of DESIRED state and an idempotent, name-keyed apply.

func NewStateServer 

func NewStateServer(store *state.Store, raft Applier, clk clock.Clock) *StateServer

NewStateServer builds the state export/apply service.

func (*StateServer) Apply 

func (s *StateServer) Apply(stream grpc.ClientStreamingServer[zatterav1.StateChunk, zatterav1.ApplyResponse]) error

Apply reassembles the streamed YAML, diffs by name, and upserts. dry-run (metadata) validates and counts without writing.

func (*StateServer) Export 

func (s *StateServer) Export(req *zatterav1.ExportRequest, stream grpc.ServerStreamingServer[zatterav1.StateChunk]) error

Export streams the YAML document for a project (or the whole cluster).

type StatsDialer 

type StatsDialer interface {
	Stats(ctx context.Context, node *zatterav1.Node, q *zatterav1.StatsQuery) (*zatterav1.StatsResponse, error)
}

StatsDialer queries one node's AgentLocalService.Stats (over the mesh in production; a fake in tests). nil disables the historical path.

type SyncServer 

type SyncServer struct {
	clusterv1.UnimplementedAgentSyncServiceServer
	// contains filtered or unexported fields
}

SyncServer implements AgentSyncService. Agents dial in and open a bidi stream; the server registers the node in livestate, pushes its full AssignmentSet on connect and whenever assignments change, records heartbeats, and folds reported status batches back into durable state.

func NewSyncServer 

func NewSyncServer(store *state.Store, applier Applier, live *livestate.Registry, clk clock.Clock, log *slog.Logger, vault *secrets.Vault) *SyncServer

NewSyncServer builds the control-side AgentSync handler. applier commits status batches (SetAssignmentsObserved) through raft; sealer decrypts env vars pushed to agents (may be nil).

func (*SyncServer) Sync 

func (s *SyncServer) Sync(stream clusterv1.AgentSyncService_SyncServer) error

Sync runs one agent connection: hello handshake, then concurrent assignment pushes, status flushing, and heartbeat/status receipt until the stream ends.

type UpgradeDialer 

type UpgradeDialer interface {
	UpgradeBinary(ctx context.Context, node *zatterav1.Node, req *clusterv1.AgentUpgradeRequest) (*clusterv1.AgentUpgradeResponse, error)
}

UpgradeDialer reaches a node's AgentLocalService to swap its binary.

type VolumeAgentDialer 

type VolumeAgentDialer interface {
	RemoveVolume(ctx context.Context, node *zatterav1.Node, envID, volumeName string) error
}

VolumeAgentDialer removes a deleted volume's docker volume on its pinned node (best effort; nil disables the cleanup). Production dials AgentLocalService.

type VolumeFileDialer 

type VolumeFileDialer interface {
	ListFiles(ctx context.Context, node *zatterav1.Node, envID, volName, path string) (*clusterv1.AgentListFilesResponse, error)
	ReadFile(ctx context.Context, node *zatterav1.Node, envID, volName, path string, emit func([]byte) error) error
}

VolumeFileDialer reaches the AgentLocalService on a volume's pinned node for read-only file browsing (T-77). Split from VolumeAgentDialer so a test (or a control node with no agent connectivity) can supply one without the other.

type VolumeServer 

type VolumeServer struct {
	zatterav1.UnimplementedVolumeServiceServer
	// contains filtered or unexported fields
}

VolumeServer implements VolumeService: CRUD (T-62), snapshot/restore (T-64/T-65) and read-only file browsing (T-77).

func NewVolumeServer 

func NewVolumeServer(store *state.Store, raft Applier, dial VolumeAgentDialer, clk clock.Clock, log *slog.Logger) *VolumeServer

NewVolumeServer builds the volume service. dial removes the docker volume on its node when the volume is deleted (best effort; nil skips that cleanup).

func (*VolumeServer) CreateVolume 

func (s *VolumeServer) CreateVolume(ctx context.Context, req *zatterav1.CreateVolumeRequest) (*zatterav1.Volume, error)

CreateVolume creates a named volume pinned to a node. When node_id is empty the least-used ALIVE worker is chosen. Names are unique within (project, env).

func (*VolumeServer) DeleteVolume 

func (s *VolumeServer) DeleteVolume(ctx context.Context, req *zatterav1.DeleteVolumeRequest) (*emptypb.Empty, error)

DeleteVolume removes a volume. It refuses while the volume is mounted (a live fencing lease or a running instance on its node).

func (*VolumeServer) ListFiles 

func (s *VolumeServer) ListFiles(ctx context.Context, req *zatterav1.ListFilesRequest) (*zatterav1.ListFilesResponse, error)

ListFiles lists one directory inside a volume. Read-only, and deliberately allowed while the volume is mounted: looking at a live volume is the main reason to browse one (unlike snapshot/restore/delete, which refuse).

func (*VolumeServer) ListSnapshots 

func (s *VolumeServer) ListSnapshots(_ context.Context, req *zatterav1.ListSnapshotsRequest) (*zatterav1.ListSnapshotsResponse, error)

ListSnapshots returns a volume's snapshots (newest first).

func (*VolumeServer) ListVolumes 

func (s *VolumeServer) ListVolumes(_ context.Context, req *zatterav1.ListVolumesRequest) (*zatterav1.ListVolumesResponse, error)

ListVolumes returns the project's volumes.

func (*VolumeServer) ReadFile 

func (s *VolumeServer) ReadFile(req *zatterav1.ReadFileRequest, stream grpc.ServerStreamingServer[zatterav1.FileChunk]) error

ReadFile streams one file out of a volume.

func (*VolumeServer) RestoreSnapshot 

func (s *VolumeServer) RestoreSnapshot(ctx context.Context, req *zatterav1.RestoreSnapshotRequest) (*zatterav1.Volume, error)

RestoreSnapshot restores a snapshot into its volume. It refuses while the volume is mounted — the service must be stopped first (single-writer).

func (*VolumeServer) SetFileDialer 

func (s *VolumeServer) SetFileDialer(d VolumeFileDialer)

SetFileDialer wires the browse/download data path. Nil leaves ListFiles and ReadFile Unimplemented.

func (*VolumeServer) SnapshotVolume 

func (s *VolumeServer) SnapshotVolume(ctx context.Context, req *zatterav1.SnapshotVolumeRequest) (*zatterav1.VolumeSnapshot, error)

SnapshotVolume takes an on-demand snapshot of a volume and returns the completed VolumeSnapshot record.

func (*VolumeServer) WithSnapshots 

func (s *VolumeServer) WithSnapshots(d *SnapshotDispatcher) *VolumeServer

WithSnapshots attaches the snapshot dispatcher (enables Snapshot/Restore).

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.