feat: remove the mutating and validating webhooks for v1alpha2 Subscr…

…iption.

.golangci.yaml

@@ -72,6 +72,8 @@ linters-settings:
         alias: kappsv1
       - pkg: k8s.io/api/rbac/v1
         alias: krbacv1
+      - pkg: k8s.io/api/batch/v1
+        alias: kbatchv1
       - pkg: k8s.io/apimachinery/pkg/runtime/schema
         alias: kschema
       - pkg: k8s.io/apimachinery/pkg/labels

Makefile

@@ -85,8 +85,8 @@ help: ## Display this help.
 ##@ Development
 
 .PHONY: manifests
-manifests: controller-gen ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
-	$(CONTROLLER_GEN) rbac:roleName=eventing-manager crd webhook paths="./..." output:crd:artifacts:config=config/crd/bases
+manifests: controller-gen ## Generate ClusterRole and CustomResourceDefinition objects.
+	$(CONTROLLER_GEN) rbac:roleName=eventing-manager crd paths="./..." output:crd:artifacts:config=config/crd/bases
 	$(MAKE) crd-docs-gen
 
 .PHONY: generate

api/eventing/v1alpha2/condition.go

@@ -10,10 +10,11 @@ import (
 type ConditionType string
 
 const (
-	ConditionSubscribed         ConditionType = "Subscribed"
-	ConditionSubscriptionActive ConditionType = "Subscription active"
-	ConditionAPIRuleStatus      ConditionType = "APIRule status"
-	ConditionWebhookCallStatus  ConditionType = "Webhook call status"
+	ConditionSubscribed            ConditionType = "Subscribed"
+	ConditionSubscriptionActive    ConditionType = "Subscription active"
+	ConditionSubscriptionSpecValid ConditionType = "Subscription spec valid"
+	ConditionAPIRuleStatus         ConditionType = "APIRule status"
+	ConditionWebhookCallStatus     ConditionType = "Webhook call status"
 
 	ConditionPublisherProxyReady ConditionType = "Publisher Proxy Ready"
 	ConditionControllerReady     ConditionType = "Subscription Controller Ready"
@@ -37,6 +38,9 @@ type Condition struct {
 type ConditionReason string
 
 const (
+	ConditionReasonSubscriptionSpecHasValidationErrors   ConditionReason = "Subscription spec has validation errors"
+	ConditionReasonSubscriptionSpecHasNoValidationErrors ConditionReason = "Subscription spec has no validation errors"
+
 	// JetStream Conditions.
 	ConditionReasonNATSSubscriptionActive    ConditionReason = "NATS Subscription active"
 	ConditionReasonNATSSubscriptionNotActive ConditionReason = "NATS Subscription not active"
@@ -133,6 +137,11 @@ func MakeSubscriptionConditions() []Condition {
 			LastTransitionTime: kmetav1.Now(),
 			Status:             kcorev1.ConditionUnknown,
 		},
+		{
+			Type:               ConditionSubscriptionSpecValid,
+			LastTransitionTime: kmetav1.Now(),
+			Status:             kcorev1.ConditionUnknown,
+		},
 	}
 	return conditions
 }
@@ -241,7 +250,7 @@ func ConditionsEquals(existing, expected []Condition) bool {
 	return true
 }
 
-// ConditionsEquals checks if two conditions are equal.
+// ConditionEquals checks if two conditions are equal.
 func ConditionEquals(existing, expected Condition) bool {
 	isTypeEqual := existing.Type == expected.Type
 	isStatusEqual := existing.Status == expected.Status
@@ -259,29 +268,112 @@ func CreateMessageForConditionReasonSubscriptionCreated(eventMeshName string) st
 	return fmt.Sprintf("EventMesh subscription name is: %s", eventMeshName)
 }
 
-// GetSubscriptionActiveCondition updates the ConditionSubscriptionActive condition based on the given error value.
-func GetSubscriptionActiveCondition(sub *Subscription, err error) []Condition {
-	subscriptionActiveCondition := Condition{
+// makeSubscriptionActiveCondition returns a new active condition based on the given error.
+func makeSubscriptionActiveCondition(err error) Condition {
+	var (
+		status  kcorev1.ConditionStatus
+		reason  ConditionReason
+		message string
+	)
+	if err != nil {
+		status = kcorev1.ConditionFalse
+		reason = ConditionReasonNATSSubscriptionNotActive
+		message = err.Error()
+	} else {
+		status = kcorev1.ConditionTrue
+		reason = ConditionReasonNATSSubscriptionActive
+	}
+	return Condition{
 		Type:               ConditionSubscriptionActive,
+		Status:             status,
 		LastTransitionTime: kmetav1.Now(),
+		Reason:             reason,
+		Message:            message,
 	}
-	if err == nil {
-		subscriptionActiveCondition.Status = kcorev1.ConditionTrue
-		subscriptionActiveCondition.Reason = ConditionReasonNATSSubscriptionActive
+}
+
+// makeSubscriptionSpecValidCondition returns a new validation condition based on the given error.
+func makeSubscriptionSpecValidCondition(err error) Condition {
+	var (
+		status  kcorev1.ConditionStatus
+		reason  ConditionReason
+		message string
+	)
+	if err != nil {
+		status = kcorev1.ConditionFalse
+		reason = ConditionReasonSubscriptionSpecHasValidationErrors
+		message = err.Error()
 	} else {
-		subscriptionActiveCondition.Message = err.Error()
-		subscriptionActiveCondition.Reason = ConditionReasonNATSSubscriptionNotActive
-		subscriptionActiveCondition.Status = kcorev1.ConditionFalse
+		status = kcorev1.ConditionTrue
+		reason = ConditionReasonSubscriptionSpecHasNoValidationErrors
 	}
-	for _, activeCond := range sub.Status.Conditions {
-		if activeCond.Type == ConditionSubscriptionActive {
-			if subscriptionActiveCondition.Status == activeCond.Status &&
-				subscriptionActiveCondition.Reason == activeCond.Reason &&
-				subscriptionActiveCondition.Message == activeCond.Message {
-				return []Condition{activeCond}
-			}
+	return Condition{
+		Type:               ConditionSubscriptionSpecValid,
+		Status:             status,
+		LastTransitionTime: kmetav1.Now(),
+		Reason:             reason,
+		Message:            message,
+	}
+}
+
+// setCondition sets the given condition in the Subscription status.
+// If the condition is already present, it will be updated.
+// If the condition is not present, it will be added.
+func (s *SubscriptionStatus) setCondition(condition Condition) {
+	isFound, isSet := false, false
+	conditions := make([]Condition, 0, len(s.Conditions))
+	for _, c := range s.Conditions {
+		if c.Type != condition.Type {
+			conditions = append(conditions, c)
+			continue
+		}
+		isFound = true
+		if !ConditionEquals(c, condition) {
+			isSet = true
+			conditions = append(conditions, condition)
 		}
 	}
+	if !isFound {
+		isSet = true
+		conditions = append(conditions, condition)
+	}
+	if isSet {
+		s.Conditions = conditions
+	}
+}
+
+// SetSubscriptionActiveCondition sets a subscription active condition based on the given error.
+// If the given error is nil, the status will have the Subscription active condition set to true,
+// otherwise it will have the Subscription active condition set to false and the error as the message.
+func SetSubscriptionActiveCondition(status *SubscriptionStatus, err error) {
+	condition := makeSubscriptionActiveCondition(err)
+	status.setCondition(condition)
+}
+
+// SetSubscriptionSpecValidCondition sets a subscription spec valid condition based on the given error.
+// If the given error is nil, the status will have the Subscription spec valid condition set to true,
+// otherwise it will have the Subscription spec valid condition set to false and the error as the message.
+func (s *SubscriptionStatus) SetSubscriptionSpecValidCondition(err error) {
+	condition := makeSubscriptionSpecValidCondition(err)
+	s.setCondition(condition)
+}
+
+// SetNotReady sets the Subscription status to not ready.
+func (s *SubscriptionStatus) SetNotReady() {
+	s.Ready = false
+}
+
+// ClearConditions sets the Subscription conditions to an empty list.
+func (s *SubscriptionStatus) ClearConditions() {
+	s.Conditions = []Condition{}
+}
+
+// ClearBackend sets the Subscription Backend to an empty struct.
+func (s *SubscriptionStatus) ClearBackend() {
+	s.Backend = Backend{}
+}
 
-	return []Condition{subscriptionActiveCondition}
+// ClearTypes sets the Subscription Types to an empty list.
+func (s *SubscriptionStatus) ClearTypes() {
+	s.Types = []EventType{}
 }

api/eventing/v1alpha2/condition_unit_test.go

@@ -48,6 +48,7 @@ func Test_InitializeSubscriptionConditions(t *testing.T) {
 				v1alpha2.ConditionSubscriptionActive,
 				v1alpha2.ConditionAPIRuleStatus,
 				v1alpha2.ConditionWebhookCallStatus,
+				v1alpha2.ConditionSubscriptionSpecValid,
 			}
 
 			// when
@@ -130,6 +131,7 @@ func Test_IsReady(t *testing.T) {
 				{Type: v1alpha2.ConditionSubscriptionActive, Status: kcorev1.ConditionTrue},
 				{Type: v1alpha2.ConditionAPIRuleStatus, Status: kcorev1.ConditionTrue},
 				{Type: v1alpha2.ConditionWebhookCallStatus, Status: kcorev1.ConditionTrue},
+				{Type: v1alpha2.ConditionSubscriptionSpecValid, Status: kcorev1.ConditionTrue},
 			},
 			wantReadyStatus: true,
 		},
@@ -208,6 +210,7 @@ func Test_ShouldUpdateReadyStatus(t *testing.T) {
 				{Type: v1alpha2.ConditionSubscriptionActive, Status: kcorev1.ConditionTrue},
 				{Type: v1alpha2.ConditionAPIRuleStatus, Status: kcorev1.ConditionTrue},
 				{Type: v1alpha2.ConditionWebhookCallStatus, Status: kcorev1.ConditionTrue},
+				{Type: v1alpha2.ConditionSubscriptionSpecValid, Status: kcorev1.ConditionTrue},
 			},
 			wantStatus: false,
 		},
@@ -230,6 +233,7 @@ func Test_ShouldUpdateReadyStatus(t *testing.T) {
 				{Type: v1alpha2.ConditionSubscriptionActive, Status: kcorev1.ConditionTrue},
 				{Type: v1alpha2.ConditionAPIRuleStatus, Status: kcorev1.ConditionTrue},
 				{Type: v1alpha2.ConditionWebhookCallStatus, Status: kcorev1.ConditionTrue},
+				{Type: v1alpha2.ConditionSubscriptionSpecValid, Status: kcorev1.ConditionTrue},
 			},
 			wantStatus: true,
 		},
@@ -540,12 +544,12 @@ func Test_SetConditionSubscriptionActive(t *testing.T) {
 			sub.Status.Conditions = testcase.givenConditions
 
 			// when
-			conditions := v1alpha2.GetSubscriptionActiveCondition(sub, testcase.givenError)
+			v1alpha2.SetSubscriptionActiveCondition(&sub.Status, testcase.givenError)
 
 			// then
-			require.True(t, v1alpha2.ConditionsEquals(conditions, testcase.wantConditions))
+			require.True(t, v1alpha2.ConditionsEquals(sub.Status.Conditions, testcase.wantConditions))
 			if testcase.wantLastTransitionTime != nil {
-				require.Equal(t, *testcase.wantLastTransitionTime, conditions[0].LastTransitionTime)
+				require.Equal(t, *testcase.wantLastTransitionTime, sub.Status.Conditions[0].LastTransitionTime)
 			}
 		})
 	}

api/eventing/v1alpha2/subscription_types.go

@@ -30,6 +30,8 @@ type SubscriptionSpec struct {
 	// Defines how types should be handled.<br />
 	// - `standard`: backend-specific logic will be applied to the configured source and types.<br />
 	// - `exact`: no further processing will be applied to the configured source and types.
+	// +kubebuilder:default:="standard"
+	// +kubebuilder:validation:XValidation:rule="self=='standard' || self=='exact'", message="typeMatching can only be set to standard or exact"
 	TypeMatching TypeMatching `json:"typeMatching,omitempty"`
 
 	// Defines the origin of the event.
@@ -40,6 +42,7 @@ type SubscriptionSpec struct {
 
 	// Map of configuration options that will be applied on the backend.
 	// +optional
+	// +kubebuilder:default={"maxInFlightMessages":"10"}
 	Config map[string]string `json:"config,omitempty"`
 }
 
@@ -114,6 +117,21 @@ func (s *Subscription) GetUniqueTypes() []string {
 	return result
 }
 
+// GetDuplicateTypes returns the duplicate types from subscription spec.
+func (s *Subscription) GetDuplicateTypes() []string {
+	if len(s.Spec.Types) == 0 {
+		return s.Spec.Types
+	}
+	types := make(map[string]int, len(s.Spec.Types))
+	duplicates := make([]string, 0, len(s.Spec.Types))
+	for _, t := range s.Spec.Types {
+		if types[t]++; types[t] == 2 {
+			duplicates = append(duplicates, t)
+		}
+	}
+	return duplicates
+}
+
 func (s *Subscription) DuplicateWithStatusDefaults() *Subscription {
 	desiredSub := s.DeepCopy()
 	desiredSub.Status = SubscriptionStatus{}
@@ -140,6 +158,3 @@ type SubscriptionList struct {
 func init() { //nolint:gochecknoinits
 	SchemeBuilder.Register(&Subscription{}, &SubscriptionList{})
 }
-
-// Hub marks this type as a conversion hub.
-func (*Subscription) Hub() {}

api/eventing/v1alpha2/subscription_types_test.go

@@ -69,3 +69,73 @@ func TestGetMaxInFlightMessages(t *testing.T) {
 		})
 	}
 }
+
+func TestGetDuplicateTypes(t *testing.T) {
+	tests := []struct {
+		name               string
+		givenTypes         []string
+		wantDuplicateTypes []string
+	}{
+		{
+			name:               "with nil types",
+			givenTypes:         nil,
+			wantDuplicateTypes: nil,
+		},
+		{
+			name:               "with empty types",
+			givenTypes:         []string{},
+			wantDuplicateTypes: []string{},
+		},
+		{
+			name: "with one type",
+			givenTypes: []string{
+				"type0",
+			},
+			wantDuplicateTypes: []string{},
+		},
+		{
+			name: "with multiple types and no duplicates",
+			givenTypes: []string{
+				"type0",
+				"type1",
+				"type2",
+			},
+			wantDuplicateTypes: []string{},
+		},
+		{
+			name: "with multiple types and consequent duplicates",
+			givenTypes: []string{
+				"type0",
+				"type1", "type1", "type1", // duplicates
+				"type2", "type2", // duplicates
+				"type3",
+				"type4", "type4", "type4", "type4", // duplicates
+				"type5",
+			},
+			wantDuplicateTypes: []string{
+				"type1", "type2", "type4",
+			},
+		},
+		{
+			name: "with multiple types and non-consequent duplicates",
+			givenTypes: []string{
+				"type5", "type0", "type1", "type2",
+				"type1", // duplicate
+				"type3", "type4",
+				"type5", // duplicate
+				"type6",
+				"type4", "type2", // duplicates
+			},
+			wantDuplicateTypes: []string{
+				"type1", "type5", "type4", "type2",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := v1alpha2.Subscription{Spec: v1alpha2.SubscriptionSpec{Types: tt.givenTypes}}
+			gotDuplicateTypes := s.GetDuplicateTypes()
+			assert.Equal(t, tt.wantDuplicateTypes, gotDuplicateTypes)
+		})
+	}
+}