diff --git a/internal/master/events/broker.go b/internal/master/events/broker.go index 994fe51..7358b5c 100644 --- a/internal/master/events/broker.go +++ b/internal/master/events/broker.go @@ -2,7 +2,10 @@ // open browser streams. package events -import "sync" +import ( + "strconv" + "sync" +) // Topic names what changed. Subscribers refetch that part of the UI. type Topic string @@ -11,8 +14,13 @@ type Topic string const ( Services Topic = "services" Agents Topic = "agents" + Jobs Topic = "jobs" // a job was created or changed status ) +// JobTopic is signalled when job id gets new output or changes status. The +// name avoids ":" so it is safe inside hx-trigger="sse:job-42". +func JobTopic(id int64) Topic { return Topic("job-" + strconv.FormatInt(id, 10)) } + // Broker is an in-process publish/subscribe hub. type Broker struct { mu sync.Mutex diff --git a/internal/master/events/broker_test.go b/internal/master/events/broker_test.go index 7613051..7043fe6 100644 --- a/internal/master/events/broker_test.go +++ b/internal/master/events/broker_test.go @@ -46,3 +46,7 @@ func TestBroker_SlowSubscriberDoesNotBlockPublish(t *testing.T) { t.Fatal("Publish blocked on a slow subscriber") } } + +func TestJobTopic(t *testing.T) { + assert.Equal(t, Topic("job-42"), JobTopic(42)) +} diff --git a/internal/master/model/job.go b/internal/master/model/job.go new file mode 100644 index 0000000..5907deb --- /dev/null +++ b/internal/master/model/job.go @@ -0,0 +1,116 @@ +package model + +import ( + "errors" + "time" +) + +// Sentinel errors for jobs. +var ( + // ErrJobActive means the service already has a queued or running job. + // CreateJob and jobs.Service.Trigger return it together with that job. + ErrJobActive = errors.New("an update is already in progress for this service") + // ErrAgentOffline means no connected agent can run the job. + ErrAgentOffline = errors.New("agent offline") + // ErrAgentTooOld means the only agent that could run the job speaks a + // protocol version without jobs. + ErrAgentTooOld = errors.New("agent too old for updates") +) + +// NotUpdatableError says why a service cannot be updated. Reason is shown +// in the UI as is. +type NotUpdatableError struct { + Reason string +} + +func (e *NotUpdatableError) Error() string { return "not updatable: " + e.Reason } + +// Reasons a service cannot be updated. +const ( + ReasonSelf = "updated manually outside the system" + ReasonContainer = "plain containers cannot be updated" + ReasonNoTag = "follows no tag" + ReasonRemoved = "service no longer exists" +) + +// JobStatus is the state of an update job. +type JobStatus string + +// Job statuses. A job moves queued -> running -> succeeded|failed, or +// straight from queued to failed. +const ( + JobQueued JobStatus = "queued" + JobRunning JobStatus = "running" + JobSucceeded JobStatus = "succeeded" + JobFailed JobStatus = "failed" +) + +// JobTrigger says what created a job. Only TriggerManual is used in phase 2. +type JobTrigger string + +// Job triggers. +const ( + TriggerManual JobTrigger = "manual" + TriggerWebhook JobTrigger = "webhook" + TriggerPoll JobTrigger = "poll" +) + +// Job is one update of one service. +type Job struct { + ID int64 + ServiceID int64 + ServiceName string // Service.DisplayName() of the job's service + AgentID *int64 // agent the job was sent to; nil if it never was + AgentName string + FromRef string + ToRef string // "repo:tag" when created; "repo:tag@digest" after success + Trigger JobTrigger + RequestedBy string // OIDC subject for manual jobs + Status JobStatus + Output string // empty in ListJobs results + OutputTruncated bool + Error string + CreatedAt time.Time + StartedAt *time.Time + FinishedAt *time.Time +} + +// Active reports whether the job is queued or running. +func (j Job) Active() bool { return j.Status == JobQueued || j.Status == JobRunning } + +// NewJob is what CreateJob needs to insert a queued job. +type NewJob struct { + ServiceID int64 + AgentID *int64 + FromRef string + ToRef string + Trigger JobTrigger + RequestedBy string + CreatedAt time.Time +} + +// JobFinish is a job's outcome as reported by the agent. +type JobFinish struct { + Status JobStatus // JobSucceeded or JobFailed + Ref string // deployed "repo:tag@digest"; set on success + Error string + Output string + OutputTruncated bool + StartedAt time.Time // agent clock + FinishedAt time.Time // agent clock +} + +// JobFilter selects jobs for ListJobs. ServiceID 0 means all services. +// Limit 0 means 100. +type JobFilter struct { + ServiceID int64 + Limit int +} + +// Route is the hub's choice of agent connection for a job. ConnID identifies +// the connection, so a job is never sent over a connection that replaced the +// one chosen. +type Route struct { + AgentID int64 + ConnID uint64 +} diff --git a/internal/master/model/model.go b/internal/master/model/model.go index e09bc4a..b1a5570 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -41,10 +41,12 @@ type Service struct { ComposeProject string ComposeService string WorkingDir string + ConfigFiles string // raw config_files label, paths joined by "," SwarmClusterID string SwarmService string ContainerName string ImageRepo string + DesiredRef string // set by the last successful job; "" if never updated ActualRef string ActualDigest string Status string @@ -52,6 +54,7 @@ type Service struct { UpdateMode string FirstSeenAt time.Time UpdatedAt time.Time + RemovedAt *time.Time } // DisplayName is the human-facing name of the service. diff --git a/internal/master/model/model_test.go b/internal/master/model/model_test.go index 0705f69..48c7947 100644 --- a/internal/master/model/model_test.go +++ b/internal/master/model/model_test.go @@ -39,3 +39,20 @@ func TestDisplayName(t *testing.T) { }) } } + +func TestJob_Active(t *testing.T) { + cases := []struct { + status JobStatus + want bool + }{ + {JobQueued, true}, + {JobRunning, true}, + {JobSucceeded, false}, + {JobFailed, false}, + } + for _, c := range cases { + t.Run(string(c.status), func(t *testing.T) { + assert.Equal(t, c.want, Job{Status: c.status}.Active()) + }) + } +} diff --git a/internal/master/store/migrations/0002_jobs.sql b/internal/master/store/migrations/0002_jobs.sql new file mode 100644 index 0000000..eaae8a6 --- /dev/null +++ b/internal/master/store/migrations/0002_jobs.sql @@ -0,0 +1,27 @@ +ALTER TABLE services ADD COLUMN config_files text NOT NULL DEFAULT ''; +ALTER TABLE services ADD COLUMN desired_ref text NOT NULL DEFAULT ''; + +CREATE TABLE update_jobs ( + id bigserial PRIMARY KEY, + service_id bigint NOT NULL REFERENCES services (id), + agent_id bigint REFERENCES agents (id), + from_ref text NOT NULL DEFAULT '', + to_ref text NOT NULL, + trigger text NOT NULL CHECK (trigger IN ('manual', 'webhook', 'poll')), + requested_by text NOT NULL DEFAULT '', + status text NOT NULL CHECK (status IN ('queued', 'running', 'succeeded', 'failed')), + output text NOT NULL DEFAULT '', + output_truncated boolean NOT NULL DEFAULT false, + error text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL, + started_at timestamptz, + finished_at timestamptz +); + +-- One queued or running job per service. CreateJob relies on this index to +-- turn a second concurrent click into ErrJobActive. +CREATE UNIQUE INDEX update_jobs_one_active ON update_jobs (service_id) + WHERE status IN ('queued', 'running'); +CREATE INDEX update_jobs_by_service ON update_jobs (service_id, id DESC); +CREATE INDEX update_jobs_active_by_agent ON update_jobs (agent_id) + WHERE status IN ('queued', 'running'); diff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go index a4aa9bb..048587b 100644 --- a/internal/protocol/protocol.go +++ b/internal/protocol/protocol.go @@ -1,33 +1,66 @@ // Package protocol defines the messages agents and the master exchange over // their WebSocket connection. Every message is an Envelope whose Payload is // one of the payload types below, selected by Type. +// +// Both sides close the connection on a message they cannot parse (bad JSON, +// or a known Type whose payload does not decode). A message with an unknown +// Type is logged and ignored, so a newer peer can add message types without +// breaking an older one. package protocol import ( "encoding/json" "fmt" + "time" ) +// Version is the protocol version this build speaks. Agents send it in +// Hello. Phase-1 agents do not send the field, which reads as 0. +const Version = 2 + +// MinJobVersion is the lowest agent protocol version the master sends jobs to. +const MinJobVersion = 2 + +// MaxJobOutput caps the job output an agent keeps and the master stores, in +// bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per +// byte) a JobResult carrying this much output still fits the 1 MiB read limit. +const MaxJobOutput = 128 << 10 + // Message types sent from agent to master. const ( - TypeHello = "hello" - TypeSnapshot = "snapshot" - TypeUpsert = "upsert" - TypeRemove = "remove" + TypeHello = "hello" + TypeSnapshot = "snapshot" + TypeUpsert = "upsert" + TypeRemove = "remove" + TypeJobProgress = "job.progress" + TypeJobResult = "job.result" ) -// Envelope wraps every message on the wire. +// Message types sent from master to agent. +const ( + TypeJobStart = "job.start" + TypeJobAck = "job.ack" +) + +// Envelope wraps every message on the wire. ID is the job ID (decimal) on +// job.* messages and empty on inventory messages. type Envelope struct { Type string `json:"type"` + ID string `json:"id,omitempty"` Payload json.RawMessage `json:"payload,omitempty"` } // Hello is the first message an agent sends on every connection. type Hello struct { - AgentVersion string `json:"agent_version"` - Hostname string `json:"hostname"` - SwarmManager bool `json:"swarm_manager"` - SwarmClusterID string `json:"swarm_cluster_id,omitempty"` + AgentVersion string `json:"agent_version"` + ProtocolVersion int `json:"protocol_version,omitempty"` + Hostname string `json:"hostname"` + SwarmManager bool `json:"swarm_manager"` + SwarmClusterID string `json:"swarm_cluster_id,omitempty"` + // KnownJobs lists every job ID the agent holds: queued, running, or + // finished but not yet acknowledged. The master fails its queued and + // running jobs for this agent that are not listed ("lost"). + KnownJobs []string `json:"known_jobs,omitempty"` } // Kind says how a service is deployed. @@ -47,12 +80,16 @@ type Service struct { ComposeProject string `json:"compose_project,omitempty"` ComposeService string `json:"compose_service,omitempty"` WorkingDir string `json:"working_dir,omitempty"` - SwarmService string `json:"swarm_service,omitempty"` - ContainerName string `json:"container_name,omitempty"` - Image string `json:"image"` - Digest string `json:"digest,omitempty"` - Status string `json:"status"` - Self bool `json:"self,omitempty"` + // ConfigFiles is the raw com.docker.compose.project.config_files label: + // absolute host paths joined by ",". Kept as a string so Service stays + // comparable with ==. + ConfigFiles string `json:"config_files,omitempty"` + SwarmService string `json:"swarm_service,omitempty"` + ContainerName string `json:"container_name,omitempty"` + Image string `json:"image"` + Digest string `json:"digest,omitempty"` + Status string `json:"status"` + Self bool `json:"self,omitempty"` } // Key identifies the service within one agent's report. It is stable across @@ -83,6 +120,50 @@ type Remove struct { Key string `json:"key"` } +// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag"; +// the agent resolves the tag's current digest and deploys "repo:tag@digest". +// The agent reads the Compose working directory and config files from its +// own Docker daemon, not from this message. +type JobStart struct { + Kind Kind `json:"kind"` + ComposeProject string `json:"compose_project,omitempty"` + ComposeService string `json:"compose_service,omitempty"` + SwarmClusterID string `json:"swarm_cluster_id,omitempty"` + SwarmService string `json:"swarm_service,omitempty"` + Ref string `json:"ref"` +} + +// JobProgress carries output produced since the previous JobProgress for the +// same job. The first JobProgress of a job (Output may be empty) tells the +// master the job is running. +type JobProgress struct { + Output string `json:"output"` +} + +// Job result statuses. +const ( + ResultSucceeded = "succeeded" + ResultFailed = "failed" +) + +// JobResult is a job's outcome. The agent keeps it until the master sends +// JobAck for the same ID, and resends it after every reconnect until then. +type JobResult struct { + Status string `json:"status"` // ResultSucceeded or ResultFailed + Ref string `json:"ref,omitempty"` // deployed "repo:tag@digest", set on success + Error string `json:"error,omitempty"` + // Output is the job's complete output, tail-capped at MaxJobOutput. It + // replaces whatever the master assembled from JobProgress messages. + Output string `json:"output"` + OutputTruncated bool `json:"output_truncated,omitempty"` + StartedAt time.Time `json:"started_at"` + FinishedAt time.Time `json:"finished_at"` +} + +// JobAck tells the agent the master has stored the JobResult with the same +// ID. It has no fields. +type JobAck struct{} + // Encode wraps payload in an Envelope of the given type. func Encode(typ string, payload any) (Envelope, error) { raw, err := json.Marshal(payload) @@ -92,6 +173,16 @@ func Encode(typ string, payload any) (Envelope, error) { return Envelope{Type: typ, Payload: raw}, nil } +// EncodeJob wraps a job message's payload in an Envelope carrying the job ID. +func EncodeJob(typ, jobID string, payload any) (Envelope, error) { + env, err := Encode(typ, payload) + if err != nil { + return Envelope{}, err + } + env.ID = jobID + return env, nil +} + // Decode unpacks env's payload into T. func Decode[T any](env Envelope) (T, error) { var v T diff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go index 4c288cb..53cc817 100644 --- a/internal/protocol/protocol_test.go +++ b/internal/protocol/protocol_test.go @@ -47,3 +47,35 @@ func TestDecodeRejectsMalformedPayload(t *testing.T) { _, err := Decode[Hello](Envelope{Type: TypeHello, Payload: json.RawMessage(`{"hostname": 5}`)}) assert.ErrorContains(t, err, "decode hello") } + +func TestEncodeJob_CarriesID(t *testing.T) { + in := JobStart{Kind: KindSwarm, SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"} + env, err := EncodeJob(TypeJobStart, "42", in) + require.NoError(t, err) + + raw, err := json.Marshal(env) + require.NoError(t, err) + var back Envelope + require.NoError(t, json.Unmarshal(raw, &back)) + + assert.Equal(t, "42", back.ID) + assert.Equal(t, TypeJobStart, back.Type) + out, err := Decode[JobStart](back) + require.NoError(t, err) + assert.Equal(t, in, out) +} + +func TestHello_PhaseOneDecodesAsVersionZero(t *testing.T) { + var h Hello + require.NoError(t, json.Unmarshal([]byte(`{"agent_version":"x","hostname":"h","swarm_manager":false}`), &h)) + assert.Equal(t, 0, h.ProtocolVersion) + assert.Nil(t, h.KnownJobs) +} + +func TestEnvelope_InventoryHasNoID(t *testing.T) { + env, err := Encode(TypeSnapshot, Snapshot{}) + require.NoError(t, err) + raw, err := json.Marshal(env) + require.NoError(t, err) + assert.NotContains(t, string(raw), `"id"`) +}