feat(controller-runtime): Add keyvalue store spec and controller (#215)

* Bump helm/kind-action from 1.10.0 to 1.11.0 (#213)

Bumps [helm/kind-action](https://github.com/helm/kind-action) from 1.10.0 to 1.11.0.
- [Release notes](https://github.com/helm/kind-action/releases)
- [Commits](helm/kind-action@v1.10.0...v1.11.0)

---
updated-dependencies:
- dependency-name: helm/kind-action
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* Bump helm/kind-action from 1.11.0 to 1.12.0 (#214)

Bumps [helm/kind-action](https://github.com/helm/kind-action) from 1.11.0 to 1.12.0.
- [Release notes](https://github.com/helm/kind-action/releases)
- [Commits](helm/kind-action@v1.11.0...v1.12.0)

---
updated-dependencies:
- dependency-name: helm/kind-action
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

* * Formatting

* Add initial definitions for KeyValue store

* Deps

* Fix test

* Add KeyValue controller

* Add KeyValue tests

* Update PreventUpdate behavior

* Minor error handling change

---------

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Makefile

@@ -2,7 +2,7 @@ export GO111MODULE := on
 
 SHELL=/usr/bin/env bash
 
-ENVTEST_K8S_VERSION = 1.29.0
+ENVTEST_K8S_VERSION = 1.31.0
 
 now := $(shell date -u +%Y-%m-%dT%H:%M:%S%z)
 gitBranch := $(shell git rev-parse --abbrev-ref HEAD)
@@ -197,7 +197,7 @@ mv "$$(echo "$(1)" | sed "s/-$(3)$$//")" $(1) ;\
 endef
 
 ENVTEST ?= $(LOCALBIN)/setup-envtest-$(ENVTEST_VERSION)
-ENVTEST_VERSION ?= release-0.17
+ENVTEST_VERSION ?= release-0.19
 
 .PHONY: envtest
 envtest: $(ENVTEST) ## Download setup-envtest locally if necessary.

cmd/jetstream-controller/main.go

@@ -18,6 +18,11 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
 	"github.com/nats-io/nack/controllers/jetstream"
 	"github.com/nats-io/nack/internal/controller"
 	jetstreamnatsiov1beta2 "github.com/nats-io/nack/pkg/jetstream/apis/jetstream/v1beta2"
@@ -29,12 +34,9 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	klog "k8s.io/klog/v2"
-	"os"
-	"os/signal"
+
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
-	"syscall"
-	"time"
 )
 
 var (
@@ -155,7 +157,6 @@ func run() error {
 }
 
 func runControlLoop(config *rest.Config, natsCfg *controller.NatsConfig, controllerCfg *controller.Config) error {
-
 	// Setup scheme
 	scheme := runtime.NewScheme()
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))

controllers/jetstream/consumer.go

@@ -244,7 +244,7 @@ func consumerSpecToOpts(spec apis.ConsumerSpec) ([]jsm.ConsumerOption, error) {
 		opts = append(opts, jsm.AcknowledgeExplicit())
 	case "":
 	default:
-		return nil, fmt.Errorf("invalid value for 'ackPolicy': '%s'. Must be one of 'none', 'all', 'explicit'.", spec.AckPolicy)
+		return nil, fmt.Errorf("invalid value for 'ackPolicy': '%s'. Must be one of 'none', 'all', 'explicit'", spec.AckPolicy)
 	}
 
 	if spec.AckWait != "" {
@@ -262,7 +262,7 @@ func consumerSpecToOpts(spec apis.ConsumerSpec) ([]jsm.ConsumerOption, error) {
 		opts = append(opts, jsm.ReplayAsReceived())
 	case "":
 	default:
-		return nil, fmt.Errorf("invalid value for 'replayPolicy': '%s'. Must be one of 'instant', 'original'.", spec.ReplayPolicy)
+		return nil, fmt.Errorf("invalid value for 'replayPolicy': '%s'. Must be one of 'instant', 'original'", spec.ReplayPolicy)
 	}
 
 	if spec.SampleFreq != "" {

controllers/jetstream/controller.go

@@ -98,11 +98,11 @@ type Controller struct {
 
 	strLister listers.StreamLister
 	strSynced cache.InformerSynced
-	strQueue  workqueue.RateLimitingInterface
+	strQueue  workqueue.TypedRateLimitingInterface[any]
 
 	cnsLister listers.ConsumerLister
 	cnsSynced cache.InformerSynced
-	cnsQueue  workqueue.RateLimitingInterface
+	cnsQueue  workqueue.TypedRateLimitingInterface[any]
 
 	accLister listers.AccountLister
 
@@ -137,8 +137,8 @@ func NewController(opt Options) *Controller {
 	}
 
 	ji := opt.JetstreamIface.JetstreamV1beta2()
-	streamQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Streams")
-	consumerQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Consumers")
+	streamQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[any](), "Streams")
+	consumerQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[any](), "Consumers")
 
 	streamInformer.Informer().AddEventHandler(
 		eventHandlers(
@@ -156,6 +156,7 @@ func NewController(opt Options) *Controller {
 	if err != nil {
 		panic(err)
 	}
+	defer os.RemoveAll(cacheDir)
 
 	return &Controller{
 		ctx:  opt.Ctx,
@@ -180,7 +181,6 @@ func NewController(opt Options) *Controller {
 }
 
 func (c *Controller) Run() error {
-
 	// Connect to NATS.
 	opts := make([]nats.Option, 0)
 	// Always attempt to have a connection to NATS.
@@ -647,7 +647,7 @@ func getStorageType(s string) (jsmapi.StorageType, error) {
 	}
 }
 
-func enqueueWork(q workqueue.RateLimitingInterface, item interface{}) (err error) {
+func enqueueWork(q workqueue.TypedRateLimitingInterface[any], item interface{}) (err error) {
 	key, err := cache.MetaNamespaceKeyFunc(item)
 	if err != nil {
 		return fmt.Errorf("failed to enqueue work: %w", err)
@@ -657,10 +657,12 @@ func enqueueWork(q workqueue.RateLimitingInterface, item interface{}) (err error
 	return nil
 }
 
-type jsmClientFunc func(*natsContext) (jsmClient, error)
-type processorFunc func(ns, name string, jmsClient jsmClientFunc) error
+type (
+	jsmClientFunc func(*natsContext) (jsmClient, error)
+	processorFunc func(ns, name string, jmsClient jsmClientFunc) error
+)
 
-func processQueueNext(q workqueue.RateLimitingInterface, jmsClient jsmClientFunc, process processorFunc) {
+func processQueueNext(q workqueue.TypedRateLimitingInterface[any], jmsClient jsmClientFunc, process processorFunc) {
 	item, shutdown := q.Get()
 	if shutdown {
 		return
@@ -730,7 +732,7 @@ func shouldEnqueue(prevObj, nextObj interface{}) bool {
 	return markedDelete || specChanged
 }
 
-func eventHandlers(q workqueue.RateLimitingInterface) cache.ResourceEventHandlerFuncs {
+func eventHandlers(q workqueue.TypedRateLimitingInterface[any]) cache.ResourceEventHandlerFuncs {
 	return cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			if err := enqueueWork(q, obj); err != nil {

controllers/jetstream/controller_test.go

@@ -63,7 +63,7 @@ func TestGetStorageType(t *testing.T) {
 func TestEnqueueWork(t *testing.T) {
 	t.Parallel()
 
-	limiter := workqueue.DefaultControllerRateLimiter()
+	limiter := workqueue.DefaultTypedControllerRateLimiter[any]()
 	q := workqueue.NewNamedRateLimitingQueue(limiter, "StreamsTest")
 	defer q.ShutDown()
 
@@ -97,7 +97,7 @@ func TestProcessQueueNext(t *testing.T) {
 	t.Run("bad item key", func(t *testing.T) {
 		t.Parallel()
 
-		limiter := workqueue.DefaultControllerRateLimiter()
+		limiter := workqueue.DefaultTypedControllerRateLimiter[any]()
 		q := workqueue.NewNamedRateLimitingQueue(limiter, "StreamsTest")
 		defer q.ShutDown()
 
@@ -122,7 +122,7 @@ func TestProcessQueueNext(t *testing.T) {
 	t.Run("process error", func(t *testing.T) {
 		t.Parallel()
 
-		limiter := workqueue.DefaultControllerRateLimiter()
+		limiter := workqueue.DefaultTypedControllerRateLimiter[any]()
 		q := workqueue.NewNamedRateLimitingQueue(limiter, "StreamsTest")
 		defer q.ShutDown()
 
@@ -156,7 +156,7 @@ func TestProcessQueueNext(t *testing.T) {
 	t.Run("process ok", func(t *testing.T) {
 		t.Parallel()
 
-		limiter := workqueue.DefaultControllerRateLimiter()
+		limiter := workqueue.DefaultTypedControllerRateLimiter[any]()
 		q := workqueue.NewNamedRateLimitingQueue(limiter, "StreamsTest")
 		defer q.ShutDown()
 

controllers/jetstream/stream.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	jsm "github.com/nats-io/jsm.go"
-	"github.com/nats-io/jsm.go/api"
 	jsmapi "github.com/nats-io/jsm.go/api"
 	apis "github.com/nats-io/nack/pkg/jetstream/apis/jetstream/v1beta2"
 	typed "github.com/nats-io/nack/pkg/jetstream/generated/clientset/versioned/typed/jetstream/v1beta2"
@@ -223,9 +222,9 @@ func createStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 
 	switch spec.Compression {
 	case "s2":
-		opts = append(opts, jsm.Compression(api.S2Compression))
+		opts = append(opts, jsm.Compression(jsmapi.S2Compression))
 	case "none":
-		opts = append(opts, jsm.Compression(api.NoCompression))
+		opts = append(opts, jsm.Compression(jsmapi.NoCompression))
 	}
 
 	if spec.NoAck {
@@ -289,7 +288,7 @@ func createStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 	}
 
 	if spec.SubjectTransform != nil {
-		opts = append(opts, func(o *api.StreamConfig) error {
+		opts = append(opts, func(o *jsmapi.StreamConfig) error {
 			o.SubjectTransform = &jsmapi.SubjectTransformConfig{
 				Source:      spec.SubjectTransform.Source,
 				Destination: spec.SubjectTransform.Dest,
@@ -425,9 +424,9 @@ func updateStream(ctx context.Context, c jsmClient, spec apis.StreamSpec) (err e
 
 	switch spec.Compression {
 	case "s2":
-		config.Compression = api.S2Compression
+		config.Compression = jsmapi.S2Compression
 	case "none":
-		config.Compression = api.NoCompression
+		config.Compression = jsmapi.NoCompression
 	}
 
 	return js.UpdateConfiguration(config)