33f888a6e2
Two plans covering SSH executor, config normalization, NATS publishing, backup scheduler, and main.go wiring for periodic RouterOS config backup. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
18 KiB
18 KiB
phase, plan, type, wave, depends_on, files_modified, autonomous, requirements, must_haves
| phase | plan | type | wave | depends_on | files_modified | autonomous | requirements | must_haves | |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 02-poller-config-collection | 02 | execute | 2 |
|
|
true |
|
|
Purpose: This is the core orchestration that ties together the SSH executor, normalizer, and NATS publisher from Plan 01 into a running backup collection system with proper scheduling, concurrency control, error handling, and retry logic. Output: BackupScheduler module fully integrated into the poller's main.go lifecycle.
<execution_context> @/Users/jasonstaack/.claude/get-shit-done/workflows/execute-plan.md @/Users/jasonstaack/.claude/get-shit-done/templates/summary.md </execution_context>
@.planning/PROJECT.md @.planning/ROADMAP.md @.planning/STATE.md @.planning/phases/02-poller-config-collection/02-CONTEXT.md @.planning/phases/02-poller-config-collection/02-01-SUMMARY.md@poller/internal/poller/scheduler.go @poller/internal/poller/worker.go @poller/internal/poller/interfaces.go @poller/cmd/poller/main.go @poller/internal/device/ssh_executor.go @poller/internal/device/normalize.go @poller/internal/bus/publisher.go @poller/internal/config/config.go @poller/internal/store/devices.go @poller/internal/observability/metrics.go
From poller/internal/device/ssh_executor.go (created in Plan 01):
type SSHErrorKind string
const (
ErrAuthFailed SSHErrorKind = "auth_failed"
ErrHostKeyMismatch SSHErrorKind = "host_key_mismatch"
ErrTimeout SSHErrorKind = "timeout"
ErrTruncatedOutput SSHErrorKind = "truncated_output"
ErrConnectionRefused SSHErrorKind = "connection_refused"
ErrUnknown SSHErrorKind = "unknown"
)
type SSHError struct { Kind SSHErrorKind; Err error; Message string }
type CommandResult struct { Stdout string; Stderr string; ExitCode int; Duration time.Duration }
func RunCommand(ctx context.Context, ip string, port int, username, password string, timeout time.Duration, knownFingerprint string, command string) (*CommandResult, string, error)
From poller/internal/device/normalize.go (created in Plan 01):
func NormalizeConfig(raw string) string
func HashConfig(normalized string) string
const NormalizationVersion = 1
From poller/internal/bus/publisher.go (modified in Plan 01):
type ConfigSnapshotEvent struct {
DeviceID string `json:"device_id"`
TenantID string `json:"tenant_id"`
RouterOSVersion string `json:"routeros_version,omitempty"`
CollectedAt string `json:"collected_at"`
SHA256Hash string `json:"sha256_hash"`
ConfigText string `json:"config_text"`
NormalizationVersion int `json:"normalization_version"`
}
func (p *Publisher) PublishConfigSnapshot(ctx context.Context, event ConfigSnapshotEvent) error
From poller/internal/store/devices.go (modified in Plan 01):
type Device struct {
// ... existing fields ...
SSHPort int
SSHHostKeyFingerprint *string
}
func (s *DeviceStore) UpdateSSHHostKey(ctx context.Context, deviceID string, fingerprint string) error
From poller/internal/config/config.go (modified in Plan 01):
type Config struct {
// ... existing fields ...
ConfigBackupIntervalSeconds int
ConfigBackupMaxConcurrent int
ConfigBackupCommandTimeoutSeconds int
}
From poller/internal/observability/metrics.go (modified in Plan 01):
var ConfigBackupTotal *prometheus.CounterVec // labels: ["status"]
var ConfigBackupDuration prometheus.Histogram
var ConfigBackupActive prometheus.Gauge
From poller/internal/poller/scheduler.go:
type Scheduler struct { ... }
func NewScheduler(...) *Scheduler
func (s *Scheduler) Run(ctx context.Context) error
func (s *Scheduler) reconcileDevices(ctx context.Context, wg *sync.WaitGroup) error
func (s *Scheduler) runDeviceLoop(ctx context.Context, dev store.Device, ds *deviceState) // per-device goroutine with ticker
From poller/internal/poller/interfaces.go:
type DeviceFetcher interface {
FetchDevices(ctx context.Context) ([]store.Device, error)
}
**2. Create backup_scheduler.go:**
Define `backupDeviceState` struct tracking per-device backup state:
- `cancel context.CancelFunc`
- `lastAttemptAt time.Time`
- `lastSuccessAt time.Time`
- `lastStatus string` — "success", "error", "skipped_offline", "auth_blocked", "hostkey_blocked"
- `lastError string`
- `consecutiveFailures int`
- `backoffUntil time.Time`
- `lastErrorKind device.SSHErrorKind` — tracks whether error is auth/hostkey (blocks retry)
Define `BackupScheduler` struct:
- `store DeviceFetcher` — reuse existing interface for FetchDevices
- `hostKeyStore SSHHostKeyUpdater` — for UpdateSSHHostKey
- `locker *redislock.Client` — per-device distributed lock
- `publisher *bus.Publisher` — for NATS publishing
- `credentialCache *vault.CredentialCache` — for decrypting device SSH creds
- `redisClient *redis.Client` — for tracking device online status
- `backupInterval time.Duration`
- `commandTimeout time.Duration`
- `refreshPeriod time.Duration` — how often to reconcile devices (reuse from existing scheduler, e.g., 60s)
- `semaphore chan struct{}` — buffered channel of size maxConcurrent
- `mu sync.Mutex`
- `activeDevices map[string]*backupDeviceState`
`NewBackupScheduler(...)` constructor — accept all dependencies, create semaphore as `make(chan struct{}, maxConcurrent)`.
`Run(ctx context.Context) error` — mirrors existing Scheduler.Run pattern:
- defer shutdown: cancel all device goroutines, wait for WaitGroup
- Loop: reconcileBackupDevices(ctx, &wg), then select on ctx.Done or time.After(refreshPeriod)
`reconcileBackupDevices(ctx, wg)` — mirrors reconcileDevices:
- FetchDevices from store
- Start backup goroutines for new devices
- Stop goroutines for removed devices
`runBackupLoop(ctx, dev, state)` — per-device backup goroutine:
- On first run: sleep for randomJitter(30, 300) seconds, then do initial backup
- After initial: ticker at backupInterval
- On each tick:
a. Check if device is online via Redis key `device:{id}:status` (set by status poll). If not online, log debug "skipped_offline", update state, increment ConfigBackupTotal("skipped_offline"), continue
b. Check if lastErrorKind is ErrAuthFailed — skip with "skipped_auth_blocked", log warning with guidance to update credentials
c. Check if lastErrorKind is ErrHostKeyMismatch — skip with "skipped_hostkey_blocked", log warning with guidance to reset host key
d. Check backoff: if time.Now().Before(state.backoffUntil), skip
e. Acquire semaphore (blocks if at max concurrency, does not drop)
f. Acquire Redis lock `backup:device:{id}` with TTL = commandTimeout + 30s
g. Call `collectAndPublish(ctx, dev, state)`
h. Release semaphore
i. Update state based on result
`collectAndPublish(ctx, dev, state) error`:
- Increment ConfigBackupActive gauge
- Defer decrement ConfigBackupActive gauge
- Start timer for ConfigBackupDuration
- Decrypt credentials via credentialCache.GetCredentials
- Call `device.RunCommand(ctx, dev.IPAddress, dev.SSHPort, username, password, commandTimeout, knownFingerprint, "/export show-sensitive")`
- On error: classify error kind, update state, apply backoff (transient: 5m/15m/1h exponential; auth/hostkey: block), return
- If new fingerprint returned (TOFU first connect): call hostKeyStore.UpdateSSHHostKey
- Validate output is non-empty and looks like RouterOS config (basic sanity: contains "/")
- Call `device.NormalizeConfig(result.Stdout)`
- Call `device.HashConfig(normalized)`
- Build `bus.ConfigSnapshotEvent` with device_id, tenant_id, routeros_version (from device or Redis), collected_at (RFC3339 now), sha256_hash, config_text, normalization_version
- Call `publisher.PublishConfigSnapshot(ctx, event)`
- On success: reset consecutiveFailures, update lastSuccessAt, increment ConfigBackupTotal("success")
- Record ConfigBackupDuration
`randomJitter(minSeconds, maxSeconds int) time.Duration` — uses math/rand for uniform distribution
Backoff for transient errors: `calculateBackupBackoff(failures int) time.Duration`:
- 1 failure: 5 min
- 2 failures: 15 min
- 3+ failures: 1 hour (cap)
Device online check via Redis: check if key `device:{id}:status` equals "online". This key is set by the existing status poll publisher flow. If key doesn't exist, assume device might be online (first poll hasn't happened yet) — allow backup attempt.
RouterOS version: read from the Device struct's RouterOSVersion field (populated by store query). If nil, use empty string in the event.
**Important implementation notes:**
- Use `log/slog` for all logging (structured JSON, matching existing pattern)
- Use existing `redislock` pattern from worker.go for per-device locking
- Semaphore pattern: `s.semaphore <- struct{}{}` to acquire, `<-s.semaphore` to release
- Do NOT share circuit breaker state with the status poll scheduler — these are independent
- Partial/truncated output (SSHError with Kind ErrTruncatedOutput) is treated as transient error — never publish, apply backoff
After the existing scheduler creation (around line 270), add a new section:
```
// -----------------------------------------------------------------------
// Start the config backup scheduler
// -----------------------------------------------------------------------
```
1. Convert config values to durations:
```go
backupInterval := time.Duration(cfg.ConfigBackupIntervalSeconds) * time.Second
backupCmdTimeout := time.Duration(cfg.ConfigBackupCommandTimeoutSeconds) * time.Second
```
2. Create BackupScheduler:
```go
backupScheduler := poller.NewBackupScheduler(
deviceStore,
deviceStore, // SSHHostKeyUpdater (DeviceStore satisfies this interface)
locker,
publisher,
credentialCache,
redisClient,
backupInterval,
backupCmdTimeout,
refreshPeriod, // reuse existing device refresh period
cfg.ConfigBackupMaxConcurrent,
)
```
3. Start in a goroutine (runs parallel with the main status poll scheduler):
```go
go func() {
slog.Info("starting config backup scheduler",
"interval", backupInterval,
"max_concurrent", cfg.ConfigBackupMaxConcurrent,
"command_timeout", backupCmdTimeout,
)
if err := backupScheduler.Run(ctx); err != nil {
slog.Error("backup scheduler exited with error", "error", err)
}
}()
```
The BackupScheduler shares the same ctx as everything else, so SIGINT/SIGTERM will trigger its shutdown via context cancellation. No additional shutdown logic needed — Run() returns when ctx is cancelled.
Log the startup with the same pattern as the existing scheduler startup log (line 273-276).
<success_criteria>
- BackupScheduler runs independently from status poll scheduler with its own per-device goroutines
- Devices get their first backup 30-300s after discovery, then every CONFIG_BACKUP_INTERVAL
- SSH command execution uses TOFU host key verification and stores fingerprints on first connect
- Config output is normalized, hashed, and published to NATS config.snapshot.create
- Concurrency limited to CONFIG_BACKUP_MAX_CONCURRENT parallel SSH sessions
- Auth/hostkey errors block retries; transient errors use exponential backoff (5m/15m/1h)
- Offline devices are skipped gracefully
- BackupScheduler is wired into main.go and starts/stops with the poller lifecycle
- All tests pass, project compiles clean </success_criteria>