diff --git a/machinery/go.mod b/machinery/go.mod index 353bf3d..2630af7 100644 --- a/machinery/go.mod +++ b/machinery/go.mod @@ -32,6 +32,7 @@ require ( github.com/pion/rtp v1.8.19 github.com/pion/webrtc/v4 v4.1.2 github.com/sirupsen/logrus v1.9.3 + github.com/stretchr/testify v1.10.0 github.com/swaggo/files v1.0.1 github.com/swaggo/gin-swagger v1.6.0 github.com/swaggo/swag v1.16.4 @@ -58,6 +59,7 @@ require ( github.com/clbanning/mxj v1.8.4 // indirect github.com/clbanning/mxj/v2 v2.7.0 // indirect github.com/cloudwego/base64x v0.1.5 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/elastic/go-windows v1.0.2 // indirect github.com/elgs/gostrgen v0.0.0-20161222160715-9d61ae07eeae // indirect github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect @@ -108,6 +110,7 @@ require ( github.com/pion/stun/v3 v3.0.0 // indirect github.com/pion/transport/v3 v3.0.7 // indirect github.com/pion/turn/v4 v4.0.0 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect diff --git a/machinery/src/onvif/events.go b/machinery/src/onvif/events.go index bcd5fb3..6c2434c 100644 --- a/machinery/src/onvif/events.go +++ b/machinery/src/onvif/events.go @@ -3,6 +3,7 @@ package onvif import ( "context" "errors" + "strings" "time" "github.com/kerberos-io/agent/machinery/src/log" @@ -10,43 +11,78 @@ import ( "github.com/kerberos-io/onvif/event/stream" ) +// initialBackoff and maxBackoff bound the wait between successive +// attempts to (re)open the event stream after a transient construction +// failure (camera reachable but ONVIF not yet ready, brief network blip +// at agent boot, etc.). The library itself handles in-stream reconnect; +// these guards cover the initial-connect path the library cannot see. +const ( + initialBackoff = time.Second + maxBackoff = 5 * time.Minute +) + // HandleONVIFEventStream opens an event/stream against the configured // ONVIF camera and routes Motion events into communication.HandleMotion // so they trigger the existing recording pipeline. // -// The goroutine returns when the stream's context is cancelled (the -// caller closes ctx when shutting the agent down) or when the camera -// is not configured for ONVIF. +// The goroutine retries construction with exponential backoff on +// transient failure (camera reachable later, credentials reloaded, +// network restored). It exits cleanly when ctx is cancelled. // -// This is a feature behind Capture.ONVIFMotion. When the flag is empty -// or "false" the goroutine returns immediately, preserving the -// pixel-diff motion detector as the only source. +// This is a feature behind Capture.ONVIFMotion. When the flag is not +// enabled the goroutine returns immediately, preserving the pixel-diff +// motion detector as the only source. Toggling the flag at runtime +// requires an agent restart (Capture.ONVIFMotion is read once at +// goroutine start). func HandleONVIFEventStream(ctx context.Context, configuration *models.Configuration, communication *models.Communication) { log.Log.Debug("onvif.HandleONVIFEventStream(): started") defer log.Log.Debug("onvif.HandleONVIFEventStream(): finished") - cfg := configuration.Config.Capture - if cfg.ONVIFMotion != "true" { + if !isONVIFMotionEnabled(configuration.Config.Capture.ONVIFMotion) { return } - camera := cfg.IPCamera - if camera.ONVIFXAddr == "" { - log.Log.Info("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do") + if configuration.Config.Capture.IPCamera.ONVIFXAddr == "" { + log.Log.Warning("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do") return } + backoff := initialBackoff + for { + if ctx.Err() != nil { + return + } + recoverable := runStreamOnce(ctx, configuration, communication) + if !recoverable { + return + } + if !sleepCtx(ctx, backoff) { + return + } + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } +} + +// runStreamOnce opens one event stream and consumes from it until ctx +// is cancelled or the stream exits. Returns true when the caller +// should retry construction (transient failure), false on clean +// ctx-driven shutdown. +func runStreamOnce(ctx context.Context, configuration *models.Configuration, communication *models.Communication) (retry bool) { + camera := configuration.Config.Capture.IPCamera + device, _, err := ConnectToOnvifDevice(&camera) if err != nil { log.Log.Error("onvif.HandleONVIFEventStream(): connect: " + err.Error()) - return + return true } - s, err := stream.NewStream(ctx, device, stream.Options{ - DeviceID: configuration.Name, - }) + deviceID := resolveDeviceID(configuration.Name, camera.ONVIFXAddr) + s, err := stream.NewStream(ctx, device, stream.Options{DeviceID: deviceID}) if err != nil { log.Log.Error("onvif.HandleONVIFEventStream(): open stream: " + err.Error()) - return + return true } defer func() { if err := s.Close(); err != nil { @@ -54,21 +90,33 @@ func HandleONVIFEventStream(ctx context.Context, configuration *models.Configura } }() - log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + configuration.Name) + log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + deviceID) + // recovering tracks whether we're in a degraded period so the + // first successful event after an error streak can log a recovery + // line for on-call operators. + var recovering bool for { select { case <-ctx.Done(): - return + return false case ev, ok := <-s.Events(): if !ok { - return + // Lib's run goroutine exited — happens only on ctx + // cancel today (library handles its own reconnect), + // so treat as clean shutdown. + return false } - dispatchEvent(ev, configuration, communication) + if recovering { + log.Log.Info("onvif.HandleONVIFEventStream(): event stream recovered for " + deviceID) + recovering = false + } + dispatchEvent(ctx, ev, configuration, communication) case e, ok := <-s.Errors(): if !ok { - return + return false } + recovering = true logStreamError(e) } } @@ -76,17 +124,20 @@ func HandleONVIFEventStream(ctx context.Context, configuration *models.Configura // dispatchEvent routes a single decoded ONVIF Event into the agent's // existing channels. Motion-active events become MotionDataPartial on -// HandleMotion, matching what the pixel-diff detector emits. Other -// event kinds are logged at debug for now; a follow-up will wire -// DigitalInput/Output into the existing inputOutputDeviceMap. -func dispatchEvent(ev stream.Event, configuration *models.Configuration, communication *models.Communication) { +// HandleMotion, matching what the pixel-diff detector emits. +// +// The ctx pre-check is the shutdown-race guard: between cancel() and +// close(communication.HandleMotion) the agent leaves a ~3s window in +// which a stale event could otherwise attempt to send on a closed +// channel and panic. If ctx is done we drop the event silently — the +// recording pipeline is already winding down. +func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.Configuration, communication *models.Communication) { if ev.Kind != stream.KindMotion { log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + ev.Topic) return } - // We only fire on the leading edge — StateActive. Motion-stop - // handling is a follow-up that needs the recorder state machine - // to accept an explicit stop signal; today the recorder uses a + // Leading-edge only. Motion-stop wiring into the recorder state + // machine is tracked as a follow-up; today the recorder uses a // fixed PostRecording timeout. if ev.State != stream.StateActive { return @@ -94,11 +145,20 @@ func dispatchEvent(ev stream.Event, configuration *models.Configuration, communi if configuration.Config.Capture.Recording == "false" { return } + if ctx.Err() != nil { + return + } + // Timestamp in seconds matches what computervision/main.go emits; + // downstream consumers (capture/main.go) tolerate either second- + // or millisecond-precision. dataToPass := models.MotionDataPartial{ Timestamp: time.Now().Unix(), - NumberOfChanges: 0, // ONVIF doesn't quantify motion area. + NumberOfChanges: 0, // ONVIF does not quantify motion area. } select { + case <-ctx.Done(): + // Closes the residual race: ctx cancelled between the + // pre-check above and reaching this select. case communication.HandleMotion <- dataToPass: default: log.Log.Debug("onvif.dispatchEvent(): HandleMotion full, dropping ONVIF motion event") @@ -106,7 +166,9 @@ func dispatchEvent(ev stream.Event, configuration *models.Configuration, communi } // logStreamError logs at a level matching the severity. Recreate -// failures are louder because they usually mean the camera is offline. +// failures are loud because they usually mean the camera is offline; +// pull/renew failures are debug because the library recovers from them +// automatically. func logStreamError(e error) { var recreate stream.ErrRecreateFailed var pull stream.ErrPullFailed @@ -122,3 +184,39 @@ func logStreamError(e error) { log.Log.Info("onvif.HandleONVIFEventStream(): stream error: " + e.Error()) } } + +// isONVIFMotionEnabled returns true when the Capture.ONVIFMotion flag +// is set to "true" with case and whitespace tolerance. The rest of the +// Capture struct uses string flags so we keep the same shape; the +// difference is that ONVIFMotion defaults to disabled (opt-in), unlike +// Recording / Motion / Snapshots which default to enabled. +func isONVIFMotionEnabled(v string) bool { + return strings.EqualFold(strings.TrimSpace(v), "true") +} + +// resolveDeviceID returns the most useful identifier for the camera +// in stream events, logs and metrics. Falls back from configuration +// name (the operator-supplied label) to the ONVIF endpoint to a +// constant placeholder so log lines always have something to grep. +func resolveDeviceID(configName, xaddr string) string { + if n := strings.TrimSpace(configName); n != "" { + return n + } + if x := strings.TrimSpace(xaddr); x != "" { + return x + } + return "unknown" +} + +// sleepCtx blocks for d or until ctx is cancelled. Returns false if +// ctx was cancelled, true if the full duration elapsed. +func sleepCtx(ctx context.Context, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/machinery/src/onvif/events_test.go b/machinery/src/onvif/events_test.go new file mode 100644 index 0000000..3d86174 --- /dev/null +++ b/machinery/src/onvif/events_test.go @@ -0,0 +1,186 @@ +package onvif + +import ( + "context" + "testing" + "time" + + "github.com/kerberos-io/agent/machinery/src/models" + "github.com/kerberos-io/onvif/event/stream" + "github.com/stretchr/testify/assert" +) + +func makeConfig(recording, onvifMotion, name string) *models.Configuration { + return &models.Configuration{ + Name: name, + Config: models.Config{ + Capture: models.Capture{ + Recording: recording, + ONVIFMotion: onvifMotion, + }, + }, + } +} + +func makeCommunication(buffer int) *models.Communication { + return &models.Communication{ + HandleMotion: make(chan models.MotionDataPartial, buffer), + } +} + +// --- dispatchEvent --------------------------------------------------- + +func TestDispatchEvent_MotionActive_SendsToHandleMotion(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case m := <-comm.HandleMotion: + assert.NotZero(t, m.Timestamp) + case <-time.After(time.Second): + t.Fatal("expected motion data on HandleMotion") + } +} + +func TestDispatchEvent_MotionInactive_DoesNotSend(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateInactive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + t.Fatal("inactive motion must not reach HandleMotion (motion-stop is a follow-up)") + case <-time.After(100 * time.Millisecond): + } +} + +func TestDispatchEvent_NonMotionKindIgnored(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindDigitalInput, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + t.Fatal("non-motion kinds must not reach HandleMotion") + case <-time.After(100 * time.Millisecond): + } +} + +func TestDispatchEvent_RecordingDisabled_DoesNotSend(t *testing.T) { + cfg := makeConfig("false", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + t.Fatal("Recording=false must gate the send (matches computervision behaviour)") + case <-time.After(100 * time.Millisecond): + } +} + +func TestDispatchEvent_HandleMotionFull_DropsRatherThanBlocks(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + // Pre-fill the buffer so the next send would block. + comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)} + comm.HandleMotion <- models.MotionDataPartial{} + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan struct{}) + go func() { + dispatchEvent(ctx, ev, cfg, comm) + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("dispatchEvent must drop when HandleMotion is full, not block") + } +} + +func TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic(t *testing.T) { + // Regression for the shutdown race: between cancel() and + // close(HandleMotion) the agent leaves a 3s window. If dispatchEvent + // runs in that window AFTER the channel is closed, a non-protected + // send would panic. The ctx pre-check must short-circuit before the + // send is attempted. + cfg := makeConfig("true", "true", "cam-1") + comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)} + close(comm.HandleMotion) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already cancelled, matching the shutdown sequence + + assert.NotPanics(t, func() { + dispatchEvent(ctx, ev, cfg, comm) + }) +} + +// --- isONVIFMotionEnabled -------------------------------------------- + +func TestIsONVIFMotionEnabled_CaseAndWhitespace(t *testing.T) { + tests := []struct { + in string + want bool + }{ + {"true", true}, + {"True", true}, + {"TRUE", true}, + {" true", true}, + {"true ", true}, + {" true ", true}, + {"false", false}, + {"False", false}, + {"", false}, + {"yes", false}, + {"1", false}, + } + for _, tc := range tests { + t.Run(tc.in, func(t *testing.T) { + assert.Equal(t, tc.want, isONVIFMotionEnabled(tc.in)) + }) + } +} + +// --- resolveDeviceID ------------------------------------------------- + +func TestResolveDeviceID_FallbackChain(t *testing.T) { + tests := []struct { + name string + cfgName string + xaddr string + want string + }{ + {"name_set", "front-door", "192.168.1.10", "front-door"}, + {"name_empty_xaddr_set", "", "192.168.1.10", "192.168.1.10"}, + {"name_whitespace_only_xaddr_set", " ", "192.168.1.10", "192.168.1.10"}, + {"both_empty", "", "", "unknown"}, + {"name_with_trailing_whitespace", "cam-2 ", "192.168.1.10", "cam-2"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, resolveDeviceID(tc.cfgName, tc.xaddr)) + }) + } +}