From 9117bed6fce24ac8dc5994fdb5d15e5f3a7e69d1 Mon Sep 17 00:00:00 2001 From: Barnabas Busa Date: Fri, 29 May 2026 15:48:33 +0200 Subject: [PATCH 1/3] feat: add /roll command for gated fleet image rollouts Adds a gated, sequential, health-checked image rollout capability across a network's nodes, surfaced as a Discord /roll slash command and a `panda-pulse roll` CLI subcommand backed by a shared pkg/roll engine. - pkg/roll: sequential engine (abort + leave-rest-untouched on failure), pluggable actuators (SSH-to-local-watchtower and watchtower vhost API), Dora-based health gating with per-node beacon fallback, cartographoor inventory resolution, and Ansible-style host selection (globs, exclusions, client/group/node, "all"). - Discord /roll: client autocomplete from the network inventory, live per-host progress in a channel message (survives the 15m interaction window), a force option to override health gating, and a completion ping. Health uses Dora as the source of truth (one unauthenticated fleet-wide call); rolls trigger watchtower's HTTP API via the watchtower- vhost, so no SSH or basic auth is required on the default path. --- cmd/main.go | 6 + cmd/roll.go | 140 ++++++++++++++ go.mod | 2 +- pkg/discord/bot.go | 2 +- pkg/discord/cmd/roll/command.go | 262 ++++++++++++++++++++++++++ pkg/discord/cmd/roll/run.go | 263 ++++++++++++++++++++++++++ pkg/roll/actuator.go | 14 ++ pkg/roll/api.go | 102 +++++++++++ pkg/roll/dora.go | 113 ++++++++++++ pkg/roll/engine.go | 316 ++++++++++++++++++++++++++++++++ pkg/roll/health.go | 89 +++++++++ pkg/roll/inventory.go | 210 +++++++++++++++++++++ pkg/roll/match.go | 115 ++++++++++++ pkg/roll/match_test.go | 108 +++++++++++ pkg/roll/provider.go | 133 ++++++++++++++ pkg/roll/ssh.go | 209 +++++++++++++++++++++ pkg/service/config.go | 4 + pkg/service/service.go | 8 + 18 files changed, 2094 insertions(+), 2 deletions(-) create mode 100644 cmd/roll.go create mode 100644 pkg/discord/cmd/roll/command.go create mode 100644 pkg/discord/cmd/roll/run.go create mode 100644 pkg/roll/actuator.go create mode 100644 pkg/roll/api.go create mode 100644 pkg/roll/dora.go create mode 100644 pkg/roll/engine.go create mode 100644 pkg/roll/health.go create mode 100644 pkg/roll/inventory.go create mode 100644 pkg/roll/match.go create mode 100644 pkg/roll/match_test.go create mode 100644 pkg/roll/provider.go create mode 100644 pkg/roll/ssh.go diff --git a/cmd/main.go b/cmd/main.go index f7e7321..8879a7b 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -75,6 +75,8 @@ func main() { setConfig(&cfg) + rootCmd.AddCommand(newRollCommand(log)) + if err := rootCmd.Execute(); err != nil { os.Exit(1) } @@ -101,6 +103,10 @@ func setConfig(cfg *service.Config) { cfg.S3EndpointURL = os.Getenv("AWS_ENDPOINT_URL") cfg.HealthCheckAddress = os.Getenv("HEALTH_CHECK_ADDRESS") cfg.MetricsAddress = os.Getenv("METRICS_ADDRESS") + cfg.RollSSHKeyPath = os.Getenv("ROLL_SSH_KEY") + cfg.WatchtowerAPIToken = os.Getenv("WATCHTOWER_HTTP_API_TOKEN") + cfg.NodeBasicAuthUser = os.Getenv("ROLL_BASIC_AUTH_USER") + cfg.NodeBasicAuthPass = os.Getenv("ROLL_BASIC_AUTH_PASS") if cfg.GrafanaBaseURL == "" { cfg.GrafanaBaseURL = grafana.DefaultGrafanaBaseURL diff --git a/cmd/roll.go b/cmd/roll.go new file mode 100644 index 0000000..22ea04d --- /dev/null +++ b/cmd/roll.go @@ -0,0 +1,140 @@ +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "syscall" + "time" + + "github.com/ethpandaops/panda-pulse/pkg/roll" + "github.com/sirupsen/logrus" + "github.com/spf13/cobra" +) + +// newRollCommand builds the `panda-pulse roll` subcommand: a gated, sequential +// image rollout across a network's nodes, resolved from cartographoor inventory. +func newRollCommand(log *logrus.Logger) *cobra.Command { + var ( + network string + client string + image string + actuatorKind string + inventoryURL string + sshKeyPath string + watchtowerPort int + watchtowerToken string + watchtowerScheme string + watchtowerPrefix string + beaconScheme string + basicAuthUser string + basicAuthPass string + doraURL string + noDora bool + skipHealth bool + dryRun bool + delay time.Duration + postTrigger time.Duration + waitTimeout time.Duration + healthInterval time.Duration + maxSyncDistance uint64 + ) + + cmd := &cobra.Command{ + Use: "roll", + Short: "Gated rolling image update across a network's nodes", + SilenceUsage: true, + RunE: func(_ *cobra.Command, _ []string) error { + if network == "" { + return fmt.Errorf("--network is required") + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + inv, err := roll.FetchInventory(ctx, inventoryURL, network) + if err != nil { + return err + } + + targets := roll.Select(roll.ResolveTargets(inv, beaconScheme), client) + if len(targets) == 0 { + return fmt.Errorf("no targets matched (network=%s client=%q)", network, client) + } + + actuator, err := buildActuator(actuatorKind, sshKeyPath, watchtowerToken, watchtowerScheme, watchtowerPrefix, watchtowerPort, log) + if err != nil { + return err + } + + doraHealthURL := "" + if !noDora { + doraHealthURL = doraURL + if doraHealthURL == "" { + doraHealthURL = roll.DoraURLForNetwork(network) + } + } + + return roll.NewEngine(actuator, log).Run(ctx, targets, roll.Options{ + Image: image, + DelayBetweenNodes: delay, + PostTriggerWait: postTrigger, + WaitTimeout: waitTimeout, + HealthCheckInterval: healthInterval, + MaxSyncDistance: maxSyncDistance, + SkipHealth: skipHealth, + DryRun: dryRun, + DoraURL: doraHealthURL, + BeaconBasicAuthUser: basicAuthUser, + BeaconBasicAuthPass: basicAuthPass, + }) + }, + } + + f := cmd.Flags() + f.StringVar(&network, "network", "", "network name as published by cartographoor (required)") + f.StringVar(&client, "client", "", "host pattern: client/group/node with globs, ! to exclude, 'all' (e.g. 'lighthouse', 'lighthouse_ethrex', 'lighthouse-*:!*-1')") + f.StringVar(&image, "image", "", "scope the roll to this image (empty = all watched containers)") + f.StringVar(&actuatorKind, "actuator", "ssh", "how to trigger the roll: ssh|api") + f.StringVar(&inventoryURL, "inventory-url", roll.DefaultInventoryBaseURL, "cartographoor inventory base URL") + f.StringVar(&sshKeyPath, "ssh-key", os.Getenv("ROLL_SSH_KEY"), "SSH private key path (ssh actuator; env ROLL_SSH_KEY)") + f.IntVar(&watchtowerPort, "watchtower-port", 0, "watchtower API port (api actuator; 0 = default for the scheme)") + f.StringVar(&watchtowerToken, "watchtower-token", os.Getenv("WATCHTOWER_HTTP_API_TOKEN"), "watchtower API token (api actuator; env WATCHTOWER_HTTP_API_TOKEN)") + f.StringVar(&watchtowerScheme, "watchtower-scheme", "https", "watchtower API scheme (api actuator)") + f.StringVar(&watchtowerPrefix, "watchtower-prefix", "watchtower-", "vhost prefix for the watchtower API (api actuator)") + f.StringVar(&doraURL, "dora-url", "", "Dora health source URL (default: https://dora..ethpandaops.io)") + f.BoolVar(&noDora, "no-dora", false, "use per-node beacon health instead of Dora") + f.StringVar(&beaconScheme, "beacon-scheme", "https", "scheme for beacon health endpoints (only used with --no-dora)") + f.StringVar(&basicAuthUser, "basic-auth-user", os.Getenv("ROLL_BASIC_AUTH_USER"), "basic auth user for beacon health endpoints (env ROLL_BASIC_AUTH_USER)") + f.StringVar(&basicAuthPass, "basic-auth-pass", os.Getenv("ROLL_BASIC_AUTH_PASS"), "basic auth password for beacon health endpoints (env ROLL_BASIC_AUTH_PASS)") + f.BoolVar(&skipHealth, "skip-health", false, "skip beacon health gating (force; trigger-and-go even if unhealthy)") + f.BoolVar(&dryRun, "dry-run", false, "log intent without triggering rolls") + f.DurationVar(&delay, "delay-roll", time.Minute, "wait between hosts (~N minutes for N hosts); overridable") + f.DurationVar(&postTrigger, "post-trigger-wait", 30*time.Second, "grace period after triggering before polling recovery") + f.DurationVar(&waitTimeout, "wait-timeout", 10*time.Minute, "per-node recovery timeout before aborting") + f.DurationVar(&healthInterval, "health-check-interval", 10*time.Second, "beacon health poll cadence during recovery") + f.Uint64Var(&maxSyncDistance, "max-sync-distance", 4, "max sync distance (slots) still considered healthy") + + return cmd +} + +func buildActuator(kind, sshKeyPath, token, scheme, prefix string, port int, log *logrus.Logger) (roll.Actuator, error) { + switch kind { + case "ssh": + return roll.NewSSHActuator(roll.SSHConfig{ + PrivateKeyPath: sshKeyPath, + ContainerName: roll.DefaultWatchtowerContainer, + Port: roll.DefaultWatchtowerPort, + Log: log, + }) + case "api": + if token == "" { + return nil, fmt.Errorf("--watchtower-token (or WATCHTOWER_HTTP_API_TOKEN) is required for the api actuator") + } + + return roll.NewAPIActuator(token, scheme, port, prefix), nil + default: + return nil, fmt.Errorf("unknown actuator %q (want ssh or api)", kind) + } +} diff --git a/go.mod b/go.mod index f1be42f..fd6a9d3 100644 --- a/go.mod +++ b/go.mod @@ -17,6 +17,7 @@ require ( github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.42.0 go.uber.org/mock v0.6.0 + golang.org/x/crypto v0.49.0 golang.org/x/text v0.37.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -96,7 +97,6 @@ require ( go.opentelemetry.io/otel/metric v1.42.0 // indirect go.opentelemetry.io/otel/trace v1.42.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect - golang.org/x/crypto v0.49.0 // indirect golang.org/x/sys v0.42.0 // indirect google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/pkg/discord/bot.go b/pkg/discord/bot.go index 5125c11..7ac1f3c 100644 --- a/pkg/discord/bot.go +++ b/pkg/discord/bot.go @@ -549,7 +549,7 @@ func (b *DiscordBot) scheduleDiscordChoiceRefresh() error { // and /hive trigger has bespoke per-subcommand handling. func commandSelfChecksPermission(cmdName string, data *discordgo.ApplicationCommandInteractionData) bool { switch cmdName { - case "build": + case "build", "roll": return true case "hive": return data != nil && len(data.Options) > 0 && data.Options[0].Name == "trigger" diff --git a/pkg/discord/cmd/roll/command.go b/pkg/discord/cmd/roll/command.go new file mode 100644 index 0000000..bc6ceef --- /dev/null +++ b/pkg/discord/cmd/roll/command.go @@ -0,0 +1,262 @@ +// Package roll provides the /roll Discord command: gated, sequential image +// rollouts across a network's nodes, resolved from cartographoor inventory and +// executed via the rollpkg engine. +package roll + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/bwmarrin/discordgo" + "github.com/ethpandaops/panda-pulse/pkg/discord/cmd/common" + rollpkg "github.com/ethpandaops/panda-pulse/pkg/roll" + "github.com/sirupsen/logrus" +) + +const ( + optionNetwork = "network" + optionClient = "client" + optionImage = "image" + optionDelay = "delay" + optionForce = "force" + optionDryRun = "dry_run" + + autocompleteLimit = 25 + providerCacheTTL = 30 * time.Second + autocompleteTime = 2 * time.Second + + actuatorSSH = "ssh" + actuatorAPI = "api" +) + +// Config configures the roll command's actuator and inventory source. +type Config struct { + // SSHKeyPath is the SSH private key used by the ssh actuator. + SSHKeyPath string + // WatchtowerToken is the bearer token used by the api actuator. + WatchtowerToken string + // InventoryURL overrides the cartographoor inventory base URL. + InventoryURL string + // Actuator selects how rolls are triggered: "ssh" or "api" (default). + Actuator string + // BasicAuthUser and BasicAuthPass authenticate beacon health checks behind + // nginx basic auth (the bn-* vhosts) — only used when Dora is unavailable. + BasicAuthUser string + BasicAuthPass string + // DoraURL overrides the Dora health source; empty derives it from the + // network (https://dora..ethpandaops.io). + DoraURL string +} + +// Command implements the /roll Discord slash command. +type Command struct { + log *logrus.Logger + bot common.BotContext + cfg Config + provider *rollpkg.InventoryProvider + guildRegistrations map[string]string +} + +// NewRollCommand creates the /roll command. +func NewRollCommand(log *logrus.Logger, bot common.BotContext, cfg Config) *Command { + if cfg.Actuator == "" { + cfg.Actuator = actuatorSSH + } + + return &Command{ + log: log, + bot: bot, + cfg: cfg, + provider: rollpkg.NewInventoryProvider(cfg.InventoryURL, "https", providerCacheTTL), + } +} + +// Name returns the command name. +func (c *Command) Name() string { return "roll" } + +func (c *Command) getCommandDefinition() *discordgo.ApplicationCommand { + return &discordgo.ApplicationCommand{ + Name: c.Name(), + Description: "Roll (re-pull + restart) client images across a network — gated and sequential", + Options: []*discordgo.ApplicationCommandOption{ + { + Name: optionNetwork, + Description: "Network name (e.g. glamsterdam-devnet-4)", + Type: discordgo.ApplicationCommandOptionString, + Required: true, + }, + { + Name: optionClient, + Description: "Host pattern: client/group/node, globs, ! to exclude, 'all'", + Type: discordgo.ApplicationCommandOptionString, + Required: true, + Autocomplete: true, + }, + { + Name: optionImage, + Description: "Scope to a specific image (optional, e.g. ethpandaops/lighthouse)", + Type: discordgo.ApplicationCommandOptionString, + Required: false, + }, + { + Name: optionDelay, + Description: "Delay between hosts, e.g. 1m or 90s (default 1m)", + Type: discordgo.ApplicationCommandOptionString, + Required: false, + }, + { + Name: optionForce, + Description: "Skip health checks — force the roll even if the node is unhealthy (e.g. known-bad node)", + Type: discordgo.ApplicationCommandOptionBoolean, + Required: false, + }, + { + Name: optionDryRun, + Description: "List what would roll without triggering", + Type: discordgo.ApplicationCommandOptionBoolean, + Required: false, + }, + }, + } +} + +// Register registers the command globally. +func (c *Command) Register(session *discordgo.Session) error { + cmd, err := session.ApplicationCommandCreate(session.State.User.ID, "", c.getCommandDefinition()) + if err != nil { + return fmt.Errorf("failed to register roll command: %w", err) + } + + if c.guildRegistrations == nil { + c.guildRegistrations = make(map[string]string, 1) + } + + c.guildRegistrations[""] = cmd.ID + + return nil +} + +// RegisterWithGuild registers the command to a specific guild. +func (c *Command) RegisterWithGuild(session *discordgo.Session, guildID string) error { + cmd, err := session.ApplicationCommandCreate(session.State.User.ID, guildID, c.getCommandDefinition()) + if err != nil { + return fmt.Errorf("failed to register roll command to guild %s: %w", guildID, err) + } + + if c.guildRegistrations == nil { + c.guildRegistrations = make(map[string]string, 2) + } + + c.guildRegistrations[guildID] = cmd.ID + + c.log.WithField("guild", guildID).Info("Registered roll command to guild") + + return nil +} + +// Handle dispatches autocomplete and command interactions. +func (c *Command) Handle(s *discordgo.Session, i *discordgo.InteractionCreate) { + switch i.Type { + case discordgo.InteractionApplicationCommandAutocomplete: + c.handleAutocomplete(s, i) + case discordgo.InteractionApplicationCommand: + c.handleCommand(s, i) + } +} + +func (c *Command) handleCommand(s *discordgo.Session, i *discordgo.InteractionCreate) { + data := i.ApplicationCommandData() + if data.Name != c.Name() { + return + } + + if !c.hasPermission(i.Member, s, i.GuildID) { + c.respondEphemeral(s, i, common.NoPermissionError(c.Name()).Error()) + + return + } + + if err := c.run(s, i, data); err != nil { + c.log.WithError(err).Error("roll command failed") + } +} + +// hasPermission allows admins or any team member to roll (mirrors /build). +func (c *Command) hasPermission(member *discordgo.Member, session *discordgo.Session, guildID string) bool { + cfg := c.bot.GetRoleConfig() + + for _, roleName := range common.GetRoleNames(member, session, guildID) { + if cfg.AdminRoles[strings.ToLower(roleName)] { + return true + } + + for _, teamRoles := range cfg.ClientRoles { + for _, teamRole := range teamRoles { + if strings.EqualFold(teamRole, roleName) { + return true + } + } + } + } + + return false +} + +// handleAutocomplete serves dynamic suggestions for the client option, sourced +// from the selected network's inventory (groups, node names, and "all"). +func (c *Command) handleAutocomplete(s *discordgo.Session, i *discordgo.InteractionCreate) { + data := i.ApplicationCommandData() + + var network, input string + + focused := false + + for _, opt := range data.Options { + switch opt.Name { + case optionNetwork: + network = opt.StringValue() + case optionClient: + if opt.Focused { + focused = true + input = opt.StringValue() + } + } + } + + var choices []*discordgo.ApplicationCommandOptionChoice + + if focused && network != "" { + ctx, cancel := context.WithTimeout(context.Background(), autocompleteTime) + defer cancel() + + if sugg, err := c.provider.Suggest(ctx, network, input, "", autocompleteLimit); err != nil { + c.log.WithError(err).Debug("roll autocomplete: suggest failed") + } else { + choices = make([]*discordgo.ApplicationCommandOptionChoice, 0, len(sugg)) + for _, tok := range sugg { + choices = append(choices, &discordgo.ApplicationCommandOptionChoice{Name: tok, Value: tok}) + } + } + } + + if err := s.InteractionRespond(i.Interaction, &discordgo.InteractionResponse{ + Type: discordgo.InteractionApplicationCommandAutocompleteResult, + Data: &discordgo.InteractionResponseData{Choices: choices}, + }); err != nil { + c.log.WithError(err).Debug("Failed to respond to roll autocomplete") + } +} + +func (c *Command) respondEphemeral(s *discordgo.Session, i *discordgo.InteractionCreate, content string) { + if err := s.InteractionRespond(i.Interaction, &discordgo.InteractionResponse{ + Type: discordgo.InteractionResponseChannelMessageWithSource, + Data: &discordgo.InteractionResponseData{ + Content: content, + Flags: discordgo.MessageFlagsEphemeral, + }, + }); err != nil { + c.log.WithError(err).Error("Failed to send ephemeral response") + } +} diff --git a/pkg/discord/cmd/roll/run.go b/pkg/discord/cmd/roll/run.go new file mode 100644 index 0000000..d888781 --- /dev/null +++ b/pkg/discord/cmd/roll/run.go @@ -0,0 +1,263 @@ +package roll + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + "github.com/bwmarrin/discordgo" + rollpkg "github.com/ethpandaops/panda-pulse/pkg/roll" +) + +// minEditInterval throttles progress edits to avoid Discord rate limits. +const minEditInterval = time.Second + +// run resolves targets, posts a live progress message to the channel, and drives +// the gated rollout. Progress is tracked in a normal bot message (not the +// interaction reply) so it survives past Discord's 15-minute interaction-token +// window — a multi-node roll can take longer than that. +func (c *Command) run(s *discordgo.Session, i *discordgo.InteractionCreate, data discordgo.ApplicationCommandInteractionData) error { + var ( + network, client, image, delayStr string + dryRun, force bool + ) + + for _, opt := range data.Options { + switch opt.Name { + case optionNetwork: + network = opt.StringValue() + case optionClient: + client = opt.StringValue() + case optionImage: + image = opt.StringValue() + case optionDelay: + delayStr = opt.StringValue() + case optionForce: + force = opt.BoolValue() + case optionDryRun: + dryRun = opt.BoolValue() + } + } + + var delay time.Duration + + if delayStr != "" { + d, err := time.ParseDuration(delayStr) + if err != nil { + c.respondEphemeral(s, i, fmt.Sprintf("❌ Invalid delay %q: %v", delayStr, err)) + + return nil + } + + delay = d + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + targets, err := c.provider.Targets(ctx, network) + if err != nil { + c.respondEphemeral(s, i, fmt.Sprintf("❌ Failed to load inventory for **%s**: %v", network, err)) + + return nil + } + + targets = rollpkg.Select(targets, client) + if len(targets) == 0 { + c.respondEphemeral(s, i, fmt.Sprintf("❌ No hosts matched `%s` on **%s**.", client, network)) + + return nil + } + + actuator, err := c.actuator() + if err != nil { + c.respondEphemeral(s, i, fmt.Sprintf("❌ Roll not configured: %v", err)) + + return nil + } + + // Ack the slash command immediately; live progress goes to a channel message + // (editable indefinitely, unlike the 15-minute interaction token). + c.respondEphemeral(s, i, fmt.Sprintf("🚀 Roll started for %d host(s) on **%s** — tracking in this channel.", len(targets), network)) + + ui := newRollUI(network, client, image, dryRun, targets) + + msg, err := s.ChannelMessageSend(i.ChannelID, ui.render()) + if err != nil { + return fmt.Errorf("failed to post progress message: %w", err) + } + + var lastEdit time.Time + + edit := func(force bool) { + if !force && time.Since(lastEdit) < minEditInterval { + return + } + + lastEdit = time.Now() + + if _, e := s.ChannelMessageEdit(i.ChannelID, msg.ID, ui.render()); e != nil { + c.log.WithError(e).Debug("roll: failed to edit progress message") + } + } + + doraURL := c.cfg.DoraURL + if doraURL == "" { + doraURL = rollpkg.DoraURLForNetwork(network) + } + + runErr := rollpkg.NewEngine(actuator, c.log).Run(ctx, targets, rollpkg.Options{ + Image: image, + DryRun: dryRun, + SkipHealth: force, + DelayBetweenNodes: delay, + DoraURL: doraURL, + BeaconBasicAuthUser: c.cfg.BasicAuthUser, + BeaconBasicAuthPass: c.cfg.BasicAuthPass, + OnProgress: func(p rollpkg.Progress) { + ui.update(p) + edit(false) + }, + }) + + ui.finish(runErr) + edit(true) + + c.notify(s, i, network, len(targets), runErr) + + return nil +} + +// notify pings the invoking user with the roll outcome (a new message, so it +// actually notifies — message edits don't). Per-host status is shown live in the +// progress message above. +func (c *Command) notify(s *discordgo.Session, i *discordgo.InteractionCreate, network string, hosts int, runErr error) { + mention := mentionUser(i) + + var content string + if runErr != nil { + content = fmt.Sprintf("%s ❌ Roll on **%s** aborted: %v", mention, network, runErr) + } else { + content = fmt.Sprintf("%s ✅ Roll on **%s** complete — %d host(s) updated and healthy.", mention, network, hosts) + } + + if _, err := s.ChannelMessageSend(i.ChannelID, strings.TrimSpace(content)); err != nil { + c.log.WithError(err).Debug("roll: failed to send completion notice") + } +} + +func mentionUser(i *discordgo.InteractionCreate) string { + switch { + case i.Member != nil && i.Member.User != nil: + return "<@" + i.Member.User.ID + ">" + case i.User != nil: + return "<@" + i.User.ID + ">" + default: + return "" + } +} + +func (c *Command) actuator() (rollpkg.Actuator, error) { + if c.cfg.Actuator == actuatorAPI { + if c.cfg.WatchtowerToken == "" { + return nil, fmt.Errorf("watchtower token not configured (WATCHTOWER_HTTP_API_TOKEN)") + } + + return rollpkg.NewAPIActuator(c.cfg.WatchtowerToken, "https", 0, "watchtower-"), nil + } + + return rollpkg.NewSSHActuator(rollpkg.SSHConfig{PrivateKeyPath: c.cfg.SSHKeyPath, Log: c.log}) +} + +// rollUI accumulates per-host progress into a renderable Discord message. +type rollUI struct { + mu sync.Mutex + header string + names []string + state []string + footer string +} + +func newRollUI(network, client, image string, dryRun bool, targets []rollpkg.Target) *rollUI { + mode := "Rolling" + if dryRun { + mode = "Dry-run" + } + + header := fmt.Sprintf("**%s `%s`** on **%s**", mode, client, network) + if image != "" { + header += fmt.Sprintf(" — image `%s`", image) + } + + header += fmt.Sprintf(" • %d host(s)", len(targets)) + + names := make([]string, len(targets)) + state := make([]string, len(targets)) + + for idx, t := range targets { + names[idx] = t.Name + state[idx] = fmt.Sprintf("⏳ `%s`", t.Name) + } + + return &rollUI{header: header, names: names, state: state} +} + +func (u *rollUI) update(p rollpkg.Progress) { + if p.Index < 1 || p.Index > len(u.state) { + return // fleet-level event (e.g. Done) is handled by finish + } + + u.mu.Lock() + defer u.mu.Unlock() + + name := u.names[p.Index-1] + + switch p.Phase { + case rollpkg.PhaseTriggering: + u.state[p.Index-1] = fmt.Sprintf("🔄 `%s` — triggering…", name) + case rollpkg.PhaseHealthy: + u.state[p.Index-1] = fmt.Sprintf("✅ `%s`", name) + case rollpkg.PhaseFailed: + u.state[p.Index-1] = fmt.Sprintf("❌ `%s` — %s", name, p.Message) + case rollpkg.PhaseSkipped: + u.state[p.Index-1] = fmt.Sprintf("• `%s` — would roll", name) + case rollpkg.PhaseDone: + } +} + +func (u *rollUI) finish(err error) { + u.mu.Lock() + defer u.mu.Unlock() + + if err != nil { + u.footer = fmt.Sprintf("**Aborted:** %v\nRemaining hosts were left untouched.", err) + + return + } + + u.footer = "**Done** ✅ — all targeted hosts rolled and healthy." +} + +func (u *rollUI) render() string { + u.mu.Lock() + defer u.mu.Unlock() + + var b strings.Builder + + b.WriteString(u.header) + b.WriteString("\n") + + for _, line := range u.state { + b.WriteString(line) + b.WriteString("\n") + } + + if u.footer != "" { + b.WriteString("\n") + b.WriteString(u.footer) + } + + return b.String() +} diff --git a/pkg/roll/actuator.go b/pkg/roll/actuator.go new file mode 100644 index 0000000..5c925e6 --- /dev/null +++ b/pkg/roll/actuator.go @@ -0,0 +1,14 @@ +package roll + +import "context" + +// Actuator triggers an image roll on a single target host. Implementations +// differ only in how they reach the node's container runtime/watchtower. +type Actuator interface { + // Name identifies the actuator (for logging). + Name() string + // Roll triggers a pull + recreate on the target. If image is non-empty it + // scopes the roll to that image (matched regardless of tag when untagged); + // empty means all of the host's watched containers. + Roll(ctx context.Context, target Target, image string) error +} diff --git a/pkg/roll/api.go b/pkg/roll/api.go new file mode 100644 index 0000000..5834b96 --- /dev/null +++ b/pkg/roll/api.go @@ -0,0 +1,102 @@ +package roll + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + "time" +) + +// APIActuator rolls by calling a node's watchtower HTTP API directly. This +// requires the watchtower API to be reachable from where this runs (i.e. +// publicly/VPN-exposed) and a shared token. Prefer SSHActuator unless you +// specifically want to avoid SSH access. +type APIActuator struct { + token string + scheme string + port int + hostPrefix string + httpClient *http.Client +} + +// NewAPIActuator returns an APIActuator targeting each node's watchtower vhost +// (hostPrefix + the node host, e.g. "watchtower-"). scheme defaults to +// "https"; hostPrefix defaults to "watchtower-"; port 0 omits the port (so 443 +// for https). The watchtower vhost is bearer-auth only — no basic auth. +func NewAPIActuator(token, scheme string, port int, hostPrefix string) *APIActuator { + if scheme == "" { + scheme = "https" + } + + if hostPrefix == "" { + hostPrefix = "watchtower-" + } + + return &APIActuator{ + token: token, + scheme: scheme, + port: port, + hostPrefix: hostPrefix, + httpClient: &http.Client{Timeout: 60 * time.Second}, + } +} + +// Name implements Actuator. +func (a *APIActuator) Name() string { return "api" } + +// Roll implements Actuator: POST /v1/update to the target's watchtower API. +func (a *APIActuator) Roll(ctx context.Context, target Target, image string) error { + host := sshHost(target.SSH) + if host == "" { + return fmt.Errorf("invalid target %q", target.SSH) + } + + host = a.hostPrefix + host + + endpoint := fmt.Sprintf("%s://%s/v1/update", a.scheme, host) + if a.port != 0 { + endpoint = fmt.Sprintf("%s://%s:%d/v1/update", a.scheme, host, a.port) + } + + if image != "" { + endpoint += "?image=" + url.QueryEscape(image) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, http.NoBody) + if err != nil { + return err + } + + req.Header.Set("Authorization", "Bearer "+a.token) + + resp, err := a.httpClient.Do(req) + if err != nil { + return fmt.Errorf("trigger update: %w", err) + } + defer resp.Body.Close() + + body, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("watchtower returned %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + return nil +} + +// sshHost extracts the host (FQDN) from an "user@host[:port]" SSH value. +func sshHost(target string) string { + host := target + if at := strings.Index(host, "@"); at >= 0 { + host = host[at+1:] + } + + if h, _, err := net.SplitHostPort(host); err == nil { + return h + } + + return host +} diff --git a/pkg/roll/dora.go b/pkg/roll/dora.go new file mode 100644 index 0000000..130bd4f --- /dev/null +++ b/pkg/roll/dora.go @@ -0,0 +1,113 @@ +package roll + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "sync" + "time" +) + +const ( + // doraReadyStatus is Dora's status for a fully-synced, healthy node. + doraReadyStatus = "ready" + doraCacheTTL = 3 * time.Second +) + +// DoraHealth determines node health from a Dora explorer — the source of truth +// the rest of the ethpandaops stack already uses. One API call covers the whole +// fleet, and the endpoint is unauthenticated, so it avoids per-node basic-auth +// beacon calls. Dora's status already accounts for the wall-clock head, so a +// node reporting "ready" is genuinely caught up. +type DoraHealth struct { + baseURL string + httpClient *http.Client + + mu sync.Mutex + cache map[string]doraClient + fetched time.Time +} + +type doraClient struct { + Name string `json:"client_name"` + Status string `json:"status"` + HeadSlot uint64 `json:"head_slot"` +} + +type doraClientsResponse struct { + Clients []doraClient `json:"clients"` +} + +// NewDoraHealth returns a DoraHealth checker for the given Dora base URL. +func NewDoraHealth(baseURL string) *DoraHealth { + return &DoraHealth{ + baseURL: strings.TrimRight(baseURL, "/"), + httpClient: &http.Client{Timeout: 15 * time.Second}, + } +} + +// DoraURLForNetwork returns the conventional Dora URL for an ethpandaops network +// (e.g. "glamsterdam-devnet-4" -> https://dora.glamsterdam-devnet-4.ethpandaops.io). +func DoraURLForNetwork(network string) string { + return fmt.Sprintf("https://dora.%s.ethpandaops.io", network) +} + +// Healthy reports whether the named node is "ready" according to Dora. +func (d *DoraHealth) Healthy(ctx context.Context, node string) (bool, string, error) { + clients, err := d.clients(ctx) + if err != nil { + return false, "", err + } + + c, ok := clients[node] + if !ok { + return false, "not found in dora", nil + } + + if !strings.EqualFold(c.Status, doraReadyStatus) { + return false, fmt.Sprintf("dora status=%s (head=%d)", c.Status, c.HeadSlot), nil + } + + return true, fmt.Sprintf("ready (head=%d)", c.HeadSlot), nil +} + +func (d *DoraHealth) clients(ctx context.Context) (map[string]doraClient, error) { + d.mu.Lock() + defer d.mu.Unlock() + + if d.cache != nil && time.Since(d.fetched) < doraCacheTTL { + return d.cache, nil + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, d.baseURL+"/api/v1/clients/consensus", nil) + if err != nil { + return nil, err + } + + resp, err := d.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("dora clients: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("dora clients returned %d", resp.StatusCode) + } + + var parsed doraClientsResponse + if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { + return nil, fmt.Errorf("decode dora clients: %w", err) + } + + clients := make(map[string]doraClient, len(parsed.Clients)) + for _, c := range parsed.Clients { + clients[c.Name] = c + } + + d.cache = clients + d.fetched = time.Now() + + return clients, nil +} diff --git a/pkg/roll/engine.go b/pkg/roll/engine.go new file mode 100644 index 0000000..faf9925 --- /dev/null +++ b/pkg/roll/engine.go @@ -0,0 +1,316 @@ +package roll + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/sirupsen/logrus" +) + +// Options tunes a rollout. +type Options struct { + // Image scopes the roll to a single image (empty = all watched containers). + Image string + // DelayBetweenNodes is the wait after a node recovers before the next. + DelayBetweenNodes time.Duration + // PostTriggerWait is the grace period after triggering before polling recovery. + PostTriggerWait time.Duration + // WaitTimeout is the per-node recovery timeout; exceeding it aborts the rollout. + WaitTimeout time.Duration + // HealthCheckInterval is how often to poll beacon health while waiting. + HealthCheckInterval time.Duration + // MaxSyncDistance is the largest sync distance (slots) still considered healthy. + MaxSyncDistance uint64 + // SkipHealth disables beacon health gating (trigger-and-go). + SkipHealth bool + // DryRun logs intent without triggering rolls. + DryRun bool + // DoraURL, if set, makes health checks use Dora (matched by node name) as the + // source of truth instead of per-node beacon calls. Preferred — one + // unauthenticated call covers the fleet. + DoraURL string + // BeaconBasicAuthUser and BeaconBasicAuthPass authenticate beacon health + // checks when Dora is not used and the beacon is behind nginx basic auth. + BeaconBasicAuthUser string + BeaconBasicAuthPass string + // OnProgress, if set, is invoked at each rollout milestone (for UIs such as + // the Discord command). It must be cheap and non-blocking. + OnProgress func(Progress) +} + +// Phase is a rollout milestone for progress reporting. +type Phase string + +const ( + PhaseTriggering Phase = "triggering" + PhaseHealthy Phase = "healthy" + PhaseFailed Phase = "failed" + PhaseSkipped Phase = "skipped" + PhaseDone Phase = "done" +) + +// Progress is a rollout milestone delivered to Options.OnProgress. +type Progress struct { + Node string + Index int // 1-based; 0 for fleet-level events + Total int + Phase Phase + Message string +} + +func (o *Options) applyDefaults() { + if o.DelayBetweenNodes == 0 { + o.DelayBetweenNodes = time.Minute + } + if o.PostTriggerWait == 0 { + o.PostTriggerWait = 30 * time.Second + } + if o.WaitTimeout == 0 { + o.WaitTimeout = 10 * time.Minute + } + if o.HealthCheckInterval == 0 { + o.HealthCheckInterval = 10 * time.Second + } + if o.MaxSyncDistance == 0 { + o.MaxSyncDistance = 4 + } +} + +// Engine performs gated, sequential rollouts via an Actuator, gating on beacon +// health between nodes and aborting on the first node that fails to recover. +type Engine struct { + actuator Actuator + health *BeaconHealth + dora *DoraHealth + log logrus.FieldLogger +} + +// NewEngine returns an Engine. +func NewEngine(actuator Actuator, log logrus.FieldLogger) *Engine { + if log == nil { + log = logrus.New() + } + + return &Engine{actuator: actuator, health: NewBeaconHealth(), log: log} +} + +// checkHealth gates on Dora when a Dora URL is configured (the preferred, +// unauthenticated source of truth), otherwise falls back to the node's beacon. +func (e *Engine) checkHealth(ctx context.Context, target Target, maxSyncDistance uint64) (bool, string, error) { + if e.dora != nil { + return e.dora.Healthy(ctx, target.Name) + } + + if target.BeaconURL == "" { + return false, "no health source (set server doraURL or node beaconUrl)", nil + } + + return e.health.Healthy(ctx, target.BeaconURL, maxSyncDistance) +} + +// Run rolls the targets in order. It aborts on the first node that fails to +// recover, leaving the remaining targets untouched. +func (e *Engine) Run(ctx context.Context, targets []Target, opts Options) error { + opts.applyDefaults() + e.health.SetBasicAuth(opts.BeaconBasicAuthUser, opts.BeaconBasicAuthPass) + + if opts.DoraURL != "" { + e.dora = NewDoraHealth(opts.DoraURL) + } + + if len(targets) == 0 { + return errors.New("no targets selected") + } + + e.log.WithFields(logrus.Fields{ + "targets": len(targets), + "actuator": e.actuator.Name(), + "image": opts.Image, + "dry_run": opts.DryRun, + }).Info("roll: starting") + + if !opts.SkipHealth { + if err := e.preflight(ctx, targets, opts); err != nil { + return fmt.Errorf("pre-flight: %w", err) + } + } + + for i, target := range targets { + if err := ctx.Err(); err != nil { + return err + } + + entry := e.log.WithFields(logrus.Fields{ + "node": target.Name, + "step": fmt.Sprintf("%d/%d", i+1, len(targets)), + }) + + progress := Progress{Node: target.Name, Index: i + 1, Total: len(targets)} + + if opts.DryRun { + entry.Info("roll: dry-run, would trigger update") + e.emit(opts, progress, PhaseSkipped, "dry-run: would trigger update") + + continue + } + + e.emit(opts, progress, PhaseTriggering, "") + + if err := e.rollOne(ctx, entry, target, opts); err != nil { + entry.WithError(err).Error("roll: aborted") + e.emit(opts, progress, PhaseFailed, err.Error()) + e.reportRemaining(targets[i:]) + + return fmt.Errorf("node %q: %w", target.Name, err) + } + + e.emit(opts, progress, PhaseHealthy, "recovered") + + if i < len(targets)-1 { + entry.WithField("delay", opts.DelayBetweenNodes).Info("roll: node healthy, waiting before next") + + if err := sleep(ctx, opts.DelayBetweenNodes); err != nil { + return err + } + } + } + + e.emit(opts, Progress{Total: len(targets)}, PhaseDone, "all targets updated and healthy") + e.log.Info("roll: complete, all targets updated and healthy") + + return nil +} + +// emit delivers a progress milestone to Options.OnProgress, if set. +func (e *Engine) emit(opts Options, p Progress, phase Phase, msg string) { + if opts.OnProgress == nil { + return + } + + p.Phase = phase + if msg != "" { + p.Message = msg + } + + opts.OnProgress(p) +} + +func (e *Engine) rollOne(ctx context.Context, entry logrus.FieldLogger, target Target, opts Options) error { + if !opts.SkipHealth { + ok, reason, err := e.checkHealth(ctx, target, opts.MaxSyncDistance) + if err != nil { + return fmt.Errorf("pre-update health check: %w", err) + } + + if !ok { + return fmt.Errorf("not healthy before update: %s", reason) + } + } + + entry.Info("roll: triggering update") + + if err := e.actuator.Roll(ctx, target, opts.Image); err != nil { + return fmt.Errorf("trigger: %w", err) + } + + if err := sleep(ctx, opts.PostTriggerWait); err != nil { + return err + } + + if opts.SkipHealth { + entry.Warn("roll: skipping recovery health check") + + return nil + } + + entry.Info("roll: waiting for recovery") + + return e.waitHealthy(ctx, entry, target, opts) +} + +func (e *Engine) waitHealthy(ctx context.Context, entry logrus.FieldLogger, target Target, opts Options) error { + deadline := time.Now().Add(opts.WaitTimeout) + ticker := time.NewTicker(opts.HealthCheckInterval) + defer ticker.Stop() + + for { + ok, reason, err := e.checkHealth(ctx, target, opts.MaxSyncDistance) + switch { + case err != nil: + entry.WithError(err).Debug("roll: health check failed, retrying") + case ok: + entry.Info("roll: recovered") + + return nil + default: + entry.WithField("status", reason).Debug("roll: not yet healthy") + } + + if time.Now().After(deadline) { + return fmt.Errorf("did not recover within %s", opts.WaitTimeout) + } + + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func (e *Engine) preflight(ctx context.Context, targets []Target, opts Options) error { + var unhealthy []string + + for _, target := range targets { + ok, reason, err := e.checkHealth(ctx, target, opts.MaxSyncDistance) + entry := e.log.WithField("node", target.Name) + + switch { + case err != nil: + entry.WithError(err).Warn("roll: pre-flight health error") + unhealthy = append(unhealthy, target.Name) + case !ok: + entry.WithField("status", reason).Warn("roll: pre-flight not healthy") + unhealthy = append(unhealthy, target.Name) + default: + entry.WithField("status", reason).Info("roll: pre-flight healthy") + } + } + + if len(unhealthy) > 0 { + return fmt.Errorf("%d node(s) not healthy: %v", len(unhealthy), unhealthy) + } + + return nil +} + +func (e *Engine) reportRemaining(remaining []Target) { + if len(remaining) <= 1 { + return + } + + names := make([]string, 0, len(remaining)-1) + for _, t := range remaining[1:] { + names = append(names, t.Name) + } + + e.log.WithField("nodes", names).Warn("roll: the following nodes were NOT updated") +} + +func sleep(ctx context.Context, d time.Duration) error { + if d <= 0 { + return nil + } + + timer := time.NewTimer(d) + defer timer.Stop() + + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/pkg/roll/health.go b/pkg/roll/health.go new file mode 100644 index 0000000..ef52aee --- /dev/null +++ b/pkg/roll/health.go @@ -0,0 +1,89 @@ +package roll + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" +) + +// BeaconHealth checks a beacon node's sync status for rollout gating. +type BeaconHealth struct { + httpClient *http.Client + user string + pass string +} + +// NewBeaconHealth returns a BeaconHealth checker. +func NewBeaconHealth() *BeaconHealth { + return &BeaconHealth{httpClient: &http.Client{Timeout: 10 * time.Second}} +} + +// SetBasicAuth sets HTTP basic auth for beacon requests, needed when the beacon +// endpoint is behind nginx basic auth (as the bn-* vhosts are). +func (b *BeaconHealth) SetBasicAuth(user, pass string) { + b.user = user + b.pass = pass +} + +type syncingResponse struct { + Data struct { + HeadSlot string `json:"head_slot"` + SyncDistance string `json:"sync_distance"` + IsSyncing bool `json:"is_syncing"` + IsOptimistic bool `json:"is_optimistic"` + ELOffline bool `json:"el_offline"` + } `json:"data"` +} + +// Healthy reports whether the beacon at beaconURL is synced within +// maxSyncDistance slots and not syncing/optimistic/EL-offline. The returned +// string is a human-readable status. +func (b *BeaconHealth) Healthy(ctx context.Context, beaconURL string, maxSyncDistance uint64) (bool, string, error) { + url := strings.TrimRight(beaconURL, "/") + "/eth/v1/node/syncing" + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return false, "", err + } + + if b.user != "" || b.pass != "" { + req.SetBasicAuth(b.user, b.pass) + } + + resp, err := b.httpClient.Do(req) + if err != nil { + return false, "", err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return false, "", fmt.Errorf("beacon /syncing returned %d", resp.StatusCode) + } + + var body syncingResponse + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + return false, "", fmt.Errorf("decode /syncing: %w", err) + } + + dist, err := strconv.ParseUint(body.Data.SyncDistance, 10, 64) + if err != nil { + return false, "", fmt.Errorf("parse sync_distance %q: %w", body.Data.SyncDistance, err) + } + + switch { + case body.Data.IsSyncing: + return false, fmt.Sprintf("syncing (distance=%d)", dist), nil + case body.Data.IsOptimistic: + return false, "optimistic (execution payload not validated)", nil + case body.Data.ELOffline: + return false, "execution layer offline", nil + case dist > maxSyncDistance: + return false, fmt.Sprintf("sync distance %d exceeds max %d", dist, maxSyncDistance), nil + } + + return true, fmt.Sprintf("synced (head=%s, distance=%d)", body.Data.HeadSlot, dist), nil +} diff --git a/pkg/roll/inventory.go b/pkg/roll/inventory.go new file mode 100644 index 0000000..7746c51 --- /dev/null +++ b/pkg/roll/inventory.go @@ -0,0 +1,210 @@ +// Package roll performs gated, sequential container image rollouts across an +// Ethereum node fleet. Targets are resolved from cartographoor's published +// per-network inventory; the roll itself is executed by a pluggable Actuator +// (SSH-to-local-watchtower by default). +package roll + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sort" + "strings" + "time" +) + +// DefaultInventoryBaseURL is where cartographoor publishes per-network inventory. +const DefaultInventoryBaseURL = "https://ethpandaops-platform-production-cartographoor.ams3.digitaloceanspaces.com" + +// clientInfo mirrors the relevant fields of cartographoor's inventory ClientInfo. +type clientInfo struct { + ClientName string `json:"clientName"` + ClientType string `json:"clientType"` + Version string `json:"version"` + DockerImage string `json:"dockerImage"` + SSH string `json:"ssh"` + BeaconAPI string `json:"bn"` + RPC string `json:"rpc"` + Status string `json:"status"` +} + +type inventoryData struct { + Network string `json:"network"` + ConsensusClients []clientInfo `json:"consensusClients"` + ExecutionClients []clientInfo `json:"executionClients"` +} + +// Target is a single host to roll, grouped from the inventory by its SSH host. +type Target struct { + // Name is the host/node identifier (derived from the SSH host). + Name string + // SSH is the cartographoor ssh value, e.g. "devops@host". + SSH string + // BeaconURL is the host's beacon API base URL for health gating (may be empty). + BeaconURL string + // Clients are the client names running on this host (CL and EL). + Clients []string + // tokens are the lowercased selectable identifiers for this host: node name, + // client types, and the cl_el group. Used by the --limit/--client matcher. + tokens []string +} + +// FetchInventory loads and parses a network's inventory from cartographoor. +func FetchInventory(ctx context.Context, baseURL, network string) (*inventoryData, error) { + if baseURL == "" { + baseURL = DefaultInventoryBaseURL + } + + endpoint := fmt.Sprintf("%s/inventory/%s.json", strings.TrimRight(baseURL, "/"), network) + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + + resp, err := (&http.Client{Timeout: 20 * time.Second}).Do(req) + if err != nil { + return nil, fmt.Errorf("fetch inventory for %s: %w", network, err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("inventory for %s: status %d", network, resp.StatusCode) + } + + var data inventoryData + if err := json.NewDecoder(resp.Body).Decode(&data); err != nil { + return nil, fmt.Errorf("decode inventory for %s: %w", network, err) + } + + return &data, nil +} + +// ResolveTargets groups the inventory by SSH host into roll targets, computing +// match tokens (node name, CL/EL client types, and the cl_el group) for +// Ansible-style selection. beaconScheme (e.g. "https") is prepended to the bare +// beacon hostname from the consensus client entry. +func ResolveTargets(inv *inventoryData, beaconScheme string) []Target { + if beaconScheme == "" { + beaconScheme = "https" + } + + type agg struct { + ssh string + beacon string + clients []string + clTypes []string + elTypes []string + } + + byHost := map[string]*agg{} + order := []string{} + + get := func(ssh string) *agg { + a, ok := byHost[ssh] + if !ok { + a = &agg{ssh: ssh} + byHost[ssh] = a + order = append(order, ssh) + } + + return a + } + + for _, c := range inv.ConsensusClients { + if c.SSH == "" { + continue + } + + a := get(c.SSH) + if c.ClientName != "" { + a.clients = append(a.clients, c.ClientName) + } + + if c.ClientType != "" { + a.clTypes = append(a.clTypes, c.ClientType) + } + + if c.BeaconAPI != "" && a.beacon == "" { + a.beacon = beaconScheme + "://" + c.BeaconAPI + } + } + + for _, c := range inv.ExecutionClients { + if c.SSH == "" { + continue + } + + a := get(c.SSH) + if c.ClientType != "" { + a.elTypes = append(a.elTypes, c.ClientType) + } + } + + targets := make([]Target, 0, len(order)) + + for _, ssh := range order { + a := byHost[ssh] + name := hostName(ssh) + + targets = append(targets, Target{ + Name: name, + SSH: ssh, + BeaconURL: a.beacon, + Clients: a.clients, + tokens: buildTokens(name, a.clTypes, a.elTypes), + }) + } + + sort.Slice(targets, func(i, j int) bool { return targets[i].Name < targets[j].Name }) + + return targets +} + +// buildTokens returns the lowercased, deduped set of selectable tokens for a +// host: its node name, each client type, and each cl_el group pairing. +func buildTokens(name string, clTypes, elTypes []string) []string { + seen := map[string]bool{} + tokens := []string{} + + add := func(s string) { + s = strings.ToLower(strings.TrimSpace(s)) + if s != "" && !seen[s] { + seen[s] = true + tokens = append(tokens, s) + } + } + + add(name) + + for _, cl := range clTypes { + add(cl) + } + + for _, el := range elTypes { + add(el) + } + + for _, cl := range clTypes { + for _, el := range elTypes { + add(cl + "_" + el) + } + } + + return tokens +} + +// hostName derives a short host identifier from an "user@host.domain" SSH value. +func hostName(ssh string) string { + host := ssh + if at := strings.Index(host, "@"); at >= 0 { + host = host[at+1:] + } + + if dot := strings.Index(host, "."); dot >= 0 { + host = host[:dot] + } + + return host +} diff --git a/pkg/roll/match.go b/pkg/roll/match.go new file mode 100644 index 0000000..bf4be33 --- /dev/null +++ b/pkg/roll/match.go @@ -0,0 +1,115 @@ +package roll + +import ( + "path" + "strings" +) + +// Select filters targets by an Ansible-style host expression (the --client +// flag): a comma/colon-separated list of glob terms, '!' to exclude, and the +// 'all' keyword. Terms match a host's node name, client types, or cl_el group. +// An empty expression selects all targets. +func Select(targets []Target, expr string) []Target { + if strings.TrimSpace(expr) == "" { + return targets + } + + return applyLimit(targets, expr) +} + +// applyLimit evaluates a comma/colon-separated expression of glob terms, with +// '!' for exclusion and the 'all' keyword. Includes are unioned; exclusions are +// subtracted. If the expression contains only exclusions, the base is all +// targets (so `!buildoor-*` means "everything except buildoor"). +func applyLimit(targets []Target, expr string) []Target { + var includes, excludes []string + + for _, term := range splitTerms(expr) { + switch { + case term == "": + case strings.HasPrefix(term, "!"): + excludes = append(excludes, strings.TrimPrefix(term, "!")) + default: + includes = append(includes, term) + } + } + + base := selectBase(targets, includes) + + out := make([]Target, 0, len(base)) + + for _, t := range base { + if !matchesAny(t, excludes) { + out = append(out, t) + } + } + + return out +} + +func selectBase(targets []Target, includes []string) []Target { + if len(includes) == 0 || containsFold(includes, "all") { + return targets + } + + out := make([]Target, 0, len(targets)) + + for _, t := range targets { + if matchesAny(t, includes) { + out = append(out, t) + } + } + + return out +} + +func matchesAny(t Target, terms []string) bool { + for _, term := range terms { + if matchTerm(t, term) { + return true + } + } + + return false +} + +// matchTerm reports whether a term (a case-insensitive glob) matches any of the +// target's tokens (node name, client types, and the cl_el group). +func matchTerm(t Target, term string) bool { + term = strings.ToLower(strings.TrimSpace(term)) + if term == "" { + return false + } + + if term == "all" { + return true + } + + for _, tok := range t.tokens { + if ok, err := path.Match(term, tok); err == nil && ok { + return true + } + } + + return false +} + +// splitTerms splits a limit expression on commas and colons. +func splitTerms(expr string) []string { + parts := strings.FieldsFunc(expr, func(r rune) bool { return r == ',' || r == ':' }) + for i := range parts { + parts[i] = strings.TrimSpace(parts[i]) + } + + return parts +} + +func containsFold(list []string, want string) bool { + for _, s := range list { + if strings.EqualFold(s, want) { + return true + } + } + + return false +} diff --git a/pkg/roll/match_test.go b/pkg/roll/match_test.go new file mode 100644 index 0000000..bde9858 --- /dev/null +++ b/pkg/roll/match_test.go @@ -0,0 +1,108 @@ +package roll + +import ( + "reflect" + "testing" +) + +func testInventory() *inventoryData { + return &inventoryData{ + Network: "glamsterdam-devnet-4", + ConsensusClients: []clientInfo{ + {ClientName: "lighthouse-ethrex-1", ClientType: "lighthouse", SSH: "devops@lighthouse-ethrex-1.example.io", BeaconAPI: "bn-lighthouse-ethrex-1.example.io"}, + {ClientName: "lighthouse-nethermind-1", ClientType: "lighthouse", SSH: "devops@lighthouse-nethermind-1.example.io", BeaconAPI: "bn-lighthouse-nethermind-1.example.io"}, + {ClientName: "prysm-ethrex-1", ClientType: "prysm", SSH: "devops@prysm-ethrex-1.example.io", BeaconAPI: "bn-prysm-ethrex-1.example.io"}, + }, + ExecutionClients: []clientInfo{ + {ClientName: "lighthouse-ethrex-1", ClientType: "ethrex", SSH: "devops@lighthouse-ethrex-1.example.io"}, + {ClientName: "lighthouse-nethermind-1", ClientType: "nethermind", SSH: "devops@lighthouse-nethermind-1.example.io"}, + {ClientName: "prysm-ethrex-1", ClientType: "ethrex", SSH: "devops@prysm-ethrex-1.example.io"}, + }, + } +} + +func targetNames(ts []Target) []string { + out := make([]string, len(ts)) + for i, t := range ts { + out[i] = t.Name + } + + return out +} + +func TestResolveTargets(t *testing.T) { + targets := ResolveTargets(testInventory(), "https") + if len(targets) != 3 { + t.Fatalf("want 3 targets, got %d (%v)", len(targets), targetNames(targets)) + } + + var lh Target + + for _, tg := range targets { + if tg.Name == "lighthouse-ethrex-1" { + lh = tg + } + } + + if lh.BeaconURL != "https://bn-lighthouse-ethrex-1.example.io" { + t.Errorf("beacon url = %q", lh.BeaconURL) + } + + got := map[string]bool{} + for _, tok := range lh.tokens { + got[tok] = true + } + + for _, want := range []string{"lighthouse-ethrex-1", "lighthouse", "ethrex", "lighthouse_ethrex"} { + if !got[want] { + t.Errorf("missing token %q in %v", want, lh.tokens) + } + } +} + +func TestSelect(t *testing.T) { + targets := ResolveTargets(testInventory(), "https") + + cases := []struct { + expr string + want []string + }{ + {"", []string{"lighthouse-ethrex-1", "lighthouse-nethermind-1", "prysm-ethrex-1"}}, + {"all", []string{"lighthouse-ethrex-1", "lighthouse-nethermind-1", "prysm-ethrex-1"}}, + {"lighthouse", []string{"lighthouse-ethrex-1", "lighthouse-nethermind-1"}}, + {"lighthouse_ethrex", []string{"lighthouse-ethrex-1"}}, + {"lighthouse-ethrex-1", []string{"lighthouse-ethrex-1"}}, + {"lighthouse-*", []string{"lighthouse-ethrex-1", "lighthouse-nethermind-1"}}, + {"*-ethrex-1", []string{"lighthouse-ethrex-1", "prysm-ethrex-1"}}, + {"all:!prysm-*", []string{"lighthouse-ethrex-1", "lighthouse-nethermind-1"}}, + {"!prysm-*", []string{"lighthouse-ethrex-1", "lighthouse-nethermind-1"}}, + {"lighthouse-*:!*nethermind*", []string{"lighthouse-ethrex-1"}}, + } + + for _, tc := range cases { + if got := targetNames(Select(targets, tc.expr)); !reflect.DeepEqual(got, tc.want) { + t.Errorf("Select(%q) = %v, want %v", tc.expr, got, tc.want) + } + } +} + +func TestSuggestionsScope(t *testing.T) { + targets := ResolveTargets(testInventory(), "https") + + set := map[string]bool{} + for _, s := range Suggestions(targets, "", "lighthouse", 25) { + set[s] = true + } + + for _, want := range []string{"lighthouse", "lighthouse-ethrex-1", "lighthouse_ethrex"} { + if !set[want] { + t.Errorf("scope lighthouse missing token %q", want) + } + } + + for _, bad := range []string{"prysm", "prysm-ethrex-1", "all"} { + if set[bad] { + t.Errorf("scope lighthouse should not include %q", bad) + } + } +} diff --git a/pkg/roll/provider.go b/pkg/roll/provider.go new file mode 100644 index 0000000..20754f6 --- /dev/null +++ b/pkg/roll/provider.go @@ -0,0 +1,133 @@ +package roll + +import ( + "context" + "sort" + "strings" + "sync" + "time" +) + +const defaultProviderTTL = 30 * time.Second + +// InventoryProvider fetches and caches per-network targets from cartographoor, +// so autocomplete and rollouts don't refetch on every Discord interaction. +type InventoryProvider struct { + baseURL string + beaconScheme string + ttl time.Duration + + mu sync.Mutex + cache map[string]cacheEntry +} + +type cacheEntry struct { + targets []Target + fetched time.Time +} + +// NewInventoryProvider returns a provider. Empty baseURL/beaconScheme use +// defaults; non-positive ttl uses a 30s default. +func NewInventoryProvider(baseURL, beaconScheme string, ttl time.Duration) *InventoryProvider { + if baseURL == "" { + baseURL = DefaultInventoryBaseURL + } + + if beaconScheme == "" { + beaconScheme = "https" + } + + if ttl <= 0 { + ttl = defaultProviderTTL + } + + return &InventoryProvider{ + baseURL: baseURL, + beaconScheme: beaconScheme, + ttl: ttl, + cache: map[string]cacheEntry{}, + } +} + +// Targets returns the cached targets for a network, refreshing if stale. +func (p *InventoryProvider) Targets(ctx context.Context, network string) ([]Target, error) { + p.mu.Lock() + if e, ok := p.cache[network]; ok && time.Since(e.fetched) < p.ttl { + targets := e.targets + p.mu.Unlock() + + return targets, nil + } + p.mu.Unlock() + + inv, err := FetchInventory(ctx, p.baseURL, network) + if err != nil { + return nil, err + } + + targets := ResolveTargets(inv, p.beaconScheme) + + p.mu.Lock() + p.cache[network] = cacheEntry{targets: targets, fetched: time.Now()} + p.mu.Unlock() + + return targets, nil +} + +// Suggest returns up to limit selectable identifiers for a network's hosts, +// filtered by input/scope (see Suggestions). +func (p *InventoryProvider) Suggest(ctx context.Context, network, input, scope string, limit int) ([]string, error) { + targets, err := p.Targets(ctx, network) + if err != nil { + return nil, err + } + + return Suggestions(targets, input, scope, limit), nil +} + +// Suggestions computes the selectable token list from resolved targets: group +// names (client types and cl_el pairings), node names, and "all". Results are +// filtered to those containing the case-insensitive input substring, and — when +// scope is non-empty — to those containing scope (e.g. only lighthouse-related +// tokens when a roll is triggered from a lighthouse build). Sorted, capped at +// limit. +func Suggestions(targets []Target, input, scope string, limit int) []string { + input = strings.ToLower(strings.TrimSpace(input)) + scope = strings.ToLower(strings.TrimSpace(scope)) + + seen := map[string]bool{} + out := []string{} + + add := func(tok string) { + if tok == "" || seen[tok] { + return + } + + if scope != "" && !strings.Contains(tok, scope) { + return + } + + if input != "" && !strings.Contains(tok, input) { + return + } + + seen[tok] = true + out = append(out, tok) + } + + add("all") + + for _, t := range targets { + for _, tok := range t.tokens { + add(tok) + } + } + + sort.Strings(out) + + if limit > 0 && len(out) > limit { + out = out[:limit] + } + + return out +} diff --git a/pkg/roll/ssh.go b/pkg/roll/ssh.go new file mode 100644 index 0000000..76f7797 --- /dev/null +++ b/pkg/roll/ssh.go @@ -0,0 +1,209 @@ +package roll + +import ( + "bytes" + "context" + "fmt" + "net" + "os" + "regexp" + "strconv" + "strings" + "time" + + "github.com/sirupsen/logrus" + "golang.org/x/crypto/ssh" +) + +// DefaultWatchtowerContainer is the watchtower container name deployed by the +// ethpandaops ethereum_node Ansible role. +const DefaultWatchtowerContainer = "ethereum-node-docker-watchtower" + +// DefaultWatchtowerPort is watchtower's default HTTP API port. +const DefaultWatchtowerPort = 8080 + +const sshDialTimeout = 15 * time.Second + +// safeRef guards values interpolated into the remote shell script. Single-quoted +// in the script, this charset (no quotes/spaces/metachars) prevents injection. +var safeRef = regexp.MustCompile(`^[A-Za-z0-9._/:@-]*$`) + +// SSHActuator rolls an image by SSHing to the host and triggering the +// node-local watchtower HTTP API. It discovers watchtower's container IP and +// API token from `docker inspect` on the node, so it needs no published port +// and no token stored centrally — only SSH access (which devops already has). +type SSHActuator struct { + signer ssh.Signer + hostKeyCB ssh.HostKeyCallback + containerName string + port int + log logrus.FieldLogger +} + +// SSHConfig configures an SSHActuator. +type SSHConfig struct { + // PrivateKeyPath is the path to the SSH private key used to authenticate. + PrivateKeyPath string + // KnownHostsCallback verifies host keys. If nil, host keys are not verified + // (acceptable for ephemeral devnets; a warning is logged). + KnownHostsCallback ssh.HostKeyCallback + // ContainerName overrides the watchtower container name. + ContainerName string + // Port overrides the watchtower HTTP API port. + Port int + // Log is the logger. + Log logrus.FieldLogger +} + +// NewSSHActuator builds an SSHActuator from the given config. +func NewSSHActuator(cfg SSHConfig) (*SSHActuator, error) { + if cfg.Log == nil { + cfg.Log = logrus.New() + } + + if cfg.PrivateKeyPath == "" { + return nil, fmt.Errorf("ssh private key path is required") + } + + keyBytes, err := os.ReadFile(cfg.PrivateKeyPath) + if err != nil { + return nil, fmt.Errorf("read ssh key: %w", err) + } + + signer, err := ssh.ParsePrivateKey(keyBytes) + if err != nil { + return nil, fmt.Errorf("parse ssh key: %w", err) + } + + hostKeyCB := cfg.KnownHostsCallback + if hostKeyCB == nil { + cfg.Log.Warn("roll: SSH host key verification disabled (no known_hosts configured)") + + hostKeyCB = ssh.InsecureIgnoreHostKey() //nolint:gosec // devnet hosts; opt-in known_hosts available + } + + container := cfg.ContainerName + if container == "" { + container = DefaultWatchtowerContainer + } + + port := cfg.Port + if port == 0 { + port = DefaultWatchtowerPort + } + + return &SSHActuator{ + signer: signer, + hostKeyCB: hostKeyCB, + containerName: container, + port: port, + log: cfg.Log, + }, nil +} + +// Name implements Actuator. +func (a *SSHActuator) Name() string { return "ssh" } + +// Roll implements Actuator: SSH to the host and trigger the local watchtower. +func (a *SSHActuator) Roll(ctx context.Context, target Target, image string) error { + if !safeRef.MatchString(image) { + return fmt.Errorf("invalid image reference %q", image) + } + + if !safeRef.MatchString(a.containerName) { + return fmt.Errorf("invalid container name %q", a.containerName) + } + + user, host := splitSSH(target.SSH) + if host == "" { + return fmt.Errorf("invalid ssh target %q", target.SSH) + } + + out, err := a.run(ctx, user, host, a.script(image)) + if err != nil { + return fmt.Errorf("ssh roll %s: %w: %s", target.Name, err, strings.TrimSpace(out)) + } + + return nil +} + +// script builds the remote shell that discovers watchtower's IP + token and +// triggers the update. image is single-quoted and charset-validated by Roll. +func (a *SSHActuator) script(image string) string { + return fmt.Sprintf(`set -e +cn='%s' +img='%s' +ip=$(docker inspect "$cn" -f '{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}' | awk '{print $1}') +[ -n "$ip" ] || { echo "watchtower container $cn not found" >&2; exit 3; } +tok=$(docker inspect "$cn" -f '{{range .Config.Env}}{{println .}}{{end}}' | sed -n 's/^WATCHTOWER_HTTP_API_TOKEN=//p') +[ -n "$tok" ] || { echo "WATCHTOWER_HTTP_API_TOKEN not set on $cn" >&2; exit 4; } +url="http://$ip:%d/v1/update" +[ -n "$img" ] && url="$url?image=$img" +curl -fsS -X POST -H "Authorization: Bearer $tok" "$url" +`, a.containerName, image, a.port) +} + +func (a *SSHActuator) run(ctx context.Context, user, host, script string) (string, error) { + clientCfg := &ssh.ClientConfig{ + User: user, + Auth: []ssh.AuthMethod{ssh.PublicKeys(a.signer)}, + HostKeyCallback: a.hostKeyCB, + Timeout: sshDialTimeout, + } + + dialer := &net.Dialer{Timeout: sshDialTimeout} + + conn, err := dialer.DialContext(ctx, "tcp", host) + if err != nil { + return "", fmt.Errorf("dial: %w", err) + } + + sshConn, chans, reqs, err := ssh.NewClientConn(conn, host, clientCfg) + if err != nil { + _ = conn.Close() + + return "", fmt.Errorf("ssh handshake: %w", err) + } + + client := ssh.NewClient(sshConn, chans, reqs) + defer client.Close() + + session, err := client.NewSession() + if err != nil { + return "", fmt.Errorf("ssh session: %w", err) + } + defer session.Close() + + var buf bytes.Buffer + session.Stdout = &buf + session.Stderr = &buf + + done := make(chan error, 1) + go func() { done <- session.Run(script) }() + + select { + case <-ctx.Done(): + _ = session.Signal(ssh.SIGKILL) + + return buf.String(), ctx.Err() + case err := <-done: + return buf.String(), err + } +} + +// splitSSH parses "user@host[:port]" into user and host:port (defaulting :22). +func splitSSH(target string) (user, hostport string) { + user = "root" + host := target + + if at := strings.Index(target, "@"); at >= 0 { + user = target[:at] + host = target[at+1:] + } + + if _, _, err := net.SplitHostPort(host); err != nil { + host = net.JoinHostPort(host, strconv.Itoa(22)) + } + + return user, host +} diff --git a/pkg/service/config.go b/pkg/service/config.go index d248f50..8b8dc2f 100644 --- a/pkg/service/config.go +++ b/pkg/service/config.go @@ -27,6 +27,10 @@ type Config struct { ClientsDataURL string MetricsAddress string // Defaults to :9091 HealthCheckAddress string // Defaults to :9191 + RollSSHKeyPath string // SSH key path for the /roll command's ssh actuator + WatchtowerAPIToken string // watchtower API token for the /roll command's api actuator + NodeBasicAuthUser string // basic auth user for beacon health endpoints (bn-* vhosts) + NodeBasicAuthPass string // basic auth password for beacon health endpoints } // AsS3Config converts the configuration to an S3Config. diff --git a/pkg/service/service.go b/pkg/service/service.go index de2e386..a2a8f4a 100644 --- a/pkg/service/service.go +++ b/pkg/service/service.go @@ -13,6 +13,7 @@ import ( "github.com/ethpandaops/panda-pulse/pkg/discord/cmd/common" cmdhive "github.com/ethpandaops/panda-pulse/pkg/discord/cmd/hive" "github.com/ethpandaops/panda-pulse/pkg/discord/cmd/mentions" + cmdroll "github.com/ethpandaops/panda-pulse/pkg/discord/cmd/roll" "github.com/ethpandaops/panda-pulse/pkg/grafana" "github.com/ethpandaops/panda-pulse/pkg/hive" httpclient "github.com/ethpandaops/panda-pulse/pkg/http" @@ -146,6 +147,13 @@ func NewService(ctx context.Context, log *logrus.Logger, cfg *Config) (*Service, mentions.NewMentionsCommand(log, bot), cmdhive.NewHiveCommand(log, bot, cfg.GithubToken, githubHTTPClient), build.NewBuildCommand(log, bot, cfg.GithubToken, githubHTTPClient), + cmdroll.NewRollCommand(log, bot, cmdroll.Config{ + Actuator: "api", + WatchtowerToken: cfg.WatchtowerAPIToken, + SSHKeyPath: cfg.RollSSHKeyPath, + BasicAuthUser: cfg.NodeBasicAuthUser, + BasicAuthPass: cfg.NodeBasicAuthPass, + }), }) return &Service{ From d8ecfc3019a99312eee7c5dd1ccc975b767b0e08 Mon Sep 17 00:00:00 2001 From: Barnabas Busa Date: Fri, 29 May 2026 16:01:06 +0200 Subject: [PATCH 2/3] refactor(roll): drop SSH actuator, fix lint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The watchtower-vhost API actuator + Dora health gating is the only path now, so the SSH actuator was dead weight. Remove it and all SSH wiring (config, flags, env, golang.org/x/crypto/ssh). Target.SSH is retained purely as the host string the API actuator derives watchtower- from. Also satisfies golangci-lint (govet shadow, wsl_v5 whitespace, and tagliatelle nolints on the beacon/Dora snake_case API structs). The reviewer's reported error-handling bug was a false positive — c.actuator()'s error is already captured and checked at the call site. --- cmd/main.go | 1 - cmd/roll.go | 30 ++--- go.mod | 2 +- pkg/discord/cmd/roll/command.go | 13 +- pkg/discord/cmd/roll/run.go | 10 +- pkg/roll/api.go | 7 +- pkg/roll/dora.go | 1 + pkg/roll/engine.go | 7 ++ pkg/roll/health.go | 5 +- pkg/roll/ssh.go | 209 -------------------------------- pkg/service/config.go | 3 +- pkg/service/service.go | 2 - 12 files changed, 27 insertions(+), 263 deletions(-) delete mode 100644 pkg/roll/ssh.go diff --git a/cmd/main.go b/cmd/main.go index 8879a7b..f7828b0 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -103,7 +103,6 @@ func setConfig(cfg *service.Config) { cfg.S3EndpointURL = os.Getenv("AWS_ENDPOINT_URL") cfg.HealthCheckAddress = os.Getenv("HEALTH_CHECK_ADDRESS") cfg.MetricsAddress = os.Getenv("METRICS_ADDRESS") - cfg.RollSSHKeyPath = os.Getenv("ROLL_SSH_KEY") cfg.WatchtowerAPIToken = os.Getenv("WATCHTOWER_HTTP_API_TOKEN") cfg.NodeBasicAuthUser = os.Getenv("ROLL_BASIC_AUTH_USER") cfg.NodeBasicAuthPass = os.Getenv("ROLL_BASIC_AUTH_PASS") diff --git a/cmd/roll.go b/cmd/roll.go index 22ea04d..a963f54 100644 --- a/cmd/roll.go +++ b/cmd/roll.go @@ -20,9 +20,7 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { network string client string image string - actuatorKind string inventoryURL string - sshKeyPath string watchtowerPort int watchtowerToken string watchtowerScheme string @@ -63,7 +61,7 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { return fmt.Errorf("no targets matched (network=%s client=%q)", network, client) } - actuator, err := buildActuator(actuatorKind, sshKeyPath, watchtowerToken, watchtowerScheme, watchtowerPrefix, watchtowerPort, log) + actuator, err := buildActuator(watchtowerToken, watchtowerScheme, watchtowerPrefix, watchtowerPort) if err != nil { return err } @@ -96,10 +94,8 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { f.StringVar(&network, "network", "", "network name as published by cartographoor (required)") f.StringVar(&client, "client", "", "host pattern: client/group/node with globs, ! to exclude, 'all' (e.g. 'lighthouse', 'lighthouse_ethrex', 'lighthouse-*:!*-1')") f.StringVar(&image, "image", "", "scope the roll to this image (empty = all watched containers)") - f.StringVar(&actuatorKind, "actuator", "ssh", "how to trigger the roll: ssh|api") f.StringVar(&inventoryURL, "inventory-url", roll.DefaultInventoryBaseURL, "cartographoor inventory base URL") - f.StringVar(&sshKeyPath, "ssh-key", os.Getenv("ROLL_SSH_KEY"), "SSH private key path (ssh actuator; env ROLL_SSH_KEY)") - f.IntVar(&watchtowerPort, "watchtower-port", 0, "watchtower API port (api actuator; 0 = default for the scheme)") + f.IntVar(&watchtowerPort, "watchtower-port", 0, "watchtower API port (0 = default for the scheme)") f.StringVar(&watchtowerToken, "watchtower-token", os.Getenv("WATCHTOWER_HTTP_API_TOKEN"), "watchtower API token (api actuator; env WATCHTOWER_HTTP_API_TOKEN)") f.StringVar(&watchtowerScheme, "watchtower-scheme", "https", "watchtower API scheme (api actuator)") f.StringVar(&watchtowerPrefix, "watchtower-prefix", "watchtower-", "vhost prefix for the watchtower API (api actuator)") @@ -119,22 +115,10 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { return cmd } -func buildActuator(kind, sshKeyPath, token, scheme, prefix string, port int, log *logrus.Logger) (roll.Actuator, error) { - switch kind { - case "ssh": - return roll.NewSSHActuator(roll.SSHConfig{ - PrivateKeyPath: sshKeyPath, - ContainerName: roll.DefaultWatchtowerContainer, - Port: roll.DefaultWatchtowerPort, - Log: log, - }) - case "api": - if token == "" { - return nil, fmt.Errorf("--watchtower-token (or WATCHTOWER_HTTP_API_TOKEN) is required for the api actuator") - } - - return roll.NewAPIActuator(token, scheme, port, prefix), nil - default: - return nil, fmt.Errorf("unknown actuator %q (want ssh or api)", kind) +func buildActuator(token, scheme, prefix string, port int) (roll.Actuator, error) { + if token == "" { + return nil, fmt.Errorf("--watchtower-token (or WATCHTOWER_HTTP_API_TOKEN) is required") } + + return roll.NewAPIActuator(token, scheme, port, prefix), nil } diff --git a/go.mod b/go.mod index fd6a9d3..f1be42f 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,6 @@ require ( github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.42.0 go.uber.org/mock v0.6.0 - golang.org/x/crypto v0.49.0 golang.org/x/text v0.37.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -97,6 +96,7 @@ require ( go.opentelemetry.io/otel/metric v1.42.0 // indirect go.opentelemetry.io/otel/trace v1.42.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect + golang.org/x/crypto v0.49.0 // indirect golang.org/x/sys v0.42.0 // indirect google.golang.org/protobuf v1.36.11 // indirect ) diff --git a/pkg/discord/cmd/roll/command.go b/pkg/discord/cmd/roll/command.go index bc6ceef..bcb8c5d 100644 --- a/pkg/discord/cmd/roll/command.go +++ b/pkg/discord/cmd/roll/command.go @@ -26,21 +26,14 @@ const ( autocompleteLimit = 25 providerCacheTTL = 30 * time.Second autocompleteTime = 2 * time.Second - - actuatorSSH = "ssh" - actuatorAPI = "api" ) // Config configures the roll command's actuator and inventory source. type Config struct { - // SSHKeyPath is the SSH private key used by the ssh actuator. - SSHKeyPath string - // WatchtowerToken is the bearer token used by the api actuator. + // WatchtowerToken is the bearer token for the watchtower HTTP API. WatchtowerToken string // InventoryURL overrides the cartographoor inventory base URL. InventoryURL string - // Actuator selects how rolls are triggered: "ssh" or "api" (default). - Actuator string // BasicAuthUser and BasicAuthPass authenticate beacon health checks behind // nginx basic auth (the bn-* vhosts) — only used when Dora is unavailable. BasicAuthUser string @@ -61,10 +54,6 @@ type Command struct { // NewRollCommand creates the /roll command. func NewRollCommand(log *logrus.Logger, bot common.BotContext, cfg Config) *Command { - if cfg.Actuator == "" { - cfg.Actuator = actuatorSSH - } - return &Command{ log: log, bot: bot, diff --git a/pkg/discord/cmd/roll/run.go b/pkg/discord/cmd/roll/run.go index d888781..594a8b6 100644 --- a/pkg/discord/cmd/roll/run.go +++ b/pkg/discord/cmd/roll/run.go @@ -160,15 +160,11 @@ func mentionUser(i *discordgo.InteractionCreate) string { } func (c *Command) actuator() (rollpkg.Actuator, error) { - if c.cfg.Actuator == actuatorAPI { - if c.cfg.WatchtowerToken == "" { - return nil, fmt.Errorf("watchtower token not configured (WATCHTOWER_HTTP_API_TOKEN)") - } - - return rollpkg.NewAPIActuator(c.cfg.WatchtowerToken, "https", 0, "watchtower-"), nil + if c.cfg.WatchtowerToken == "" { + return nil, fmt.Errorf("watchtower token not configured (WATCHTOWER_HTTP_API_TOKEN)") } - return rollpkg.NewSSHActuator(rollpkg.SSHConfig{PrivateKeyPath: c.cfg.SSHKeyPath, Log: c.log}) + return rollpkg.NewAPIActuator(c.cfg.WatchtowerToken, "https", 0, "watchtower-"), nil } // rollUI accumulates per-host progress into a renderable Discord message. diff --git a/pkg/roll/api.go b/pkg/roll/api.go index 5834b96..78bf081 100644 --- a/pkg/roll/api.go +++ b/pkg/roll/api.go @@ -11,10 +11,9 @@ import ( "time" ) -// APIActuator rolls by calling a node's watchtower HTTP API directly. This -// requires the watchtower API to be reachable from where this runs (i.e. -// publicly/VPN-exposed) and a shared token. Prefer SSHActuator unless you -// specifically want to avoid SSH access. +// APIActuator rolls by calling a node's watchtower HTTP API at its public vhost +// (e.g. watchtower-) with a bearer token. It requires the watchtower API +// to be reachable (vhost-exposed); no SSH access is needed. type APIActuator struct { token string scheme string diff --git a/pkg/roll/dora.go b/pkg/roll/dora.go index 130bd4f..9ebec45 100644 --- a/pkg/roll/dora.go +++ b/pkg/roll/dora.go @@ -30,6 +30,7 @@ type DoraHealth struct { fetched time.Time } +//nolint:tagliatelle // Dora API uses snake_case type doraClient struct { Name string `json:"client_name"` Status string `json:"status"` diff --git a/pkg/roll/engine.go b/pkg/roll/engine.go index faf9925..8611b2f 100644 --- a/pkg/roll/engine.go +++ b/pkg/roll/engine.go @@ -64,15 +64,19 @@ func (o *Options) applyDefaults() { if o.DelayBetweenNodes == 0 { o.DelayBetweenNodes = time.Minute } + if o.PostTriggerWait == 0 { o.PostTriggerWait = 30 * time.Second } + if o.WaitTimeout == 0 { o.WaitTimeout = 10 * time.Minute } + if o.HealthCheckInterval == 0 { o.HealthCheckInterval = 10 * time.Second } + if o.MaxSyncDistance == 0 { o.MaxSyncDistance = 4 } @@ -232,6 +236,7 @@ func (e *Engine) rollOne(ctx context.Context, entry logrus.FieldLogger, target T func (e *Engine) waitHealthy(ctx context.Context, entry logrus.FieldLogger, target Target, opts Options) error { deadline := time.Now().Add(opts.WaitTimeout) + ticker := time.NewTicker(opts.HealthCheckInterval) defer ticker.Stop() @@ -270,9 +275,11 @@ func (e *Engine) preflight(ctx context.Context, targets []Target, opts Options) switch { case err != nil: entry.WithError(err).Warn("roll: pre-flight health error") + unhealthy = append(unhealthy, target.Name) case !ok: entry.WithField("status", reason).Warn("roll: pre-flight not healthy") + unhealthy = append(unhealthy, target.Name) default: entry.WithField("status", reason).Info("roll: pre-flight healthy") diff --git a/pkg/roll/health.go b/pkg/roll/health.go index ef52aee..8676d0e 100644 --- a/pkg/roll/health.go +++ b/pkg/roll/health.go @@ -29,6 +29,7 @@ func (b *BeaconHealth) SetBasicAuth(user, pass string) { b.pass = pass } +//nolint:tagliatelle // beacon API uses snake_case type syncingResponse struct { Data struct { HeadSlot string `json:"head_slot"` @@ -65,8 +66,8 @@ func (b *BeaconHealth) Healthy(ctx context.Context, beaconURL string, maxSyncDis } var body syncingResponse - if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { - return false, "", fmt.Errorf("decode /syncing: %w", err) + if decErr := json.NewDecoder(resp.Body).Decode(&body); decErr != nil { + return false, "", fmt.Errorf("decode /syncing: %w", decErr) } dist, err := strconv.ParseUint(body.Data.SyncDistance, 10, 64) diff --git a/pkg/roll/ssh.go b/pkg/roll/ssh.go deleted file mode 100644 index 76f7797..0000000 --- a/pkg/roll/ssh.go +++ /dev/null @@ -1,209 +0,0 @@ -package roll - -import ( - "bytes" - "context" - "fmt" - "net" - "os" - "regexp" - "strconv" - "strings" - "time" - - "github.com/sirupsen/logrus" - "golang.org/x/crypto/ssh" -) - -// DefaultWatchtowerContainer is the watchtower container name deployed by the -// ethpandaops ethereum_node Ansible role. -const DefaultWatchtowerContainer = "ethereum-node-docker-watchtower" - -// DefaultWatchtowerPort is watchtower's default HTTP API port. -const DefaultWatchtowerPort = 8080 - -const sshDialTimeout = 15 * time.Second - -// safeRef guards values interpolated into the remote shell script. Single-quoted -// in the script, this charset (no quotes/spaces/metachars) prevents injection. -var safeRef = regexp.MustCompile(`^[A-Za-z0-9._/:@-]*$`) - -// SSHActuator rolls an image by SSHing to the host and triggering the -// node-local watchtower HTTP API. It discovers watchtower's container IP and -// API token from `docker inspect` on the node, so it needs no published port -// and no token stored centrally — only SSH access (which devops already has). -type SSHActuator struct { - signer ssh.Signer - hostKeyCB ssh.HostKeyCallback - containerName string - port int - log logrus.FieldLogger -} - -// SSHConfig configures an SSHActuator. -type SSHConfig struct { - // PrivateKeyPath is the path to the SSH private key used to authenticate. - PrivateKeyPath string - // KnownHostsCallback verifies host keys. If nil, host keys are not verified - // (acceptable for ephemeral devnets; a warning is logged). - KnownHostsCallback ssh.HostKeyCallback - // ContainerName overrides the watchtower container name. - ContainerName string - // Port overrides the watchtower HTTP API port. - Port int - // Log is the logger. - Log logrus.FieldLogger -} - -// NewSSHActuator builds an SSHActuator from the given config. -func NewSSHActuator(cfg SSHConfig) (*SSHActuator, error) { - if cfg.Log == nil { - cfg.Log = logrus.New() - } - - if cfg.PrivateKeyPath == "" { - return nil, fmt.Errorf("ssh private key path is required") - } - - keyBytes, err := os.ReadFile(cfg.PrivateKeyPath) - if err != nil { - return nil, fmt.Errorf("read ssh key: %w", err) - } - - signer, err := ssh.ParsePrivateKey(keyBytes) - if err != nil { - return nil, fmt.Errorf("parse ssh key: %w", err) - } - - hostKeyCB := cfg.KnownHostsCallback - if hostKeyCB == nil { - cfg.Log.Warn("roll: SSH host key verification disabled (no known_hosts configured)") - - hostKeyCB = ssh.InsecureIgnoreHostKey() //nolint:gosec // devnet hosts; opt-in known_hosts available - } - - container := cfg.ContainerName - if container == "" { - container = DefaultWatchtowerContainer - } - - port := cfg.Port - if port == 0 { - port = DefaultWatchtowerPort - } - - return &SSHActuator{ - signer: signer, - hostKeyCB: hostKeyCB, - containerName: container, - port: port, - log: cfg.Log, - }, nil -} - -// Name implements Actuator. -func (a *SSHActuator) Name() string { return "ssh" } - -// Roll implements Actuator: SSH to the host and trigger the local watchtower. -func (a *SSHActuator) Roll(ctx context.Context, target Target, image string) error { - if !safeRef.MatchString(image) { - return fmt.Errorf("invalid image reference %q", image) - } - - if !safeRef.MatchString(a.containerName) { - return fmt.Errorf("invalid container name %q", a.containerName) - } - - user, host := splitSSH(target.SSH) - if host == "" { - return fmt.Errorf("invalid ssh target %q", target.SSH) - } - - out, err := a.run(ctx, user, host, a.script(image)) - if err != nil { - return fmt.Errorf("ssh roll %s: %w: %s", target.Name, err, strings.TrimSpace(out)) - } - - return nil -} - -// script builds the remote shell that discovers watchtower's IP + token and -// triggers the update. image is single-quoted and charset-validated by Roll. -func (a *SSHActuator) script(image string) string { - return fmt.Sprintf(`set -e -cn='%s' -img='%s' -ip=$(docker inspect "$cn" -f '{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}' | awk '{print $1}') -[ -n "$ip" ] || { echo "watchtower container $cn not found" >&2; exit 3; } -tok=$(docker inspect "$cn" -f '{{range .Config.Env}}{{println .}}{{end}}' | sed -n 's/^WATCHTOWER_HTTP_API_TOKEN=//p') -[ -n "$tok" ] || { echo "WATCHTOWER_HTTP_API_TOKEN not set on $cn" >&2; exit 4; } -url="http://$ip:%d/v1/update" -[ -n "$img" ] && url="$url?image=$img" -curl -fsS -X POST -H "Authorization: Bearer $tok" "$url" -`, a.containerName, image, a.port) -} - -func (a *SSHActuator) run(ctx context.Context, user, host, script string) (string, error) { - clientCfg := &ssh.ClientConfig{ - User: user, - Auth: []ssh.AuthMethod{ssh.PublicKeys(a.signer)}, - HostKeyCallback: a.hostKeyCB, - Timeout: sshDialTimeout, - } - - dialer := &net.Dialer{Timeout: sshDialTimeout} - - conn, err := dialer.DialContext(ctx, "tcp", host) - if err != nil { - return "", fmt.Errorf("dial: %w", err) - } - - sshConn, chans, reqs, err := ssh.NewClientConn(conn, host, clientCfg) - if err != nil { - _ = conn.Close() - - return "", fmt.Errorf("ssh handshake: %w", err) - } - - client := ssh.NewClient(sshConn, chans, reqs) - defer client.Close() - - session, err := client.NewSession() - if err != nil { - return "", fmt.Errorf("ssh session: %w", err) - } - defer session.Close() - - var buf bytes.Buffer - session.Stdout = &buf - session.Stderr = &buf - - done := make(chan error, 1) - go func() { done <- session.Run(script) }() - - select { - case <-ctx.Done(): - _ = session.Signal(ssh.SIGKILL) - - return buf.String(), ctx.Err() - case err := <-done: - return buf.String(), err - } -} - -// splitSSH parses "user@host[:port]" into user and host:port (defaulting :22). -func splitSSH(target string) (user, hostport string) { - user = "root" - host := target - - if at := strings.Index(target, "@"); at >= 0 { - user = target[:at] - host = target[at+1:] - } - - if _, _, err := net.SplitHostPort(host); err != nil { - host = net.JoinHostPort(host, strconv.Itoa(22)) - } - - return user, host -} diff --git a/pkg/service/config.go b/pkg/service/config.go index 8b8dc2f..e6b1822 100644 --- a/pkg/service/config.go +++ b/pkg/service/config.go @@ -27,8 +27,7 @@ type Config struct { ClientsDataURL string MetricsAddress string // Defaults to :9091 HealthCheckAddress string // Defaults to :9191 - RollSSHKeyPath string // SSH key path for the /roll command's ssh actuator - WatchtowerAPIToken string // watchtower API token for the /roll command's api actuator + WatchtowerAPIToken string // watchtower API token for the /roll command NodeBasicAuthUser string // basic auth user for beacon health endpoints (bn-* vhosts) NodeBasicAuthPass string // basic auth password for beacon health endpoints } diff --git a/pkg/service/service.go b/pkg/service/service.go index a2a8f4a..0f98141 100644 --- a/pkg/service/service.go +++ b/pkg/service/service.go @@ -148,9 +148,7 @@ func NewService(ctx context.Context, log *logrus.Logger, cfg *Config) (*Service, cmdhive.NewHiveCommand(log, bot, cfg.GithubToken, githubHTTPClient), build.NewBuildCommand(log, bot, cfg.GithubToken, githubHTTPClient), cmdroll.NewRollCommand(log, bot, cmdroll.Config{ - Actuator: "api", WatchtowerToken: cfg.WatchtowerAPIToken, - SSHKeyPath: cfg.RollSSHKeyPath, BasicAuthUser: cfg.NodeBasicAuthUser, BasicAuthPass: cfg.NodeBasicAuthPass, }), From 80782ea3b9249c7708d9fbbf6aa61ec305f2724c Mon Sep 17 00:00:00 2001 From: Barnabas Busa Date: Fri, 29 May 2026 16:11:54 +0200 Subject: [PATCH 3/3] refactor(roll): Dora-only health, watchtower-vhost trigger Simplify the rollout to a single, clean path with no SSH and no direct (basic-auth) beacon access: - Health is sourced solely from Dora (status: ready), keyed by node name. Remove the beacon health checker and all basic-auth wiring (config, env, flags, Options fields). - The API actuator targets each node's watchtower vhost (https://watchtower-.srv..ethpandaops.io) with a bearer token; drop SSH host derivation. - Inventory targets are grouped by node name (clientName); drop the unused ssh/beacon/rpc fields and the beaconScheme plumbing. The only inputs are now the network, the cartographoor inventory, the Dora URL (derived from the network), and the watchtower API token. --- cmd/main.go | 2 - cmd/roll.go | 83 ++++++++--------------- pkg/discord/cmd/roll/command.go | 6 +- pkg/discord/cmd/roll/run.go | 18 +++-- pkg/roll/api.go | 67 +++++-------------- pkg/roll/engine.go | 89 ++++++++++--------------- pkg/roll/health.go | 90 ------------------------- pkg/roll/inventory.go | 114 ++++++++++---------------------- pkg/roll/match_test.go | 22 +++--- pkg/roll/provider.go | 24 +++---- pkg/service/config.go | 2 - pkg/service/service.go | 2 - 12 files changed, 139 insertions(+), 380 deletions(-) delete mode 100644 pkg/roll/health.go diff --git a/cmd/main.go b/cmd/main.go index f7828b0..303b7e6 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -104,8 +104,6 @@ func setConfig(cfg *service.Config) { cfg.HealthCheckAddress = os.Getenv("HEALTH_CHECK_ADDRESS") cfg.MetricsAddress = os.Getenv("METRICS_ADDRESS") cfg.WatchtowerAPIToken = os.Getenv("WATCHTOWER_HTTP_API_TOKEN") - cfg.NodeBasicAuthUser = os.Getenv("ROLL_BASIC_AUTH_USER") - cfg.NodeBasicAuthPass = os.Getenv("ROLL_BASIC_AUTH_PASS") if cfg.GrafanaBaseURL == "" { cfg.GrafanaBaseURL = grafana.DefaultGrafanaBaseURL diff --git a/cmd/roll.go b/cmd/roll.go index a963f54..1ed0ede 100644 --- a/cmd/roll.go +++ b/cmd/roll.go @@ -14,29 +14,22 @@ import ( ) // newRollCommand builds the `panda-pulse roll` subcommand: a gated, sequential -// image rollout across a network's nodes, resolved from cartographoor inventory. +// image rollout across a network's nodes, resolved from cartographoor inventory, +// health-gated on Dora, and triggered via each node's watchtower vhost. func newRollCommand(log *logrus.Logger) *cobra.Command { var ( - network string - client string - image string - inventoryURL string - watchtowerPort int - watchtowerToken string - watchtowerScheme string - watchtowerPrefix string - beaconScheme string - basicAuthUser string - basicAuthPass string - doraURL string - noDora bool - skipHealth bool - dryRun bool - delay time.Duration - postTrigger time.Duration - waitTimeout time.Duration - healthInterval time.Duration - maxSyncDistance uint64 + network string + client string + image string + inventoryURL string + doraURL string + watchtowerToken string + skipHealth bool + dryRun bool + delay time.Duration + postTrigger time.Duration + waitTimeout time.Duration + healthInterval time.Duration ) cmd := &cobra.Command{ @@ -48,6 +41,10 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { return fmt.Errorf("--network is required") } + if watchtowerToken == "" { + return fmt.Errorf("--watchtower-token (or WATCHTOWER_HTTP_API_TOKEN) is required") + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -56,36 +53,24 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { return err } - targets := roll.Select(roll.ResolveTargets(inv, beaconScheme), client) + targets := roll.Select(roll.ResolveTargets(inv), client) if len(targets) == 0 { return fmt.Errorf("no targets matched (network=%s client=%q)", network, client) } - actuator, err := buildActuator(watchtowerToken, watchtowerScheme, watchtowerPrefix, watchtowerPort) - if err != nil { - return err - } - - doraHealthURL := "" - if !noDora { - doraHealthURL = doraURL - if doraHealthURL == "" { - doraHealthURL = roll.DoraURLForNetwork(network) - } + if doraURL == "" { + doraURL = roll.DoraURLForNetwork(network) } - return roll.NewEngine(actuator, log).Run(ctx, targets, roll.Options{ + return roll.NewEngine(roll.NewAPIActuator(watchtowerToken, network), log).Run(ctx, targets, roll.Options{ Image: image, DelayBetweenNodes: delay, PostTriggerWait: postTrigger, WaitTimeout: waitTimeout, HealthCheckInterval: healthInterval, - MaxSyncDistance: maxSyncDistance, SkipHealth: skipHealth, DryRun: dryRun, - DoraURL: doraHealthURL, - BeaconBasicAuthUser: basicAuthUser, - BeaconBasicAuthPass: basicAuthPass, + DoraURL: doraURL, }) }, } @@ -95,30 +80,14 @@ func newRollCommand(log *logrus.Logger) *cobra.Command { f.StringVar(&client, "client", "", "host pattern: client/group/node with globs, ! to exclude, 'all' (e.g. 'lighthouse', 'lighthouse_ethrex', 'lighthouse-*:!*-1')") f.StringVar(&image, "image", "", "scope the roll to this image (empty = all watched containers)") f.StringVar(&inventoryURL, "inventory-url", roll.DefaultInventoryBaseURL, "cartographoor inventory base URL") - f.IntVar(&watchtowerPort, "watchtower-port", 0, "watchtower API port (0 = default for the scheme)") - f.StringVar(&watchtowerToken, "watchtower-token", os.Getenv("WATCHTOWER_HTTP_API_TOKEN"), "watchtower API token (api actuator; env WATCHTOWER_HTTP_API_TOKEN)") - f.StringVar(&watchtowerScheme, "watchtower-scheme", "https", "watchtower API scheme (api actuator)") - f.StringVar(&watchtowerPrefix, "watchtower-prefix", "watchtower-", "vhost prefix for the watchtower API (api actuator)") f.StringVar(&doraURL, "dora-url", "", "Dora health source URL (default: https://dora..ethpandaops.io)") - f.BoolVar(&noDora, "no-dora", false, "use per-node beacon health instead of Dora") - f.StringVar(&beaconScheme, "beacon-scheme", "https", "scheme for beacon health endpoints (only used with --no-dora)") - f.StringVar(&basicAuthUser, "basic-auth-user", os.Getenv("ROLL_BASIC_AUTH_USER"), "basic auth user for beacon health endpoints (env ROLL_BASIC_AUTH_USER)") - f.StringVar(&basicAuthPass, "basic-auth-pass", os.Getenv("ROLL_BASIC_AUTH_PASS"), "basic auth password for beacon health endpoints (env ROLL_BASIC_AUTH_PASS)") - f.BoolVar(&skipHealth, "skip-health", false, "skip beacon health gating (force; trigger-and-go even if unhealthy)") + f.StringVar(&watchtowerToken, "watchtower-token", os.Getenv("WATCHTOWER_HTTP_API_TOKEN"), "watchtower API token (env WATCHTOWER_HTTP_API_TOKEN)") + f.BoolVar(&skipHealth, "skip-health", false, "skip health gating (force; trigger-and-go even if unhealthy)") f.BoolVar(&dryRun, "dry-run", false, "log intent without triggering rolls") f.DurationVar(&delay, "delay-roll", time.Minute, "wait between hosts (~N minutes for N hosts); overridable") f.DurationVar(&postTrigger, "post-trigger-wait", 30*time.Second, "grace period after triggering before polling recovery") f.DurationVar(&waitTimeout, "wait-timeout", 10*time.Minute, "per-node recovery timeout before aborting") - f.DurationVar(&healthInterval, "health-check-interval", 10*time.Second, "beacon health poll cadence during recovery") - f.Uint64Var(&maxSyncDistance, "max-sync-distance", 4, "max sync distance (slots) still considered healthy") + f.DurationVar(&healthInterval, "health-check-interval", 10*time.Second, "Dora health poll cadence during recovery") return cmd } - -func buildActuator(token, scheme, prefix string, port int) (roll.Actuator, error) { - if token == "" { - return nil, fmt.Errorf("--watchtower-token (or WATCHTOWER_HTTP_API_TOKEN) is required") - } - - return roll.NewAPIActuator(token, scheme, port, prefix), nil -} diff --git a/pkg/discord/cmd/roll/command.go b/pkg/discord/cmd/roll/command.go index bcb8c5d..7410862 100644 --- a/pkg/discord/cmd/roll/command.go +++ b/pkg/discord/cmd/roll/command.go @@ -34,10 +34,6 @@ type Config struct { WatchtowerToken string // InventoryURL overrides the cartographoor inventory base URL. InventoryURL string - // BasicAuthUser and BasicAuthPass authenticate beacon health checks behind - // nginx basic auth (the bn-* vhosts) — only used when Dora is unavailable. - BasicAuthUser string - BasicAuthPass string // DoraURL overrides the Dora health source; empty derives it from the // network (https://dora..ethpandaops.io). DoraURL string @@ -58,7 +54,7 @@ func NewRollCommand(log *logrus.Logger, bot common.BotContext, cfg Config) *Comm log: log, bot: bot, cfg: cfg, - provider: rollpkg.NewInventoryProvider(cfg.InventoryURL, "https", providerCacheTTL), + provider: rollpkg.NewInventoryProvider(cfg.InventoryURL, providerCacheTTL), } } diff --git a/pkg/discord/cmd/roll/run.go b/pkg/discord/cmd/roll/run.go index 594a8b6..e183a2b 100644 --- a/pkg/discord/cmd/roll/run.go +++ b/pkg/discord/cmd/roll/run.go @@ -71,7 +71,7 @@ func (c *Command) run(s *discordgo.Session, i *discordgo.InteractionCreate, data return nil } - actuator, err := c.actuator() + actuator, err := c.actuator(network) if err != nil { c.respondEphemeral(s, i, fmt.Sprintf("❌ Roll not configured: %v", err)) @@ -109,13 +109,11 @@ func (c *Command) run(s *discordgo.Session, i *discordgo.InteractionCreate, data } runErr := rollpkg.NewEngine(actuator, c.log).Run(ctx, targets, rollpkg.Options{ - Image: image, - DryRun: dryRun, - SkipHealth: force, - DelayBetweenNodes: delay, - DoraURL: doraURL, - BeaconBasicAuthUser: c.cfg.BasicAuthUser, - BeaconBasicAuthPass: c.cfg.BasicAuthPass, + Image: image, + DryRun: dryRun, + SkipHealth: force, + DelayBetweenNodes: delay, + DoraURL: doraURL, OnProgress: func(p rollpkg.Progress) { ui.update(p) edit(false) @@ -159,12 +157,12 @@ func mentionUser(i *discordgo.InteractionCreate) string { } } -func (c *Command) actuator() (rollpkg.Actuator, error) { +func (c *Command) actuator(network string) (rollpkg.Actuator, error) { if c.cfg.WatchtowerToken == "" { return nil, fmt.Errorf("watchtower token not configured (WATCHTOWER_HTTP_API_TOKEN)") } - return rollpkg.NewAPIActuator(c.cfg.WatchtowerToken, "https", 0, "watchtower-"), nil + return rollpkg.NewAPIActuator(c.cfg.WatchtowerToken, network), nil } // rollUI accumulates per-host progress into a renderable Discord message. diff --git a/pkg/roll/api.go b/pkg/roll/api.go index 78bf081..e55bfc9 100644 --- a/pkg/roll/api.go +++ b/pkg/roll/api.go @@ -4,63 +4,42 @@ import ( "context" "fmt" "io" - "net" "net/http" "net/url" "strings" "time" ) -// APIActuator rolls by calling a node's watchtower HTTP API at its public vhost -// (e.g. watchtower-) with a bearer token. It requires the watchtower API -// to be reachable (vhost-exposed); no SSH access is needed. +// WatchtowerURLForNode returns the watchtower API base URL for a node on an +// ethpandaops network, e.g. +// https://watchtower-lighthouse-ethrex-1.srv.glamsterdam-devnet-4.ethpandaops.io +func WatchtowerURLForNode(network, node string) string { + return fmt.Sprintf("https://watchtower-%s.srv.%s.ethpandaops.io", node, network) +} + +// APIActuator rolls by calling each node's watchtower HTTP API at its public +// vhost with a bearer token. The watchtower vhost is bearer-auth only. type APIActuator struct { token string - scheme string - port int - hostPrefix string + network string httpClient *http.Client } -// NewAPIActuator returns an APIActuator targeting each node's watchtower vhost -// (hostPrefix + the node host, e.g. "watchtower-"). scheme defaults to -// "https"; hostPrefix defaults to "watchtower-"; port 0 omits the port (so 443 -// for https). The watchtower vhost is bearer-auth only — no basic auth. -func NewAPIActuator(token, scheme string, port int, hostPrefix string) *APIActuator { - if scheme == "" { - scheme = "https" - } - - if hostPrefix == "" { - hostPrefix = "watchtower-" - } - +// NewAPIActuator returns an APIActuator for the given network and bearer token. +func NewAPIActuator(token, network string) *APIActuator { return &APIActuator{ token: token, - scheme: scheme, - port: port, - hostPrefix: hostPrefix, + network: network, httpClient: &http.Client{Timeout: 60 * time.Second}, } } // Name implements Actuator. -func (a *APIActuator) Name() string { return "api" } +func (a *APIActuator) Name() string { return "watchtower" } -// Roll implements Actuator: POST /v1/update to the target's watchtower API. +// Roll implements Actuator: POST /v1/update to the node's watchtower vhost. func (a *APIActuator) Roll(ctx context.Context, target Target, image string) error { - host := sshHost(target.SSH) - if host == "" { - return fmt.Errorf("invalid target %q", target.SSH) - } - - host = a.hostPrefix + host - - endpoint := fmt.Sprintf("%s://%s/v1/update", a.scheme, host) - if a.port != 0 { - endpoint = fmt.Sprintf("%s://%s:%d/v1/update", a.scheme, host, a.port) - } - + endpoint := WatchtowerURLForNode(a.network, target.Name) + "/v1/update" if image != "" { endpoint += "?image=" + url.QueryEscape(image) } @@ -85,17 +64,3 @@ func (a *APIActuator) Roll(ctx context.Context, target Target, image string) err return nil } - -// sshHost extracts the host (FQDN) from an "user@host[:port]" SSH value. -func sshHost(target string) string { - host := target - if at := strings.Index(host, "@"); at >= 0 { - host = host[at+1:] - } - - if h, _, err := net.SplitHostPort(host); err == nil { - return h - } - - return host -} diff --git a/pkg/roll/engine.go b/pkg/roll/engine.go index 8611b2f..79e592c 100644 --- a/pkg/roll/engine.go +++ b/pkg/roll/engine.go @@ -19,47 +19,19 @@ type Options struct { PostTriggerWait time.Duration // WaitTimeout is the per-node recovery timeout; exceeding it aborts the rollout. WaitTimeout time.Duration - // HealthCheckInterval is how often to poll beacon health while waiting. + // HealthCheckInterval is how often to poll Dora health while waiting. HealthCheckInterval time.Duration - // MaxSyncDistance is the largest sync distance (slots) still considered healthy. - MaxSyncDistance uint64 - // SkipHealth disables beacon health gating (trigger-and-go). + // DoraURL is the Dora explorer used as the health source of truth. + DoraURL string + // SkipHealth disables health gating (force; trigger-and-go even if unhealthy). SkipHealth bool // DryRun logs intent without triggering rolls. DryRun bool - // DoraURL, if set, makes health checks use Dora (matched by node name) as the - // source of truth instead of per-node beacon calls. Preferred — one - // unauthenticated call covers the fleet. - DoraURL string - // BeaconBasicAuthUser and BeaconBasicAuthPass authenticate beacon health - // checks when Dora is not used and the beacon is behind nginx basic auth. - BeaconBasicAuthUser string - BeaconBasicAuthPass string // OnProgress, if set, is invoked at each rollout milestone (for UIs such as // the Discord command). It must be cheap and non-blocking. OnProgress func(Progress) } -// Phase is a rollout milestone for progress reporting. -type Phase string - -const ( - PhaseTriggering Phase = "triggering" - PhaseHealthy Phase = "healthy" - PhaseFailed Phase = "failed" - PhaseSkipped Phase = "skipped" - PhaseDone Phase = "done" -) - -// Progress is a rollout milestone delivered to Options.OnProgress. -type Progress struct { - Node string - Index int // 1-based; 0 for fleet-level events - Total int - Phase Phase - Message string -} - func (o *Options) applyDefaults() { if o.DelayBetweenNodes == 0 { o.DelayBetweenNodes = time.Minute @@ -76,17 +48,32 @@ func (o *Options) applyDefaults() { if o.HealthCheckInterval == 0 { o.HealthCheckInterval = 10 * time.Second } +} - if o.MaxSyncDistance == 0 { - o.MaxSyncDistance = 4 - } +// Phase is a rollout milestone for progress reporting. +type Phase string + +const ( + PhaseTriggering Phase = "triggering" + PhaseHealthy Phase = "healthy" + PhaseFailed Phase = "failed" + PhaseSkipped Phase = "skipped" + PhaseDone Phase = "done" +) + +// Progress is a rollout milestone delivered to Options.OnProgress. +type Progress struct { + Node string + Index int + Total int + Phase Phase + Message string } -// Engine performs gated, sequential rollouts via an Actuator, gating on beacon +// Engine performs gated, sequential rollouts via an Actuator, gating on Dora // health between nodes and aborting on the first node that fails to recover. type Engine struct { actuator Actuator - health *BeaconHealth dora *DoraHealth log logrus.FieldLogger } @@ -97,28 +84,22 @@ func NewEngine(actuator Actuator, log logrus.FieldLogger) *Engine { log = logrus.New() } - return &Engine{actuator: actuator, health: NewBeaconHealth(), log: log} + return &Engine{actuator: actuator, log: log} } -// checkHealth gates on Dora when a Dora URL is configured (the preferred, -// unauthenticated source of truth), otherwise falls back to the node's beacon. -func (e *Engine) checkHealth(ctx context.Context, target Target, maxSyncDistance uint64) (bool, string, error) { - if e.dora != nil { - return e.dora.Healthy(ctx, target.Name) - } - - if target.BeaconURL == "" { - return false, "no health source (set server doraURL or node beaconUrl)", nil +// checkHealth reports whether a node is healthy according to Dora. +func (e *Engine) checkHealth(ctx context.Context, target Target) (bool, string, error) { + if e.dora == nil { + return false, "", errors.New("no health source configured (set DoraURL)") } - return e.health.Healthy(ctx, target.BeaconURL, maxSyncDistance) + return e.dora.Healthy(ctx, target.Name) } // Run rolls the targets in order. It aborts on the first node that fails to // recover, leaving the remaining targets untouched. func (e *Engine) Run(ctx context.Context, targets []Target, opts Options) error { opts.applyDefaults() - e.health.SetBasicAuth(opts.BeaconBasicAuthUser, opts.BeaconBasicAuthPass) if opts.DoraURL != "" { e.dora = NewDoraHealth(opts.DoraURL) @@ -136,7 +117,7 @@ func (e *Engine) Run(ctx context.Context, targets []Target, opts Options) error }).Info("roll: starting") if !opts.SkipHealth { - if err := e.preflight(ctx, targets, opts); err != nil { + if err := e.preflight(ctx, targets); err != nil { return fmt.Errorf("pre-flight: %w", err) } } @@ -203,7 +184,7 @@ func (e *Engine) emit(opts Options, p Progress, phase Phase, msg string) { func (e *Engine) rollOne(ctx context.Context, entry logrus.FieldLogger, target Target, opts Options) error { if !opts.SkipHealth { - ok, reason, err := e.checkHealth(ctx, target, opts.MaxSyncDistance) + ok, reason, err := e.checkHealth(ctx, target) if err != nil { return fmt.Errorf("pre-update health check: %w", err) } @@ -241,7 +222,7 @@ func (e *Engine) waitHealthy(ctx context.Context, entry logrus.FieldLogger, targ defer ticker.Stop() for { - ok, reason, err := e.checkHealth(ctx, target, opts.MaxSyncDistance) + ok, reason, err := e.checkHealth(ctx, target) switch { case err != nil: entry.WithError(err).Debug("roll: health check failed, retrying") @@ -265,11 +246,11 @@ func (e *Engine) waitHealthy(ctx context.Context, entry logrus.FieldLogger, targ } } -func (e *Engine) preflight(ctx context.Context, targets []Target, opts Options) error { +func (e *Engine) preflight(ctx context.Context, targets []Target) error { var unhealthy []string for _, target := range targets { - ok, reason, err := e.checkHealth(ctx, target, opts.MaxSyncDistance) + ok, reason, err := e.checkHealth(ctx, target) entry := e.log.WithField("node", target.Name) switch { diff --git a/pkg/roll/health.go b/pkg/roll/health.go deleted file mode 100644 index 8676d0e..0000000 --- a/pkg/roll/health.go +++ /dev/null @@ -1,90 +0,0 @@ -package roll - -import ( - "context" - "encoding/json" - "fmt" - "net/http" - "strconv" - "strings" - "time" -) - -// BeaconHealth checks a beacon node's sync status for rollout gating. -type BeaconHealth struct { - httpClient *http.Client - user string - pass string -} - -// NewBeaconHealth returns a BeaconHealth checker. -func NewBeaconHealth() *BeaconHealth { - return &BeaconHealth{httpClient: &http.Client{Timeout: 10 * time.Second}} -} - -// SetBasicAuth sets HTTP basic auth for beacon requests, needed when the beacon -// endpoint is behind nginx basic auth (as the bn-* vhosts are). -func (b *BeaconHealth) SetBasicAuth(user, pass string) { - b.user = user - b.pass = pass -} - -//nolint:tagliatelle // beacon API uses snake_case -type syncingResponse struct { - Data struct { - HeadSlot string `json:"head_slot"` - SyncDistance string `json:"sync_distance"` - IsSyncing bool `json:"is_syncing"` - IsOptimistic bool `json:"is_optimistic"` - ELOffline bool `json:"el_offline"` - } `json:"data"` -} - -// Healthy reports whether the beacon at beaconURL is synced within -// maxSyncDistance slots and not syncing/optimistic/EL-offline. The returned -// string is a human-readable status. -func (b *BeaconHealth) Healthy(ctx context.Context, beaconURL string, maxSyncDistance uint64) (bool, string, error) { - url := strings.TrimRight(beaconURL, "/") + "/eth/v1/node/syncing" - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) - if err != nil { - return false, "", err - } - - if b.user != "" || b.pass != "" { - req.SetBasicAuth(b.user, b.pass) - } - - resp, err := b.httpClient.Do(req) - if err != nil { - return false, "", err - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return false, "", fmt.Errorf("beacon /syncing returned %d", resp.StatusCode) - } - - var body syncingResponse - if decErr := json.NewDecoder(resp.Body).Decode(&body); decErr != nil { - return false, "", fmt.Errorf("decode /syncing: %w", decErr) - } - - dist, err := strconv.ParseUint(body.Data.SyncDistance, 10, 64) - if err != nil { - return false, "", fmt.Errorf("parse sync_distance %q: %w", body.Data.SyncDistance, err) - } - - switch { - case body.Data.IsSyncing: - return false, fmt.Sprintf("syncing (distance=%d)", dist), nil - case body.Data.IsOptimistic: - return false, "optimistic (execution payload not validated)", nil - case body.Data.ELOffline: - return false, "execution layer offline", nil - case dist > maxSyncDistance: - return false, fmt.Sprintf("sync distance %d exceeds max %d", dist, maxSyncDistance), nil - } - - return true, fmt.Sprintf("synced (head=%s, distance=%d)", body.Data.HeadSlot, dist), nil -} diff --git a/pkg/roll/inventory.go b/pkg/roll/inventory.go index 7746c51..e077017 100644 --- a/pkg/roll/inventory.go +++ b/pkg/roll/inventory.go @@ -1,7 +1,7 @@ // Package roll performs gated, sequential container image rollouts across an // Ethereum node fleet. Targets are resolved from cartographoor's published -// per-network inventory; the roll itself is executed by a pluggable Actuator -// (SSH-to-local-watchtower by default). +// per-network inventory, health is gated on Dora, and rolls are triggered via +// each node's watchtower HTTP API vhost. package roll import ( @@ -17,16 +17,10 @@ import ( // DefaultInventoryBaseURL is where cartographoor publishes per-network inventory. const DefaultInventoryBaseURL = "https://ethpandaops-platform-production-cartographoor.ams3.digitaloceanspaces.com" -// clientInfo mirrors the relevant fields of cartographoor's inventory ClientInfo. +// clientInfo mirrors the fields we need from a cartographoor inventory client. type clientInfo struct { - ClientName string `json:"clientName"` - ClientType string `json:"clientType"` - Version string `json:"version"` - DockerImage string `json:"dockerImage"` - SSH string `json:"ssh"` - BeaconAPI string `json:"bn"` - RPC string `json:"rpc"` - Status string `json:"status"` + ClientName string `json:"clientName"` + ClientType string `json:"clientType"` } type inventoryData struct { @@ -35,18 +29,14 @@ type inventoryData struct { ExecutionClients []clientInfo `json:"executionClients"` } -// Target is a single host to roll, grouped from the inventory by its SSH host. +// Target is a single node to roll. type Target struct { - // Name is the host/node identifier (derived from the SSH host). + // Name is the node identifier (e.g. lighthouse-ethrex-1). Name string - // SSH is the cartographoor ssh value, e.g. "devops@host". - SSH string - // BeaconURL is the host's beacon API base URL for health gating (may be empty). - BeaconURL string - // Clients are the client names running on this host (CL and EL). + // Clients are the client types running on this node (CL and EL). Clients []string - // tokens are the lowercased selectable identifiers for this host: node name, - // client types, and the cl_el group. Used by the --limit/--client matcher. + // tokens are the lowercased selectable identifiers for this node — node + // name, client types, and the cl_el group — used by the --client matcher. tokens []string } @@ -81,79 +71,59 @@ func FetchInventory(ctx context.Context, baseURL, network string) (*inventoryDat return &data, nil } -// ResolveTargets groups the inventory by SSH host into roll targets, computing -// match tokens (node name, CL/EL client types, and the cl_el group) for -// Ansible-style selection. beaconScheme (e.g. "https") is prepended to the bare -// beacon hostname from the consensus client entry. -func ResolveTargets(inv *inventoryData, beaconScheme string) []Target { - if beaconScheme == "" { - beaconScheme = "https" - } - +// ResolveTargets groups the inventory by node into roll targets, computing match +// tokens (node name, CL/EL client types, and the cl_el group) for selection. +func ResolveTargets(inv *inventoryData) []Target { type agg struct { - ssh string - beacon string - clients []string clTypes []string elTypes []string } - byHost := map[string]*agg{} + byNode := map[string]*agg{} order := []string{} - get := func(ssh string) *agg { - a, ok := byHost[ssh] + get := func(name string) *agg { + a, ok := byNode[name] if !ok { - a = &agg{ssh: ssh} - byHost[ssh] = a - order = append(order, ssh) + a = &agg{} + byNode[name] = a + order = append(order, name) } return a } for _, c := range inv.ConsensusClients { - if c.SSH == "" { + if c.ClientName == "" || c.ClientType == "" { continue } - a := get(c.SSH) - if c.ClientName != "" { - a.clients = append(a.clients, c.ClientName) - } - - if c.ClientType != "" { - a.clTypes = append(a.clTypes, c.ClientType) - } - - if c.BeaconAPI != "" && a.beacon == "" { - a.beacon = beaconScheme + "://" + c.BeaconAPI - } + a := get(c.ClientName) + a.clTypes = append(a.clTypes, c.ClientType) } for _, c := range inv.ExecutionClients { - if c.SSH == "" { + if c.ClientName == "" || c.ClientType == "" { continue } - a := get(c.SSH) - if c.ClientType != "" { - a.elTypes = append(a.elTypes, c.ClientType) - } + a := get(c.ClientName) + a.elTypes = append(a.elTypes, c.ClientType) } targets := make([]Target, 0, len(order)) - for _, ssh := range order { - a := byHost[ssh] - name := hostName(ssh) + for _, name := range order { + a := byNode[name] + + clients := make([]string, 0, len(a.clTypes)+len(a.elTypes)) + clients = append(clients, a.clTypes...) + clients = append(clients, a.elTypes...) targets = append(targets, Target{ - Name: name, - SSH: ssh, - BeaconURL: a.beacon, - Clients: a.clients, - tokens: buildTokens(name, a.clTypes, a.elTypes), + Name: name, + Clients: clients, + tokens: buildTokens(name, a.clTypes, a.elTypes), }) } @@ -163,7 +133,7 @@ func ResolveTargets(inv *inventoryData, beaconScheme string) []Target { } // buildTokens returns the lowercased, deduped set of selectable tokens for a -// host: its node name, each client type, and each cl_el group pairing. +// node: its name, each client type, and each cl_el group pairing. func buildTokens(name string, clTypes, elTypes []string) []string { seen := map[string]bool{} tokens := []string{} @@ -194,17 +164,3 @@ func buildTokens(name string, clTypes, elTypes []string) []string { return tokens } - -// hostName derives a short host identifier from an "user@host.domain" SSH value. -func hostName(ssh string) string { - host := ssh - if at := strings.Index(host, "@"); at >= 0 { - host = host[at+1:] - } - - if dot := strings.Index(host, "."); dot >= 0 { - host = host[:dot] - } - - return host -} diff --git a/pkg/roll/match_test.go b/pkg/roll/match_test.go index bde9858..3e941e4 100644 --- a/pkg/roll/match_test.go +++ b/pkg/roll/match_test.go @@ -9,14 +9,14 @@ func testInventory() *inventoryData { return &inventoryData{ Network: "glamsterdam-devnet-4", ConsensusClients: []clientInfo{ - {ClientName: "lighthouse-ethrex-1", ClientType: "lighthouse", SSH: "devops@lighthouse-ethrex-1.example.io", BeaconAPI: "bn-lighthouse-ethrex-1.example.io"}, - {ClientName: "lighthouse-nethermind-1", ClientType: "lighthouse", SSH: "devops@lighthouse-nethermind-1.example.io", BeaconAPI: "bn-lighthouse-nethermind-1.example.io"}, - {ClientName: "prysm-ethrex-1", ClientType: "prysm", SSH: "devops@prysm-ethrex-1.example.io", BeaconAPI: "bn-prysm-ethrex-1.example.io"}, + {ClientName: "lighthouse-ethrex-1", ClientType: "lighthouse"}, + {ClientName: "lighthouse-nethermind-1", ClientType: "lighthouse"}, + {ClientName: "prysm-ethrex-1", ClientType: "prysm"}, }, ExecutionClients: []clientInfo{ - {ClientName: "lighthouse-ethrex-1", ClientType: "ethrex", SSH: "devops@lighthouse-ethrex-1.example.io"}, - {ClientName: "lighthouse-nethermind-1", ClientType: "nethermind", SSH: "devops@lighthouse-nethermind-1.example.io"}, - {ClientName: "prysm-ethrex-1", ClientType: "ethrex", SSH: "devops@prysm-ethrex-1.example.io"}, + {ClientName: "lighthouse-ethrex-1", ClientType: "ethrex"}, + {ClientName: "lighthouse-nethermind-1", ClientType: "nethermind"}, + {ClientName: "prysm-ethrex-1", ClientType: "ethrex"}, }, } } @@ -31,7 +31,7 @@ func targetNames(ts []Target) []string { } func TestResolveTargets(t *testing.T) { - targets := ResolveTargets(testInventory(), "https") + targets := ResolveTargets(testInventory()) if len(targets) != 3 { t.Fatalf("want 3 targets, got %d (%v)", len(targets), targetNames(targets)) } @@ -44,10 +44,6 @@ func TestResolveTargets(t *testing.T) { } } - if lh.BeaconURL != "https://bn-lighthouse-ethrex-1.example.io" { - t.Errorf("beacon url = %q", lh.BeaconURL) - } - got := map[string]bool{} for _, tok := range lh.tokens { got[tok] = true @@ -61,7 +57,7 @@ func TestResolveTargets(t *testing.T) { } func TestSelect(t *testing.T) { - targets := ResolveTargets(testInventory(), "https") + targets := ResolveTargets(testInventory()) cases := []struct { expr string @@ -87,7 +83,7 @@ func TestSelect(t *testing.T) { } func TestSuggestionsScope(t *testing.T) { - targets := ResolveTargets(testInventory(), "https") + targets := ResolveTargets(testInventory()) set := map[string]bool{} for _, s := range Suggestions(targets, "", "lighthouse", 25) { diff --git a/pkg/roll/provider.go b/pkg/roll/provider.go index 20754f6..329f2bd 100644 --- a/pkg/roll/provider.go +++ b/pkg/roll/provider.go @@ -13,9 +13,8 @@ const defaultProviderTTL = 30 * time.Second // InventoryProvider fetches and caches per-network targets from cartographoor, // so autocomplete and rollouts don't refetch on every Discord interaction. type InventoryProvider struct { - baseURL string - beaconScheme string - ttl time.Duration + baseURL string + ttl time.Duration mu sync.Mutex cache map[string]cacheEntry @@ -26,26 +25,21 @@ type cacheEntry struct { fetched time.Time } -// NewInventoryProvider returns a provider. Empty baseURL/beaconScheme use -// defaults; non-positive ttl uses a 30s default. -func NewInventoryProvider(baseURL, beaconScheme string, ttl time.Duration) *InventoryProvider { +// NewInventoryProvider returns a provider. Empty baseURL uses the default; +// non-positive ttl uses a 30s default. +func NewInventoryProvider(baseURL string, ttl time.Duration) *InventoryProvider { if baseURL == "" { baseURL = DefaultInventoryBaseURL } - if beaconScheme == "" { - beaconScheme = "https" - } - if ttl <= 0 { ttl = defaultProviderTTL } return &InventoryProvider{ - baseURL: baseURL, - beaconScheme: beaconScheme, - ttl: ttl, - cache: map[string]cacheEntry{}, + baseURL: baseURL, + ttl: ttl, + cache: map[string]cacheEntry{}, } } @@ -65,7 +59,7 @@ func (p *InventoryProvider) Targets(ctx context.Context, network string) ([]Targ return nil, err } - targets := ResolveTargets(inv, p.beaconScheme) + targets := ResolveTargets(inv) p.mu.Lock() p.cache[network] = cacheEntry{targets: targets, fetched: time.Now()} diff --git a/pkg/service/config.go b/pkg/service/config.go index e6b1822..d09b0a6 100644 --- a/pkg/service/config.go +++ b/pkg/service/config.go @@ -28,8 +28,6 @@ type Config struct { MetricsAddress string // Defaults to :9091 HealthCheckAddress string // Defaults to :9191 WatchtowerAPIToken string // watchtower API token for the /roll command - NodeBasicAuthUser string // basic auth user for beacon health endpoints (bn-* vhosts) - NodeBasicAuthPass string // basic auth password for beacon health endpoints } // AsS3Config converts the configuration to an S3Config. diff --git a/pkg/service/service.go b/pkg/service/service.go index 0f98141..e7617bc 100644 --- a/pkg/service/service.go +++ b/pkg/service/service.go @@ -149,8 +149,6 @@ func NewService(ctx context.Context, log *logrus.Logger, cfg *Config) (*Service, build.NewBuildCommand(log, bot, cfg.GithubToken, githubHTTPClient), cmdroll.NewRollCommand(log, bot, cmdroll.Config{ WatchtowerToken: cfg.WatchtowerAPIToken, - BasicAuthUser: cfg.NodeBasicAuthUser, - BasicAuthPass: cfg.NodeBasicAuthPass, }), })