Skip to content

Commit 29e4f20

Browse files
knative-prow-robotastelmashenko
knative-prow-robot
and
astelmashenko
authored
[release-1.18] remove orphaned subscribers from nats jetstream channels (#685)
* added feature to remove orphaned subscribers from channels * fixed dispatcher test * added messaging client assign in tests * switched off check orphaned subs in tests * renamed orphaned subs check flag * changed flags values * removed unnecessary logs * included orphaned subs into checks --------- Co-authored-by: Andrii Stelmashenko <andrey.stelmashenko@gmail.com>
1 parent 1ffa71b commit 29e4f20

5 files changed

Lines changed: 124 additions & 18 deletions

File tree

‎config/jetstream/200-jsm-dispatcher-clusterrole.yaml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,20 @@ rules:
2424
resources:
2525
- natsjetstreamchannels
2626
- natsjetstreamchannels/status
27+
- natsjetstreamchannels/spec
2728
verbs:
2829
- get
2930
- list
3031
- watch
3132
- update
3233
- patch
34+
- apiGroups:
35+
- messaging.knative.dev
36+
resources:
37+
- subscriptions
38+
verbs:
39+
- get
40+
- list
3341
- apiGroups:
3442
- eventing.knative.dev
3543
resources:

‎pkg/channel/jetstream/dispatcher/controller.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
jsminformer "knative.dev/eventing-natss/pkg/client/injection/informers/messaging/v1alpha1/natsjetstreamchannel"
2828
"knative.dev/eventing-natss/pkg/common/configloader/fsloader"
2929
"knative.dev/eventing/pkg/apis/eventing"
30+
messagingv1client "knative.dev/eventing/pkg/client/injection/client"
3031
"knative.dev/pkg/injection"
3132
pkgreconciler "knative.dev/pkg/reconciler"
3233

@@ -101,11 +102,13 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
101102
}
102103

103104
r := &Reconciler{
104-
clientSet: clientinject.Get(ctx),
105-
js: js,
106-
dispatcher: dispatcher,
107-
streamNameFunc: utils.StreamName,
108-
consumerNameFunc: dispatcher.consumerNameFunc,
105+
skipOrphanedSubscriptions: false,
106+
msgingClient: messagingv1client.Get(ctx),
107+
clientSet: clientinject.Get(ctx),
108+
js: js,
109+
dispatcher: dispatcher,
110+
streamNameFunc: utils.StreamName,
111+
consumerNameFunc: dispatcher.consumerNameFunc,
109112
}
110113

111114
impl := natsjetstreamchannel.NewImpl(ctx, r)

