121 lines
3.5 KiB
Go
121 lines
3.5 KiB
Go
// Command updater-agent reports this machine's Docker services to the
|
|
// package-updater master.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/docker/docker/client"
|
|
|
|
"github.com/shcizo/package-updater/internal/agent/config"
|
|
"github.com/shcizo/package-updater/internal/agent/conn"
|
|
"github.com/shcizo/package-updater/internal/agent/inventory"
|
|
"github.com/shcizo/package-updater/internal/agent/updater"
|
|
"github.com/shcizo/package-updater/internal/logging"
|
|
)
|
|
|
|
// Build info is injected at link time via -ldflags="-X main.version=...".
|
|
var version = "dev"
|
|
|
|
func main() {
|
|
if err := run(); err != nil {
|
|
fmt.Fprintln(os.Stderr, "fatal:", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func run() error {
|
|
cfg, err := config.Load()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
logger := logging.New(cfg.LogLevel)
|
|
|
|
dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
|
|
if err != nil {
|
|
return fmt.Errorf("docker client: %w", err)
|
|
}
|
|
defer dockerCli.Close()
|
|
|
|
auth, err := updater.LoadRegistryAuth(cfg.DockerConfig, logger)
|
|
if err != nil {
|
|
return fmt.Errorf("registry auth: %w", err)
|
|
}
|
|
resolver := updater.NewResolver(dockerCli, auth)
|
|
exec := updater.NewDispatch(
|
|
updater.NewComposeExecutor(dockerCli, resolver, cfg.SelfID, updater.ExecDocker, updater.ExecDockerOutput),
|
|
updater.NewSwarmExecutor(dockerCli, resolver, auth),
|
|
)
|
|
runner := updater.NewRunner(exec, cfg.UpdateTimeout)
|
|
|
|
collector := inventory.New(dockerCli, cfg.SelfID).
|
|
WithBuildCheck(updater.NewBuildChecker(dockerCli, cfg.SelfID, updater.ExecDockerOutput), logger)
|
|
agent := conn.New(conn.Config{
|
|
URL: cfg.MasterURL, Token: cfg.Token, Fingerprint: cfg.MasterKey, Version: version,
|
|
}, collector, runner, logger)
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
logMounts(ctx, dockerCli, cfg.SelfID, logger)
|
|
go runner.Run(ctx)
|
|
go watchLoop(ctx, dockerCli, agent, logger)
|
|
logger.Info("agent_starting", "version", version, "master_url", cfg.MasterURL,
|
|
"update_timeout", cfg.UpdateTimeout.String(), "registry_auth", auth != nil)
|
|
return agent.Run(ctx)
|
|
}
|
|
|
|
// logMounts warns at startup when Compose updates cannot work. Each job reads
|
|
// the mounts again itself; this is only for the operator.
|
|
func logMounts(ctx context.Context, cli updater.SelfInspector, selfID string, log *slog.Logger) {
|
|
roots, err := updater.MountRoots(ctx, cli, selfID)
|
|
if err != nil || len(roots) == 0 {
|
|
attrs := []any{"hint", "Compose updates will fail; mount each stacks directory at the same path, read-write"}
|
|
if err != nil {
|
|
attrs = append(attrs, "error", err)
|
|
}
|
|
log.Warn("no_stack_mounts", attrs...)
|
|
return
|
|
}
|
|
log.Info("stack_mounts", "roots", roots)
|
|
}
|
|
|
|
// watchLoop keeps a Docker event subscription alive. When the stream breaks
|
|
// (daemon restart), it resubscribes and asks for a refresh, because events
|
|
// may have been missed in between. It also asks for a full re-inventory every
|
|
// 60 seconds, since a Swarm replica dying on another node emits no event here.
|
|
func watchLoop(ctx context.Context, src inventory.EventSource, agent *conn.Agent, log *slog.Logger) {
|
|
go func() {
|
|
ticker := time.NewTicker(60 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
agent.Notify()
|
|
}
|
|
}
|
|
}()
|
|
|
|
for {
|
|
err := inventory.Watch(ctx, src, 2*time.Second, agent.Notify)
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
log.Warn("docker_events_lost", "error", err)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(5 * time.Second):
|
|
}
|
|
agent.Notify()
|
|
}
|
|
}
|