Move left over delivery helpers from pkg/buses to pkg/provisioners

config/provisioners/in-memory-channel/in-memory-channel.yaml

@@ -59,6 +59,22 @@ rules:
       - list
       - watch
       - create
+  - apiGroups:
+        - "" # Core API group.
+    resources:
+      - services
+    resourceNames:
+      - in-memory-channel-clusterbus
+    verbs:
+      - delete
+  - apiGroups:
+      - "" # Core API group.
+    resources:
+      - services
+    resourceNames:
+      - in-memory-channel-dispatcher
+    verbs:
+      - update
   - apiGroups:
       - "" # Core API Group.
     resources:
@@ -76,6 +92,7 @@ rules:
       - list
       - watch
       - create
+      - update
 
 ---
 

config/provisioners/kafka/kafka.yaml

@@ -51,6 +51,14 @@ rules:
       - list
       - watch
       - create
+  - apiGroups:
+      - "" # Core API group.
+    resources:
+      - services
+    resourceNames:
+      - kafka-dispatcher
+    verbs:
+      - update
   - apiGroups:
       - "" # Core API Group.
     resources:
@@ -68,6 +76,7 @@ rules:
       - list
       - watch
       - create
+      - update
 ---
 
 apiVersion: rbac.authorization.k8s.io/v1beta1

pkg/controller/eventing/inmemory/channel/reconcile_test.go

@@ -661,7 +661,7 @@ func makeVirtualService() *istiov1alpha3.VirtualService {
 				},
 				Route: []istiov1alpha3.DestinationWeight{{
 					Destination: istiov1alpha3.Destination{
-						Host: "in-memory-channel-clusterbus.knative-eventing.svc.cluster.local",
+						Host: "in-memory-channel-dispatcher.knative-eventing.svc.cluster.local",
 						Port: istiov1alpha3.PortSelector{
 							Number: util.PortNumber,
 						},

pkg/controller/eventing/inmemory/clusterchannelprovisioner/reconcile.go

@@ -18,17 +18,20 @@ package clusterchannelprovisioner
 
 import (
 	"context"
+	"fmt"
 
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
 	util "github.com/knative/eventing/pkg/provisioners"
+	"github.com/knative/eventing/pkg/system"
 )
 
 const (
@@ -139,6 +142,32 @@ func (r *reconciler) reconcile(ctx context.Context, ccp *eventingv1alpha1.Cluste
 		logger.Warn("ClusterChannelProvisioner's K8s Service is not owned by the ClusterChannelProvisioner", zap.Any("clusterChannelProvisioner", ccp), zap.Any("service", svc))
 	}
 
+	// The name of the svc has changed since version 0.2.1. Hence, delete old dispatcher service (in-memory-channel-clusterbus)
+	// that was created previously in version 0.2.0 to ensure backwards compatibility.
+	err = r.deleteOldDispatcherService(ctx, ccp)
+	if err != nil {
+		logger.Info("Error deleting the old ClusterChannelProvisioner's K8s Service", zap.Error(err))
+		return err
+	}
+
 	ccp.Status.MarkReady()
 	return nil
 }
+
+func (r *reconciler) deleteOldDispatcherService(ctx context.Context, ccp *eventingv1alpha1.ClusterChannelProvisioner) error {
+	svcName := fmt.Sprintf("%s-clusterbus", ccp.Name)
+	svcKey := types.NamespacedName{
+		Namespace: system.Namespace,
+		Name:      svcName,
+	}
+	svc := &corev1.Service{}
+	err := r.client.Get(ctx, svcKey, svc)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			return nil
+		}
+		return err
+	}
+
+	return r.client.Delete(ctx, svc)
+}

pkg/controller/eventing/inmemory/clusterchannelprovisioner/reconcile_test.go

@@ -178,6 +178,20 @@ func TestReconcile(t *testing.T) {
 				makeReadyClusterChannelProvisioner(),
 			},
 		},
