api

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 13, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package api holds the operator-facing Connect handlers: tasks and node administration. Identity comes from the transport middleware; nothing here authenticates anyone.

Index

Constants

View Source 
const AgentLoginHeader = "X-Podium-Login"

AgentLoginHeader is what the proxy tells the conductor about the human on the other end.

It is a plain header, not a signed assertion. The conductor trusts it only because the bearer proves the request came through this server, which is the only party that knows the token and the only one that authenticated the caller. If the conductor is ever exposed beyond loopback or the compose network, this has to become a signed assertion — docs/security.md says so.

View Source 
const ArtifactDownloadPrefix = "/artifacts/"

ArtifactDownloadPrefix is the HTTP path the server proxies artifact bytes on. It is not a Connect procedure on purpose: a 512 MB artifact must stream, and a unary response would have to hold all of it in memory at both ends.

View Source 
const MaxNodeSlots = 256

MaxNodeSlots is the largest slot count SetNodeSlots accepts. It is a guard against a typo, not a considered limit on what a machine can do: the scheduler assigns straight up to this number, so a stray zero on the end would pile hundreds of containers onto one box before anybody noticed. An operator who really wants more can raise max_tasks in the node's own configuration, where the number sits next to the machine it describes.

Variables

This section is empty.

Functions

func NewAgentProxy 

func NewAgentProxy(agentURL, agentToken string, logger *slog.Logger) (http.Handler, error)

NewAgentProxy reverse-proxies the conductor's Connect service.

Mount it behind transport.WithIdentity on /podium.agent.v1.AgentService/ only. Two things happen on every request and both matter: any client-supplied Authorization and X-Podium-Login are deleted, and the server's own pair is set. The browser's credential is the dev token, which the conductor must never see; the login is the server's word about who is calling, which a client must never be able to write.

The conductor's /healthz, /readyz and /metrics are deliberately not proxied. They are its own operational surface, they are unauthenticated on its listener, and publishing them through an authenticated origin would put a second, differently-shaped health story in front of an operator.

func NewArtifactDownloadHandler 

func NewArtifactDownloadHandler(a *artifacts.Service, logger *slog.Logger) http.Handler

NewArtifactDownloadHandler streams an artifact's bytes through the server. It is what `podium artifact get` uses by default: the CLI is guaranteed a route to the control plane and is not guaranteed one to the object store.

func NewUIHandler 

func NewUIHandler(logger *slog.Logger) http.Handler

NewUIHandler serves the embedded single-page app with an SPA fallback: any path that is not a real file is answered with index.html so the client router can take it. Mount it on "/" after the RPC prefixes — it is the last resort, and it is deliberately unauthenticated, because the bundle is not a secret and the API behind it still demands a token.

A binary built with `-tags noui`, or one built before `make web` ever ran, has no index.html; it gets a handler that says so instead of a confusing 404.

Types

type ArtifactService 

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

ArtifactService implements podium.v1.ArtifactService.

func NewArtifactService 

func NewArtifactService(a *artifacts.Service, logger *slog.Logger) *ArtifactService

NewArtifactService returns the operator-facing artifact handlers.

func (*ArtifactService) GetArtifactURL 

func (s *ArtifactService) GetArtifactURL(
	ctx context.Context,
	req *connect.Request[podiumv1.GetArtifactURLRequest],
) (*connect.Response[podiumv1.GetArtifactURLResponse], error)

GetArtifactURL mints a presigned GET. A caller with no route to the object store fetches the same bytes from ArtifactDownloadPrefix instead.

func (*ArtifactService) ListArtifacts 

func (s *ArtifactService) ListArtifacts(
	ctx context.Context,
	req *connect.Request[podiumv1.ListArtifactsRequest],
) (*connect.Response[podiumv1.ListArtifactsResponse], error)

ListArtifacts returns everything stored for a task, rolled-up logs included.

type Canceller 

type Canceller interface {
	Cancel(ctx context.Context, nodeID, taskID, reason string) error
}

Canceller is the node registry's cancel path.

type Drainer 

