Add ApplicationCRD Enabled ENV in Publisher deployment

cmd/main.go

@@ -39,6 +39,7 @@ import (
 	"github.com/kyma-project/kyma/components/eventing-controller/logger"
 	"github.com/kyma-project/kyma/components/eventing-controller/options"
 	backendmetrics "github.com/kyma-project/kyma/components/eventing-controller/pkg/backend/metrics"
+	apiclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
@@ -122,7 +123,15 @@ func main() { //nolint:funlen // main function needs to initialize many object
 
 	// init custom kube client wrapper
 	k8sClient := mgr.GetClient()
-	kubeClient := k8s.NewKubeClient(k8sClient, "eventing-manager")
+
+	// init custom kube client wrapper
+	apiClientSet, err := apiclientset.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		setupLog.Error(err, "failed to create new k8s clientset")
+		os.Exit(1)
+	}
+
+	kubeClient := k8s.NewKubeClient(k8sClient, apiClientSet, "eventing-manager")
 	recorder := mgr.GetEventRecorderFor("eventing-manager")
 	ctx := context.Background()
 

config/crd/kustomization.yaml

@@ -4,7 +4,6 @@
 resources:
 - bases/operator.kyma-project.io_eventings.yaml
 - external/subscriptions.eventing.kyma-project.io.crd.yaml
-- external/applications.applicationconnector.crd.yaml
 #+kubebuilder:scaffold:crdkustomizeresource
 
 patchesStrategicMerge:

config/manager/manager.yaml

@@ -90,7 +90,7 @@ spec:
           - name: PUBLISHER_LIMITS_MEMORY
             value: 128Mi
           - name: PUBLISHER_IMAGE
-            value: "europe-docker.pkg.dev/kyma-project/prod/event-publisher-proxy:v20230817-c3e3291f"
+            value: "europe-docker.pkg.dev/kyma-project/dev/event-publisher-proxy:PR-18259"
           - name: PUBLISHER_IMAGE_PULL_POLICY
             value: "IfNotPresent"
           - name: PUBLISHER_REPLICAS

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/stretchr/testify v1.8.4
 	go.uber.org/zap v1.25.0
 	k8s.io/api v0.27.4
+	k8s.io/apiextensions-apiserver v0.27.2
 	k8s.io/apimachinery v0.27.3
 	k8s.io/client-go v0.27.2
 	k8s.io/utils v0.0.0-20230220204549-a5ecb0141aa5
@@ -85,7 +86,6 @@ require (
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiextensions-apiserver v0.27.2 // indirect
 	k8s.io/component-base v0.27.2 // indirect
 	k8s.io/klog/v2 v2.90.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20230501164219-8b0f38b5fd1f // indirect

internal/controller/eventing/controller.go

@@ -177,7 +177,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		return r.handleEventingDeletion(ctx, eventing, log)
 	}
 
-	// check if the Eveting CR is allowed to be created.
+	// check if the Eventing CR is allowed to be created.
 	if r.allowedEventingCR != nil {
 		if result, err := r.handleEventingCRAllowedCheck(ctx, eventing, log); !result || err != nil {
 			return ctrl.Result{}, err
@@ -333,6 +333,14 @@ func (r *Reconciler) handleEventingReconcile(ctx context.Context,
 	// update ActiveBackend in status.
 	eventing.SyncStatusActiveBackend()
 
+	// check if Application CRD is installed.
+	isApplicationCRDEnabled, err := r.kubeClient.ApplicationCRDExists(ctx)
+	if err != nil {
+		return ctrl.Result{}, r.syncStatusWithSubscriptionManagerErr(ctx, eventing, err, log)
+	}
+	r.backendConfig.PublisherConfig.ApplicationCRDEnabled = isApplicationCRDEnabled
+	r.eventingManager.SetBackendConfig(r.backendConfig)
+
 	// reconcile for specified backend.
 	switch eventing.Spec.Backend.Type {
 	case eventingv1alpha1.NatsBackendType:

internal/controller/eventing/integrationtests/controller/integration_test.go

@@ -661,6 +661,7 @@ func ensureEPPDeploymentAndHPAResources(t *testing.T, givenEventing *eventingv1a
 	testEnvironment.EnsureEventingSpecPublisherReflected(t, givenEventing)
 	testEnvironment.EnsureEventingReplicasReflected(t, givenEventing)
 	testEnvironment.EnsureDeploymentOwnerReferenceSet(t, givenEventing)
+	testEnvironment.EnsurePublisherDeploymentENVSet(t, givenEventing)
 }
 
 func ensureK8sResources(t *testing.T, givenEventing *eventingv1alpha1.Eventing, testEnvironment *testutils.TestEnvironment) {

internal/controller/eventing/nats.go

@@ -3,6 +3,7 @@ package eventing
 import (
 	"context"
 	"fmt"
+
 	eventingv1alpha1 "github.com/kyma-project/eventing-manager/api/v1alpha1"
 	ecenv "github.com/kyma-project/kyma/components/eventing-controller/pkg/env"
 

internal/controller/eventing/unit_test.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"testing"
 
+	apiclientsetfake "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
+
 	"github.com/kyma-project/eventing-manager/pkg/k8s"
 
 	"github.com/kyma-project/eventing-manager/pkg/env"
@@ -63,8 +65,9 @@ func NewMockedUnitTestEnvironment(t *testing.T, objs ...client.Object) *MockedUn
 	// TODO: once controller-runtime version is upgraded to >=0.15.x, use the following.
 	//fakeClient := fakeClientBuilder.WithObjects(objs...).WithStatusSubresource(objs...).Build()
 	fakeClient := fakeClientBuilder.WithObjects(objs...).Build()
+	fakeClientSet := apiclientsetfake.NewSimpleClientset()
 	recorder := &record.FakeRecorder{}
-	kubeClient := k8s.NewKubeClient(fakeClient, "eventing-manager")
+	kubeClient := k8s.NewKubeClient(fakeClient, fakeClientSet, "eventing-manager")
 
 	// setup custom mocks
 	eventingManager := new(managermocks.Manager)

internal/controller/eventing/utils.go

@@ -2,6 +2,7 @@ package eventing
 
 import (
 	"context"
+
 	eventingv1alpha1 "github.com/kyma-project/eventing-manager/api/v1alpha1"
 	"github.com/kyma-project/eventing-manager/pkg/env"
 	ecenv "github.com/kyma-project/kyma/components/eventing-controller/pkg/env"

pkg/env/backend_config.go

@@ -40,7 +40,8 @@ type PublisherConfig struct {
 	RequestTimeout    string `envconfig:"PUBLISHER_REQUEST_TIMEOUT" default:"5s"`
 	PriorityClassName string `envconfig:"PUBLISHER_PRIORITY_CLASS_NAME" default:""`
 	// publisher takes the controller values
-	AppLogFormat string `envconfig:"APP_LOG_FORMAT" default:"json"`
+	AppLogFormat          string `envconfig:"APP_LOG_FORMAT" default:"json"`
+	ApplicationCRDEnabled bool
 }
 
 type DefaultSubscriptionConfig struct {

pkg/eventing/deployment.go

@@ -203,6 +203,7 @@ func getNATSEnvVars(natsConfig env.NATSConfig, publisherConfig env.PublisherConf
 		{Name: "REQUEST_TIMEOUT", Value: publisherConfig.RequestTimeout},
 		{Name: "LEGACY_NAMESPACE", Value: "kyma"},
 		{Name: "EVENT_TYPE_PREFIX", Value: eventing.Spec.Backend.Config.EventTypePrefix},
+		{Name: "APPLICATION_CRD_ENABLED", Value: strconv.FormatBool(publisherConfig.ApplicationCRDEnabled)},
 		// JetStream-specific config
 		{Name: "JS_STREAM_NAME", Value: natsConfig.JSStreamName},
 	}
@@ -312,6 +313,7 @@ func getEventMeshEnvVars(publisherName string, publisherConfig env.PublisherConf
 		{Name: "BACKEND", Value: "beb"},
 		{Name: "PORT", Value: strconv.Itoa(int(publisherPortNum))},
 		{Name: "EVENT_TYPE_PREFIX", Value: eventing.Spec.Backend.Config.EventTypePrefix},
+		{Name: "APPLICATION_CRD_ENABLED", Value: strconv.FormatBool(publisherConfig.ApplicationCRDEnabled)},
 		{Name: "REQUEST_TIMEOUT", Value: publisherConfig.RequestTimeout},
 		{
 			Name: "CLIENT_ID",

pkg/eventing/deployment_test.go

@@ -134,6 +134,7 @@ func Test_GetNATSEnvVars(t *testing.T) {
 				{Name: "REQUEST_TIMEOUT", Value: "10s"},
 				{Name: "LEGACY_NAMESPACE", Value: "kyma"},
 				{Name: "EVENT_TYPE_PREFIX", Value: ""},
+				{Name: "APPLICATION_CRD_ENABLED", Value: "false"},
 				{Name: "JS_STREAM_NAME", Value: ""},
 			},
 		},
@@ -154,6 +155,7 @@ func Test_GetNATSEnvVars(t *testing.T) {
 				{Name: "REQUEST_TIMEOUT", Value: "10s"},
 				{Name: "LEGACY_NAMESPACE", Value: "kyma"},
 				{Name: "EVENT_TYPE_PREFIX", Value: ""},
+				{Name: "APPLICATION_CRD_ENABLED", Value: "false"},
 				{Name: "JS_STREAM_NAME", Value: "sap"},
 			},
 		},
@@ -252,6 +254,17 @@ func Test_GetEventMeshEnvVars(t *testing.T) {
 				"REQUEST_TIMEOUT":   "10s",
 			},
 		},
+		{
+			name:      "APPLICATION_CRD_ENABLED should be set",
+			givenEnvs: map[string]string{},
+			givenEventing: testutils.NewEventingCR(
+				testutils.WithEventMeshBackend("test-namespace/test-name"),
+				testutils.WithEventingEventTypePrefix(eventTypePrefix),
+			),
+			wantEnvs: map[string]string{
+				"APPLICATION_CRD_ENABLED": "false",
+			},
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {

pkg/eventing/manager.go

@@ -38,6 +38,7 @@ type Manager interface {
 		backendType v1alpha1.BackendType) (*appsv1.Deployment, error)
 	DeployPublisherProxyResources(context.Context, *v1alpha1.Eventing, *appsv1.Deployment) error
 	GetBackendConfig() *env.BackendConfig
+	SetBackendConfig(env.BackendConfig)
 }
 
 type EventingManager struct {
@@ -57,7 +58,7 @@ func NewEventingManager(
 	logger *logger.Logger,
 	recorder record.EventRecorder,
 ) Manager {
-	return EventingManager{
+	return &EventingManager{
 		ctx:           ctx,
 		Client:        client,
 		backendConfig: backendConfig,
@@ -168,6 +169,10 @@ func (em EventingManager) GetBackendConfig() *env.BackendConfig {
 	return &em.backendConfig
 }
 
+func (em *EventingManager) SetBackendConfig(config env.BackendConfig) {
+	em.backendConfig = config
+}
+
 func (em EventingManager) DeployPublisherProxyResources(
 	ctx context.Context,
 	eventing *v1alpha1.Eventing,

pkg/k8s/client.go

@@ -10,6 +10,8 @@ import (
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbac "k8s.io/api/rbac/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	k8sclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -29,16 +31,20 @@ type Client interface {
 		name string) (*admissionv1.MutatingWebhookConfiguration, error)
 	GetValidatingWebHookConfiguration(ctx context.Context,
 		name string) (*admissionv1.ValidatingWebhookConfiguration, error)
+	GetCRD(context.Context, string) (*apiextensionsv1.CustomResourceDefinition, error)
+	ApplicationCRDExists(context.Context) (bool, error)
 }
 
 type KubeClient struct {
 	fieldManager string
 	client       client.Client
+	clientset    k8sclientset.Interface
 }
 
-func NewKubeClient(client client.Client, fieldManager string) Client {
+func NewKubeClient(client client.Client, clientset k8sclientset.Interface, fieldManager string) Client {
 	return &KubeClient{
 		client:       client,
+		clientset:    clientset,
 		fieldManager: fieldManager,
 	}
 }
@@ -130,6 +136,18 @@ func (c *KubeClient) GetSecret(ctx context.Context, namespacedName string) (*cor
 	return secret, nil
 }
 
+func (c *KubeClient) GetCRD(ctx context.Context, name string) (*apiextensionsv1.CustomResourceDefinition, error) {
+	return c.clientset.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, name, metav1.GetOptions{})
+}
+
+func (c *KubeClient) ApplicationCRDExists(ctx context.Context) (bool, error) {
+	_, err := c.GetCRD(ctx, ApplicationCrdName)
+	if err != nil {
+		return false, client.IgnoreNotFound(err)
+	}
+	return true, nil
+}
+
 // GetMutatingWebHookConfiguration returns the MutatingWebhookConfiguration k8s resource.
 func (c *KubeClient) GetMutatingWebHookConfiguration(ctx context.Context,
 	name string) (*admissionv1.MutatingWebhookConfiguration, error) {