add stop and more logs

cmd/process.go

@@ -128,6 +128,7 @@ func process(ctx context.Context, logger *zap.SugaredLogger) error {
 	}
 
 	cx, cancel := context.WithCancel(ctx)
+	defer cancel()
 
 	eSrv, err := echox.NewServer(
 		logger.Desugar(),
@@ -194,6 +195,16 @@ func process(ctx context.Context, logger *zap.SugaredLogger) error {
 		logger.Fatalw("failed to initialize tracer", "error", err)
 	}
 
+	go func() {
+		defer cancel()
+		err := server.ReconcileTimer(ctx, config.AppConfig.ReconcilerInterval)
+		if err != nil {
+			logger.Error("error reconciling load balancers", zap.Error(err))
+		}
+
+		logger.Warn("reconciler stopped")
+	}()
+
 	if err := server.Run(cx); err != nil {
 		logger.Fatalw("failed starting server", "error", err)
 		cancel()
@@ -202,10 +213,13 @@ func process(ctx context.Context, logger *zap.SugaredLogger) error {
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, os.Interrupt)
 
-	recvSig := <-sigCh
-	signal.Stop(sigCh)
-	cancel()
-	logger.Infof("exiting. Performing necessary cleanup", recvSig)
+	select {
+	case <-server.Context.Done():
+		logger.Info("context done signal received")
+	case recvSig := <-sigCh:
+		signal.Stop(sigCh)
+		logger.Infof("exiting. Performing necessary cleanup", recvSig)
+	}
 
 	if err := server.Shutdown(); err != nil {
 		logger.Errorw("failed to shutdown server", zap.Error(err))

internal/srv/errors.go

@@ -6,13 +6,14 @@ var (
 	// ErrPortsRequired is returned when a healthcheck port has not been provided
 	ErrPortsRequired = errors.New("no ports provided")
 
-	errInvalidObjectNameLength = errors.New("object name must be less than 53 characters")
-	errSubscriptionCreate      = errors.New("unable to subscribe to topic")
-	errInvalidHelmClient       = errors.New("unable to create helm client")
-	errInvalidNamespace        = errors.New("unable to create namespace")
-	errInvalidRoleBinding      = errors.New("unable to create namespace role binding")
-	errInvalidHelmValues       = errors.New("unable to create helm values")
-	errLoadBalancerInit        = errors.New("unable to initialize loadbalancer data")
-	errNotMyMessage            = errors.New("message not for this location")
-	errMissingLocation         = errors.New("at least one location is required")
+	errInvalidObjectNameLength   = errors.New("object name must be less than 53 characters")
+	errSubscriptionCreate        = errors.New("unable to subscribe to topic")
+	errInvalidHelmClient         = errors.New("unable to create helm client")
+	errInvalidNamespace          = errors.New("unable to create namespace")
+	errInvalidRoleBinding        = errors.New("unable to create namespace role binding")
+	errInvalidHelmValues         = errors.New("unable to create helm values")
+	errLoadBalancerInit          = errors.New("unable to initialize loadbalancer data")
+	errNotMyMessage              = errors.New("message not for this location")
+	errMissingLocation           = errors.New("at least one location is required")
+	errMissingReconcilerInterval = errors.New("reconciler interval not set")
 )

internal/srv/reconciler.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/hex"
 	"errors"
+	"time"
 
 	lbapi "go.infratographer.com/load-balancer-api/pkg/client"
 	"go.infratographer.com/x/events"
@@ -17,6 +18,28 @@ import (
 // OperatorManagedLabelKey is the label key for the operator managed namespaces
 var OperatorManagedLabelKey = "com.infratographer.lb-operator/managed"
 
+func (s *Server) ReconcileTimer(ctx context.Context, interval time.Duration) error {
+	if interval <= 0 {
+		return errMissingReconcilerInterval
+	}
+
+	ticker := time.NewTicker(interval)
+
+	for {
+		select {
+		case <-ctx.Done():
+			s.Logger.Info("reconciler done")
+
+			return nil
+		case <-ticker.C:
+			s.Logger.Info("starting reconciler run")
+			if err := s.Reconcile(ctx); err != nil {
+				return err
+			}
+		}
+	}
+}
+
 // Reconcile will reconcile out of sync load balancers
 func (s *Server) Reconcile(ctx context.Context) error {
 	s.Logger.Info("starting reconciler")
@@ -70,13 +93,15 @@ func (s *Server) Reconcile(ctx context.Context) error {
 
 	s.Logger.Debugf("actual lbs: %+v\n", actualLBs)
 
+	// add load balancers
 	toAddLBs := difference(expectedLBs, actualLBs)
 	s.Logger.Infof("LBs to add: %+v", toAddLBs)
 
 	if err = s.addLBs(ctx, toAddLBs); err != nil {
 		s.Logger.Error("Error reconciling (adding) missing lbs:", err)
 	}
 
+	// remove load balancers
 	toRemoveLBs := difference(actualLBs, expectedLBs)
 	s.Logger.Infof("LBs to remove: %+v", toRemoveLBs)
 
@@ -161,7 +186,7 @@ func (s *Server) addLBs(ctx context.Context, lbs []string) error {
 		lb, err := s.getLoadBalancer(ctx, lbID, evt)
 
 		if err != nil {
-			s.Logger.Debug("Error getting lb from api: ", err)
+			s.Logger.Warn("Error getting lb from api: ", err)
 			e = errors.Join(e, err)
 
 			continue
@@ -177,7 +202,7 @@ func (s *Server) addLBs(ctx context.Context, lbs []string) error {
 		s.Runner.writer <- task
 	}
 
-	return nil
+	return e
 }
 
 func (s *Server) getLoadBalancer(ctx context.Context, lbID string, evt events.ChangeType) (*loadBalancer, error) {

internal/srv/server.go

@@ -2,7 +2,6 @@ package srv
 
 import (
 	"context"
-	"time"
 
 	"github.com/lestrrat-go/backoff/v2"
 	lbapi "go.infratographer.com/load-balancer-api/pkg/client"
@@ -14,8 +13,6 @@ import (
 	"k8s.io/client-go/rest"
 
 	"go.infratographer.com/ipam-api/pkg/ipamclient"
-
-	"go.infratographer.com/load-balancer-operator/internal/config"
 )
 
 // instrumentationName is a unique package name used for tracing
@@ -59,28 +56,6 @@ func (s *Server) Run(ctx context.Context) error {
 		}
 	}()
 
-	go func() {
-		if config.AppConfig.ReconcilerInterval <= 0 {
-			s.Logger.Warn("reconciler interval not set")
-			return
-		}
-
-		ticker := time.NewTicker(config.AppConfig.ReconcilerInterval)
-
-		for {
-			select {
-			case <-ctx.Done():
-				s.Logger.Info("reconciler done")
-
-				return
-			case <-ticker.C:
-				if err := s.Reconcile(ctx); err != nil {
-					s.Logger.Error("error reconciling load balancers", zap.Error(err))
-				}
-			}
-		}
-	}()
-
 	s.Logger.Infow("starting subscribers")
 
 	if err := s.configureSubscribers(ctx); err != nil {

internal/srv/task.go

@@ -15,6 +15,7 @@ type runner struct {
 	quit       chan struct{}
 	buffer     []*lbTask
 	taskRunner func(*lbTask)
+	ctx        context.Context
 }
 
 type lbTask struct {
@@ -40,6 +41,8 @@ func (r *runner) run() {
 
 	for {
 		select {
+		case <-r.ctx.Done():
+			return
 		case <-r.quit:
 			return
 		default:
@@ -72,6 +75,7 @@ func NewRunner(ctx context.Context, tr taskRunner) *runner {
 		buffer:     make([]*lbTask, 0),
 		quit:       make(chan struct{}),
 		taskRunner: tr,
+		ctx:        ctx,
 	}
 
 	go r.run()