998 lines
35 KiB
Go
998 lines
35 KiB
Go
package updateagent
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/root-kdw12/hetzner-api-dyndns-records-webinterface/internal/buildinfo"
|
|
"github.com/root-kdw12/hetzner-api-dyndns-records-webinterface/internal/updateprotocol"
|
|
)
|
|
|
|
const defaultImageRepository = "git.j-mahler.de/admin/domain-control-webinterface-prod"
|
|
|
|
const (
|
|
labelImageVersion = "org.opencontainers.image.version"
|
|
labelMinimumVersion = "de.j-mahler.domain-control.minimum-update-version"
|
|
labelMinimumAgent = "de.j-mahler.domain-control.minimum-agent-protocol"
|
|
labelRuntimeContract = "de.j-mahler.domain-control.runtime-contract"
|
|
labelDatabaseContract = "de.j-mahler.domain-control.database-contract"
|
|
labelSchemaVersion = "de.j-mahler.domain-control.schema-version"
|
|
runtimeEntrypoint = "/usr/local/bin/domain-control"
|
|
runtimeCommand = "serve"
|
|
runtimeUser = "dyndns:dyndns"
|
|
runtimeWorkingDirectory = "/data"
|
|
)
|
|
|
|
type DockerRunner struct {
|
|
docker *dockerClient
|
|
imageRepository string
|
|
targetLabel string
|
|
logger *slog.Logger
|
|
}
|
|
|
|
type updateError struct {
|
|
message string
|
|
err error
|
|
}
|
|
|
|
func (e *updateError) Error() string { return e.err.Error() }
|
|
func (e *updateError) Unwrap() error { return e.err }
|
|
func (e *updateError) PublicMessage() string { return e.message }
|
|
|
|
func NewDockerRunner(socket, imageRepository, targetLabel string, logger *slog.Logger) *DockerRunner {
|
|
if imageRepository == "" {
|
|
imageRepository = defaultImageRepository
|
|
}
|
|
return &DockerRunner{docker: newDockerClient(socket), imageRepository: imageRepository, targetLabel: targetLabel, logger: logger}
|
|
}
|
|
|
|
func (r *DockerRunner) Update(ctx context.Context, request updateprotocol.Request, progress func(phase, message string)) (resultErr error) {
|
|
if !updateprotocol.ValidVersion(request.Version) {
|
|
return errors.New("invalid target version")
|
|
}
|
|
if err := waitForDockerAPI(ctx, 30*time.Second, time.Second, r.docker.negotiate); err != nil {
|
|
return &updateError{message: "Containerverwaltung ist nicht erreichbar; der Dienst läuft unverändert weiter", err: err}
|
|
}
|
|
target, err := r.docker.findTarget(ctx, r.targetLabel)
|
|
if err != nil {
|
|
return &updateError{message: "Laufender Anwendungscontainer wurde nicht eindeutig gefunden", err: err}
|
|
}
|
|
inspected, err := r.docker.inspectContainer(ctx, target.ID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if inspected.State.Health == nil {
|
|
return &updateError{message: "Die laufende Instanz ist nicht healthy; das Update wurde sicherheitshalber nicht gestartet", err: errors.New("current container is not healthy")}
|
|
}
|
|
switch inspected.State.Health.Status {
|
|
case "healthy":
|
|
case "starting":
|
|
progress(updateprotocol.PhaseQueued, "Laufender Container beendet noch seinen Start-Healthcheck")
|
|
inspected, err = r.docker.waitCurrentHealthy(ctx, target.ID, inspected.Image, 100*time.Second, time.Second)
|
|
if err != nil {
|
|
return &updateError{message: "Die laufende Instanz wurde nach dem Host-Start nicht rechtzeitig healthy; das Update wurde nicht gestartet", err: err}
|
|
}
|
|
default:
|
|
return &updateError{message: "Die laufende Instanz ist nicht healthy; das Update wurde sicherheitshalber nicht gestartet", err: errors.New("current container is not healthy")}
|
|
}
|
|
currentVersion, err := resolveCurrentVersion(request.CurrentVersion, inspected.Config, r.imageRepository)
|
|
if err != nil {
|
|
return &updateError{message: "Die laufende Version konnte nicht sicher bestimmt werden; bitte einmal mit Docker Compose aktualisieren", err: err}
|
|
}
|
|
if comparison, compareErr := updateprotocol.CompareVersions(currentVersion, request.Version); compareErr != nil || comparison >= 0 {
|
|
return &updateError{message: "Die Zielversion ist nicht neuer als die laufende Version", err: fmt.Errorf("invalid version transition %s to %s", currentVersion, request.Version)}
|
|
}
|
|
currentImage, err := r.docker.inspectImage(ctx, inspected.Image)
|
|
if err != nil {
|
|
return &updateError{message: "Die laufende Container-Basis konnte nicht geprüft werden", err: err}
|
|
}
|
|
|
|
targetReference := r.imageRepository + ":" + request.Version
|
|
progress(updateprotocol.PhasePulling, "Container-Image wird geladen und geprüft")
|
|
if err := r.docker.pullImage(ctx, r.imageRepository, request.Version, request.RegistryUsername, request.RegistryToken); err != nil {
|
|
if errors.Is(err, errRegistryAuthentication) {
|
|
return &updateError{message: "Container-Image ist nicht öffentlich zugänglich oder der Zugang wurde abgelehnt. Bitte einen gültigen Gitea-Zugang mit Paket-Leserechten hinterlegen; der Dienst läuft unverändert weiter", err: err}
|
|
}
|
|
return &updateError{message: "Image konnte nicht geladen werden; der Dienst läuft unverändert weiter", err: err}
|
|
}
|
|
targetImage, err := r.docker.inspectImage(ctx, targetReference)
|
|
if err != nil {
|
|
return &updateError{message: "Das geladene Image konnte nicht geprüft werden; der Dienst läuft unverändert weiter", err: err}
|
|
}
|
|
if err := validateUpdateCompatibility(currentVersion, request.Version, currentImage, targetImage); err != nil {
|
|
return &updateError{message: "Dieses Release unterstützt kein sicheres Direktupdate von der laufenden Version; bitte Docker Compose verwenden", err: err}
|
|
}
|
|
replacementConfig, err := buildReplacementConfig(inspected, currentImage.Config, targetImage.Config, targetReference)
|
|
if err != nil {
|
|
return &updateError{message: "Die neue Container-Konfiguration konnte nicht sicher vorbereitet werden", err: err}
|
|
}
|
|
originalName := strings.TrimPrefix(inspected.Name, "/")
|
|
if originalName == "" {
|
|
return errors.New("target container has no name")
|
|
}
|
|
backupName := originalName + "-rollback-" + fmt.Sprint(time.Now().UTC().UnixNano())
|
|
var oldStopped, oldRenamed bool
|
|
var newID string
|
|
defer func() {
|
|
if resultErr == nil || (!oldStopped && !oldRenamed && newID == "") {
|
|
return
|
|
}
|
|
rollbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
if rollbackErr := r.rollback(rollbackCtx, target.ID, originalName, inspected.Image, oldStopped, oldRenamed, newID); rollbackErr != nil {
|
|
r.logger.Error("software update rollback failed", "error", rollbackErr)
|
|
resultErr = &updateError{message: "Update und automatische Wiederherstellung sind fehlgeschlagen; Containerstatus am Host prüfen", err: fmt.Errorf("%w; rollback: %v", resultErr, rollbackErr)}
|
|
}
|
|
}()
|
|
|
|
progress(updateprotocol.PhaseRestarting, "Anwendung wird kontrolliert neu gestartet")
|
|
// A lost Docker response does not mean the stop was not executed. Arm
|
|
// rollback first so a stopped application is restarted on this error path.
|
|
oldStopped = true
|
|
if err := r.docker.stopContainer(ctx, target.ID, 20); err != nil {
|
|
return err
|
|
}
|
|
if err := r.docker.renameContainer(ctx, target.ID, backupName); err != nil {
|
|
return err
|
|
}
|
|
oldRenamed = true
|
|
newID, err = r.docker.createReplacement(ctx, inspected, originalName, replacementConfig)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := r.docker.startContainer(ctx, newID); err != nil {
|
|
return err
|
|
}
|
|
progress(updateprotocol.PhaseVerifying, "Neuer Container wird geprüft")
|
|
if err := r.docker.waitHealthy(ctx, newID, targetImage.ID, request.Version, 100*time.Second); err != nil {
|
|
return err
|
|
}
|
|
if err := r.docker.removeContainer(ctx, target.ID, false); err != nil {
|
|
r.logger.Warn("old update rollback container could not be removed", "container", target.ID, "error", err)
|
|
}
|
|
oldStopped, oldRenamed, newID = false, false, ""
|
|
return nil
|
|
}
|
|
|
|
func (r *DockerRunner) rollback(ctx context.Context, oldID, originalName, oldImageID string, oldStopped, oldRenamed bool, newID string) error {
|
|
var failures []string
|
|
if newID != "" {
|
|
if err := r.docker.stopContainer(ctx, newID, 5); err != nil && !isDockerNotModified(err) {
|
|
failures = append(failures, "stop replacement: "+err.Error())
|
|
}
|
|
if err := r.docker.removeContainer(ctx, newID, true); err != nil {
|
|
failures = append(failures, "remove replacement: "+err.Error())
|
|
}
|
|
}
|
|
if oldRenamed {
|
|
if err := r.docker.renameContainer(ctx, oldID, originalName); err != nil {
|
|
failures = append(failures, "restore name: "+err.Error())
|
|
}
|
|
}
|
|
if oldStopped {
|
|
if err := r.docker.startContainer(ctx, oldID); err != nil && !isDockerNotModified(err) {
|
|
failures = append(failures, "restart previous container: "+err.Error())
|
|
} else if err := r.docker.waitHealthy(ctx, oldID, oldImageID, "", 60*time.Second); err != nil {
|
|
failures = append(failures, "verify previous container: "+err.Error())
|
|
}
|
|
}
|
|
if len(failures) > 0 {
|
|
return errors.New(strings.Join(failures, "; "))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type dockerClient struct {
|
|
socket string
|
|
client *http.Client
|
|
apiPrefix string
|
|
}
|
|
|
|
type containerSummary struct {
|
|
ID string `json:"Id"`
|
|
Names []string `json:"Names"`
|
|
State string `json:"State"`
|
|
}
|
|
|
|
type containerInspect struct {
|
|
ID string `json:"Id"`
|
|
Image string `json:"Image"`
|
|
Name string `json:"Name"`
|
|
Config json.RawMessage `json:"Config"`
|
|
HostConfig json.RawMessage `json:"HostConfig"`
|
|
NetworkSettings struct {
|
|
Networks map[string]endpointSettings `json:"Networks"`
|
|
} `json:"NetworkSettings"`
|
|
State containerState `json:"State"`
|
|
}
|
|
|
|
type imageInspect struct {
|
|
ID string `json:"Id"`
|
|
Config json.RawMessage `json:"Config"`
|
|
}
|
|
|
|
type imageRuntimeConfig struct {
|
|
User string `json:"User"`
|
|
Entrypoint []string `json:"Entrypoint"`
|
|
Cmd []string `json:"Cmd"`
|
|
WorkingDir string `json:"WorkingDir"`
|
|
Labels map[string]string `json:"Labels"`
|
|
ExposedPorts map[string]any `json:"ExposedPorts"`
|
|
Healthcheck *struct {
|
|
Test []string `json:"Test"`
|
|
} `json:"Healthcheck"`
|
|
}
|
|
|
|
type endpointSettings struct {
|
|
IPAMConfig json.RawMessage `json:"IPAMConfig,omitempty"`
|
|
Links []string `json:"Links,omitempty"`
|
|
Aliases []string `json:"Aliases,omitempty"`
|
|
MacAddress string `json:"MacAddress,omitempty"`
|
|
DriverOpts map[string]string `json:"DriverOpts,omitempty"`
|
|
GwPriority int `json:"GwPriority,omitempty"`
|
|
}
|
|
|
|
type containerState struct {
|
|
Running bool `json:"Running"`
|
|
ExitCode int `json:"ExitCode"`
|
|
Health *struct {
|
|
Status string `json:"Status"`
|
|
} `json:"Health"`
|
|
}
|
|
|
|
func newDockerClient(socket string) *dockerClient {
|
|
transport := &http.Transport{
|
|
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
|
return (&net.Dialer{Timeout: 2 * time.Second}).DialContext(ctx, "unix", socket)
|
|
},
|
|
MaxIdleConns: 2,
|
|
MaxIdleConnsPerHost: 2,
|
|
IdleConnTimeout: 30 * time.Second,
|
|
}
|
|
return &dockerClient{socket: socket, client: &http.Client{Transport: transport}}
|
|
}
|
|
|
|
func (c *dockerClient) negotiate(ctx context.Context) error {
|
|
response, err := c.request(ctx, http.MethodGet, "/version", nil, nil, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode != http.StatusOK {
|
|
return dockerStatusError(response)
|
|
}
|
|
var version struct {
|
|
APIVersion string `json:"ApiVersion"`
|
|
}
|
|
if err := json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(&version); err != nil {
|
|
return fmt.Errorf("decode Docker API version: %w", err)
|
|
}
|
|
for _, character := range version.APIVersion {
|
|
if (character < '0' || character > '9') && character != '.' {
|
|
return errors.New("Docker returned an invalid API version")
|
|
}
|
|
}
|
|
if version.APIVersion == "" {
|
|
return errors.New("Docker returned an empty API version")
|
|
}
|
|
c.apiPrefix = "/v" + version.APIVersion
|
|
return nil
|
|
}
|
|
|
|
func (c *dockerClient) findTarget(ctx context.Context, label string) (containerSummary, error) {
|
|
filters, _ := json.Marshal(map[string][]string{"label": {label}})
|
|
query := url.Values{"all": {"1"}, "filters": {string(filters)}}
|
|
response, err := c.request(ctx, http.MethodGet, "/containers/json?"+query.Encode(), nil, nil, true)
|
|
if err != nil {
|
|
return containerSummary{}, err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode != http.StatusOK {
|
|
return containerSummary{}, dockerStatusError(response)
|
|
}
|
|
var containers []containerSummary
|
|
if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&containers); err != nil {
|
|
return containerSummary{}, err
|
|
}
|
|
var running []containerSummary
|
|
for _, container := range containers {
|
|
if container.State == "running" {
|
|
running = append(running, container)
|
|
}
|
|
}
|
|
if len(running) != 1 {
|
|
return containerSummary{}, fmt.Errorf("expected exactly one running target container, found %d", len(running))
|
|
}
|
|
return running[0], nil
|
|
}
|
|
|
|
var errRegistryAuthentication = errors.New("registry authentication required")
|
|
|
|
func (c *dockerClient) pullImage(ctx context.Context, repository, tag, username, credential string) error {
|
|
// Public packages must work without credentials, including stale saved ones.
|
|
err := c.pullImageOnce(ctx, repository, tag, "", "")
|
|
if errors.Is(err, errRegistryAuthentication) && username != "" && credential != "" {
|
|
return c.pullImageOnce(ctx, repository, tag, username, credential)
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (c *dockerClient) pullImageOnce(ctx context.Context, repository, tag, username, credential string) error {
|
|
var headers http.Header
|
|
if username != "" && credential != "" {
|
|
authJSON, _ := json.Marshal(map[string]string{"username": username, "password": credential, "serveraddress": strings.Split(repository, "/")[0]})
|
|
headers = http.Header{"X-Registry-Auth": {base64.URLEncoding.EncodeToString(authJSON)}}
|
|
}
|
|
query := url.Values{"fromImage": {repository}, "tag": {tag}}
|
|
response, err := c.request(ctx, http.MethodPost, "/images/create?"+query.Encode(), nil, headers, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode != http.StatusOK {
|
|
err := dockerStatusError(response)
|
|
if response.StatusCode == http.StatusUnauthorized || response.StatusCode == http.StatusForbidden || registryAuthFailure(err.Error()) {
|
|
return fmt.Errorf("%w: %v", errRegistryAuthentication, err)
|
|
}
|
|
return err
|
|
}
|
|
scanner := bufio.NewScanner(response.Body)
|
|
scanner.Buffer(make([]byte, 64<<10), 1<<20)
|
|
for scanner.Scan() {
|
|
var event struct {
|
|
Error string `json:"error"`
|
|
ErrorDetail struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
} `json:"errorDetail"`
|
|
}
|
|
if err := json.Unmarshal(scanner.Bytes(), &event); err != nil {
|
|
return errors.New("Docker returned invalid pull progress")
|
|
}
|
|
message := event.Error
|
|
if message == "" {
|
|
message = event.ErrorDetail.Message
|
|
}
|
|
if message != "" {
|
|
if event.ErrorDetail.Code == http.StatusUnauthorized || event.ErrorDetail.Code == http.StatusForbidden || registryAuthFailure(message) {
|
|
return fmt.Errorf("%w: %s", errRegistryAuthentication, message)
|
|
}
|
|
return errors.New(message)
|
|
}
|
|
}
|
|
return scanner.Err()
|
|
}
|
|
|
|
func registryAuthFailure(message string) bool {
|
|
message = strings.ToLower(message)
|
|
for _, marker := range []string{"unauthorized", "authentication required", "access denied", "access to the resource is denied", "pull access denied", "no basic auth credentials"} {
|
|
if strings.Contains(message, marker) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (c *dockerClient) inspectContainer(ctx context.Context, id string) (containerInspect, error) {
|
|
response, err := c.request(ctx, http.MethodGet, "/containers/"+url.PathEscape(id)+"/json", nil, nil, true)
|
|
if err != nil {
|
|
return containerInspect{}, err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode != http.StatusOK {
|
|
return containerInspect{}, dockerStatusError(response)
|
|
}
|
|
var inspected containerInspect
|
|
if err := json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&inspected); err != nil {
|
|
return containerInspect{}, err
|
|
}
|
|
return inspected, nil
|
|
}
|
|
|
|
func (c *dockerClient) inspectImage(ctx context.Context, reference string) (imageInspect, error) {
|
|
if strings.TrimSpace(reference) == "" {
|
|
return imageInspect{}, errors.New("image reference is empty")
|
|
}
|
|
response, err := c.request(ctx, http.MethodGet, "/images/"+url.PathEscape(reference)+"/json", nil, nil, true)
|
|
if err != nil {
|
|
return imageInspect{}, err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode != http.StatusOK {
|
|
return imageInspect{}, dockerStatusError(response)
|
|
}
|
|
var inspected imageInspect
|
|
if err := json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&inspected); err != nil {
|
|
return imageInspect{}, err
|
|
}
|
|
if inspected.ID == "" || len(inspected.Config) == 0 {
|
|
return imageInspect{}, errors.New("Docker returned incomplete image metadata")
|
|
}
|
|
return inspected, nil
|
|
}
|
|
|
|
func (c *dockerClient) stopContainer(ctx context.Context, id string, seconds int) error {
|
|
query := url.Values{"t": {fmt.Sprint(seconds)}}
|
|
response, err := c.request(ctx, http.MethodPost, "/containers/"+url.PathEscape(id)+"/stop?"+query.Encode(), nil, nil, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotModified {
|
|
return nil
|
|
}
|
|
return dockerStatusError(response)
|
|
}
|
|
|
|
func (c *dockerClient) renameContainer(ctx context.Context, id, name string) error {
|
|
query := url.Values{"name": {name}}
|
|
response, err := c.request(ctx, http.MethodPost, "/containers/"+url.PathEscape(id)+"/rename?"+query.Encode(), nil, nil, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode == http.StatusNoContent {
|
|
return nil
|
|
}
|
|
return dockerStatusError(response)
|
|
}
|
|
|
|
func resolveCurrentVersion(reported string, rawConfig json.RawMessage, repository string) (string, error) {
|
|
var config struct {
|
|
Image string `json:"Image"`
|
|
Labels map[string]string `json:"Labels"`
|
|
}
|
|
if err := json.Unmarshal(rawConfig, &config); err != nil {
|
|
return "", fmt.Errorf("decode running container version: %w", err)
|
|
}
|
|
prefix := strings.TrimSuffix(repository, ":") + ":"
|
|
var imageVersion string
|
|
if strings.HasPrefix(config.Image, prefix) {
|
|
if version := strings.TrimPrefix(config.Image, prefix); updateprotocol.ValidVersion(version) {
|
|
imageVersion = version
|
|
}
|
|
}
|
|
labelVersion := strings.TrimSpace(config.Labels[labelImageVersion])
|
|
if labelVersion != "" && !updateprotocol.ValidVersion(labelVersion) {
|
|
return "", errors.New("running container has an invalid version label")
|
|
}
|
|
if imageVersion == "" || labelVersion == "" {
|
|
return "", errors.New("running container has no trustworthy version metadata")
|
|
}
|
|
if imageVersion != labelVersion {
|
|
return "", fmt.Errorf("running image reference %s does not match version label %s", imageVersion, labelVersion)
|
|
}
|
|
actualVersion := imageVersion
|
|
if reported != "" {
|
|
if !updateprotocol.ValidVersion(reported) || reported != actualVersion {
|
|
return "", fmt.Errorf("reported version %q does not match running version %q", reported, actualVersion)
|
|
}
|
|
}
|
|
return actualVersion, nil
|
|
}
|
|
|
|
func validateUpdateCompatibility(currentVersion, targetVersion string, currentImage, targetImage imageInspect) error {
|
|
if comparison, err := updateprotocol.CompareVersions(currentVersion, targetVersion); err != nil || comparison >= 0 {
|
|
return fmt.Errorf("target %s is not newer than %s", targetVersion, currentVersion)
|
|
}
|
|
var current, target imageRuntimeConfig
|
|
if err := json.Unmarshal(currentImage.Config, ¤t); err != nil {
|
|
return fmt.Errorf("decode current image config: %w", err)
|
|
}
|
|
if err := json.Unmarshal(targetImage.Config, &target); err != nil {
|
|
return fmt.Errorf("decode target image config: %w", err)
|
|
}
|
|
if current.Labels[labelImageVersion] != currentVersion {
|
|
return fmt.Errorf("current image version label is %q", current.Labels[labelImageVersion])
|
|
}
|
|
if target.Labels[labelImageVersion] != targetVersion {
|
|
return fmt.Errorf("target image version label is %q", target.Labels[labelImageVersion])
|
|
}
|
|
minimumVersion := target.Labels[labelMinimumVersion]
|
|
if !updateprotocol.ValidVersion(minimumVersion) {
|
|
return errors.New("target image has no valid minimum update version")
|
|
}
|
|
if comparison, err := updateprotocol.CompareVersions(currentVersion, minimumVersion); err != nil || comparison < 0 {
|
|
return fmt.Errorf("target requires at least %s", minimumVersion)
|
|
}
|
|
minimumAgent, err := strconv.Atoi(target.Labels[labelMinimumAgent])
|
|
if err != nil || minimumAgent < buildinfo.MinimumAgentProtocol || minimumAgent > buildinfo.UpdateProtocolVersion {
|
|
return fmt.Errorf("target requires unsupported update protocol %q", target.Labels[labelMinimumAgent])
|
|
}
|
|
runtimeContract := strconv.Itoa(buildinfo.RuntimeContractVersion)
|
|
if current.Labels[labelRuntimeContract] != runtimeContract || target.Labels[labelRuntimeContract] != runtimeContract {
|
|
return fmt.Errorf("unsupported runtime contract transition %q to %q", current.Labels[labelRuntimeContract], target.Labels[labelRuntimeContract])
|
|
}
|
|
databaseContract := strconv.Itoa(buildinfo.DatabaseContractVersion)
|
|
if current.Labels[labelDatabaseContract] != databaseContract || target.Labels[labelDatabaseContract] != databaseContract {
|
|
return fmt.Errorf("unsupported database contract transition %q to %q", current.Labels[labelDatabaseContract], target.Labels[labelDatabaseContract])
|
|
}
|
|
currentSchema, currentSchemaErr := strconv.Atoi(current.Labels[labelSchemaVersion])
|
|
targetSchema, targetSchemaErr := strconv.Atoi(target.Labels[labelSchemaVersion])
|
|
if currentSchemaErr != nil || targetSchemaErr != nil || currentSchema < buildinfo.CurrentSchemaVersion || targetSchema < currentSchema {
|
|
return fmt.Errorf("unsupported schema transition %q to %q", current.Labels[labelSchemaVersion], target.Labels[labelSchemaVersion])
|
|
}
|
|
if !compatibleRuntimeConfig(current) {
|
|
return errors.New("current image violates the stable runtime contract")
|
|
}
|
|
if !compatibleRuntimeConfig(target) {
|
|
return errors.New("target image violates the stable runtime contract")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func compatibleRuntimeConfig(config imageRuntimeConfig) bool {
|
|
if config.User != runtimeUser || config.WorkingDir != runtimeWorkingDirectory || len(config.Entrypoint) != 1 || config.Entrypoint[0] != runtimeEntrypoint || len(config.Cmd) != 1 || config.Cmd[0] != runtimeCommand {
|
|
return false
|
|
}
|
|
if _, ok := config.ExposedPorts["8080/tcp"]; !ok {
|
|
return false
|
|
}
|
|
return config.Healthcheck != nil && strings.Join(config.Healthcheck.Test, "\x00") == strings.Join([]string{"CMD", runtimeEntrypoint, "healthcheck"}, "\x00")
|
|
}
|
|
|
|
func buildReplacementConfig(inspected containerInspect, oldImageConfig, targetImageConfig json.RawMessage, targetReference string) (map[string]json.RawMessage, error) {
|
|
oldContainer, err := decodeConfigMap(inspected.Config)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode running container config: %w", err)
|
|
}
|
|
oldImage, err := decodeConfigMap(oldImageConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode running image config: %w", err)
|
|
}
|
|
target, err := decodeConfigMap(targetImageConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode target image config: %w", err)
|
|
}
|
|
|
|
for _, field := range []string{
|
|
"Domainname", "AttachStdin", "AttachStdout", "AttachStderr", "Tty", "OpenStdin", "StdinOnce",
|
|
"User", "Cmd", "Entrypoint", "WorkingDir", "Healthcheck", "StopSignal", "StopTimeout", "Shell",
|
|
"NetworkDisabled", "MacAddress",
|
|
} {
|
|
preserveConfigOverride(target, oldContainer, oldImage, field)
|
|
}
|
|
preserveCustomHostname(target, oldContainer, inspected.ID)
|
|
if err := mergeEnvironment(target, oldContainer, oldImage); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := mergeLabels(target, oldContainer, oldImage); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, field := range []string{"ExposedPorts", "Volumes"} {
|
|
if err := mergeAddedMapKeys(target, oldContainer, oldImage, field); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
encodedReference, _ := json.Marshal(targetReference)
|
|
target["Image"] = encodedReference
|
|
return target, nil
|
|
}
|
|
|
|
func decodeConfigMap(raw json.RawMessage) (map[string]json.RawMessage, error) {
|
|
var config map[string]json.RawMessage
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil, errors.New("empty config")
|
|
}
|
|
if err := json.Unmarshal(raw, &config); err != nil {
|
|
return nil, err
|
|
}
|
|
return config, nil
|
|
}
|
|
|
|
func preserveConfigOverride(target, container, image map[string]json.RawMessage, field string) {
|
|
containerValue, exists := container[field]
|
|
if !exists || rawJSONEqual(containerValue, image[field]) {
|
|
return
|
|
}
|
|
target[field] = append(json.RawMessage(nil), containerValue...)
|
|
}
|
|
|
|
func preserveCustomHostname(target, container map[string]json.RawMessage, containerID string) {
|
|
var hostname string
|
|
if err := json.Unmarshal(container["Hostname"], &hostname); err != nil || hostname == "" {
|
|
return
|
|
}
|
|
shortID := containerID
|
|
if len(shortID) > 12 {
|
|
shortID = shortID[:12]
|
|
}
|
|
if hostname == containerID || hostname == shortID {
|
|
delete(target, "Hostname")
|
|
return
|
|
}
|
|
target["Hostname"] = append(json.RawMessage(nil), container["Hostname"]...)
|
|
}
|
|
|
|
func mergeEnvironment(target, container, image map[string]json.RawMessage) error {
|
|
targetValues, err := decodeStringSlice(target["Env"])
|
|
if err != nil {
|
|
return fmt.Errorf("decode target environment: %w", err)
|
|
}
|
|
containerValues, err := decodeStringSlice(container["Env"])
|
|
if err != nil {
|
|
return fmt.Errorf("decode container environment: %w", err)
|
|
}
|
|
imageValues, err := decodeStringSlice(image["Env"])
|
|
if err != nil {
|
|
return fmt.Errorf("decode image environment: %w", err)
|
|
}
|
|
result, positions := indexedEnvironment(targetValues)
|
|
imageEnvironment := make(map[string]string, len(imageValues))
|
|
for _, value := range imageValues {
|
|
imageEnvironment[environmentKey(value)] = value
|
|
}
|
|
for _, value := range containerValues {
|
|
key := environmentKey(value)
|
|
if previous, exists := imageEnvironment[key]; exists && previous == value {
|
|
continue
|
|
}
|
|
if position, exists := positions[key]; exists {
|
|
result[position] = value
|
|
} else {
|
|
positions[key] = len(result)
|
|
result = append(result, value)
|
|
}
|
|
}
|
|
encoded, _ := json.Marshal(result)
|
|
target["Env"] = encoded
|
|
return nil
|
|
}
|
|
|
|
func decodeStringSlice(raw json.RawMessage) ([]string, error) {
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil, nil
|
|
}
|
|
var values []string
|
|
if err := json.Unmarshal(raw, &values); err != nil {
|
|
return nil, err
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func indexedEnvironment(values []string) ([]string, map[string]int) {
|
|
result := append([]string(nil), values...)
|
|
positions := make(map[string]int, len(values))
|
|
for index, value := range values {
|
|
positions[environmentKey(value)] = index
|
|
}
|
|
return result, positions
|
|
}
|
|
|
|
func environmentKey(value string) string {
|
|
if index := strings.IndexByte(value, '='); index >= 0 {
|
|
return value[:index]
|
|
}
|
|
return value
|
|
}
|
|
|
|
func mergeLabels(target, container, image map[string]json.RawMessage) error {
|
|
targetLabels, err := decodeStringMap(target["Labels"])
|
|
if err != nil {
|
|
return fmt.Errorf("decode target labels: %w", err)
|
|
}
|
|
containerLabels, err := decodeStringMap(container["Labels"])
|
|
if err != nil {
|
|
return fmt.Errorf("decode container labels: %w", err)
|
|
}
|
|
imageLabels, err := decodeStringMap(image["Labels"])
|
|
if err != nil {
|
|
return fmt.Errorf("decode image labels: %w", err)
|
|
}
|
|
if targetLabels == nil {
|
|
targetLabels = make(map[string]string)
|
|
}
|
|
for key, value := range containerLabels {
|
|
if protectedImageLabel(key) {
|
|
continue
|
|
}
|
|
if imageValue, exists := imageLabels[key]; exists && imageValue == value {
|
|
continue
|
|
}
|
|
targetLabels[key] = value
|
|
}
|
|
encoded, _ := json.Marshal(targetLabels)
|
|
target["Labels"] = encoded
|
|
return nil
|
|
}
|
|
|
|
func protectedImageLabel(key string) bool {
|
|
switch key {
|
|
case labelImageVersion, labelMinimumVersion, labelMinimumAgent, labelRuntimeContract, labelDatabaseContract, labelSchemaVersion:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func decodeStringMap(raw json.RawMessage) (map[string]string, error) {
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil, nil
|
|
}
|
|
var values map[string]string
|
|
if err := json.Unmarshal(raw, &values); err != nil {
|
|
return nil, err
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func mergeAddedMapKeys(target, container, image map[string]json.RawMessage, field string) error {
|
|
targetValues, err := decodeRawMap(target[field])
|
|
if err != nil {
|
|
return fmt.Errorf("decode target %s: %w", field, err)
|
|
}
|
|
containerValues, err := decodeRawMap(container[field])
|
|
if err != nil {
|
|
return fmt.Errorf("decode container %s: %w", field, err)
|
|
}
|
|
imageValues, err := decodeRawMap(image[field])
|
|
if err != nil {
|
|
return fmt.Errorf("decode image %s: %w", field, err)
|
|
}
|
|
if targetValues == nil {
|
|
targetValues = make(map[string]json.RawMessage)
|
|
}
|
|
for key, value := range containerValues {
|
|
if imageValue, exists := imageValues[key]; exists && rawJSONEqual(value, imageValue) {
|
|
continue
|
|
}
|
|
targetValues[key] = value
|
|
}
|
|
encoded, _ := json.Marshal(targetValues)
|
|
target[field] = encoded
|
|
return nil
|
|
}
|
|
|
|
func decodeRawMap(raw json.RawMessage) (map[string]json.RawMessage, error) {
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return nil, nil
|
|
}
|
|
var values map[string]json.RawMessage
|
|
if err := json.Unmarshal(raw, &values); err != nil {
|
|
return nil, err
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func rawJSONEqual(left, right json.RawMessage) bool {
|
|
if len(left) == 0 && len(right) == 0 {
|
|
return true
|
|
}
|
|
var a, b any
|
|
if err := json.Unmarshal(left, &a); err != nil {
|
|
return false
|
|
}
|
|
if err := json.Unmarshal(right, &b); err != nil {
|
|
return false
|
|
}
|
|
return reflect.DeepEqual(a, b)
|
|
}
|
|
|
|
func (c *dockerClient) createReplacement(ctx context.Context, inspected containerInspect, name string, body map[string]json.RawMessage) (string, error) {
|
|
body["HostConfig"] = inspected.HostConfig
|
|
endpoints := make(map[string]endpointSettings, len(inspected.NetworkSettings.Networks))
|
|
oldIDPrefix := inspected.ID
|
|
if len(oldIDPrefix) > 12 {
|
|
oldIDPrefix = oldIDPrefix[:12]
|
|
}
|
|
for network, endpoint := range inspected.NetworkSettings.Networks {
|
|
aliases := endpoint.Aliases[:0]
|
|
for _, alias := range endpoint.Aliases {
|
|
if alias != inspected.ID && alias != oldIDPrefix {
|
|
aliases = append(aliases, alias)
|
|
}
|
|
}
|
|
endpoint.Aliases = aliases
|
|
endpoints[network] = endpoint
|
|
}
|
|
networkingJSON, _ := json.Marshal(map[string]any{"EndpointsConfig": endpoints})
|
|
body["NetworkingConfig"] = networkingJSON
|
|
encoded, err := json.Marshal(body)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
query := url.Values{"name": {name}}
|
|
response, err := c.request(ctx, http.MethodPost, "/containers/create?"+query.Encode(), bytes.NewReader(encoded), http.Header{"Content-Type": {"application/json"}}, true)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode != http.StatusCreated {
|
|
return "", dockerStatusError(response)
|
|
}
|
|
var created struct {
|
|
ID string `json:"Id"`
|
|
}
|
|
if err := json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(&created); err != nil {
|
|
return "", err
|
|
}
|
|
if created.ID == "" {
|
|
return "", errors.New("Docker created a replacement without an ID")
|
|
}
|
|
return created.ID, nil
|
|
}
|
|
|
|
func (c *dockerClient) startContainer(ctx context.Context, id string) error {
|
|
response, err := c.request(ctx, http.MethodPost, "/containers/"+url.PathEscape(id)+"/start", nil, nil, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotModified {
|
|
return nil
|
|
}
|
|
return dockerStatusError(response)
|
|
}
|
|
|
|
func (c *dockerClient) removeContainer(ctx context.Context, id string, force bool) error {
|
|
query := url.Values{"force": {fmt.Sprint(force)}, "v": {"0"}}
|
|
response, err := c.request(ctx, http.MethodDelete, "/containers/"+url.PathEscape(id)+"?"+query.Encode(), nil, nil, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode == http.StatusNoContent || response.StatusCode == http.StatusNotFound {
|
|
return nil
|
|
}
|
|
return dockerStatusError(response)
|
|
}
|
|
|
|
func (c *dockerClient) waitHealthy(ctx context.Context, id, expectedImageID, expectedVersion string, timeout time.Duration) error {
|
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
ticker := time.NewTicker(time.Second)
|
|
defer ticker.Stop()
|
|
consecutiveHealthy := 0
|
|
startedAt := time.Now()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-ticker.C:
|
|
inspected, err := c.inspectContainer(ctx, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !inspected.State.Running {
|
|
return fmt.Errorf("replacement container stopped with exit code %d", inspected.State.ExitCode)
|
|
}
|
|
if expectedImageID != "" && inspected.Image != expectedImageID {
|
|
return fmt.Errorf("container uses image %s instead of %s", inspected.Image, expectedImageID)
|
|
}
|
|
if expectedVersion != "" {
|
|
var config struct {
|
|
Labels map[string]string `json:"Labels"`
|
|
}
|
|
if err := json.Unmarshal(inspected.Config, &config); err != nil {
|
|
return fmt.Errorf("decode replacement version: %w", err)
|
|
}
|
|
version := strings.TrimSpace(config.Labels[labelImageVersion])
|
|
if version != expectedVersion {
|
|
return fmt.Errorf("container version is %q instead of %q", version, expectedVersion)
|
|
}
|
|
}
|
|
if inspected.State.Health == nil {
|
|
if time.Since(startedAt) >= 8*time.Second {
|
|
return errors.New("replacement container has no healthcheck")
|
|
}
|
|
continue
|
|
}
|
|
if inspected.State.Health.Status == "healthy" {
|
|
consecutiveHealthy++
|
|
if consecutiveHealthy >= 3 {
|
|
return nil
|
|
}
|
|
} else {
|
|
consecutiveHealthy = 0
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *dockerClient) waitCurrentHealthy(ctx context.Context, id, expectedImageID string, timeout, pollInterval time.Duration) (containerInspect, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
ticker := time.NewTicker(pollInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return containerInspect{}, ctx.Err()
|
|
case <-ticker.C:
|
|
inspected, err := c.inspectContainer(ctx, id)
|
|
if err != nil {
|
|
return containerInspect{}, err
|
|
}
|
|
if !inspected.State.Running {
|
|
return containerInspect{}, fmt.Errorf("current container stopped with exit code %d", inspected.State.ExitCode)
|
|
}
|
|
if expectedImageID != "" && inspected.Image != expectedImageID {
|
|
return containerInspect{}, fmt.Errorf("current container image changed from %s to %s", expectedImageID, inspected.Image)
|
|
}
|
|
if inspected.State.Health == nil {
|
|
return containerInspect{}, errors.New("current container has no healthcheck")
|
|
}
|
|
switch inspected.State.Health.Status {
|
|
case "healthy":
|
|
return inspected, nil
|
|
case "starting":
|
|
continue
|
|
default:
|
|
return containerInspect{}, fmt.Errorf("current container health is %q", inspected.State.Health.Status)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func waitForDockerAPI(ctx context.Context, timeout, interval time.Duration, negotiate func(context.Context) error) error {
|
|
retryCtx, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
var lastErr error
|
|
for {
|
|
attemptCtx, attemptCancel := context.WithTimeout(retryCtx, 3*time.Second)
|
|
lastErr = negotiate(attemptCtx)
|
|
attemptCancel()
|
|
if lastErr == nil {
|
|
return nil
|
|
}
|
|
timer := time.NewTimer(interval)
|
|
select {
|
|
case <-retryCtx.Done():
|
|
if !timer.Stop() {
|
|
select {
|
|
case <-timer.C:
|
|
default:
|
|
}
|
|
}
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
return lastErr
|
|
case <-timer.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *dockerClient) request(ctx context.Context, method, path string, body io.Reader, headers http.Header, versioned bool) (*http.Response, error) {
|
|
if versioned {
|
|
path = c.apiPrefix + path
|
|
}
|
|
request, err := http.NewRequestWithContext(ctx, method, "http://docker"+path, body)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for name, values := range headers {
|
|
for _, value := range values {
|
|
request.Header.Add(name, value)
|
|
}
|
|
}
|
|
return c.client.Do(request)
|
|
}
|
|
|
|
func dockerStatusError(response *http.Response) error {
|
|
message, _ := io.ReadAll(io.LimitReader(response.Body, 8<<10))
|
|
trimmed := strings.TrimSpace(string(message))
|
|
if trimmed == "" {
|
|
trimmed = http.StatusText(response.StatusCode)
|
|
}
|
|
return fmt.Errorf("Docker API HTTP %d: %s", response.StatusCode, trimmed)
|
|
}
|
|
|
|
func isDockerNotModified(err error) bool {
|
|
return err != nil && strings.Contains(err.Error(), "HTTP 304")
|
|
}
|