type Drainer interface {
	SetDrain(ctx context.Context, nodeID string, draining bool) error
	SetSlots(ctx context.Context, nodeID string, maxTasks int32) error
}

Drainer pushes an operator's standing instructions at a connected node: what it may not take, and how much of it.

type Events 

type Events interface {
	Subscribe(ctx context.Context, taskID string, fromSeq uint64) (<-chan *podiumv1.TaskEvent, error)
}

Events is the logs service, as much of it as the API needs.

type IdentityService 

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

IdentityService implements podium.v1.IdentityService. It reads nothing and writes nothing: the whole answer is the Identity the transport already put in the context, plus one fact about how this control plane is configured.

func NewIdentityService 

func NewIdentityService(agentEnabled bool) *IdentityService

NewIdentityService returns the identity handler. agentEnabled is cfg.AgentEnabled(): whether this server proxies the conductor's API, which is how the web UI knows to show its Agent screen at all.

func (*IdentityService) WhoAmI 

func (s *IdentityService) WhoAmI(
	ctx context.Context,
	_ *connect.Request[podiumv1.WhoAmIRequest],
) (*connect.Response[podiumv1.WhoAmIResponse], error)

WhoAmI reports who the transport says the caller is.

The call is behind the same identity middleware as everything else, and that is the point: a client that gets an answer without presenting a credential knows it is on a tailnet and needs no login step, and a client that gets 401 knows it must supply the dev token. The web UI uses exactly that to decide whether to prompt.

type Injector 

type Injector interface {
	Inject(ctx context.Context, nodeID, taskID, text string) error
}

Injector is the node registry's path for a human message into a running task.

type NodeAdminService 

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

NodeAdminService implements podium.v1.NodeAdminService.

func NewNodeAdminService 

func NewNodeAdminService(st *store.Store, sessions Sessions, drains Drainer, logger *slog.Logger) *NodeAdminService

NewNodeAdminService returns the node administration API.

func (*NodeAdminService) CreateEnrollmentToken 

func (s *NodeAdminService) CreateEnrollmentToken(
	ctx context.Context,
	req *connect.Request[podiumv1.CreateEnrollmentTokenRequest],
) (*connect.Response[podiumv1.CreateEnrollmentTokenResponse], error)

CreateEnrollmentToken mints a single-use token. The plaintext is returned once and never stored; only its SHA-256 reaches Postgres, so it is never recoverable and never logged.

func (*NodeAdminService) DeleteNode 

func (s *NodeAdminService) DeleteNode(
	ctx context.Context,
	req *connect.Request[podiumv1.DeleteNodeRequest],
) (*connect.Response[podiumv1.DeleteNodeResponse], error)

DeleteNode forgets a node. It refuses one that still holds a stream unless it has been drained, and refuses one that still has tasks on it whatever its state: tasks.node_id is a foreign key, and a task whose node row vanished is a task nobody can explain.

func (*NodeAdminService) DrainNode 

func (s *NodeAdminService) DrainNode(
	ctx context.Context,
	req *connect.Request[podiumv1.DrainNodeRequest],
) (*connect.Response[podiumv1.DrainNodeResponse], error)

DrainNode stops a node being given new work. Whatever it is running finishes, which is the whole point: a drain is how a machine is taken out of service without killing the jobs that are already on it.

The flag is a column, so it survives both daemons restarting and applies to a node that is offline right now — draining a machine before you power it back on is a reasonable thing to want. The live session is told as well, so the scheduler stops considering the node immediately rather than at its next heartbeat.

func (*NodeAdminService) ListNodes 

func (s *NodeAdminService) ListNodes(
	ctx context.Context,
	_ *connect.Request[podiumv1.ListNodesRequest],
) (*connect.Response[podiumv1.ListNodesResponse], error)

ListNodes returns every enrolled node, oldest first, with live slot counts for the ones that currently hold a stream on this server.

func (*NodeAdminService) RekeyNode 

