Skip to content

remove orphaned subscribers from nats jetstream channels #679

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
knative-prow merged 8 commits into from
Sep 11, 2025
Merged

remove orphaned subscribers from nats jetstream channels #679

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
1025c2c
added feature to remove orphaned subscribers from channels
astelmashenko Sep 3, 2025
d9968f6
fixed dispatcher test
astelmashenko Sep 3, 2025
24aa40d
added messaging client assign in tests
astelmashenko Sep 3, 2025
1b2d978
switched off check orphaned subs in tests
astelmashenko Sep 3, 2025
3158d9e
renamed orphaned subs check flag
astelmashenko Sep 3, 2025
7968d6d
changed flags values
astelmashenko Sep 3, 2025
f5cb206
removed unnecessary logs
astelmashenko Sep 3, 2025
eeb17b2
included orphaned subs into checks
astelmashenko Sep 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions config/jetstream/200-jsm-dispatcher-clusterrole.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,20 @@ rules:
resources:
- natsjetstreamchannels
- natsjetstreamchannels/status
- natsjetstreamchannels/spec
verbs:
- get
- list
- watch
- update
- patch
- apiGroups:
- messaging.knative.dev
resources:
- subscriptions
verbs:
- get
- list
- apiGroups:
- eventing.knative.dev
resources:
Expand Down
13 changes: 8 additions & 5 deletions pkg/channel/jetstream/dispatcher/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
jsminformer "knative.dev/eventing-natss/pkg/client/injection/informers/messaging/v1alpha1/natsjetstreamchannel"
"knative.dev/eventing-natss/pkg/common/configloader/fsloader"
"knative.dev/eventing/pkg/apis/eventing"
messagingv1client "knative.dev/eventing/pkg/client/injection/client"
"knative.dev/pkg/injection"
pkgreconciler "knative.dev/pkg/reconciler"

Expand Down Expand Up @@ -95,11 +96,13 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
}

r := &Reconciler{
clientSet: clientinject.Get(ctx),
js: js,
dispatcher: dispatcher,
streamNameFunc: utils.StreamName,
consumerNameFunc: dispatcher.consumerNameFunc,
skipOrphanedSubscriptions: false,
msgingClient: messagingv1client.Get(ctx),
clientSet: clientinject.Get(ctx),
js: js,
dispatcher: dispatcher,
streamNameFunc: utils.StreamName,
consumerNameFunc: dispatcher.consumerNameFunc,
}

