Skip to content

Commit

Permalink
feat(controller-runtime): Add keyvalue store spec and controller (#215)
Browse files Browse the repository at this point in the history 
* 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] <[email protected]>
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] <[email protected]>
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] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
  • Loading branch information
@samuelattwood @dependabot
samuelattwood and dependabot[bot] authored Jan 7, 2025
1 parent 88f61b2 commit c690327
Show file tree
Hide file tree
Showing 85 changed files with 2,693 additions and 974 deletions.
2 changes: 1 addition & 1 deletion .github/workflows/e2e.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ jobs:
chmod +x /usr/local/bin/kubectl-kuttl
- name: create kind cluster
uses: helm/kind-action@v1.10.0
uses: helm/kind-action@v1.12.0

- name: set up helm
uses: azure/setup-helm@v4
Expand Down
4 changes: 2 additions & 2 deletions Makefile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -191,7 +191,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.
Expand Down
11 changes: 6 additions & 5 deletions cmd/jetstream-controller/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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 (
Expand Down Expand Up @@ -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))
Expand Down
8 changes: 4 additions & 4 deletions controllers/jetstream/consumer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ func (c *Controller) processConsumerObject(cns *apis.Consumer, jsm jsmClientFunc

// Write this to the cacheDir
accDir := filepath.Join(c.cacheDir, ns, spec.Account)
if err := os.MkdirAll(accDir, 0755); err != nil {
if err := os.MkdirAll(accDir, 0o755); err != nil {
return err
}

Expand All @@ -101,7 +101,7 @@ func (c *Controller) processConsumerObject(cns *apis.Consumer, jsm jsmClientFunc

// Write the user credentials to the cache dir.
accDir := filepath.Join(c.cacheDir, ns, spec.Account)
if err := os.MkdirAll(accDir, 0755); err != nil {
if err := os.MkdirAll(accDir, 0o755); err != nil {
return err
}
for k, v := range secret.Data {
Expand Down Expand Up @@ -352,7 +352,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 != "" {
Expand All @@ -370,7 +370,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 != "" {
Expand Down
22 changes: 12 additions & 10 deletions controllers/jetstream/controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,11 +97,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

Expand Down Expand Up @@ -136,8 +136,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(
Expand All @@ -155,6 +155,7 @@ func NewController(opt Options) *Controller {
if err != nil {
panic(err)
}
defer os.RemoveAll(cacheDir)

return &Controller{
ctx: opt.Ctx,
Expand All @@ -179,7 +180,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.
Expand Down Expand Up @@ -447,7 +447,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)
Expand All @@ -457,10 +457,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
Expand Down Expand Up @@ -530,7 +532,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 {
Expand Down
8 changes: 4 additions & 4 deletions controllers/jetstream/controller_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down Expand Up @@ -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()

Expand All @@ -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()

Expand Down Expand Up @@ -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()

Expand Down
11 changes: 5 additions & 6 deletions controllers/jetstream/stream.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,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"
Expand Down Expand Up @@ -331,9 +330,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 {
Expand Down Expand Up @@ -397,7 +396,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,
Expand Down Expand Up @@ -533,9 +532,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)
Expand Down
Loading

0 comments on commit c690327

Please sign in to comment.