‎pkg/channel/jetstream/dispatcher/dispatcher_test.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ import (
3535
"knative.dev/eventing-natss/pkg/client/injection/client"
3636
fakeclientset "knative.dev/eventing-natss/pkg/client/injection/client/fake"
3737
reconcilertesting "knative.dev/eventing-natss/pkg/reconciler/testing"
38+
messagingv1client "knative.dev/eventing/pkg/client/injection/client"
3839
fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
3940
fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
4041
"knative.dev/pkg/controller"
@@ -138,11 +139,13 @@ func TestDispatcher_ReconcileConsumers(t *testing.T) {
138139
require.NoError(t, err)
139140

140141
reconciler := &Reconciler{
141-
clientSet: client.Get(ctx),
142-
js: js,
143-
dispatcher: d,
144-
streamNameFunc: utils.StreamName,
145-
consumerNameFunc: utils.ConsumerName,
142+
skipOrphanedSubscriptions: false,
143+
msgingClient: messagingv1client.Get(ctx),
144+
clientSet: client.Get(ctx),
145+
js: js,
146+
dispatcher: d,
147+
streamNameFunc: utils.StreamName,
148+
consumerNameFunc: utils.ConsumerName,
146149
}
147150
_ = reconciler.reconcileStream(ctx, nc)
148151

‎pkg/channel/jetstream/dispatcher/natsjetstreamchannel_test.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ import (
3939
"knative.dev/eventing-natss/pkg/client/injection/client"
4040
natsjschannelreconciler "knative.dev/eventing-natss/pkg/client/injection/reconciler/messaging/v1alpha1/natsjetstreamchannel"
4141
reconciletesting "knative.dev/eventing-natss/pkg/reconciler/testing"
42+
messagingv1client "knative.dev/eventing/pkg/client/injection/client"
4243
)
4344

4445
const (
@@ -175,11 +176,13 @@ func createReconciler(ctx context.Context, listers *reconciletesting.Listers, js
175176
listers.GetNatsJetstreamChannelLister(),
176177
controller.GetEventRecorder(ctx),
177178
&Reconciler{
178-
clientSet: client.Get(ctx),
179-
js: js,
180-
dispatcher: dispatcherFactory(),
181-
streamNameFunc: utils.StreamName,
182-
consumerNameFunc: utils.ConsumerName,
179+
skipOrphanedSubscriptions: true,
180+
msgingClient: messagingv1client.Get(ctx),
181+
clientSet: client.Get(ctx),
182+
js: js,
183+
dispatcher: dispatcherFactory(),
184+
streamNameFunc: utils.StreamName,
185+
consumerNameFunc: utils.ConsumerName,
183186
},
184187
controller.Options{
185188
FinalizerName: "jetstream-ch-dispatcher",

‎pkg/channel/jetstream/dispatcher/reconciler.go‎

Lines changed: 92 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
corev1 "k8s.io/api/core/v1"
2828
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
2929
"k8s.io/apimachinery/pkg/types"
30+
"k8s.io/apimachinery/pkg/util/sets"
3031
v1 "knative.dev/eventing/pkg/apis/duck/v1"
3132
"knative.dev/eventing/pkg/channel"
3233
"knative.dev/eventing/pkg/channel/fanout"
@@ -38,6 +39,7 @@ import (
3839

3940
"knative.dev/eventing-natss/pkg/client/clientset/versioned"
4041
commonerr "knative.dev/eventing-natss/pkg/common/error"
42+
msgingversioned "knative.dev/eventing/pkg/client/clientset/versioned"
4143

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

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

85+
if err := r.reconcileOrphanedSubscriptions(ctx, nc); err != nil {
86+
logger.Errorw("failed to reconcile orphaned subscriptions", zap.Error(err))
87+
return err
88+
}
89+
8190
if err := r.syncChannel(ctx, nc, true); err != nil {
8291
logger.Errorw("failed to syncChannel", zap.Error(err))
8392
return err
@@ -86,6 +95,86 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, nc *v1alpha1.NatsJetStre
8695
return r.reconcileSubscriberStatuses(ctx, nc)
8796
}
8897

98+
func (r *Reconciler) reconcileOrphanedSubscriptions(ctx context.Context, nc *v1alpha1.NatsJetStreamChannel) error {
99+
logger := logging.FromContext(ctx)
100+
if r.skipOrphanedSubscriptions {
101+
return nil
102+
}
103+
var allSubsInNs, err = r.msgingClient.MessagingV1().Subscriptions(nc.Namespace).List(ctx, metav1.ListOptions{})
104+
if err != nil {
105+
logger.Errorw("failed to get subs", zap.Error(err))
106+
return err
107+
}
108+
109+
allSubsInNsUids := sets.New[string]()
110+
for _, s := range allSubsInNs.Items {
111+
allSubsInNsUids.Insert(string(s.UID))
112+
}
113+
114+
subsUids := sets.New[string]()
115+
for _, s := range nc.Spec.Subscribers {
116+
subsUids.Insert(string(s.UID))
117+
}
118+
119+
orphanedSubs := subsUids.Difference(allSubsInNsUids)
120+
if len(orphanedSubs) == 0 {
121+
return nil
122+
}
123+
logger.Warnw("orpaned subscriptions found", zap.Any("orphaned_subs", orphanedSubs))
124+
125+
after := nc.DeepCopy()
126+
after.Status.Subscribers = make([]v1.SubscriberStatus, len(nc.Status.Subscribers)-len(orphanedSubs))
127+
after.Spec.Subscribers = make([]v1.SubscriberSpec, len(nc.Spec.Subscribers)-len(orphanedSubs))
128+
i := 0
129+
for _, s := range nc.Spec.Subscribers {
130+
if orphanedSubs.Has(string(s.UID)) {
131+
continue
132+
}
133+
134+
after.Spec.Subscribers[i] = *s.DeepCopy()
135+
i = i + 1
136+
}
137+
138+
i = 0
139+
for _, s := range nc.Status.Subscribers {
140+
if orphanedSubs.Has(string(s.UID)) {
141+
continue
142+
}
143+
144+
after.Status.Subscribers[i] = *s.DeepCopy()
145+
i = i + 1
146+
}
147+
148+
logger.Debugw("reconciling orphaned subscribers")
149+
150+
jsonPatch, err := duck.CreatePatch(nc, after)
151+
if err != nil {
152+
return fmt.Errorf("failed to create JSON patch: %w", err)
153+
}
154+
155+
// If there is nothing to patch, we are good, just return.
156+
// Empty patch is [], hence we check for that.
157+
if len(jsonPatch) == 0 {
158+
return nil
159+
}
160+
161+
patch, err := jsonPatch.MarshalJSON()
162+
if err != nil {
163+
return fmt.Errorf("failed to marshal patch to JSON: %w", err)
164+
}
165+
166+
patched, err := r.clientSet.MessagingV1alpha1().
167+
NatsJetStreamChannels(nc.Namespace).
168+
Patch(ctx, nc.Name, types.JSONPatchType, patch, metav1.PatchOptions{})
169+
if err != nil {
170+
return fmt.Errorf("failed to patch subscriber: %w", err)
171+
}
172+
173+
logger.Debugw("patched resource", zap.Any("patch", patch), zap.Any("patched", patched))
174+
175+
return nil
176+
}
177+
89178
func (r *Reconciler) reconcileSubscriberStatuses(ctx context.Context, nc *v1alpha1.NatsJetStreamChannel) error {
90179
logger := logging.FromContext(ctx)
91180
after := nc.DeepCopy()

0 commit comments

Comments
 (0)