// Package jobs decides whether a service may be updated and starts update // jobs on the right agent. package jobs import ( "context" "errors" "log/slog" "time" "github.com/shcizo/package-updater/internal/imageref" "github.com/shcizo/package-updater/internal/master/events" "github.com/shcizo/package-updater/internal/master/model" "github.com/shcizo/package-updater/internal/protocol" ) // Store is the storage jobs needs. *store.Store satisfies it. type Store interface { GetService(ctx context.Context, id int64) (model.Service, error) CreateJob(ctx context.Context, j model.NewJob) (model.Job, error) FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error) GetJob(ctx context.Context, id int64) (model.Job, error) } // Hub routes jobs to agent connections. *hub.Hub satisfies it. type Hub interface { // Route picks the connection for svc: model.ErrAgentOffline or // model.ErrAgentTooOld when there is none. Route(svc model.Service) (model.Route, error) // StartJob sends job.start over exactly the routed connection. // model.ErrAgentOffline if that connection is gone or was replaced. StartJob(r model.Route, jobID int64, start protocol.JobStart) error } // Publisher receives change signals. *events.Broker satisfies it. type Publisher interface{ Publish(t events.Topic) } // Service starts update jobs. type Service struct { store Store hub Hub pub Publisher selfRepo string log *slog.Logger now func() time.Time } // New returns a Service. agentImage is the master's AGENT_IMAGE; services // whose image repo equals its repo are the updater itself and are refused. func New(store Store, hub Hub, pub Publisher, agentImage string, log *slog.Logger) *Service { return &Service{ store: store, hub: hub, pub: pub, log: log, now: time.Now, selfRepo: imageref.Parse(agentImage).Repo, } } // target splits a service's actual ref into the repo and tag to deploy. A bare // image ("postgres", no tag and no digest) means "latest", as it does for // Docker. A ref with a digest but no tag follows no tag (ok is false). func target(actualRef string) (repo, tag string, ok bool) { ref := imageref.Parse(actualRef) switch { case ref.Tag != "": return ref.Repo, ref.Tag, true case ref.Digest == "": return ref.Repo, "latest", true } return ref.Repo, "", false } // FollowedRef returns "repo:tag" for the tag a service with this actual ref // follows (a bare image follows "latest"), or "" when it follows no tag. func FollowedRef(actualRef string) string { repo, tag, ok := target(actualRef) if !ok { return "" } return repo + ":" + tag } // Updatable returns nil or a *model.NotUpdatableError. It looks only at the // service row, so the UI can call it for every page render. It answers for an // update to the service's own followed tag (the Update button). func (s *Service) Updatable(svc model.Service) error { return s.updatable(svc, false) } // AutoAllowed returns nil or a *model.NotUpdatableError saying why the // service can never be updated automatically. Unlike Updatable it does not // refuse a service that follows no tag: a webhook that names a tag can still // update it. func (s *Service) AutoAllowed(svc model.Service) error { return s.updatable(svc, true) } // updatable checks the service row. explicitRef means the caller supplies the // target ref, so a service that follows no tag is fine. func (s *Service) updatable(svc model.Service, explicitRef bool) error { switch { case svc.RemovedAt != nil: return &model.NotUpdatableError{Reason: model.ReasonRemoved} case svc.Self || imageref.Parse(svc.ActualRef).Repo == s.selfRepo: return &model.NotUpdatableError{Reason: model.ReasonSelf} case svc.Built: return &model.NotUpdatableError{Reason: model.ReasonBuilt} case svc.Kind == protocol.KindContainer: return &model.NotUpdatableError{Reason: model.ReasonContainer} } if _, _, ok := target(svc.ActualRef); !ok && !explicitRef { return &model.NotUpdatableError{Reason: model.ReasonNoTag} } return nil } // StartRequest says who asks for an update and what to deploy. type StartRequest struct { Trigger model.JobTrigger RequestedBy string // OIDC subject for manual jobs; "" otherwise // Ref is the ref to deploy ("repo:tag"). Empty means the service's // followed tag, derived from its actual ref. Ref string } // Trigger starts a manual update of the service to its current tag. It // returns what Start returns, and model.ErrNotFound for an unknown service. func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) { svc, err := s.store.GetService(ctx, serviceID) if err != nil { return model.Job{}, err } return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy}) } // Start starts an update of svc. It returns the job, which may already be // failed ("agent offline", "agent too old for updates"); a // *model.NotUpdatableError when the service cannot be updated (no job // created); the active job and model.ErrJobActive when one is queued or // running. func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) { if err := s.updatable(svc, req.Ref != ""); err != nil { return model.Job{}, err } ref := req.Ref if ref == "" { ref = FollowedRef(svc.ActualRef) // updatable checked the tag } r, routeErr := s.hub.Route(svc) nj := model.NewJob{ ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref, Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(), } if routeErr == nil { nj.AgentID = &r.AgentID } job, err := s.store.CreateJob(ctx, nj) if errors.Is(err, model.ErrAgentOffline) { // The routed agent was revoked after Route. Record the attempt as a // failed job that was never sent anywhere. routeErr, nj.AgentID = err, nil job, err = s.store.CreateJob(ctx, nj) } if err != nil { return job, err } failure := routeErr if failure == nil { start := protocol.JobStart{Kind: svc.Kind, Ref: ref} if svc.Kind == protocol.KindSwarm { start.SwarmClusterID, start.SwarmService = svc.SwarmClusterID, svc.SwarmService } else { start.ComposeProject, start.ComposeService = svc.ComposeProject, svc.ComposeService } if err := s.hub.StartJob(r, job.ID, start); err != nil { s.log.Warn("job_start_failed", "job_id", job.ID, "error", err) failure = model.ErrAgentOffline } } if failure != nil { if _, err := s.store.FailJob(ctx, job.ID, failure.Error(), s.now()); err != nil { return model.Job{}, err } } s.pub.Publish(events.Jobs) s.pub.Publish(events.JobTopic(job.ID)) return s.store.GetJob(ctx, job.ID) }