Create configmap for starting dispatcher pods (knative-extensions#4027)…

… (#1299) Signed-off-by: Pierangelo Di Pilato <[email protected]>
openshift-knative · Oct 1, 2024 · 1e7271f · 1e7271f
1 parent 4146637
commit 1e7271f
Show file tree

Hide file tree

Showing 10 changed files with 159 additions and 16 deletions.
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
@@ -33,6 +33,7 @@ import (
 	"knative.dev/eventing/pkg/auth"
 	"knative.dev/eventing/pkg/eventingtls"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/clientpool"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
@@ -70,7 +71,7 @@ func main() {
 	ctx = filteredFactory.WithSelectors(ctx,
 		eventingtls.TrustBundleLabelSelector,
 		auth.OIDCLabelSelector,
-		"app.kubernetes.io/kind=kafka-dispatcher",
+		eventing.DispatcherLabelSelectorStr,
 	)
 
 	if v := os.Getenv("ENABLE_SARAMA_LOGGER"); strings.EqualFold(v, "true") {

diff --git a/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
@@ -27,6 +27,11 @@ const (
 	ConfigMapVolumeName = "kafka-resources"
 
 	DispatcherVolumeName = "contract-resources"
+
+	DataPlanePodKindLabelKey    = "app.kubernetes.io/kind"
+	DispatcherPodKindLabelValue = "kafka-dispatcher"
+
+	DispatcherLabelSelectorStr = DataPlanePodKindLabelKey + "=" + DispatcherPodKindLabelValue
 )
 
 func ConfigMapNameFromPod(p *corev1.Pod) (string, error) {

diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/register.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/register.go
@@ -17,6 +17,8 @@
 package v1alpha1
 
 import (
+	"strings"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -75,3 +77,16 @@ func IsKnownStatefulSet(name string) bool {
 		name == ChannelStatefulSetName ||
 		name == BrokerStatefulSetName
 }
+
+func GetOwnerKindFromStatefulSetPrefix(name string) (string, bool) {
+	if strings.HasPrefix(name, SourceStatefulSetName) {
+		return "KafkaSource", true
+	}
+	if strings.HasPrefix(name, ChannelStatefulSetName) {
+		return "KafkaChannel", true
+	}
+	if strings.HasPrefix(name, BrokerStatefulSetName) {
+		return "Trigger", true
+	}
+	return "", false
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -27,13 +27,15 @@ import (
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/storage/names"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/clientpool"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/offset"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
@@ -203,6 +205,47 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 
 	ResyncOnStatefulSetChange(ctx, globalResync)
 
+	dispatcherPodInformer.Informer().AddEventHandler(controller.HandleAll(func(obj interface{}) {
+		pod, ok := obj.(*corev1.Pod)
+		if !ok {
+			return
+		}
+
+		kind, ok := kafkainternals.GetOwnerKindFromStatefulSetPrefix(pod.Name)
+		if !ok {
+			return
+		}
+
+		cmName, err := eventing.ConfigMapNameFromPod(pod)
+		if err != nil {
+			logger.Warnw("Failed to get ConfigMap name from pod", zap.String("pod", pod.Name), zap.Error(err))
+			return
+		}
+		if err := r.ensureContractConfigMapExists(ctx, pod, cmName); err != nil {
+			logger.Warnw("Failed to ensure ConfigMap for pod exists", zap.String("pod", pod.Name), zap.String("configmap", cmName), zap.Error(err))
+			return
+		}
+
+		impl.FilteredGlobalResync(
+			func(obj interface{}) bool {
+				cg, ok := obj.(*kafkainternals.ConsumerGroup)
+				if !ok {
+					return false
+				}
+
+				uf := cg.GetUserFacingResourceRef()
+				if uf == nil {
+					return false
+				}
+				if strings.EqualFold(kind, uf.Kind) {
+					return true
+				}
+				return false
+			},
+			consumerGroupInformer.Informer(),
+		)
+	}))
+
 	//Todo: ScaledObject informer when KEDA is installed
 
 	return impl

diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -17,24 +17,30 @@
 package consumergroup
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/clientpool"
 	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/filtered/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	filteredFactory "knative.dev/pkg/client/injection/kube/informers/factory/filtered"
+	_ "knative.dev/pkg/client/injection/kube/informers/factory/filtered/fake"
 	"knative.dev/pkg/configmap"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/clientpool"
+
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumer/fake"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
@@ -55,7 +61,11 @@ const (
 )
 
 func TestNewController(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, func(ctx context.Context) context.Context {
+		return filteredFactory.WithSelectors(ctx,
+			eventing.DispatcherLabelSelectorStr,
+		)
+	})
 	ctx, _ = kedaclient.With(ctx)
 
 	t.Setenv("SYSTEM_NAMESPACE", systemNamespace)

diff --git a/control-plane/pkg/reconciler/consumergroup/evictor_test.go b/control-plane/pkg/reconciler/consumergroup/evictor_test.go
@@ -17,6 +17,7 @@
 package consumergroup
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -26,21 +27,23 @@ import (
 	"k8s.io/utils/pointer"
 	eventingduckv1alpha1 "knative.dev/eventing/pkg/apis/duck/v1alpha1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	filteredFactory "knative.dev/pkg/client/injection/kube/informers/factory/filtered"
 	reconcilertesting "knative.dev/pkg/reconciler/testing"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 	kafkainternalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client/fake"
 )
 
 func TestNewEvictor(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	require.NotPanics(t, func() { newEvictor(ctx, zap.String("k", "n")) })
 }
 
 func TestEvictorNilPodNoPanic(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	var pod *corev1.Pod
 
@@ -74,7 +77,7 @@ func TestEvictorNilPodNoPanic(t *testing.T) {
 }
 
 func TestEvictorEvictSuccess(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -111,7 +114,7 @@ func TestEvictorEvictSuccess(t *testing.T) {
 }
 
 func TestEvictorNoEvictionEmptyPlacement(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -145,7 +148,7 @@ func TestEvictorNoEvictionEmptyPlacement(t *testing.T) {
 }
 
 func TestEvictorNoEviction(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -181,7 +184,7 @@ func TestEvictorNoEviction(t *testing.T) {
 }
 
 func TestEvictorEvictSuccessConsumerGroupSchedulingInProgress(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -212,7 +215,7 @@ func TestEvictorEvictSuccessConsumerGroupSchedulingInProgress(t *testing.T) {
 }
 
 func TestEvictorEvictPodNotFound(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -238,7 +241,7 @@ func TestEvictorEvictPodNotFound(t *testing.T) {
 	require.Nil(t, err)
 }
 func TestEvictorEvictConsumerGroupNotFound(t *testing.T) {
-	ctx, _ := reconcilertesting.SetupFakeContext(t)
+	ctx, _ := reconcilertesting.SetupFakeContext(t, withFilteredSelectors)
 
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "name"},
@@ -263,3 +266,9 @@ func TestEvictorEvictConsumerGroupNotFound(t *testing.T) {
 
 	require.Nil(t, err)
 }
+
+func withFilteredSelectors(ctx context.Context) context.Context {
+	return filteredFactory.WithSelectors(ctx,
+		eventing.DispatcherLabelSelectorStr,
+	)
+}
diff --git a/openshift/ci-operator/build-image/Dockerfile b/openshift/ci-operator/build-image/Dockerfile
@@ -19,6 +19,7 @@ RUN wget https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 &&
 RUN ./get-helm-3 --version v3.11.3 --no-sudo && helm version
 
 RUN GOFLAGS='' go install github.com/mikefarah/yq/v3@latest
+RUN GOFLAGS='' go install -tags="exclude_graphdriver_btrfs containers_image_openpgp" github.com/containers/skopeo/cmd/[email protected]
 
 # go install creates $GOPATH/.cache with root permissions, we delete it here
 # to avoid permission issues with the runtime users

diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/filtered/fake/fake.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/pod/filtered/fake/fake.go
diff --git a/vendor/modules.txt b/vendor/modules.txt
@@ -1527,6 +1527,7 @@ knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod/filtered
+knative.dev/pkg/client/injection/kube/informers/core/v1/pod/filtered/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret
 knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/service