diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e912ee..29a47ca 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -19,6 +19,7 @@ concurrency: cancel-in-progress: true env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true REGISTRY: ghcr.io IMAGE_NAME: ${{ github.repository }} DOCKERHUB_IMAGE: gmcintire/towerops-agent diff --git a/agent.go b/agent.go index f9c43bc..4f6af1d 100644 --- a/agent.go +++ b/agent.go @@ -22,6 +22,7 @@ import ( var doSelfUpdate = selfUpdate var errRestartRequested = fmt.Errorf("restart requested") +var errChannelReloaded = fmt.Errorf("channel reloaded") var joinTimeout = 10 * time.Second var heartbeatInterval = 60 * time.Second var channelHeartbeatInterval = 25 * time.Second @@ -320,9 +321,10 @@ func runSession(ctx context.Context, baseURL, token string) error { slog.Warn("invalid message", "error", err) continue } - if handleMessage(sessionCtx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh) { + shouldEnd, endErr := handleMessage(sessionCtx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh) + if shouldEnd { flushSnmpBatch() - return errRestartRequested + return endErr } case result := <-snmpResultCh: @@ -382,7 +384,7 @@ func runSession(ctx context.Context, baseURL, token string) error { } // handleMessage dispatches incoming channel messages. -// Returns true if the session should end (e.g. restart requested). +// Returns whether the session should end and the reason for reconnecting. func handleMessage( ctx context.Context, msg channelMsg, @@ -393,32 +395,38 @@ func handleMessage( monitoringCheckCh chan<- *pb.MonitoringCheck, checkResultCh chan<- *pb.CheckResult, lldpTopologyResultCh chan<- *pb.LldpTopologyResult, -) bool { +) (bool, error) { switch msg.Event { case "phx_reply": slog.Debug("channel reply", "topic", msg.Topic) + case "phx_error", "phx_close": + slog.Warn("phoenix channel ended, reconnecting", + "event", msg.Event, + "topic", msg.Topic) + return true, errChannelReloaded + case "jobs", "discovery_job", "backup_job": var payload struct { Binary string `json:"binary"` } if err := json.Unmarshal(msg.Payload, &payload); err != nil { slog.Error("decode job payload", "error", err) - return false + return false, nil } if len(payload.Binary) > maxJobPayloadBytes { slog.Error("job payload too large", "size", len(payload.Binary), "max", maxJobPayloadBytes) - return false + return false, nil } bin, err := base64.StdEncoding.DecodeString(payload.Binary) if err != nil { slog.Error("decode base64", "error", err) - return false + return false, nil } var jobList pb.AgentJobList if err := proto.Unmarshal(bin, &jobList); err != nil { slog.Error("unmarshal job list", "error", err) - return false + return false, nil } slog.Info("received jobs", "count", len(jobList.Jobs)) for _, job := range jobList.Jobs { @@ -431,21 +439,21 @@ func handleMessage( } if err := json.Unmarshal(msg.Payload, &payload); err != nil { slog.Error("decode check_jobs payload", "error", err) - return false + return false, nil } if len(payload.Binary) > maxJobPayloadBytes { slog.Error("check_jobs payload too large", "size", len(payload.Binary), "max", maxJobPayloadBytes) - return false + return false, nil } bin, err := base64.StdEncoding.DecodeString(payload.Binary) if err != nil { slog.Error("decode check_jobs base64", "error", err) - return false + return false, nil } var checkList pb.CheckList if err := proto.Unmarshal(bin, &checkList); err != nil { slog.Error("unmarshal check list", "error", err) - return false + return false, nil } slog.Info("received checks", "count", len(checkList.Checks)) for _, check := range checkList.Checks { @@ -454,7 +462,7 @@ func handleMessage( case "restart": slog.Info("restart requested by server") - return true + return true, errRestartRequested case "update": var payload struct { @@ -463,7 +471,7 @@ func handleMessage( } if err := json.Unmarshal(msg.Payload, &payload); err != nil || payload.URL == "" || payload.Checksum == "" { slog.Error("invalid update payload") - return false + return false, nil } slog.Info("update requested", "url", sanitizeURL(payload.URL)) if err := doSelfUpdate(payload.URL, payload.Checksum); err != nil { @@ -473,7 +481,7 @@ func handleMessage( default: slog.Debug("ignoring event", "event", msg.Event) } - return false + return false, nil } // jobPools holds the worker pools for each job type. diff --git a/agent_test.go b/agent_test.go index 187cfbb..62dac8e 100644 --- a/agent_test.go +++ b/agent_test.go @@ -109,7 +109,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "phx_reply", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "phx_reply", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Just verify it doesn't panic }) @@ -137,7 +137,7 @@ func TestHandleMessage(t *testing.T) { SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1", Port: 161}, }) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Wait for goroutine to finish select { case <-snmpCh: @@ -152,7 +152,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -163,7 +163,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": "not-base64!!!"}) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) }) t.Run("invalid protobuf", func(t *testing.T) { @@ -173,7 +173,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": base64.StdEncoding.EncodeToString([]byte{0xFF, 0xFF, 0xFF})}) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) }) t.Run("restart", func(t *testing.T) { @@ -182,11 +182,45 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - shouldEnd := handleMessage(context.Background(), channelMsg{Event: "restart", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + shouldEnd, reason := handleMessage(context.Background(), channelMsg{Event: "restart", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if !shouldEnd { t.Error("expected handleMessage to return true for restart") } + if reason != errRestartRequested { + t.Fatalf("expected restart reason %v, got %v", errRestartRequested, reason) + } + }) + + t.Run("phoenix channel teardown reconnects", func(t *testing.T) { + events := []string{"phx_error", "phx_close"} + + for _, event := range events { + snmpCh := make(chan *pb.SnmpResult, 1) + mtCh := make(chan *pb.MikrotikResult, 1) + credCh := make(chan *pb.CredentialTestResult, 1) + monCh := make(chan *pb.MonitoringCheck, 1) + checkCh := make(chan *pb.CheckResult, 1) + + shouldEnd, reason := handleMessage( + context.Background(), + channelMsg{Topic: "agent:test", Event: event, Payload: json.RawMessage(`{}`)}, + testPools(t), + snmpCh, + mtCh, + credCh, + monCh, + checkCh, + make(chan *pb.LldpTopologyResult, 1), + ) + + if !shouldEnd { + t.Fatalf("expected handleMessage to end session for %s", event) + } + if reason != errChannelReloaded { + t.Fatalf("expected reload reason %v for %s, got %v", errChannelReloaded, event, reason) + } + } }) t.Run("update success", func(t *testing.T) { @@ -205,7 +239,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"url": "https://example.com/agent", "checksum": "abc123"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if calledURL != "https://example.com/agent" { t.Errorf("expected update URL %q, got %q", "https://example.com/agent", calledURL) @@ -229,7 +263,7 @@ func TestHandleMessage(t *testing.T) { checkCh := make(chan *pb.CheckResult, 1) // Missing URL field payload, _ := json.Marshal(map[string]string{"checksum": "abc123"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if called { t.Error("selfUpdate should not be called with empty URL") @@ -250,7 +284,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"url": "https://example.com/agent", "checksum": "abc123"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -270,7 +304,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"url": "https://example.com/agent"}) - handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "update", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) if called { t.Error("selfUpdate should not be called with empty checksum") @@ -283,7 +317,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "some_unknown_event", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "some_unknown_event", Payload: json.RawMessage(`{}`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should just log and not panic }) @@ -300,7 +334,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case <-checkCh: case <-time.After(5 * time.Second): @@ -314,7 +348,7 @@ func TestHandleMessage(t *testing.T) { credCh := make(chan *pb.CredentialTestResult, 1) monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: json.RawMessage(`not json`)}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -325,7 +359,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": "not-base64!!!"}) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -336,7 +370,7 @@ func TestHandleMessage(t *testing.T) { monCh := make(chan *pb.MonitoringCheck, 1) checkCh := make(chan *pb.CheckResult, 1) payload, _ := json.Marshal(map[string]string{"binary": base64.StdEncoding.EncodeToString([]byte{0xFF, 0xFF, 0xFF})}) - handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "check_jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Should log error but not panic }) @@ -363,7 +397,7 @@ func TestHandleMessage(t *testing.T) { JobType: pb.JobType_DISCOVER, SnmpDevice: &pb.SnmpDevice{Ip: "10.0.0.1"}, }) - handleMessage(context.Background(), channelMsg{Event: "discovery_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "discovery_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case <-snmpCh: case <-time.After(2 * time.Second): @@ -391,7 +425,7 @@ func TestHandleMessage(t *testing.T) { JobType: pb.JobType_MIKROTIK, MikrotikDevice: &pb.MikrotikDevice{Ip: "10.0.0.1", SshPort: 22, Username: "admin", Password: "pass"}, }) - handleMessage(context.Background(), channelMsg{Event: "backup_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "backup_job", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) select { case result := <-mtCh: if result.Error != "" { @@ -415,7 +449,7 @@ func TestHandleMessageRejectsOversizedPayload(t *testing.T) { encoded := base64.StdEncoding.EncodeToString(oversized) payload, _ := json.Marshal(map[string]string{"binary": encoded}) - handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) + _, _ = handleMessage(context.Background(), channelMsg{Event: "jobs", Payload: payload}, testPools(t), snmpCh, mtCh, credCh, monCh, checkCh, make(chan *pb.LldpTopologyResult, 1)) // Verify no jobs were dispatched select {