Skip to content

Commit

Permalink
Fix: Parallels updates subscription where possible, instead of recrea…
Browse files Browse the repository at this point in the history
…ting it
  • Loading branch information
7h3-3mp7y-m4n committed Jun 5, 2024
1 parent 3ee2400 commit 1ee3b82
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 11 deletions.
13 changes: 11 additions & 2 deletions pkg/reconciler/parallel/parallel.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@ import (

"go.uber.org/zap"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/equality"

//"k8s.io/apimachinery/pkg/api/equality"
apierrs "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand All @@ -45,6 +46,7 @@ import (
messaginglisters "knative.dev/eventing/pkg/client/listers/messaging/v1"
ducklib "knative.dev/eventing/pkg/duck"
"knative.dev/eventing/pkg/reconciler/parallel/resources"
"knative.dev/pkg/kmp"
)

type Reconciler struct {
Expand Down Expand Up @@ -216,7 +218,7 @@ func (r *Reconciler) reconcileSubscription(ctx context.Context, branchNumber int
// TODO: Send events here, or elsewhere?
//r.Recorder.Eventf(p, corev1.EventTypeWarning, subscriptionCreateFailed, "Create Parallels's subscription failed: %v", err)
return nil, fmt.Errorf("failed to get Subscription: %s", err)
} else if !equality.Semantic.DeepDerivative(expected.Spec, sub.Spec) {
} else if immutableFieldsChanged := expected.CheckImmutableFields(ctx, sub); immutableFieldsChanged != nil {
// Given that spec.channel is immutable, we cannot just update the subscription. We delete
// it instead, and re-create it.
err = r.eventingClientSet.MessagingV1().Subscriptions(sub.Namespace).Delete(ctx, sub.Name, metav1.DeleteOptions{})
Expand All @@ -230,6 +232,13 @@ func (r *Reconciler) reconcileSubscription(ctx context.Context, branchNumber int
return nil, err
}
return newSub, nil
} else if equal, err := kmp.SafeEqual(sub.Spec, expected.Spec); !equal || err != nil {
updatedSub, err := r.eventingClientSet.MessagingV1().Subscriptions(sub.Namespace).Update(ctx, expected, metav1.UpdateOptions{})
if err != nil {
logging.FromContext(ctx).Infow("Cannot update subscription", zap.Error(err))
return nil, err
}
return updatedSub, nil
}
return sub, nil
}
Expand Down
21 changes: 12 additions & 9 deletions pkg/reconciler/parallel/parallel_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -427,22 +427,25 @@ func TestAllBranches(t *testing.T) {
{Subscriber: createSubscriber(0)},
})))},
WantErr: false,
WantDeletes: []clientgotesting.DeleteActionImpl{{
ActionImpl: clientgotesting.ActionImpl{
Namespace: testNS,
Resource: v1.SchemeGroupVersion.WithResource("subscriptions"),
WantUpdates: []clientgotesting.UpdateActionImpl{
{
ActionImpl: clientgotesting.ActionImpl{
Namespace: testNS,
Resource: v1.SchemeGroupVersion.WithResource("subscriptions"),
},
Object: resources.NewSubscription(0, NewFlowsParallel(parallelName, testNS,
WithFlowsParallelChannelTemplateSpec(imc),
WithFlowsParallelBranches([]v1.ParallelBranch{
{Subscriber: createSubscriber(1)},
}))),
},
Name: resources.ParallelBranchChannelName(parallelName, 0),
}},
},
WantCreates: []runtime.Object{
createChannel(parallelName),
createBranchChannel(parallelName, 0),
resources.NewFilterSubscription(0, NewFlowsParallel(parallelName, testNS, WithFlowsParallelChannelTemplateSpec(imc), WithFlowsParallelBranches([]v1.ParallelBranch{
{Subscriber: createSubscriber(1)},
}))),
resources.NewSubscription(0, NewFlowsParallel(parallelName, testNS, WithFlowsParallelChannelTemplateSpec(imc), WithFlowsParallelBranches([]v1.ParallelBranch{
{Subscriber: createSubscriber(1)},
}))),
},
WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
Object: NewFlowsParallel(parallelName, testNS,
Expand Down

0 comments on commit 1ee3b82

Please sign in to comment.