func (s *NodeAdminService) RekeyNode(
	ctx context.Context,
	req *connect.Request[podiumv1.RekeyNodeRequest],
) (*connect.Response[podiumv1.RekeyNodeResponse], error)

RekeyNode unbinds a node from the Tailscale device it enrolled from. It is what an operator runs when a worker is rebuilt or replaced: the node keeps its ID, its labels and its history, and the next Hello binds it to whatever device it arrives from. Until then the node key alone is enough to connect, which is exactly the exposure the binding removes — so rekey a node when you are about to move it, not as a matter of routine.

func (*NodeAdminService) SetNodeLabels 

func (s *NodeAdminService) SetNodeLabels(
	ctx context.Context,
	req *connect.Request[podiumv1.SetNodeLabelsRequest],
) (*connect.Response[podiumv1.SetNodeLabelsResponse], error)

SetNodeLabels changes what a node is eligible for after it has enrolled. Until this existed the labels a node was given at enrollment were the labels it had for good, and tagging a machine meant editing nodes.labels by hand and restarting the daemon so its session re-read the row.

It adds and removes rather than replacing, because a whole-list write is a race between two operators who each meant to change one label. The live session is retagged too, so the scheduler routes on the new set immediately rather than at the next reconnect.

func (*NodeAdminService) SetNodeSlots 

func (s *NodeAdminService) SetNodeSlots(
	ctx context.Context,
	req *connect.Request[podiumv1.SetNodeSlotsRequest],
) (*connect.Response[podiumv1.SetNodeSlotsResponse], error)

SetNodeSlots changes how many tasks a node runs at once. Zero clears the instruction, and the node goes back to the max_tasks in its own configuration file.

It is stored against the node rather than sent and forgotten, for the same reason a drain is: an operator who caps a machine at two tasks means it for the machine, and a value that evaporated on the next reconnect would be worse than no feature at all. A node that is offline right now is a legitimate thing to configure — every stream is sent the count just after its HelloAck, so the node picks it up on its next connection.

func (*NodeAdminService) UndrainNode 

func (s *NodeAdminService) UndrainNode(
	ctx context.Context,
	req *connect.Request[podiumv1.UndrainNodeRequest],
) (*connect.Response[podiumv1.UndrainNodeResponse], error)

UndrainNode puts a drained node back in the pool.

type RegistryService 

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

RegistryService implements podium.v1.RegistryService. It is the secret store's second table under the same rules: no read endpoint, any authenticated caller, actor recorded.

func NewRegistryService 

func NewRegistryService(svc *secrets.Service, logger *slog.Logger) *RegistryService

NewRegistryService returns the registry API.

func (*RegistryService) DeleteRegistry 

func (s *RegistryService) DeleteRegistry(
	ctx context.Context,
	req *connect.Request[podiumv1.DeleteRegistryRequest],
) (*connect.Response[podiumv1.DeleteRegistryResponse], error)

DeleteRegistry removes a registry login. The next pull from that host is anonymous.

func (*RegistryService) ListRegistries 

func (s *RegistryService) ListRegistries(
	ctx context.Context,
	_ *connect.Request[podiumv1.ListRegistriesRequest],
) (*connect.Response[podiumv1.ListRegistriesResponse], error)

ListRegistries returns metadata only: hosts, usernames, key ids and who set them.

func (*RegistryService) SetRegistry 

func (s *RegistryService) SetRegistry(
	ctx context.Context,
	req *connect.Request[podiumv1.SetRegistryRequest],
) (*connect.Response[podiumv1.SetRegistryResponse], error)

SetRegistry creates or replaces the login for one registry host. The request's password is zeroed before the handler returns.

type SecretChecker 

type SecretChecker interface {
	CheckRefs(ctx context.Context, refs []spec.SecretRef) error
}

SecretChecker answers whether the secrets a spec names exist, without reading their values. It is optional: a server with no master key has no secrets to check against and refuses the task at assignment instead.

type SecretService 

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

SecretService implements podium.v1.SecretService.