+		{
+			Name: "Delete old dispatcher",
+			InitialState: []runtime.Object{
+				makeClusterChannelProvisioner(),
+				makeOldK8sService(),
+			},
+			WantPresent: []runtime.Object{
+				makeReadyClusterChannelProvisioner(),
+				makeK8sService(),
+			},
+			WantAbsent: []runtime.Object{
+				makeOldK8sService(),
+			},
+		},
 		{
 			Name: "Create dispatcher - not owned by CCP",
 			InitialState: []runtime.Object{
@@ -279,7 +293,7 @@ func makeK8sService() *corev1.Service {
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: system.Namespace,
-			Name:      fmt.Sprintf("%s-clusterbus", Name),
+			Name:      fmt.Sprintf("%s-dispatcher", Name),
 			OwnerReferences: []metav1.OwnerReference{
 				{
 					APIVersion:         eventingv1alpha1.SchemeGroupVersion.String(),
@@ -305,6 +319,12 @@ func makeK8sService() *corev1.Service {
 	}
 }
 
+func makeOldK8sService() *corev1.Service {
+	svc := makeK8sService()
+	svc.ObjectMeta.Name = fmt.Sprintf("%s-clusterbus", Name)
+	return svc
+}
+
 func makeK8sServiceNotOwnedByClusterChannelProvisioner() *corev1.Service {
 	svc := makeK8sService()
 	svc.OwnerReferences = nil

pkg/controller/eventing/inmemory/controller/main.go

@@ -20,9 +20,9 @@ import (
 	"flag"
 
 	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
-	"github.com/knative/eventing/pkg/buses"
 	"github.com/knative/eventing/pkg/controller/eventing/inmemory/channel"
 	"github.com/knative/eventing/pkg/controller/eventing/inmemory/clusterchannelprovisioner"
+	"github.com/knative/eventing/pkg/provisioners"
 	istiov1alpha3 "github.com/knative/pkg/apis/istio/v1alpha3"
 	"github.com/knative/pkg/signals"
 	"go.uber.org/zap"
@@ -31,8 +31,8 @@ import (
 )
 
 func main() {
-	logConfig := buses.NewLoggingConfig()
-	logger := buses.NewBusLoggerFromConfig(logConfig)
+	logConfig := provisioners.NewLoggingConfig()
+	logger := provisioners.NewProvisionerLoggerFromConfig(logConfig)
 	defer logger.Sync()
 	logger = logger.With(
 		zap.String("eventing.knative.dev/clusterChannelProvisioner", clusterchannelprovisioner.Name),

pkg/controller/names.go

@@ -18,22 +18,6 @@ package controller
 
 import "fmt"
 
-func ClusterBusDispatcherServiceName(clusterBusName string) string {
-	return fmt.Sprintf("%s-clusterbus", clusterBusName)
-}
-
-func ChannelVirtualServiceName(channelName string) string {
-	return fmt.Sprintf("%s-channel", channelName)
-}
-
-func ChannelServiceName(channelName string) string {
-	return fmt.Sprintf("%s-channel", channelName)
-}
-
-func ChannelHostName(channelName, namespace string) string {
-	return fmt.Sprintf("%s.%s.channels.cluster.local", channelName, namespace)
-}
-
 func ServiceHostName(serviceName, namespace string) string {
 	return fmt.Sprintf("%s.%s.svc.cluster.local", serviceName, namespace)
 }

pkg/controller/names_test.go

@@ -26,30 +26,6 @@ func TestNames(t *testing.T) {
 		F    func() string
 		Want string
 	}{{
-		Name: "ClusterBusDispatcherServiceName",
-		F: func() string {
-			return ClusterBusDispatcherServiceName("foo")
-		},
-		Want: "foo-clusterbus",
-	}, {
-		Name: "ChannelVirtualServiceName",
-		F: func() string {
-			return ChannelVirtualServiceName("foo")
-		},
-		Want: "foo-channel",
-	}, {
-		Name: "ChannelServiceName",
-		F: func() string {
-			return ChannelServiceName("foo")
-		},
-		Want: "foo-channel",
-	}, {
-		Name: "ChannelHostName",
-		F: func() string {
-			return ChannelHostName("foo", "namespace")
-		},
-		Want: "foo.namespace.channels.cluster.local",
-	}, {
 		Name: "ServiceHostName",
 		F: func() string {
 			return ServiceHostName("foo", "namespace")