api

package
v0.1.0-alpha.6 Latest Latest
Warning

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

 Go to latest Published: Jul 17, 2026 License: Apache-2.0 Imports: 61 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, sealer secrets.Sealer, clk clock.Clock, log *slog.Logger) http.Handler

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 nil sealer or missing key degrades gracefully (deploys are skipped).

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 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 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, sealer secrets.Sealer) *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) QueryAudit 

func (a *Auditor) QueryAudit(_ 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) 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) *AuthServer

NewAuthServer builds the auth service. clusterDomain is cfg.Domain.

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) 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, sealer secrets.Sealer, 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.

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 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 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, keyring *secrets.Keyring, cfg JoinConfig, log *slog.Logger) *JoinServer

NewJoinServer builds the join service. keyring is the cluster data key (may be nil on a node that never bootstrapped/unsealed); it is handed to joining control nodes so they come up already unsealed (T-55).

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 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) 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.

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

	// Node↔control services (mTLS node identity, no REST gateway).
	AgentSyncService clusterv1.AgentSyncServiceServer
	// JoinService is token-authenticated (no mTLS), no REST gateway.
	JoinService clusterv1.JoinServiceServer
	// MeshService distributes WireGuard peer sets (mTLS node identity).
	MeshService clusterv1.MeshServiceServer
	// RouteService streams route snapshots to node proxies (mTLS node identity).
	RouteService clusterv1.RouteServiceServer

	// 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, sealer secrets.Sealer, dataKey []byte, connect func(context.Context, *zatterav1.Node) (*grpc.ClientConn, error), clk clock.Clock, log *slog.Logger) *SnapshotDispatcher

NewSnapshotDispatcher builds the dispatcher. sealer/dataKey come 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, sealer secrets.Sealer) *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 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 VolumeServer 

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

VolumeServer implements the CRUD subset of VolumeService (T-62). Snapshot, restore and file operations land in later tasks (T-64/T-65/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) 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) 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) 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.