There is deliberately no read endpoint. A value goes in through SetSecret and only ever comes back out inside an Assign, on its way to the node about to run the task that referenced it. Authorization is "any authenticated caller" in this slice; the actor is recorded on the row and in the audit log so RBAC can be layered on later.

func NewSecretService 

func NewSecretService(svc *secrets.Service, logger *slog.Logger) *SecretService

NewSecretService returns the secret API.

func (*SecretService) DeleteSecret 

func (s *SecretService) DeleteSecret(
	ctx context.Context,
	req *connect.Request[podiumv1.DeleteSecretRequest],
) (*connect.Response[podiumv1.DeleteSecretResponse], error)

DeleteSecret removes a secret. Tasks already assigned keep the copy inside their Assign; the next task that references the name fails to resolve.

func (*SecretService) ListSecrets 

func (s *SecretService) ListSecrets(
	ctx context.Context,
	_ *connect.Request[podiumv1.ListSecretsRequest],
) (*connect.Response[podiumv1.ListSecretsResponse], error)

ListSecrets returns metadata only: names, versions, key ids and who set them.

func (*SecretService) SetSecret 

func (s *SecretService) SetSecret(
	ctx context.Context,
	req *connect.Request[podiumv1.SetSecretRequest],
) (*connect.Response[podiumv1.SetSecretResponse], error)

SetSecret creates or replaces a secret and returns its metadata. The request's value is zeroed before the handler returns.

type Sessions 

type Sessions interface {
	SnapshotOf(nodeID string) (nodes.Snapshot, bool)
	SetLabels(nodeID string, labels []string) bool
}

Sessions is the live-session half of the node registry.

type TaskService 

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

TaskService implements podium.v1.TaskService.

func NewTaskService 

func NewTaskService(st *store.Store, events Events, canceller Canceller, checker SecretChecker, logger *slog.Logger) *TaskService

NewTaskService returns the task API. injector may be the same object as canceller (the node registry implements both); nil means InjectTask is unavailable.

func (*TaskService) CancelTask 

func (s *TaskService) CancelTask(
	ctx context.Context,
	req *connect.Request[podiumv1.CancelTaskRequest],
) (*connect.Response[podiumv1.CancelTaskResponse], error)

CancelTask cancels a queued task outright and asks the node to stop anything further along. It never waits for the container to die: with the task command as PID 1 a default-disposition SIGTERM is discarded, so a cancel routinely costs the node's full 30s grace period. The terminal status lands when the exited/finished events arrive.

func (*TaskService) CreateTask 

func (s *TaskService) CreateTask(
	ctx context.Context,
	req *connect.Request[podiumv1.CreateTaskRequest],
) (*connect.Response[podiumv1.CreateTaskResponse], error)

CreateTask validates the spec and queues the task. Scheduling is the scheduler's problem.

func (*TaskService) GetTask 

func (s *TaskService) GetTask(
	ctx context.Context,
	req *connect.Request[podiumv1.GetTaskRequest],
) (*connect.Response[podiumv1.GetTaskResponse], error)

GetTask returns one task.

func (*TaskService) InjectTask 

func (s *TaskService) InjectTask(
	ctx context.Context,
	req *connect.Request[podiumv1.InjectTaskRequest],
) (*connect.Response[podiumv1.InjectTaskResponse], error)

InjectTask delivers one human message into a running task. The container stays up.

func (*TaskService) ListTasks 

func (s *TaskService) ListTasks(
	ctx context.Context,
	req *connect.Request[podiumv1.ListTasksRequest],
) (*connect.Response[podiumv1.ListTasksResponse], error)

ListTasks pages newest-first. next_cursor is empty on the last page; feed it back verbatim.

func (*TaskService) StreamTaskEvents 

func (s *TaskService) StreamTaskEvents(
	ctx context.Context,
	req *connect.Request[podiumv1.StreamTaskEventsRequest],
	stream *connect.ServerStream[podiumv1.TaskEvent],
) error

StreamTaskEvents replays everything stored after from_seq and then follows the task live. The two storage tables share the node's single per-task seq space, so the merged stream is in strict seq order. It ends when the task is terminal and every event has been sent.

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.