React to EventMesh backend config changes

api/v1alpha1/eventing_types.go

@@ -88,7 +88,7 @@ type Eventing struct {
 // EventingStatus defines the observed state of Eventing
 type EventingStatus struct {
 	ActiveBackend     BackendType        `json:"activeBackend"`
-	BackendConfigHash uint64             `json:"specHash"`
+	BackendConfigHash int64              `json:"specHash"`
 	State             string             `json:"state"`
 	Conditions        []metav1.Condition `json:"conditions,omitempty"`
 }

config/crd/external/apirules.gateway.crd.yaml

@@ -14,205 +14,8 @@ spec:
     plural: apirules
     singular: apirule
   scope: Namespaced
-  conversion:
-    strategy: Webhook
-    webhook:
-      clientConfig:
-        service:
-          namespace: kyma-system
-          name: api-gateway-webhook-service
-          path: /convert
-      conversionReviewVersions:
-        - v1beta1
-        - v1alpha1
   preserveUnknownFields: false
   versions:
-  - deprecated: true
-    deprecationWarning: Since Kyma 2.5.X, APIRule in version v1alpha1 has been deprecated.
-      Consider using v1beta1.
-    name: v1alpha1
-    schema:
-      openAPIV3Schema:
-        description: APIRule is the Schema for ApiRule APIs.
-        properties:
-          apiVersion:
-            description: 'APIVersion defines the versioned schema of this representation
-              of an object. Servers should convert recognized schemas to the latest
-              internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources'
-            type: string
-          kind:
-            description: 'Kind is a string value representing the REST resource this
-              object represents. Servers may infer this from the endpoint the client
-              submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
-            type: string
-          metadata:
-            type: object
-          spec:
-            description: Defines the desired state of ApiRule.
-            properties:
-              gateway:
-                description: Specifies the Istio Gateway to be used.
-                pattern: ^[0-9a-z-_]+(\/[0-9a-z-_]+|(\.[0-9a-z-_]+)*)$
-                type: string
-              rules:
-                description: Represents the array of Oathkeeper access rules to be
-                  applied.
-                items:
-                  properties:
-                    accessStrategies:
-                      description: Specifies the list of access strategies. All strategies
-                        listed in [Oathkeeper documentation](https://www.ory.sh/docs/oathkeeper/pipeline/authn)
-                        are supported.
-                      items:
-                        description: Represents a handler that authenticates provided
-                          credentials. See the corresponding type in the oathkeeper-maester
-                          project.
-                        properties:
-                          config:
-                            description: Configures the handler. Configuration keys
-                              vary per handler.
-                            type: object
-                            x-kubernetes-preserve-unknown-fields: true
-                            properties:
-                              jwks_urls:
-                                description: Specifies the array of URLs from which Ory Oathkeeper can retrieve JSON Web Keys for validating JSON Web Token.
-                                  The value must begin with either `http://`, `https://`, or `file://`.
-                                type: array
-                                items:
-                                  type: string
-                                  pattern: ^(http://|https://|file://).*$
-                              trusted_issuers:
-                                description: If the **trusted_issuers** field is set, the JWT must contain a value for the claim `iss`
-                                  that matches exactly (case-sensitive) one of the values of **trusted_issuers**.
-                                  The value must begin with either `http://`, `https://`, or `file://`.
-                                type: array
-                                items:
-                                  type: string
-                                  pattern: ^(http://|https://|file://).*$
-                          handler:
-                            description: Specifies the name of the handler.
-                            type: string
-                        required:
-                        - handler
-                        type: object
-                      minItems: 1
-                      type: array
-                    methods:
-                      description: Represents the list of allowed HTTP request methods
-                        available for the **spec.rules.path**.
-                      items:
-                        type: string
-                      minItems: 1
-                      type: array
-                    mutators:
-                      description: Specifies the list of [Ory Oathkeeper mutators](https://www.ory.sh/docs/oathkeeper/pipeline/mutator).
-                      items:
-                        description: Mutator represents a handler that transforms
-                          the HTTP request before forwarding it. See the corresponding
-                          in the oathkeeper-maester project.
-                        properties:
-                          config:
-                            description: Configures the handler. Configuration keys
-                              vary per handler.
-                            type: object
-                            x-kubernetes-preserve-unknown-fields: true
-                          handler:
-                            description: Specifies the name of the handler.
-                            type: string
-                        required:
-                        - handler
-                        type: object
-                      type: array
-                    path:
-                      description: Specifies the path of the exposed service.
-                      pattern: ^([0-9a-zA-Z./*()?!\\_-]+)
-                      type: string
-                  required:
-                  - accessStrategies
-                  - methods
-                  - path
-                  type: object
-                minItems: 1
-                type: array
-              service:
-                description: Describes the service to expose.
-                properties:
-                  external:
-                    description: Specifies if the service is internal (in cluster) or
-                      external.
-                    type: boolean
-                  host:
-                    description: Specifies the URL of the exposed service.
-                    maxLength: 256
-                    minLength: 3
-                    pattern: ^([a-zA-Z0-9][a-zA-Z0-9-_]*\.)*[a-zA-Z0-9]*[a-zA-Z0-9-_]*[[a-zA-Z0-9]+$
-                    type: string
-                  name:
-                    description: Specifies the name of the exposed service.
-                    type: string
-                  port:
-                    description: Specifies the communication port of the exposed service.
-                    format: int32
-                    maximum: 65535
-                    minimum: 1
-                    type: integer
-                required:
-                - host
-                - name
-                - port
-                type: object
-            required:
-            - gateway
-            - rules
-            - service
-            type: object
-          status:
-            description: Describes the observed state of ApiRule.
-            properties:
-              APIRuleStatus:
-                description: Describes the status of APIRule.
-                properties:
-                  code:
-                    description: Status code describing APIRule.
-                    type: string
-                  desc:
-                    description: Explains the status of APIRule.
-                    type: string
-                type: object
-              accessRuleStatus:
-                description: Describes the status of ORY Oathkeeper Rule.
-                properties:
-                  code:
-                    description: Status code describing ORY Oathkeeper Rule.
-                    type: string
-                  desc:
-                    description: Explains the status of ORY Oathkeeper Rule.
-                    type: string
-                type: object
-              lastProcessedTime:
-                description: Indicates the timestamp when the API Gateway controller last processed APIRule.
-                format: date-time
-                type: string
-              observedGeneration:
-                description: Specifies the generation of the resource that was observed by the API Gateway controller.
-                format: int64
-                type: integer
-              virtualServiceStatus:
-                description: Describes the status of Istio VirtualService.
-                properties:
-                  code:
-                    description: Status code describing Istio VirtualService.
-                    type: string
-                  desc:
-                    description: Explains the status of Istio VirtualService.
-                    type: string
-                type: object
-            type: object
-        type: object
-    served: true
-    storage: false
-    subresources:
-      status: {}
   - additionalPrinterColumns:
     - jsonPath: .status.APIRuleStatus.code
       name: Status

go.mod

@@ -17,10 +17,14 @@ require (
 	github.com/mitchellh/hashstructure/v2 v2.0.2
 	github.com/nats-io/nats-server/v2 v2.9.21
 	github.com/nats-io/nats.go v1.28.0
+	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/gomega v1.27.10
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.16.0
 	github.com/stretchr/testify v1.8.4
+	go.uber.org/atomic v1.11.0
 	go.uber.org/zap v1.25.0
+	golang.org/x/oauth2 v0.10.0
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2
 	k8s.io/api v0.28.1
 	k8s.io/apiextensions-apiserver v0.28.0
@@ -62,19 +66,15 @@ require (
 	github.com/nats-io/nkeys v0.4.4 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
-	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.16.0 // indirect
 	github.com/prometheus/client_model v0.4.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.10.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
-	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/crypto v0.11.0 // indirect
 	golang.org/x/net v0.13.0 // indirect
-	golang.org/x/oauth2 v0.10.0 // indirect
 	golang.org/x/sys v0.11.0 // indirect
 	golang.org/x/term v0.10.0 // indirect
 	golang.org/x/text v0.11.0 // indirect

internal/controller/eventing/eventmesh.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/kyma-project/eventing-manager/pkg/env"
 	"os"
 
 	"github.com/kyma-project/eventing-manager/pkg/eventing"
@@ -41,7 +42,6 @@ func (r *Reconciler) reconcileEventMeshSubManager(ctx context.Context, eventing
 	if err != nil {
 		return errors.Errorf("failed to sync OAuth secret: %v", err)
 	}
-
 	// retrieve secret to authenticate with EventMesh
 	eventMeshSecret, err := r.kubeClient.GetSecret(ctx, eventing.Spec.Backend.Config.EventMeshSecret)
 	if err != nil {
@@ -59,6 +59,25 @@ func (r *Reconciler) reconcileEventMeshSubManager(ctx context.Context, eventing
 		return fmt.Errorf("failed to setup environment variables for EventMesh controller: %v", err)
 	}
 
+	// get the subscription config
+	defaultSubsConfig := r.getDefaultSubscriptionConfig()
+	// get the subManager parameters
+	eventMeshSubMgrParams := r.getEventMeshSubManagerParams()
+	// get the hash of current config
+	specHash, err := r.getEventMeshBackendConfigHash(eventing.Spec.Backend.Config.EventMeshSecret,
+		eventing.Spec.Backend.Config.EventTypePrefix)
+	if err != nil {
+		return err
+	}
+
+	// update the config if hashes differ
+	if eventing.Status.BackendConfigHash != specHash {
+		// stop the subsManager without cleanup
+		if err := r.stopEventMeshSubManager(false, r.namedLogger()); err != nil {
+			return err
+		}
+	}
+
 	if r.eventMeshSubManager == nil {
 		// create instance of EventMesh subscription manager
 		eventMeshSubManager, err := r.subManagerFactory.NewEventMeshManager()
@@ -81,19 +100,36 @@ func (r *Reconciler) reconcileEventMeshSubManager(ctx context.Context, eventing
 		return nil
 	}
 
-	defaultSubsConfig := r.eventingManager.GetBackendConfig().DefaultSubscriptionConfig
-	eventMeshSubMgrParams := subscriptionmanager.Params{
+	err = r.startEventMeshSubManager(defaultSubsConfig, eventMeshSubMgrParams)
+	if err != nil {
+		return err
+	}
+
+	// update the hash of the current config only once subManager is started
+	eventing.Status.BackendConfigHash = specHash
+	r.namedLogger().Info(fmt.Sprintf("NATS subscription-manager has been updated, new hash: %d", specHash))
+
+	return nil
+}
+
+func (r *Reconciler) getEventMeshSubManagerParams() subscriptionmanager.Params {
+	return subscriptionmanager.Params{
 		subscriptionmanager.ParamNameClientID:     r.oauth2credentials.clientID,
 		subscriptionmanager.ParamNameClientSecret: r.oauth2credentials.clientSecret,
 		subscriptionmanager.ParamNameTokenURL:     r.oauth2credentials.tokenURL,
 		subscriptionmanager.ParamNameCertsURL:     r.oauth2credentials.certsURL,
 	}
-	if err = r.eventMeshSubManager.Start(defaultSubsConfig, eventMeshSubMgrParams); err != nil {
+}
+
+func (r *Reconciler) startEventMeshSubManager(defaultSubsConfig env.DefaultSubscriptionConfig,
+	eventMeshSubMgrParams subscriptionmanager.Params) error {
+	if err := r.eventMeshSubManager.Start(defaultSubsConfig, eventMeshSubMgrParams); err != nil {
 		return err
 	}
+
 	r.namedLogger().Info("EventMesh subscription-manager started")
+	// update flag so it does not try to start the manager again
 	r.isEventMeshSubManagerStarted = true
-
 	return nil
 }