impl := natsjetstreamchannel.NewImpl(ctx, r)
Expand Down
13 changes: 8 additions & 5 deletions pkg/channel/jetstream/dispatcher/dispatcher_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import (
"knative.dev/eventing-natss/pkg/client/injection/client"
fakeclientset "knative.dev/eventing-natss/pkg/client/injection/client/fake"
reconcilertesting "knative.dev/eventing-natss/pkg/reconciler/testing"
messagingv1client "knative.dev/eventing/pkg/client/injection/client"
fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
"knative.dev/pkg/controller"
Expand Down Expand Up @@ -138,11 +139,13 @@ func TestDispatcher_ReconcileConsumers(t *testing.T) {
require.NoError(t, err)

reconciler := &Reconciler{
clientSet: client.Get(ctx),
js: js,
dispatcher: d,
streamNameFunc: utils.StreamName,
consumerNameFunc: utils.ConsumerName,
skipOrphanedSubscriptions: true,
msgingClient: messagingv1client.Get(ctx),
clientSet: client.Get(ctx),
js: js,
dispatcher: d,
streamNameFunc: utils.StreamName,
consumerNameFunc: utils.ConsumerName,
}
_ = reconciler.reconcileStream(ctx, nc)

Expand Down
13 changes: 8 additions & 5 deletions pkg/channel/jetstream/dispatcher/natsjetstreamchannel_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import (
"knative.dev/eventing-natss/pkg/client/injection/client"
natsjschannelreconciler "knative.dev/eventing-natss/pkg/client/injection/reconciler/messaging/v1alpha1/natsjetstreamchannel"
reconciletesting "knative.dev/eventing-natss/pkg/reconciler/testing"
messagingv1client "knative.dev/eventing/pkg/client/injection/client"
)

const (
Expand Down Expand Up @@ -175,11 +176,13 @@ func createReconciler(ctx context.Context, listers *reconciletesting.Listers, js
listers.GetNatsJetstreamChannelLister(),
controller.GetEventRecorder(ctx),
&Reconciler{
clientSet: client.Get(ctx),
js: js,
dispatcher: dispatcherFactory(),
streamNameFunc: utils.StreamName,
consumerNameFunc: utils.ConsumerName,
skipOrphanedSubscriptions: true,
msgingClient: messagingv1client.Get(ctx),
clientSet: client.Get(ctx),
js: js,
dispatcher: dispatcherFactory(),
streamNameFunc: utils.StreamName,
consumerNameFunc: utils.ConsumerName,
},
controller.Options{
FinalizerName: "jetstream-ch-dispatcher",
Expand Down
95 changes: 92 additions & 3 deletions pkg/channel/jetstream/dispatcher/reconciler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
v1 "knative.dev/eventing/pkg/apis/duck/v1"
"knative.dev/eventing/pkg/channel"
"knative.dev/eventing/pkg/channel/fanout"
Expand All @@ -38,6 +39,7 @@ import (

"knative.dev/eventing-natss/pkg/client/clientset/versioned"
commonerr "knative.dev/eventing-natss/pkg/common/error"
msgingversioned "knative.dev/eventing/pkg/client/clientset/versioned"

"knative.dev/eventing-natss/pkg/apis/messaging/v1alpha1"
jsmreconciler "knative.dev/eventing-natss/pkg/client/injection/reconciler/messaging/v1alpha1/natsjetstreamchannel"
Expand All @@ -57,9 +59,11 @@ const (
// - Creates a HTTP listener which publishes received events to the Stream
// - Creates a consumer for each .spec.subscribers[] and forwards events to the subscriber address
type Reconciler struct {
clientSet versioned.Interface
js nats.JetStreamManager
dispatcher *Dispatcher
skipOrphanedSubscriptions bool
msgingClient msgingversioned.Interface
clientSet versioned.Interface
js nats.JetStreamManager
dispatcher *Dispatcher

streamNameFunc StreamNameFunc
consumerNameFunc ConsumerNameFunc
Expand All @@ -78,6 +82,11 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, nc *v1alpha1.NatsJetStre
return err
}

if err := r.reconcileOrphanedSubscriptions(ctx, nc); err != nil {
logger.Errorw("failed to reconcile orphaned subscriptions", zap.Error(err))
return err
}

if err := r.syncChannel(ctx, nc, true); err != nil {
logger.Errorw("failed to syncChannel", zap.Error(err))
return err
Expand All @@ -86,6 +95,86 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, nc *v1alpha1.NatsJetStre
return r.reconcileSubscriberStatuses(ctx, nc)
}

func (r *Reconciler) reconcileOrphanedSubscriptions(ctx context.Context, nc *v1alpha1.NatsJetStreamChannel) error {
logger := logging.FromContext(ctx)
if r.skipOrphanedSubscriptions {
return nil
}
var allSubsInNs, err = r.msgingClient.MessagingV1().Subscriptions(nc.Namespace).List(ctx, metav1.ListOptions{})
if err != nil {
logger.Errorw("failed to get subs", zap.Error(err))
return err
}

allSubsInNsUids := sets.New[string]()
for _, s := range allSubsInNs.Items {
allSubsInNsUids.Insert(string(s.UID))
}

subsUids := sets.New[string]()
for _, s := range nc.Spec.Subscribers {
subsUids.Insert(string(s.UID))
}

orphanedSubs := subsUids.Difference(allSubsInNsUids)

@Cali0707 Cali0707 Sep 3, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should have some sort of check for the type of Channel the subscription is referencing, as there are scenarios where users have multiple types of Channels (e.g. In Memory + NATS)

My concern is that we would be deleting too many subscriptions in this scenario

@astelmashenko astelmashenko Sep 4, 2025 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@Cali0707 , it is difference, it works the next way:

Difference returns a set of objects that are not in s2. For example: s1 = {a1, a2, a3} s2 = {a1, a2, a4, a5} s1.Difference(s2) = {a3} s2.Difference(s1) = {a4, a5}

s1 in our case is Channel.Spec.Subscriptions, s2 is all sub in a namespace, so there is no issue if we take all subs in a namespace, making difference of UIDS we just remove all subs from s1 which are actually exist and leftovers are those do not exist. And we do not delete any CRDs, we only delete orphaned subscriptions from a Channel.Spec.Subscriptions and Channel.Status.Subscriptions.

if len(orphanedSubs) == 0 {
return nil
}
logger.Warnw("orpaned subscriptions found", zap.Any("orphaned_subs", orphanedSubs))

after := nc.DeepCopy()
after.Status.Subscribers = make([]v1.SubscriberStatus, len(nc.Status.Subscribers)-len(orphanedSubs))
after.Spec.Subscribers = make([]v1.SubscriberSpec, len(nc.Spec.Subscribers)-len(orphanedSubs))
i := 0
for _, s := range nc.Spec.Subscribers {
if orphanedSubs.Has(string(s.UID)) {
continue
}

after.Spec.Subscribers[i] = *s.DeepCopy()
i = i + 1
}

i = 0
for _, s := range nc.Status.Subscribers {
if orphanedSubs.Has(string(s.UID)) {
continue
}

after.Status.Subscribers[i] = *s.DeepCopy()
i = i + 1
}

logger.Debugw("reconciling orphaned subscribers")

jsonPatch, err := duck.CreatePatch(nc, after)
if err != nil {
return fmt.Errorf("failed to create JSON patch: %w", err)
}

// If there is nothing to patch, we are good, just return.
// Empty patch is [], hence we check for that.
if len(jsonPatch) == 0 {
return nil
}

patch, err := jsonPatch.MarshalJSON()
if err != nil {
return fmt.Errorf("failed to marshal patch to JSON: %w", err)
}

patched, err := r.clientSet.MessagingV1alpha1().
NatsJetStreamChannels(nc.Namespace).
Patch(ctx, nc.Name, types.JSONPatchType, patch, metav1.PatchOptions{})
if err != nil {
return fmt.Errorf("failed to patch subscriber: %w", err)
}

logger.Debugw("patched resource", zap.Any("patch", patch), zap.Any("patched", patched))

return nil
}

func (r *Reconciler) reconcileSubscriberStatuses(ctx context.Context, nc *v1alpha1.NatsJetStreamChannel) error {
logger := logging.FromContext(ctx)
after := nc.DeepCopy()
Expand Down
Loading