outofband/worker: set kv replicas through cli params for consistency

with other controllers

Author: joelrebel

cmd/outofband.go

@@ -36,6 +36,9 @@ var (
 
 	// asWorker when true runs Alloy as a worker listening on the NATS JS for Conditions to act on.
 	asWorker bool
+
+	// The number of replicaCount to use NATS KV data
+	replicaCount int
 )
 
 // outofband inventory, bios configuration collection command
@@ -90,7 +93,7 @@ func runWorker(ctx context.Context, alloy *app.App) {
 		alloy.Logger.Fatal(err)
 	}
 
-	w, err := worker.New(ctx, facilityCode, stream, alloy.Config, alloy.SyncWg, alloy.Logger)
+	w, err := worker.New(ctx, facilityCode, replicaCount, stream, alloy.Config, alloy.SyncWg, alloy.Logger)
 	if err != nil {
 		alloy.Logger.Fatal(err)
 	}
@@ -127,6 +130,7 @@ func init() {
 	cmdOutofband.PersistentFlags().StringVar(&csvFile, "csv-file", "assets.csv", "CSV file containing BMC credentials for assets.")
 	cmdOutofband.PersistentFlags().StringVar(&facilityCode, "facility-code", "sandbox", "The facility code this Alloy instance is associated with")
 	cmdOutofband.PersistentFlags().BoolVar(&asWorker, "worker", false, "Run Alloy as a worker listening for conditions on NATS")
+	cmdOutofband.PersistentFlags().IntVarP(&replicaCount, "replica-count", "r", 3, "The number of replicaCount to use for NATS KV data") // nolint:gomnd // obvious int is obvious
 
 	rootCmd.AddCommand(cmdOutofband)
 }

internal/app/config.go

@@ -240,14 +240,6 @@ func (a *App) envVarNatsOverrides() error {
 		a.Config.NatsOptions.Consumer.Name = a.v.GetString("nats.consumer.name")
 	}
 
-	if a.v.GetInt("nats.kv.replicaCount") != 0 {
-		if a.Config.NatsOptions.KV == nil {
-			a.Config.NatsOptions.KV = &events.NatsKVOptions{}
-		}
-
-		a.Config.NatsOptions.KV.ReplicaCount = a.v.GetInt("nats.kv.replicaCount")
-	}
-
 	if len(a.v.GetStringSlice("nats.consumer.subscribeSubjects")) != 0 {
 		a.Config.NatsOptions.Consumer.SubscribeSubjects = a.v.GetStringSlice("nats.consumer.subscribeSubjects")
 	}

internal/worker/kv_status.go

@@ -29,11 +29,11 @@ var (
 	}
 )
 
-func createOrBindKVBucketWithOpts(s events.Stream, opts *events.NatsKVOptions) (nats.KeyValue, error) {
+func createOrBindKVBucketWithOpts(s events.Stream, replicaCount int) (nats.KeyValue, error) {
 	kvOptions := defaultKVOpts
 
-	if opts.ReplicaCount > 1 {
-		kvOptions = append(kvOptions, kv.WithReplicas(opts.ReplicaCount))
+	if replicaCount > 1 {
+		kvOptions = append(kvOptions, kv.WithReplicas(replicaCount))
 	}
 
 	js, ok := s.(*events.NatsJetstream)
@@ -52,8 +52,8 @@ type statusKVPublisher struct {
 	facility string
 }
 
-func newStatusKVPublisher(s events.Stream, log *logrus.Logger, workerID, facility string, opts *events.NatsKVOptions) (*statusKVPublisher, error) {
-	statusKV, err := createOrBindKVBucketWithOpts(s, opts)
+func newStatusKVPublisher(s events.Stream, log *logrus.Logger, workerID, facility string, replicaCount int) (*statusKVPublisher, error) {
+	statusKV, err := createOrBindKVBucketWithOpts(s, replicaCount)
 	if err != nil {
 		return nil, err
 	}

internal/worker/liveness.go

@@ -34,7 +34,7 @@ func (w *Worker) startWorkerLivenessCheckin(ctx context.Context) {
 			kv.WithTTL(livenessTTL),
 		}
 
-		// any setting of replicas (even 1) chokes NATS in non-clustered mode
+		// any setting of replicaCount (even 1) chokes NATS in non-clustered mode
 		if w.replicaCount != 1 {
 			opts = append(opts, kv.WithReplicas(w.replicaCount))
 		}

internal/worker/worker.go

@@ -64,14 +64,15 @@ type Worker struct {
 	logger       *logrus.Logger
 	name         string
 	facilityCode string
-	replicaCount int
 	concurrency  int
+	replicaCount int
 	dispatched   int32
 }
 
 func New(
 	ctx context.Context,
 	facilityCode string,
+	replicaCount int,
 	stream events.Stream,
 	cfg *app.Configuration,
 	syncWG *sync.WaitGroup,
@@ -92,6 +93,7 @@ func New(
 	return &Worker{
 		name:         id,
 		facilityCode: facilityCode,
+		replicaCount: replicaCount,
 		cfg:          cfg,
 		syncWG:       syncWG,
 		logger:       logger,
@@ -121,7 +123,7 @@ func (w *Worker) Run(ctx context.Context) {
 	// register worker in NATS active-controllers kv bucket
 	w.startWorkerLivenessCheckin(ctx)
 
-	if _, err := createOrBindKVBucketWithOpts(w.stream, w.cfg.NatsOptions.KV); err != nil {
+	if _, err := createOrBindKVBucketWithOpts(w.stream, w.replicaCount); err != nil {
 		w.logger.WithError(err).Error("failed to create/bind to status kv" + inventoryStatusKVBucket)
 	}
 
@@ -286,7 +288,7 @@ func (w *Worker) doWork(ctx context.Context, condition *rctypes.Condition, e eve
 
 	startTS := time.Now()
 
-	publisher, err := newStatusKVPublisher(w.stream, w.logger, w.id.String(), w.facilityCode, w.cfg.NatsOptions.KV)
+	publisher, err := newStatusKVPublisher(w.stream, w.logger, w.id.String(), w.facilityCode, w.replicaCount)
 	if err != nil {
 		w.logger.WithError(err).Warn("status KV init - internal error")