diff --git a/machinery/go.mod b/machinery/go.mod index e571134..8bd0e75 100644 --- a/machinery/go.mod +++ b/machinery/go.mod @@ -2,6 +2,9 @@ module github.com/kerberos-io/agent/machinery go 1.19 +// replace github.com/kerberos-io/joy4 v1.0.57 => ../../../../github.com/kerberos-io/joy4 +// replace github.com/kerberos-io/onvif v0.0.5 => ../../../../github.com/kerberos-io/onvif + require ( github.com/InVisionApp/conjungo v1.1.0 github.com/appleboy/gin-jwt/v2 v2.9.1 @@ -21,7 +24,7 @@ require ( github.com/golang-module/carbon/v2 v2.2.3 github.com/gorilla/websocket v1.5.0 github.com/kellydunn/golang-geo v0.7.0 - github.com/kerberos-io/joy4 v1.0.57 + github.com/kerberos-io/joy4 v1.0.58 github.com/kerberos-io/onvif v0.0.5 github.com/minio/minio-go/v6 v6.0.57 github.com/nsmith5/mjpeg v0.0.0-20200913181537-54b8ada0e53e diff --git a/machinery/go.sum b/machinery/go.sum index 8e52797..e96e66e 100644 --- a/machinery/go.sum +++ b/machinery/go.sum @@ -264,8 +264,8 @@ github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+MrarZYNNg= github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU= -github.com/kerberos-io/joy4 v1.0.57 h1:/8epNAJv4cOzBG8pFiM9hVNXfwsgA+8/2nHQ2yOeyII= -github.com/kerberos-io/joy4 v1.0.57/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU= +github.com/kerberos-io/joy4 v1.0.58 h1:R8EECSF+bG7o2yHC6cX/lF77Z+bDVGl6OioLZ3+5MN4= +github.com/kerberos-io/joy4 v1.0.58/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU= github.com/kerberos-io/onvif v0.0.5 h1:kq9mnHZkih9Jl4DyIJ4Rzt++Y3DDKy3nI8S2ESEfZ5w= github.com/kerberos-io/onvif v0.0.5/go.mod h1:Hr2dJOH2LM5SpYKk17gYZ1CMjhGhUl+QlT5kwYogrW0= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= diff --git a/machinery/main.go b/machinery/main.go index 1858800..7a44250 100644 --- a/machinery/main.go +++ b/machinery/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "os" "time" @@ -16,7 +17,6 @@ import ( var VERSION = "3.0.0" func main() { - // You might be interested in debugging the agent. if os.Getenv("DATADOG_AGENT_ENABLED") == "true" { if os.Getenv("DATADOG_AGENT_K8S_ENABLED") == "true" { @@ -111,8 +111,14 @@ func main() { } } + // Create a cancelable context, which will be used to cancel and restart. + // This is used to restart the agent when the configuration is updated. + ctx, cancel := context.WithCancel(context.Background()) + // Bootstrapping the agent communication := models.Communication{ + Context: &ctx, + CancelContext: &cancel, HandleBootstrap: make(chan string, 1), } go components.Bootstrap(&configuration, &communication) diff --git a/machinery/src/capture/IPCamera.go b/machinery/src/capture/IPCamera.go index be9664f..b109bcc 100644 --- a/machinery/src/capture/IPCamera.go +++ b/machinery/src/capture/IPCamera.go @@ -1,6 +1,7 @@ package capture import ( + "context" "strconv" "sync" "time" @@ -15,9 +16,9 @@ import ( "github.com/kerberos-io/joy4/format" ) -func OpenRTSP(url string) (av.DemuxCloser, []av.CodecData, error) { +func OpenRTSP(ctx context.Context, url string) (av.DemuxCloser, []av.CodecData, error) { format.RegisterAll() - infile, err := avutil.Open(url) + infile, err := avutil.Open(ctx, url) if err == nil { streams, errstreams := infile.Streams() return infile, streams, errstreams diff --git a/machinery/src/capture/main.go b/machinery/src/capture/main.go index 064642b..2b52144 100644 --- a/machinery/src/capture/main.go +++ b/machinery/src/capture/main.go @@ -2,6 +2,7 @@ package capture import ( + "context" "os" "strconv" "time" @@ -431,6 +432,10 @@ func VerifyCamera(c *gin.Context) { var cameraStreams models.CameraStreams err := c.BindJSON(&cameraStreams) + // Should return in 5 seconds. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err == nil { streamType := c.Param("streamType") @@ -442,7 +447,7 @@ func VerifyCamera(c *gin.Context) { if streamType == "secondary" { rtspUrl = cameraStreams.SubRTSP } - _, codecs, err := OpenRTSP(rtspUrl) + _, codecs, err := OpenRTSP(ctx, rtspUrl) if err == nil { videoIdx := -1 diff --git a/machinery/src/components/Kerberos.go b/machinery/src/components/Kerberos.go index 49a7a86..0476ad5 100644 --- a/machinery/src/components/Kerberos.go +++ b/machinery/src/components/Kerberos.go @@ -1,6 +1,7 @@ package components import ( + "context" "runtime" "strconv" "sync" @@ -73,14 +74,22 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun for { // This will blocking until receiving a signal to be restarted, reconfigured, stopped, etc. status := RunAgent(configuration, communication, uptimeStart, cameraSettings, decoder, subDecoder) + if status == "stop" { break } + // We will re open the configuration, might have changed :O! OpenConfig(configuration) // We will override the configuration with the environment variables OverrideWithEnvironmentVariables(configuration) + + // We will create a new cancelable context, which will be used to cancel and restart. + // This is used to restart the agent when the configuration is updated. + ctx, cancel := context.WithCancel(context.Background()) + communication.Context = &ctx + communication.CancelContext = &cancel } log.Log.Debug("Bootstrap: finished") } @@ -90,10 +99,27 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi log.Log.Debug("RunAgent: bootstrapping agent") config := configuration.Config + status := "not started" + // Currently only support H264 encoded cameras, this will change. // Establishing the camera connection rtspUrl := config.Capture.IPCamera.RTSP - infile, streams, err := capture.OpenRTSP(rtspUrl) + infile, streams, err := capture.OpenRTSP(context.Background(), rtspUrl) + + // We will initialise the camera settings object + // so we can check if the camera settings have changed, and we need + // to reload the decoders. + + videoStream, _ := capture.GetVideoStream(streams) + if videoStream == nil { + log.Log.Error("RunAgent: no video stream found, might be the wrong codec (we only support H264 for the moment)") + time.Sleep(time.Second * 3) + return status + } + + num, denum := videoStream.(av.VideoCodecData).Framerate() + width := videoStream.(av.VideoCodecData).Width() + height := videoStream.(av.VideoCodecData).Height() var queue *pubsub.Queue var subQueue *pubsub.Queue @@ -101,8 +127,6 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi var decoderMutex sync.Mutex var subDecoderMutex sync.Mutex - status := "not started" - if err == nil { log.Log.Info("RunAgent: opened RTSP stream" + rtspUrl) @@ -113,23 +137,21 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi subStreamEnabled := false subRtspUrl := config.Capture.IPCamera.SubRTSP if subRtspUrl != "" && subRtspUrl != rtspUrl { - subInfile, subStreams, err = capture.OpenRTSP(subRtspUrl) + subInfile, subStreams, err = capture.OpenRTSP(context.Background(), subRtspUrl) if err == nil { log.Log.Info("RunAgent: opened RTSP sub stream " + subRtspUrl) subStreamEnabled = true } + + videoStream, _ := capture.GetVideoStream(subStreams) + if videoStream == nil { + log.Log.Error("RunAgent: no video substream found, might be the wrong codec (we only support H264 for the moment)") + time.Sleep(time.Second * 3) + return status + } } - // We will initialise the camera settings object - // so we can check if the camera settings have changed, and we need - // to reload the decoders. - videoStream, _ := capture.GetVideoStream(streams) - num, denum := videoStream.(av.VideoCodecData).Framerate() - width := videoStream.(av.VideoCodecData).Width() - height := videoStream.(av.VideoCodecData).Height() - if cameraSettings.RTSP != rtspUrl || cameraSettings.SubRTSP != subRtspUrl || cameraSettings.Width != width || cameraSettings.Height != height || cameraSettings.Num != num || cameraSettings.Denum != denum || cameraSettings.Codec != videoStream.(av.VideoCodecData).Type() { - if cameraSettings.Initialized { decoder.Close() if subStreamEnabled { @@ -240,8 +262,12 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! // This will go into a blocking state, once this channel is triggered // the agent will cleanup and restart. + status = <-communication.HandleBootstrap + // Cancel the main context, this will stop all the other goroutines. + (*communication.CancelContext)() + // Here we are cleaning up everything! if configuration.Config.Offline != "true" { communication.HandleUpload <- "stop" diff --git a/machinery/src/computervision/main.go b/machinery/src/computervision/main.go index 47e6156..2602f46 100644 --- a/machinery/src/computervision/main.go +++ b/machinery/src/computervision/main.go @@ -10,15 +10,12 @@ import ( "time" mqtt "github.com/eclipse/paho.mqtt.golang" + geo "github.com/kellydunn/golang-geo" "github.com/kerberos-io/agent/machinery/src/capture" "github.com/kerberos-io/agent/machinery/src/log" "github.com/kerberos-io/agent/machinery/src/models" - "github.com/kerberos-io/joy4/av/pubsub" - - //"github.com/whorfin/go-libjpeg/jpeg" - - geo "github.com/kellydunn/golang-geo" "github.com/kerberos-io/joy4/av" + "github.com/kerberos-io/joy4/av/pubsub" "github.com/kerberos-io/joy4/cgo/ffmpeg" ) diff --git a/machinery/src/models/Communication.go b/machinery/src/models/Communication.go index 40147c5..93c526d 100644 --- a/machinery/src/models/Communication.go +++ b/machinery/src/models/Communication.go @@ -1,6 +1,7 @@ package models import ( + "context" "sync" "sync/atomic" @@ -12,6 +13,8 @@ import ( // The communication struct that is managing // all the communication between the different goroutines. type Communication struct { + Context *context.Context + CancelContext *context.CancelFunc PackageCounter *atomic.Value LastPacketTimer *atomic.Value CloudTimestamp *atomic.Value