chore: remove mqtt dependency (#68)

Author: leoparente

agent/agent.go

@@ -7,7 +7,6 @@ import (
 	"runtime"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/google/uuid"
 	"github.com/mitchellh/mapstructure"
 	"go.uber.org/zap"
@@ -32,8 +31,6 @@ type Agent interface {
 type orbAgent struct {
 	logger            *zap.Logger
 	config            config.Config
-	client            mqtt.Client
-	agentID           string
 	backends          map[string]backend.Backend
 	backendState      map[string]*backend.State
 	backendsCommon    config.BackendCommons
@@ -45,10 +42,6 @@ type orbAgent struct {
 	heartbeatCtx    context.Context
 	heartbeatCancel context.CancelFunc
 
-	// Agent RPC channel, configured from command line
-	baseTopic        string
-	rpcFromCoreTopic string
-
 	// Retry Mechanism to ensure the Request is received
 	groupRequestSucceeded  context.CancelFunc
 	policyRequestSucceeded context.CancelFunc
@@ -149,13 +142,6 @@ func (a *orbAgent) Start(ctx context.Context, cancelFunc context.CancelFunc) err
 	a.rpcFromCancelFunc = cancelAllAsync
 	a.cancelFunction = cancelFunc
 	a.logger.Info("agent started", zap.String("version", version.GetBuildVersion()), zap.Any("routine", agentCtx.Value(routineKey)))
-	mqtt.CRITICAL = &agentLoggerCritical{a: a}
-	mqtt.ERROR = &agentLoggerError{a: a}
-
-	if a.config.OrbAgent.Debug.Enable {
-		a.logger.Info("debug logging enabled")
-		mqtt.DEBUG = &agentLoggerDebug{a: a}
-	}
 
 	if err := a.startBackends(ctx); err != nil {
 		return err
@@ -180,11 +166,6 @@ func (a *orbAgent) logoffWithHeartbeat(ctx context.Context) {
 	if a.heartbeatCtx != nil {
 		a.heartbeatCancel()
 	}
-	if a.client != nil && a.client.IsConnected() {
-		if token := a.client.Unsubscribe(a.rpcFromCoreTopic); token.Wait() && token.Error() != nil {
-			a.logger.Warn("failed to unsubscribe to RPC channel", zap.Error(token.Error()))
-		}
-	}
 }
 
 func (a *orbAgent) Stop(ctx context.Context) {
@@ -201,9 +182,6 @@ func (a *orbAgent) Stop(ctx context.Context) {
 		}
 	}
 	a.logoffWithHeartbeat(ctx)
-	if a.client != nil && a.client.IsConnected() {
-		a.client.Disconnect(0)
-	}
 	a.logger.Debug("stopping agent with number of go routines and go calls", zap.Int("goroutines", runtime.NumGoroutine()), zap.Int64("gocalls", runtime.NumCgoCall()))
 	if a.policyRequestSucceeded != nil {
 		a.policyRequestSucceeded()
@@ -237,7 +215,6 @@ func (a *orbAgent) RestartBackend(ctx context.Context, name string, reason strin
 		a.backendState[name].LastError = fmt.Sprintf("failed to reset backend: %v", err)
 		a.logger.Error("failed to reset backend", zap.String("backend", name), zap.Error(err))
 	}
-	be.SetCommsClient(a.agentID, &a.client, fmt.Sprintf("%s/?/%s", a.baseTopic, name))
 
 	return nil
 }

agent/backend/backend.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"go.uber.org/zap"
 
 	"github.com/netboxlabs/orb-agent/agent/config"
@@ -49,7 +48,6 @@ func (s RunningStatus) String() string {
 // Backend is the interface that all backends must implement
 type Backend interface {
 	Configure(*zap.Logger, policies.PolicyRepo, map[string]interface{}, config.BackendCommons) error
-	SetCommsClient(string, *mqtt.Client, string)
 	Version() (string, error)
 	Start(ctx context.Context, cancelFunc context.CancelFunc) error
 	Stop(ctx context.Context) error

agent/backend/devicediscovery/device_discovery.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/go-cmd/cmd"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
@@ -50,8 +49,6 @@ type deviceDiscoveryBackend struct {
 	statusChan <-chan cmd.Status
 	cancelFunc context.CancelFunc
 	ctx        context.Context
-
-	mqttClient *mqtt.Client
 }
 
 type info struct {
@@ -87,10 +84,6 @@ func (d *deviceDiscoveryBackend) Configure(logger *zap.Logger, repo policies.Pol
 	return nil
 }
 
-func (d *deviceDiscoveryBackend) SetCommsClient(_ string, client *mqtt.Client, _ string) {
-	d.mqttClient = client
-}
-
 func (d *deviceDiscoveryBackend) Version() (string, error) {
 	var info info
 	err := d.request("status", &info, http.MethodGet, http.NoBody, "application/json", versionTimeout)

agent/backend/networkdiscovery/network_discovery.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/go-cmd/cmd"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
@@ -50,8 +49,6 @@ type networkDiscoveryBackend struct {
 	statusChan <-chan cmd.Status
 	cancelFunc context.CancelFunc
 	ctx        context.Context
-
-	mqttClient *mqtt.Client
 }
 
 type info struct {
@@ -87,10 +84,6 @@ func (d *networkDiscoveryBackend) Configure(logger *zap.Logger, repo policies.Po
 	return nil
 }
 
-func (d *networkDiscoveryBackend) SetCommsClient(_ string, client *mqtt.Client, _ string) {
-	d.mqttClient = client
-}
-
 func (d *networkDiscoveryBackend) Version() (string, error) {
 	var info info
 	err := d.request("status", &info, http.MethodGet, http.NoBody, "application/json", versionTimeout)

agent/backend/otel/otel.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/go-cmd/cmd"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
@@ -47,8 +46,6 @@ type openTelemetryBackend struct {
 	statusChan  <-chan cmd.Status
 	cancelFunc  context.CancelFunc
 	ctx         context.Context
-
-	mqttClient *mqtt.Client
 }
 
 type info struct {
@@ -248,10 +245,6 @@ func (o *openTelemetryBackend) GetRunningStatus() (backend.RunningStatus, string
 	return runningStatus, "", nil
 }
 
-func (o *openTelemetryBackend) SetCommsClient(_ string, client *mqtt.Client, _ string) {
-	o.mqttClient = client
-}
-
 func (o *openTelemetryBackend) ApplyPolicy(data policies.PolicyData, updatePolicy bool) error {
 	if updatePolicy {
 		// To update a policy it's necessary first remove it and then apply a new version

agent/backend/pktvisor/pktvisor.go

@@ -3,14 +3,11 @@ package pktvisor
 import (
 	"context"
 	"errors"
-	"fmt"
 	"net/http"
 	"os/exec"
 	"strconv"
-	"strings"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/go-cmd/cmd"
 	"go.uber.org/zap"
 
@@ -53,11 +50,7 @@ type pktvisorBackend struct {
 	startTime       time.Time
 	cancelFunc      context.CancelFunc
 	ctx             context.Context
-
-	mqttClient       *mqtt.Client
-	metricsTopic     string
-	otlpMetricsTopic string
-	policyRepo       policies.PolicyRepo
+	policyRepo      policies.PolicyRepo
 
 	adminAPIHost     string
 	adminAPIPort     string
@@ -79,14 +72,6 @@ func (p *pktvisorBackend) GetInitialState() backend.RunningStatus {
 	return backend.Unknown
 }
 
-func (p *pktvisorBackend) SetCommsClient(agentID string, client *mqtt.Client, baseTopic string) {
-	p.mqttClient = client
-	metricsTopic := strings.Replace(baseTopic, "?", "be", 1)
-	otelMetricsTopic := strings.Replace(baseTopic, "?", "otlp", 1)
-	p.metricsTopic = fmt.Sprintf("%s/m/%c", metricsTopic, agentID[0])
-	p.otlpMetricsTopic = fmt.Sprintf("%s/m/%c", otelMetricsTopic, agentID[0])
-}
-
 func (p *pktvisorBackend) GetRunningStatus() (backend.RunningStatus, string, error) {
 	// first check process status
 	runningStatus, errMsg, err := p.getProcRunningStatus()

agent/backend/worker/worker.go

@@ -8,7 +8,6 @@ import (
 	"net/http"
 	"time"
 
-	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/go-cmd/cmd"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
@@ -50,8 +49,6 @@ type workerBackend struct {
 	statusChan <-chan cmd.Status
 	cancelFunc context.CancelFunc
 	ctx        context.Context
-
-	mqttClient *mqtt.Client
 }
 
 type info struct {
@@ -87,10 +84,6 @@ func (d *workerBackend) Configure(logger *zap.Logger, repo policies.PolicyRepo,
 	return nil
 }
 
-func (d *workerBackend) SetCommsClient(_ string, client *mqtt.Client, _ string) {
-	d.mqttClient = client
-}
-
 func (d *workerBackend) Version() (string, error) {
 	var info info
 	err := d.request("status", &info, http.MethodGet, http.NoBody, "application/json", versionTimeout)

agent/logging.go

@@ -3,7 +3,6 @@ module github.com/netboxlabs/orb-agent
 go 1.23.2
 
 require (
-	github.com/eclipse/paho.mqtt.golang v1.5.0
 	github.com/go-cmd/cmd v1.4.3
 	github.com/go-co-op/gocron/v2 v2.15.0
 	github.com/go-git/go-git/v5 v5.13.2
@@ -34,7 +33,6 @@ require (
 	github.com/go-gorp/gorp/v3 v3.1.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/pprof v0.0.0-20230228050547-1710fef4ab10 // indirect
-	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect

go.mod

@@ -23,8 +23,6 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
-github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
 github.com/elazarl/goproxy v1.4.0 h1:4GyuSbFa+s26+3rmYNSuUVsx+HgPrV1bk1jXI0l9wjM=
 github.com/elazarl/goproxy v1.4.0/go.mod h1:X/5W/t+gzDyLfHW4DrMdpjqYjpXsURlBt9lpBDxZZZQ=
 github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc=
@@ -64,8 +62,6 @@ github.com/google/pprof v0.0.0-20230228050547-1710fef4ab10 h1:CqYfpuYIjnlNxM3msd
 github.com/google/pprof v0.0.0-20230228050547-1710fef4ab10/go.mod h1:79YE0hCXdHag9sBkw2o+N/YnZtTkXi0UT9Nnixa5eYk=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
-github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w=