feat(master): POST /update webhook with auto-mode updates (#8)
This commit is contained in:
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Config holds the master's runtime configuration.
|
||||
@@ -20,8 +21,18 @@ type Config struct {
|
||||
Port string
|
||||
AgentPort string
|
||||
LogLevel string
|
||||
|
||||
// UpdaterAPIKey is the bearer token for POST /update (UPDATER_API_KEY).
|
||||
// Empty disables the webhook.
|
||||
UpdaterAPIKey string
|
||||
// UpdateTimeout bounds how long one webhook request waits for its auto
|
||||
// jobs (UPDATE_TIMEOUT, default 5m).
|
||||
UpdateTimeout time.Duration
|
||||
}
|
||||
|
||||
// defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset.
|
||||
const defaultUpdateTimeout = 5 * time.Minute
|
||||
|
||||
// Load reads and validates the master configuration, failing fast on
|
||||
// anything missing so a half-configured master never starts.
|
||||
func Load() (*Config, error) {
|
||||
@@ -37,6 +48,8 @@ func Load() (*Config, error) {
|
||||
Port: getenvDefault("PORT", "8080"),
|
||||
AgentPort: getenvDefault("AGENT_PORT", "8090"),
|
||||
LogLevel: getenvDefault("LOG_LEVEL", "info"),
|
||||
UpdaterAPIKey: os.Getenv("UPDATER_API_KEY"),
|
||||
UpdateTimeout: defaultUpdateTimeout,
|
||||
}
|
||||
for _, r := range []struct{ name, value string }{
|
||||
{"DATABASE_URL", cfg.DatabaseURL},
|
||||
@@ -57,6 +70,13 @@ func Load() (*Config, error) {
|
||||
if !strings.HasPrefix(cfg.AgentURL, "wss://") {
|
||||
return nil, fmt.Errorf("AGENT_URL must start with wss://")
|
||||
}
|
||||
if v := os.Getenv("UPDATE_TIMEOUT"); v != "" {
|
||||
d, err := time.ParseDuration(v)
|
||||
if err != nil || d <= 0 {
|
||||
return nil, fmt.Errorf("UPDATE_TIMEOUT must be a positive duration such as 5m, got %q", v)
|
||||
}
|
||||
cfg.UpdateTimeout = d
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,7 @@ package config
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -21,7 +22,7 @@ func setAll(t *testing.T) {
|
||||
for k, v := range required {
|
||||
t.Setenv(k, v)
|
||||
}
|
||||
for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL"} {
|
||||
for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT"} {
|
||||
t.Setenv(k, "")
|
||||
}
|
||||
}
|
||||
@@ -35,6 +36,29 @@ func TestLoad_Defaults(t *testing.T) {
|
||||
assert.Equal(t, "gitea.shcizo.se/shcizo/package-updater:latest", cfg.AgentImage)
|
||||
assert.Equal(t, "info", cfg.LogLevel)
|
||||
assert.Equal(t, "updater-admins", cfg.AllowedGroup)
|
||||
assert.Empty(t, cfg.UpdaterAPIKey)
|
||||
assert.Equal(t, 5*time.Minute, cfg.UpdateTimeout)
|
||||
}
|
||||
|
||||
func TestLoad_WebhookSettings(t *testing.T) {
|
||||
setAll(t)
|
||||
t.Setenv("UPDATER_API_KEY", "s3cret")
|
||||
t.Setenv("UPDATE_TIMEOUT", "90s")
|
||||
cfg, err := Load()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "s3cret", cfg.UpdaterAPIKey)
|
||||
assert.Equal(t, 90*time.Second, cfg.UpdateTimeout)
|
||||
}
|
||||
|
||||
func TestLoad_InvalidUpdateTimeout(t *testing.T) {
|
||||
for _, v := range []string{"abc", "0s", "-5m", "30"} {
|
||||
t.Run(v, func(t *testing.T) {
|
||||
setAll(t)
|
||||
t.Setenv("UPDATE_TIMEOUT", v)
|
||||
_, err := Load()
|
||||
assert.ErrorContains(t, err, "UPDATE_TIMEOUT must be a positive duration")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoad_EachRequiredVariable(t *testing.T) {
|
||||
|
||||
@@ -68,9 +68,34 @@ func target(actualRef string) (repo, tag string, ok bool) {
|
||||
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.
|
||||
// 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}
|
||||
@@ -81,32 +106,49 @@ func (s *Service) Updatable(svc model.Service) error {
|
||||
case svc.Kind == protocol.KindContainer:
|
||||
return &model.NotUpdatableError{Reason: model.ReasonContainer}
|
||||
}
|
||||
if _, _, ok := target(svc.ActualRef); !ok {
|
||||
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 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; model.ErrNotFound for an unknown service.
|
||||
// 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
|
||||
}
|
||||
if err := s.Updatable(svc); err != nil {
|
||||
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
|
||||
}
|
||||
repo, tag, _ := target(svc.ActualRef) // Updatable checked the tag
|
||||
ref := repo + ":" + tag
|
||||
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: model.TriggerManual, RequestedBy: requestedBy, CreatedAt: s.now(),
|
||||
Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(),
|
||||
}
|
||||
if routeErr == nil {
|
||||
nj.AgentID = &r.AgentID
|
||||
|
||||
@@ -242,3 +242,102 @@ func TestTrigger_NotUpdatableCreatesNoJob(t *testing.T) {
|
||||
assert.Empty(t, h.starts)
|
||||
assert.Empty(t, pub.topics)
|
||||
}
|
||||
|
||||
func TestStart_WebhookTriggerWithExplicitRef(t *testing.T) {
|
||||
st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
|
||||
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
||||
s, _ := newService(st, h)
|
||||
|
||||
job, err := s.Start(context.Background(), composeSvc, StartRequest{
|
||||
Trigger: model.TriggerWebhook, Ref: "gitea.shcizo.se/shcizo/slusk:2.0.0"})
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, model.JobQueued, job.Status)
|
||||
require.Len(t, st.created, 1)
|
||||
assert.Equal(t, model.TriggerWebhook, st.created[0].Trigger)
|
||||
assert.Empty(t, st.created[0].RequestedBy)
|
||||
assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", st.created[0].ToRef)
|
||||
require.Len(t, h.starts, 1)
|
||||
assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", h.starts[0].start.Ref)
|
||||
}
|
||||
|
||||
func TestStart_ExplicitRefBypassesOnlyTheNoTagRefusal(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
svc model.Service
|
||||
reason string // "" means the job starts
|
||||
}{
|
||||
{"digest only", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""},
|
||||
{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
|
||||
{"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf},
|
||||
{"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt},
|
||||
{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
tt.svc.ID = 1
|
||||
st := &fakeStore{services: map[int64]model.Service{1: tt.svc}, jobs: map[int64]model.Job{}}
|
||||
h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
|
||||
s, _ := newService(st, h)
|
||||
|
||||
_, err := s.Start(context.Background(), tt.svc, StartRequest{Trigger: model.TriggerWebhook, Ref: "x/y:2"})
|
||||
|
||||
if tt.reason == "" {
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, st.created, 1)
|
||||
return
|
||||
}
|
||||
var nu *model.NotUpdatableError
|
||||
require.True(t, errors.As(err, &nu))
|
||||
assert.Equal(t, tt.reason, nu.Reason)
|
||||
assert.Empty(t, st.created)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStart_WithoutRefStillRefusesNoTag(t *testing.T) {
|
||||
svc := model.Service{ID: 1, Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}
|
||||
s, _ := newService(&fakeStore{jobs: map[int64]model.Job{}}, &fakeHub{})
|
||||
|
||||
_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerWebhook})
|
||||
|
||||
var nu *model.NotUpdatableError
|
||||
require.True(t, errors.As(err, &nu))
|
||||
assert.Equal(t, model.ReasonNoTag, nu.Reason)
|
||||
}
|
||||
|
||||
func TestAutoAllowed(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
svc model.Service
|
||||
reason string
|
||||
}{
|
||||
{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
|
||||
{"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf},
|
||||
{"agent image repo", model.Service{Kind: protocol.KindCompose, ActualRef: selfImage}, model.ReasonSelf},
|
||||
{"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt},
|
||||
{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
|
||||
{"digest without tag is allowed", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""},
|
||||
{"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""},
|
||||
}
|
||||
s, _ := newService(&fakeStore{}, &fakeHub{})
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := s.AutoAllowed(tt.svc)
|
||||
if tt.reason == "" {
|
||||
assert.NoError(t, err)
|
||||
return
|
||||
}
|
||||
var nu *model.NotUpdatableError
|
||||
require.True(t, errors.As(err, &nu))
|
||||
assert.Equal(t, tt.reason, nu.Reason)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFollowedRef(t *testing.T) {
|
||||
assert.Equal(t, "postgres:16", FollowedRef("postgres:16@sha256:abc"))
|
||||
assert.Equal(t, "postgres:latest", FollowedRef("postgres"))
|
||||
assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo"))
|
||||
assert.Empty(t, FollowedRef("redis@sha256:abc"))
|
||||
}
|
||||
|
||||
@@ -0,0 +1,331 @@
|
||||
// Package webhook implements POST /update, the endpoint CI calls after
|
||||
// pushing an image. It records the pushed tag as "available" on every service
|
||||
// that uses the image, starts an update for the services in auto mode and
|
||||
// waits for those updates to finish.
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/imageref"
|
||||
"github.com/shcizo/package-updater/internal/logging"
|
||||
"github.com/shcizo/package-updater/internal/master/events"
|
||||
"github.com/shcizo/package-updater/internal/master/jobs"
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
)
|
||||
|
||||
// maxBodyBytes caps the request body; a valid request is a few hundred bytes.
|
||||
const maxBodyBytes = 64 << 10
|
||||
|
||||
// defaultPollEvery is how often pending jobs are re-read from the store when
|
||||
// no broker signal arrives (signals to a slow subscriber can be dropped).
|
||||
const defaultPollEvery = 5 * time.Second
|
||||
|
||||
// tagPattern is Docker's tag grammar. It keeps spaces, colons and "@" out of
|
||||
// the ref the agent is asked to deploy.
|
||||
var tagPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$`)
|
||||
|
||||
// Result statuses.
|
||||
const (
|
||||
statusSucceeded = "succeeded"
|
||||
statusFailed = "failed"
|
||||
statusTimeout = "timeout"
|
||||
statusMarked = "marked"
|
||||
statusBusy = "busy"
|
||||
statusRefused = "refused"
|
||||
)
|
||||
|
||||
// Store is the storage the webhook needs. *store.Store satisfies it.
|
||||
type Store interface {
|
||||
ServicesByRepo(ctx context.Context, repo string) ([]model.Service, error)
|
||||
SetAvailable(ctx context.Context, id int64, a model.Available) error
|
||||
GetJob(ctx context.Context, id int64) (model.Job, error)
|
||||
}
|
||||
|
||||
// Jobs starts updates. *jobs.Service satisfies it.
|
||||
type Jobs interface {
|
||||
Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error)
|
||||
}
|
||||
|
||||
// Broker delivers change signals. *events.Broker satisfies it.
|
||||
type Broker interface {
|
||||
Subscribe() (<-chan events.Topic, func())
|
||||
Publish(t events.Topic)
|
||||
}
|
||||
|
||||
// Config holds the webhook settings.
|
||||
type Config struct {
|
||||
Token string // bearer token; empty disables the webhook
|
||||
Timeout time.Duration // bounds the whole wait of one request
|
||||
PublicURL string // base of job_url, without trailing slash
|
||||
}
|
||||
|
||||
// Handler serves POST /update.
|
||||
type Handler struct {
|
||||
store Store
|
||||
jobs Jobs
|
||||
broker Broker
|
||||
cfg Config
|
||||
log *slog.Logger
|
||||
|
||||
now func() time.Time
|
||||
pollEvery time.Duration
|
||||
}
|
||||
|
||||
// New returns the handler for POST /update. It does its own bearer-token
|
||||
// check, so mount it outside the OIDC login.
|
||||
func New(store Store, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) *Handler {
|
||||
return &Handler{store: store, jobs: jobs, broker: broker, cfg: cfg, log: log,
|
||||
now: time.Now, pollEvery: defaultPollEvery}
|
||||
}
|
||||
|
||||
type request struct {
|
||||
Image string `json:"image"`
|
||||
Tag string `json:"tag"`
|
||||
}
|
||||
|
||||
type result struct {
|
||||
ServiceID int64 `json:"service_id"`
|
||||
Service string `json:"service"`
|
||||
Kind string `json:"kind"`
|
||||
Mode string `json:"mode"`
|
||||
Status string `json:"status"`
|
||||
JobID int64 `json:"job_id,omitempty"`
|
||||
JobURL string `json:"job_url,omitempty"`
|
||||
ToRef string `json:"to_ref,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
DurationMS int64 `json:"duration_ms"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
RequestID string `json:"request_id"`
|
||||
Image string `json:"image"`
|
||||
Tag string `json:"tag,omitempty"`
|
||||
Matched int `json:"matched"`
|
||||
Results []result `json:"results"`
|
||||
}
|
||||
|
||||
// ServeHTTP handles one webhook call. See the README for the contract.
|
||||
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
started := h.now()
|
||||
requestID := r.Header.Get("X-Request-ID")
|
||||
if requestID == "" {
|
||||
requestID = newRequestID()
|
||||
}
|
||||
w.Header().Set("X-Request-ID", requestID)
|
||||
|
||||
if h.cfg.Token == "" {
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "webhook disabled: UPDATER_API_KEY not set"})
|
||||
return
|
||||
}
|
||||
if !h.authorized(r) {
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
||||
return
|
||||
}
|
||||
|
||||
var req request
|
||||
r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON body"})
|
||||
return
|
||||
}
|
||||
if req.Image == "" {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "image is required"})
|
||||
return
|
||||
}
|
||||
if req.Tag != "" && !tagPattern.MatchString(req.Tag) {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid tag"})
|
||||
return
|
||||
}
|
||||
|
||||
log := logging.FromContext(logging.WithRequestID(r.Context(), requestID), h.log)
|
||||
repo := imageref.Parse(req.Image).Repo
|
||||
// A client that disconnects must not abort marking or starting: that
|
||||
// could leave a queued job no agent knows about.
|
||||
work := context.WithoutCancel(r.Context())
|
||||
|
||||
services, err := h.store.ServicesByRepo(work, repo)
|
||||
if err != nil {
|
||||
log.Error("webhook_lookup_failed", "image", req.Image, "error", err)
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal error"})
|
||||
return
|
||||
}
|
||||
if len(services) == 0 {
|
||||
writeJSON(w, http.StatusNotFound, map[string]string{
|
||||
"error": "no service uses image " + repo, "request_id": requestID})
|
||||
return
|
||||
}
|
||||
|
||||
// Subscribe before any job exists so no completion signal is missed.
|
||||
sub, cancel := h.broker.Subscribe()
|
||||
defer cancel()
|
||||
|
||||
resp := response{RequestID: requestID, Image: req.Image, Tag: req.Tag, Matched: len(services)}
|
||||
pending := map[int]int64{} // index into resp.Results -> job ID
|
||||
for _, svc := range services {
|
||||
res, job := h.handle(work, svc, req.Tag, started)
|
||||
resp.Results = append(resp.Results, res)
|
||||
if job != nil {
|
||||
pending[len(resp.Results)-1] = job.ID
|
||||
}
|
||||
}
|
||||
h.broker.Publish(events.Services)
|
||||
|
||||
h.wait(r.Context(), sub, &resp, pending, started, log)
|
||||
|
||||
status := http.StatusOK
|
||||
counts := map[string]int{}
|
||||
for _, res := range resp.Results {
|
||||
counts[res.Status]++
|
||||
if res.Status == statusFailed || res.Status == statusTimeout {
|
||||
status = http.StatusInternalServerError
|
||||
}
|
||||
}
|
||||
log.Info("webhook_request", "image", req.Image, "tag", req.Tag, "matched", resp.Matched,
|
||||
"succeeded", counts[statusSucceeded], "failed", counts[statusFailed], "timeout", counts[statusTimeout],
|
||||
"marked", counts[statusMarked], "busy", counts[statusBusy], "refused", counts[statusRefused])
|
||||
writeJSON(w, status, resp)
|
||||
}
|
||||
|
||||
// handle records the available ref on svc and, in auto mode, starts an update.
|
||||
// It returns the result and, if the job is still active, that job.
|
||||
func (h *Handler) handle(ctx context.Context, svc model.Service, tag string, started time.Time) (result, *model.Job) {
|
||||
res := result{ServiceID: svc.ID, Service: svc.DisplayName(), Kind: string(svc.Kind), Mode: svc.UpdateMode}
|
||||
|
||||
// The ref is built from the service's own repo, never from the request
|
||||
// string, so a request cannot point a service at another registry.
|
||||
var ref string
|
||||
if tag != "" {
|
||||
ref = svc.ImageRepo + ":" + tag
|
||||
} else {
|
||||
ref = jobs.FollowedRef(svc.ActualRef)
|
||||
}
|
||||
res.ToRef = ref
|
||||
if ref != "" {
|
||||
err := h.store.SetAvailable(ctx, svc.ID, model.Available{Ref: ref, Source: model.SourceWebhook, At: h.now()})
|
||||
if err != nil {
|
||||
h.log.Error("webhook_set_available_failed", "service_id", svc.ID, "error", err)
|
||||
return h.finish(res, statusFailed, "could not record available version: "+err.Error(), started), nil
|
||||
}
|
||||
}
|
||||
if svc.UpdateMode != model.ModeAuto {
|
||||
return h.finish(res, statusMarked, "", started), nil
|
||||
}
|
||||
|
||||
job, err := h.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerWebhook, Ref: ref})
|
||||
var notUpdatable *model.NotUpdatableError
|
||||
switch {
|
||||
case errors.As(err, ¬Updatable):
|
||||
return h.finish(res, statusRefused, notUpdatable.Reason, started), nil
|
||||
case errors.Is(err, model.ErrJobActive):
|
||||
h.setJob(&res, job)
|
||||
return h.finish(res, statusBusy, err.Error(), started), nil
|
||||
case err != nil:
|
||||
h.log.Error("webhook_start_failed", "service_id", svc.ID, "error", err)
|
||||
return h.finish(res, statusFailed, err.Error(), started), nil
|
||||
}
|
||||
h.setJob(&res, job)
|
||||
if job.Active() {
|
||||
return res, &job
|
||||
}
|
||||
h.resolveJob(&res, job, started)
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (h *Handler) setJob(res *result, job model.Job) {
|
||||
res.JobID = job.ID
|
||||
res.JobURL = h.cfg.PublicURL + "/jobs/" + strconv.FormatInt(job.ID, 10)
|
||||
}
|
||||
|
||||
func (h *Handler) finish(res result, status, msg string, started time.Time) result {
|
||||
res.Status, res.Error = status, msg
|
||||
res.DurationMS = h.now().Sub(started).Milliseconds()
|
||||
return res
|
||||
}
|
||||
|
||||
// resolveJob sets the result from a job that is no longer active.
|
||||
func (h *Handler) resolveJob(res *result, job model.Job, started time.Time) {
|
||||
if job.Status == model.JobSucceeded {
|
||||
*res = h.finish(*res, statusSucceeded, "", started)
|
||||
return
|
||||
}
|
||||
*res = h.finish(*res, statusFailed, job.Error, started)
|
||||
}
|
||||
|
||||
// wait blocks until every pending job has finished, the timeout passes or
|
||||
// the client goes away. Jobs still pending then get status "timeout"; the
|
||||
// master never cancels or fails them.
|
||||
func (h *Handler) wait(parent context.Context, sub <-chan events.Topic, resp *response,
|
||||
pending map[int]int64, started time.Time, log *slog.Logger) {
|
||||
if len(pending) == 0 {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(parent, h.cfg.Timeout)
|
||||
defer cancel()
|
||||
ticker := time.NewTicker(h.pollEvery)
|
||||
defer ticker.Stop()
|
||||
|
||||
for len(pending) > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
for i := range pending {
|
||||
resp.Results[i] = h.finish(resp.Results[i], statusTimeout, "still running when the request timed out", started)
|
||||
}
|
||||
return
|
||||
case t := <-sub:
|
||||
if t != events.Jobs {
|
||||
continue
|
||||
}
|
||||
case <-ticker.C:
|
||||
}
|
||||
for i, id := range pending {
|
||||
job, err := h.store.GetJob(ctx, id)
|
||||
if err != nil {
|
||||
log.Warn("webhook_get_job_failed", "job_id", id, "error", err)
|
||||
continue
|
||||
}
|
||||
if !job.Active() {
|
||||
h.resolveJob(&resp.Results[i], job, started)
|
||||
delete(pending, i)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// authorized reports whether the request carries the configured bearer token.
|
||||
// Both sides are hashed so the comparison takes the same time whatever the
|
||||
// length of the supplied token.
|
||||
func (h *Handler) authorized(r *http.Request) bool {
|
||||
const prefix = "Bearer "
|
||||
got, ok := strings.CutPrefix(r.Header.Get("Authorization"), prefix)
|
||||
if !ok || got == "" {
|
||||
return false
|
||||
}
|
||||
a, b := sha256.Sum256([]byte(got)), sha256.Sum256([]byte(h.cfg.Token))
|
||||
return subtle.ConstantTimeCompare(a[:], b[:]) == 1
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, body any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(body)
|
||||
}
|
||||
|
||||
// newRequestID returns a random 128-bit hex ID.
|
||||
func newRequestID() string {
|
||||
var b [16]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
return hex.EncodeToString(b[:])
|
||||
}
|
||||
@@ -0,0 +1,475 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/master/events"
|
||||
"github.com/shcizo/package-updater/internal/master/jobs"
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
const token = "s3cret"
|
||||
|
||||
type availCall struct {
|
||||
id int64
|
||||
a model.Available
|
||||
}
|
||||
|
||||
type fakeStore struct {
|
||||
mu sync.Mutex
|
||||
services []model.Service // returned for any repo that equals their ImageRepo
|
||||
jobs map[int64]model.Job
|
||||
avail []availCall
|
||||
repos []string
|
||||
}
|
||||
|
||||
func (f *fakeStore) ServicesByRepo(_ context.Context, repo string) ([]model.Service, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.repos = append(f.repos, repo)
|
||||
var out []model.Service
|
||||
for _, s := range f.services {
|
||||
if s.ImageRepo == repo {
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) SetAvailable(_ context.Context, id int64, a model.Available) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.avail = append(f.avail, availCall{id, a})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.jobs[id], nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) setJob(j model.Job) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.jobs[j.ID] = j
|
||||
}
|
||||
|
||||
type startCall struct {
|
||||
svc model.Service
|
||||
req jobs.StartRequest
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
// fakeJobs hands out job IDs and applies per-service behaviour.
|
||||
type fakeJobs struct {
|
||||
mu sync.Mutex
|
||||
store *fakeStore
|
||||
calls []startCall
|
||||
errFor map[int64]error // returned together with errJob[id] if set
|
||||
jobFor map[int64]model.Job // status of the job returned by Start
|
||||
next int64
|
||||
}
|
||||
|
||||
func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.calls = append(f.calls, startCall{svc, req, ctx})
|
||||
if err, ok := f.errFor[svc.ID]; ok {
|
||||
return f.jobFor[svc.ID], err
|
||||
}
|
||||
f.next++
|
||||
j := model.Job{ID: f.next, ServiceID: svc.ID, Status: model.JobQueued, Trigger: req.Trigger, ToRef: req.Ref}
|
||||
if o, ok := f.jobFor[svc.ID]; ok {
|
||||
o.ID = j.ID
|
||||
j = o
|
||||
}
|
||||
f.store.mu.Lock()
|
||||
f.store.jobs[j.ID] = j
|
||||
f.store.mu.Unlock()
|
||||
return j, nil
|
||||
}
|
||||
|
||||
func (f *fakeJobs) startCalls() []startCall {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return append([]startCall(nil), f.calls...)
|
||||
}
|
||||
|
||||
type env struct {
|
||||
store *fakeStore
|
||||
jobs *fakeJobs
|
||||
broker *events.Broker
|
||||
h *Handler
|
||||
}
|
||||
|
||||
func newEnv(t *testing.T, services ...model.Service) *env {
|
||||
t.Helper()
|
||||
st := &fakeStore{services: services, jobs: map[int64]model.Job{}}
|
||||
fj := &fakeJobs{store: st, errFor: map[int64]error{}, jobFor: map[int64]model.Job{}}
|
||||
b := events.NewBroker()
|
||||
h := New(st, fj, b, Config{Token: token, Timeout: 2 * time.Second, PublicURL: "https://updater.example.com"},
|
||||
slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
h.now = func() time.Time { return time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) }
|
||||
return &env{store: st, jobs: fj, broker: b, h: h}
|
||||
}
|
||||
|
||||
func (e *env) post(body string, headers ...string) *httptest.ResponseRecorder {
|
||||
return e.postCtx(context.Background(), body, headers...)
|
||||
}
|
||||
|
||||
func (e *env) postCtx(ctx context.Context, body string, headers ...string) *httptest.ResponseRecorder {
|
||||
req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(body)).WithContext(ctx)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
for i := 0; i+1 < len(headers); i += 2 {
|
||||
req.Header.Set(headers[i], headers[i+1])
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
e.h.ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
func decode(t *testing.T, rec *httptest.ResponseRecorder) response {
|
||||
t.Helper()
|
||||
var r response
|
||||
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &r), rec.Body.String())
|
||||
return r
|
||||
}
|
||||
|
||||
func compose(id int64, mode, actualRef string) model.Service {
|
||||
return model.Service{ID: id, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web" + string(rune('a'+id)),
|
||||
ImageRepo: "gitea.shcizo.se/shcizo/slusk", ActualRef: actualRef, UpdateMode: mode}
|
||||
}
|
||||
|
||||
const slusk = "gitea.shcizo.se/shcizo/slusk"
|
||||
|
||||
func TestAuth(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
serverKey string
|
||||
header string
|
||||
wantStatus int
|
||||
wantBody string
|
||||
}{
|
||||
{"missing header", token, "", 401, `{"error":"unauthorized"}`},
|
||||
{"wrong scheme", token, "Basic " + token, 401, `{"error":"unauthorized"}`},
|
||||
{"wrong token", token, "Bearer nope", 401, `{"error":"unauthorized"}`},
|
||||
{"empty bearer", token, "Bearer ", 401, `{"error":"unauthorized"}`},
|
||||
{"key unset", "", "Bearer " + token, 503, `{"error":"webhook disabled: UPDATER_API_KEY not set"}`},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeManual, slusk+":1"))
|
||||
e.h.cfg.Token = tt.serverKey
|
||||
req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"`+slusk+`"}`))
|
||||
if tt.header != "" {
|
||||
req.Header.Set("Authorization", tt.header)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
e.h.ServeHTTP(rec, req)
|
||||
assert.Equal(t, tt.wantStatus, rec.Code)
|
||||
assert.JSONEq(t, tt.wantBody, rec.Body.String())
|
||||
assert.Empty(t, e.store.repos, "nothing may be looked up before auth")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBadRequest(t *testing.T) {
|
||||
tests := []struct {
|
||||
name, body, wantError string
|
||||
}{
|
||||
{"invalid JSON", `{`, "invalid JSON body"},
|
||||
{"empty body", ``, "invalid JSON body"},
|
||||
{"empty image", `{"image":""}`, "image is required"},
|
||||
{"missing image", `{"tag":"v1"}`, "image is required"},
|
||||
{"oversized", `{"image":"` + strings.Repeat("a", 70<<10) + `"}`, "invalid JSON body"},
|
||||
{"tag with digest", `{"image":"` + slusk + `","tag":"v1@sha256:abc"}`, "invalid tag"},
|
||||
{"tag with space", `{"image":"` + slusk + `","tag":"v 1"}`, "invalid tag"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1"))
|
||||
rec := e.post(tt.body)
|
||||
assert.Equal(t, 400, rec.Code)
|
||||
assert.JSONEq(t, `{"error":"`+tt.wantError+`"}`, rec.Body.String())
|
||||
assert.Empty(t, e.jobs.startCalls())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoMatch(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1"))
|
||||
for _, image := range []string{"gitea.shcizo.se/shcizo/other", "gitea.shcizo.se/shcizo/Slusk"} {
|
||||
rec := e.post(`{"image":"` + image + `"}`)
|
||||
assert.Equal(t, 404, rec.Code, image)
|
||||
assert.Contains(t, rec.Body.String(), "no service uses image "+image)
|
||||
}
|
||||
assert.Empty(t, e.jobs.startCalls())
|
||||
assert.Empty(t, e.store.avail)
|
||||
}
|
||||
|
||||
func TestImageIsNormalisedBeforeMatching(t *testing.T) {
|
||||
svc := compose(1, model.ModeManual, slusk+":1")
|
||||
local := compose(2, model.ModeManual, "localhost:5000/foo:v1")
|
||||
local.ImageRepo = "localhost:5000/foo"
|
||||
for _, tt := range []struct{ image, wantRepo string }{
|
||||
{slusk, slusk},
|
||||
{slusk + ":2.0", slusk},
|
||||
{slusk + "@sha256:abc", slusk},
|
||||
{slusk + ":2.0@sha256:abc", slusk},
|
||||
{"localhost:5000/foo:v1", "localhost:5000/foo"},
|
||||
} {
|
||||
e := newEnv(t, svc, local)
|
||||
rec := e.post(`{"image":"` + tt.image + `"}`)
|
||||
assert.Equal(t, 200, rec.Code, tt.image)
|
||||
assert.Equal(t, []string{tt.wantRepo}, e.store.repos, tt.image)
|
||||
assert.Equal(t, 1, decode(t, rec).Matched, tt.image)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManualServiceIsMarkedNotUpdated(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeManual, slusk+":1.4.2@sha256:old"))
|
||||
sub, cancel := e.broker.Subscribe()
|
||||
defer cancel()
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `:ignored-in-ref","tag":"1.5.0"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
assert.Empty(t, e.jobs.startCalls())
|
||||
require.Len(t, e.store.avail, 1)
|
||||
assert.Equal(t, int64(1), e.store.avail[0].id)
|
||||
assert.Equal(t, model.Available{Ref: slusk + ":1.5.0", Source: model.SourceWebhook, At: e.h.now()}, e.store.avail[0].a)
|
||||
resp := decode(t, rec)
|
||||
assert.Equal(t, 1, resp.Matched)
|
||||
require.Len(t, resp.Results, 1)
|
||||
r := resp.Results[0]
|
||||
assert.Equal(t, statusMarked, r.Status)
|
||||
assert.Equal(t, "slusk/webb", r.Service)
|
||||
assert.Equal(t, "compose", r.Kind)
|
||||
assert.Equal(t, "manual", r.Mode)
|
||||
assert.Equal(t, slusk+":1.5.0", r.ToRef)
|
||||
assert.Zero(t, r.JobID)
|
||||
assert.NotEmpty(t, resp.RequestID)
|
||||
assert.Equal(t, resp.RequestID, rec.Header().Get("X-Request-ID"))
|
||||
assert.Equal(t, events.Services, <-sub)
|
||||
}
|
||||
|
||||
func TestRequestIDIsEchoed(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeManual, slusk+":1"))
|
||||
rec := e.post(`{"image":"`+slusk+`"}`, "X-Request-ID", "ci-123")
|
||||
assert.Equal(t, "ci-123", decode(t, rec).RequestID)
|
||||
assert.Equal(t, "ci-123", rec.Header().Get("X-Request-ID"))
|
||||
}
|
||||
|
||||
func TestAvailableRefWithoutTag(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
actualRef string
|
||||
wantRef string // "" means SetAvailable must not be called
|
||||
}{
|
||||
{"followed tag", slusk + ":1.4.2@sha256:old", slusk + ":1.4.2"},
|
||||
{"bare image follows latest", slusk, slusk + ":latest"},
|
||||
{"digest only", slusk + "@sha256:old", ""},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeManual, tt.actualRef))
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
if tt.wantRef == "" {
|
||||
assert.Empty(t, e.store.avail)
|
||||
return
|
||||
}
|
||||
require.Len(t, e.store.avail, 1)
|
||||
assert.Equal(t, tt.wantRef, e.store.avail[0].a.Ref)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAutoDigestOnlyServiceWithoutTagIsRefused(t *testing.T) {
|
||||
svc := compose(1, model.ModeAuto, slusk+"@sha256:old")
|
||||
e := newEnv(t, svc)
|
||||
e.jobs.errFor[1] = &model.NotUpdatableError{Reason: model.ReasonNoTag}
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
assert.Empty(t, e.store.avail)
|
||||
calls := e.jobs.startCalls()
|
||||
require.Len(t, calls, 1)
|
||||
assert.Empty(t, calls[0].req.Ref, "no request tag and nothing followed: the job layer decides")
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusRefused, r.Status)
|
||||
assert.Equal(t, model.ReasonNoTag, r.Error)
|
||||
}
|
||||
|
||||
func TestAutoJobTargetIsBuiltFromServiceRepo(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2@sha256:old"))
|
||||
e.jobs.jobFor[1] = model.Job{Status: model.JobSucceeded}
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `:other","tag":"2.0.0"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
calls := e.jobs.startCalls()
|
||||
require.Len(t, calls, 1)
|
||||
assert.Equal(t, slusk+":2.0.0", calls[0].req.Ref)
|
||||
assert.Equal(t, model.TriggerWebhook, calls[0].req.Trigger)
|
||||
assert.Empty(t, calls[0].req.RequestedBy)
|
||||
require.Len(t, e.store.avail, 1)
|
||||
assert.Equal(t, slusk+":2.0.0", e.store.avail[0].a.Ref)
|
||||
}
|
||||
|
||||
func TestAutoJobSucceeded(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
|
||||
go func() {
|
||||
assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond)
|
||||
e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded})
|
||||
e.broker.Publish(events.JobTopic(1)) // progress topic is ignored
|
||||
e.broker.Publish(events.Jobs)
|
||||
}()
|
||||
rec := e.post(`{"image":"` + slusk + `","tag":"1.5.0"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusSucceeded, r.Status)
|
||||
assert.Equal(t, int64(1), r.JobID)
|
||||
assert.Equal(t, "https://updater.example.com/jobs/1", r.JobURL)
|
||||
assert.Equal(t, "auto", r.Mode)
|
||||
}
|
||||
|
||||
func TestAutoJobFailedIs500(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
go func() {
|
||||
assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond)
|
||||
e.store.setJob(model.Job{ID: 1, Status: model.JobFailed, Error: "pull access denied"})
|
||||
e.broker.Publish(events.Jobs)
|
||||
}()
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
|
||||
assert.Equal(t, 500, rec.Code)
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusFailed, r.Status)
|
||||
assert.Equal(t, "pull access denied", r.Error)
|
||||
}
|
||||
|
||||
func TestJobFailedAtStartIsNotWaitedFor(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
e.jobs.jobFor[1] = model.Job{Status: model.JobFailed, Error: "agent offline"}
|
||||
e.h.cfg.Timeout = time.Minute
|
||||
|
||||
begin := time.Now()
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
|
||||
assert.Less(t, time.Since(begin), 5*time.Second)
|
||||
assert.Equal(t, 500, rec.Code)
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusFailed, r.Status)
|
||||
assert.Equal(t, "agent offline", r.Error)
|
||||
assert.NotZero(t, r.JobID)
|
||||
}
|
||||
|
||||
func TestTimeout(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
e.h.cfg.Timeout = 50 * time.Millisecond
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
|
||||
assert.Equal(t, 500, rec.Code)
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusTimeout, r.Status)
|
||||
assert.Equal(t, int64(1), r.JobID)
|
||||
assert.Equal(t, model.JobQueued, e.store.jobs[1].Status, "the master must not touch a timed-out job")
|
||||
}
|
||||
|
||||
func TestDroppedSignalIsRecoveredByPolling(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
e.h.pollEvery = 5 * time.Millisecond
|
||||
go func() {
|
||||
assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond)
|
||||
e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded}) // no Publish
|
||||
}()
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
assert.Equal(t, statusSucceeded, decode(t, rec).Results[0].Status)
|
||||
}
|
||||
|
||||
func TestBusyIsNotAFailure(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
e.jobs.errFor[1] = model.ErrJobActive
|
||||
e.jobs.jobFor[1] = model.Job{ID: 42, Status: model.JobRunning}
|
||||
e.h.cfg.Timeout = time.Minute
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusBusy, r.Status)
|
||||
assert.Equal(t, int64(42), r.JobID)
|
||||
assert.Equal(t, "https://updater.example.com/jobs/42", r.JobURL)
|
||||
assert.NotEmpty(t, e.store.avail, "the available ref is still recorded")
|
||||
}
|
||||
|
||||
func TestRefusedIsNotAFailure(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
e.jobs.errFor[1] = &model.NotUpdatableError{Reason: model.ReasonSelf}
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `","tag":"2"}`)
|
||||
|
||||
assert.Equal(t, 200, rec.Code)
|
||||
r := decode(t, rec).Results[0]
|
||||
assert.Equal(t, statusRefused, r.Status)
|
||||
assert.Equal(t, model.ReasonSelf, r.Error)
|
||||
assert.Zero(t, r.JobID)
|
||||
assert.NotEmpty(t, e.store.avail)
|
||||
}
|
||||
|
||||
func TestMixedAutoAndManual(t *testing.T) {
|
||||
e := newEnv(t,
|
||||
compose(1, model.ModeManual, slusk+":1"),
|
||||
compose(2, model.ModeAuto, slusk+":1"),
|
||||
compose(3, model.ModeAuto, slusk+":1"))
|
||||
e.jobs.jobFor[2] = model.Job{Status: model.JobSucceeded}
|
||||
e.jobs.jobFor[3] = model.Job{Status: model.JobFailed, Error: "boom"}
|
||||
|
||||
rec := e.post(`{"image":"` + slusk + `","tag":"2"}`)
|
||||
|
||||
assert.Equal(t, 500, rec.Code)
|
||||
resp := decode(t, rec)
|
||||
assert.Equal(t, 3, resp.Matched)
|
||||
got := []string{resp.Results[0].Status, resp.Results[1].Status, resp.Results[2].Status}
|
||||
assert.Equal(t, []string{statusMarked, statusSucceeded, statusFailed}, got)
|
||||
assert.Len(t, e.store.avail, 3)
|
||||
}
|
||||
|
||||
func TestClientCancelStopsWaitingButNotStarting(t *testing.T) {
|
||||
e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2"))
|
||||
e.h.cfg.Timeout = time.Minute
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel() // the client is already gone when the handler runs
|
||||
|
||||
begin := time.Now()
|
||||
rec := e.postCtx(ctx, `{"image":"`+slusk+`"}`)
|
||||
|
||||
assert.Less(t, time.Since(begin), 5*time.Second)
|
||||
assert.Equal(t, statusTimeout, decode(t, rec).Results[0].Status)
|
||||
calls := e.jobs.startCalls()
|
||||
require.Len(t, calls, 1, "the job must still be started")
|
||||
assert.NoError(t, calls[0].ctx.Err(), "Start must run on a context the client cannot cancel")
|
||||
}
|
||||
Reference in New Issue
Block a user