Repository: knative-sandbox/eventing-kafka (go)

Issue:
Improve Dispatcher Naming To Avoid Conflicts
Currently the "dispatcher" deployments are named by convention based on the KafkaChannel's Namespace/Name values.  Since we're also limited to 63 char DNS safe K8S names we have truncation logic in place.  Therefore, IF there are multiple namespaces with long names which only differ after the first 20 chars, AND there are multiple KafkaChannels in those namespaces with long names which only differ after the first 30 chars, THEN we will have a conflict and the second Dispatcher Deployment will not be created.  Therefore we should improve the naming logic so that it is less likely to have a conflict while still maintaining some readability via kubectl queries.  We do not have to guarantee there will never be a conflict as we don't want to use generated names and have to track their affiliation with labels or some other mechanism.   A simple hashing should provide a reasonable implementation whereby the risk is effectively minimized.

Note - This was discovered by virtue of the e2e tests creating long namespace/kafkachannel names which only differ in their suffix.  


Pull request description:
Improve dispatcher naming
Fixes #52
Change the ChannelDnsSafeName and DispatcherDnsSafeName functions to return a value that includes a hash of the channel name and namespace, to prevent naming collisions.

Patch:
diff --git a/pkg/controller/util/channel.go b/pkg/controller/util/channel.go
index 4719f2f5b..d0c3c231f 100644
--- a/pkg/controller/util/channel.go
+++ b/pkg/controller/util/channel.go
@@ -52,10 +52,11 @@ func NewChannelOwnerReference(channel *kafkav1beta1.KafkaChannel) metav1.OwnerRe
 func ChannelDnsSafeName(kafkaSecretName string) string {
 
 	// In order for the resulting name to be a valid DNS component it's length must be no more than 63 characters.
-	// We are consuming 9 chars for the component separators, and the Channel suffix, which reduces the available
-	// length to 54. We will allocate 50 characters to the kafka secret name leaving an extra buffer.
-	safeSecretName := GenerateValidDnsName(kafkaSecretName, 50, true, false)
-	return fmt.Sprintf("%s-channel", safeSecretName)
+	// We are consuming 17 chars for the component separators, hash, and Channel suffix, which reduces the
+	// available length to 46. We will allocate 41 characters to the kafka secret name leaving an extra buffer.
+	safeSecretName := GenerateValidDnsName(kafkaSecretName, 41, true, false)
+
+	return fmt.Sprintf("%s-%s-channel", safeSecretName, GenerateHash(kafkaSecretName, 8))
 }
 
 // Channel Host Naming Utility
diff --git a/pkg/controller/util/dispatcher.go b/pkg/controller/util/dispatcher.go
index 91ca806d7..38632f6c7 100644
--- a/pkg/controller/util/dispatcher.go
+++ b/pkg/controller/util/dispatcher.go
@@ -2,16 +2,19 @@ package util
 
 import (
 	"fmt"
+
 	kafkav1beta1 "knative.dev/eventing-contrib/kafka/channel/pkg/apis/messaging/v1beta1"
 )
 
 // Create A DNS Safe Name For The Specified KafkaChannel Suitable For Use With K8S Services
 func DispatcherDnsSafeName(channel *kafkav1beta1.KafkaChannel) string {
 
-	// In order for the resulting name to be a valid DNS component is 63 characters.  We are appending 12 characters to separate
-	// the components and to indicate this is a Dispatcher which further reduces the available length to 51.  We will allocate 30
-	// characters to the channel and 20 to the namespace, leaving some extra buffer.
-	safeChannelName := GenerateValidDnsName(channel.Name, 30, true, false)
-	safeChannelNamespace := GenerateValidDnsName(channel.Namespace, 20, false, false)
-	return fmt.Sprintf("%s-%s-dispatcher", safeChannelName, safeChannelNamespace)
+	// In order for the resulting name to be a valid DNS component is 63 characters.  We are appending 13 characters to
+	// separate the components and to indicate this is a Dispatcher, and adding 8 hash characters, which further reduces
+	// the available length to 42.
+	// We will allocate 26 characters to the channel and 16 to the namespace, leaving some extra buffer.
+	safeChannelName := GenerateValidDnsName(channel.Name, 26, true, false)
+	safeChannelNamespace := GenerateValidDnsName(channel.Namespace, 16, false, false)
+	hash := GenerateHash(channel.Name+channel.Namespace, 8)
+	return fmt.Sprintf("%s-%s-%s-dispatcher", safeChannelName, safeChannelNamespace, hash)
 }
diff --git a/pkg/controller/util/hash.go b/pkg/controller/util/hash.go
new file mode 100644
index 000000000..d55b0db02
--- /dev/null
+++ b/pkg/controller/util/hash.go
@@ -0,0 +1,12 @@
+package util
+
+import (
+	"crypto/md5"
+	"fmt"
+)
+
+// Generate An MD5 Hash Of A String And Return Desired Number Of Characters
+func GenerateHash(stringToHash string, length int) string {
+	// Create an MD5 hash and return however many characters the caller wants (note that the max is 32 for MD5)
+	return fmt.Sprintf(fmt.Sprintf("%%.%ds", length), fmt.Sprintf("%x", md5.Sum([]byte(stringToHash))))
+}


Test patch:
diff --git a/pkg/controller/test/data.go b/pkg/controller/test/data.go
index 0c723de8a..b7131c584 100644
--- a/pkg/controller/test/data.go
+++ b/pkg/controller/test/data.go
@@ -59,7 +59,7 @@ const (
 	KafkaSecretNamespace  = commonconstants.KnativeEventingNamespace // Needs To Match Hardcoded Value In Reconciliation
 	KafkaSecretName       = "kafkasecret-name"
 	KafkaSecretKey        = KafkaSecretNamespace + "/" + KafkaSecretName
-	ChannelDeploymentName = KafkaSecretName + "-channel"
+	ChannelDeploymentName = KafkaSecretName + "-b9176d5f-channel" // Truncated MD5 Hash Of KafkaSecretName
 	ChannelServiceName    = ChannelDeploymentName
 	TopicName             = KafkaChannelNamespace + "." + KafkaChannelName
 
diff --git a/pkg/controller/util/channel_test.go b/pkg/controller/util/channel_test.go
index 04b9ae9b7..305995f18 100644
--- a/pkg/controller/util/channel_test.go
+++ b/pkg/controller/util/channel_test.go
@@ -84,7 +84,7 @@ func TestChannelDeploymentDnsSafeName(t *testing.T) {
 	actualResult := ChannelDnsSafeName(kafkaSecret)
 
 	// Verify The Results
-	expectedResult := fmt.Sprintf("%s-channel", strings.ToLower(kafkaSecret))
+	expectedResult := fmt.Sprintf("%s-%s-channel", strings.ToLower(kafkaSecret), GenerateHash(kafkaSecret, 8))
 	assert.Equal(t, expectedResult, actualResult)
 }
 
diff --git a/pkg/controller/util/dispatcher_test.go b/pkg/controller/util/dispatcher_test.go
index 11d495f42..dce2ba7ac 100644
--- a/pkg/controller/util/dispatcher_test.go
+++ b/pkg/controller/util/dispatcher_test.go
@@ -2,11 +2,12 @@ package util
 
 import (
 	"fmt"
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kafkav1beta1 "knative.dev/eventing-contrib/kafka/channel/pkg/apis/messaging/v1beta1"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
-	"testing"
 )
 
 // Test The newControllerRef() Functionality
@@ -28,13 +29,114 @@ func TestNewControllerRef(t *testing.T) {
 // Test The DispatcherDnsSafeName() Functionality
 func TestDispatcherDnsSafeName(t *testing.T) {
 
-	// Test Data
-	channel := &kafkav1beta1.KafkaChannel{ObjectMeta: metav1.ObjectMeta{Name: channelName, Namespace: channelNamespace}}
+	// Define The TestCase Struct
+	type TestCase struct {
+		Name      string
+		Namespace string
+	}
 
-	// Perform The Test
-	actualResult := DispatcherDnsSafeName(channel)
+	// Create The TestCases
+	testCases := []TestCase{
+		{Name: channelName, Namespace: channelNamespace},
+		{Name: "kafkachannel-kne-trigger", Namespace: "test-broker-redelivery-kafka-channel-messaging-knative-devg4l9d"},
+		{Name: "short", Namespace: "kubernetes-maximum-length-for-namespace-with-sixty-three-chars"},
+		{Name: "kubernetes-maximum-length-of-channel-name-is-sixty-three-chars", Namespace: "short"},
+		{Name: "kubernetes-maximum-length-of-channel-name-is-sixty-three-chars", Namespace: "kubernetes-maximum-length-for-namespace-with-sixty-three-chars"},
+		{Name: "a", Namespace: "b"},
+	}
 
-	// Verify The Results
-	expectedResult := fmt.Sprintf("%s-%s-dispatcher", channelName, channelNamespace)
-	assert.Equal(t, expectedResult, actualResult)
+	// Run The TestCases
+	for _, testCase := range testCases {
+		// Test Data
+		channel := &kafkav1beta1.KafkaChannel{ObjectMeta: metav1.ObjectMeta{Name: testCase.Name, Namespace: testCase.Namespace}}
+
+		// Perform The Test
+		actualResult := DispatcherDnsSafeName(channel)
+		hash := GenerateHash(testCase.Name+testCase.Namespace, 8)
+		truncateName := fmt.Sprintf("%.26s", testCase.Name)
+		truncateNamespace := fmt.Sprintf("%.16s", testCase.Namespace)
+		expectedResult := fmt.Sprintf("%s-%s-%s-dispatcher", truncateName, truncateNamespace, hash)
+
+		// Verify The Results
+		assert.Equal(t, expectedResult, actualResult)
+	}
+}
+
+// Test The DispatcherDnsSafeName() Functionality
+func TestDispatcherDnsSafeName_LongNamesDifferent(t *testing.T) {
+
+	// This test ensures that the DnsSafeName for two channels with similar prefixes are, in fact, different
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		Name1      string
+		Namespace1 string
+		Name2      string
+		Namespace2 string
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{
+			Name1:      channelName,
+			Namespace1: channelNamespace,
+			Name2:      channelName + "suffix1",
+			Namespace2: channelNamespace + "suffix1",
+		},
+		{
+			Name1:      "kafkachannel-kne-trigger-with-long-prefix-one",
+			Namespace1: "test-broker-redelivery-kafka-channel-messaging-knative-one",
+			Name2:      "kafkachannel-kne-trigger-with-long-prefix-two",
+			Namespace2: "test-broker-redelivery-kafka-channel-messaging-knative-two",
+		},
+		{
+			Name1:      "short-1",
+			Namespace1: "kubernetes-maximum-length-for-namespace-with-sixty-three-char1",
+			Name2:      "short-2",
+			Namespace2: "kubernetes-maximum-length-for-namespace-with-sixty-three-char2",
+		},
+		{
+			Name1:      "kubernetes-maximum-length-of-channel-name-is-sixty-three-char1",
+			Namespace1: "short-1",
+			Name2:      "kubernetes-maximum-length-of-channel-name-is-sixty-three-char2",
+			Namespace2: "short-2",
+		},
+		{
+			Name1:      "kubernetes-maximum-length-of-channel-name-is-sixty-three-char1",
+			Namespace1: "kubernetes-maximum-length-for-namespace-with-sixty-three-char1",
+			Name2:      "kubernetes-maximum-length-of-channel-name-is-sixty-three-char2",
+			Namespace2: "kubernetes-maximum-length-for-namespace-with-sixty-three-char2",
+		},
+		{
+			Name1:      "a-first",
+			Namespace1: "b-first",
+			Name2:      "a-second",
+			Namespace2: "b-second",
+		},
+	}
+
+	// Run The TestCases
+	for _, testCase := range testCases {
+		// Test Data
+		channel1 := &kafkav1beta1.KafkaChannel{ObjectMeta: metav1.ObjectMeta{Name: testCase.Name1, Namespace: testCase.Namespace1}}
+		channel2 := &kafkav1beta1.KafkaChannel{ObjectMeta: metav1.ObjectMeta{Name: testCase.Name2, Namespace: testCase.Namespace2}}
+
+		// Perform The Test
+		actualResult1 := DispatcherDnsSafeName(channel1)
+		hash1 := GenerateHash(testCase.Name1+testCase.Namespace1, 8)
+		truncateName1 := fmt.Sprintf("%.26s", testCase.Name1)
+		truncateNamespace1 := fmt.Sprintf("%.16s", testCase.Namespace1)
+		expectedResult1 := fmt.Sprintf("%s-%s-%s-dispatcher", truncateName1, truncateNamespace1, hash1)
+
+		actualResult2 := DispatcherDnsSafeName(channel2)
+		hash2 := GenerateHash(testCase.Name2+testCase.Namespace2, 8)
+		truncateName2 := fmt.Sprintf("%.26s", testCase.Name2)
+		truncateNamespace2 := fmt.Sprintf("%.16s", testCase.Namespace2)
+		expectedResult2 := fmt.Sprintf("%s-%s-%s-dispatcher", truncateName2, truncateNamespace2, hash2)
+
+		// Verify The Results
+		assert.Equal(t, expectedResult1, actualResult1)
+		assert.Equal(t, expectedResult2, actualResult2)
+		assert.NotEqual(t, actualResult1, actualResult2)
+	}
 }
diff --git a/pkg/controller/util/hash_test.go b/pkg/controller/util/hash_test.go
new file mode 100644
index 000000000..2ad3759d7
--- /dev/null
+++ b/pkg/controller/util/hash_test.go
@@ -0,0 +1,36 @@
+package util
+
+import (
+	"fmt"
+	"testing"
+
+	"crypto/md5"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// Test the GenerateHash Functionality
+func TestGenerateHash(t *testing.T) {
+	// Define The TestCase Struct
+	type TestCase struct {
+		Name   string
+		Length int
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{Name: "", Length: 32},
+		{Name: "short string", Length: 4},
+		{Name: "long string, 8-character hash", Length: 8},
+		{Name: "odd hash length, 13-characters", Length: 13},
+		{Name: "very long string with 16-character hash and more than 64 characters total", Length: 16},
+	}
+
+	// Run The TestCases
+	for _, testCase := range testCases {
+		hash := GenerateHash(testCase.Name, testCase.Length)
+		expected := fmt.Sprintf("%x", md5.Sum([]byte(testCase.Name)))[0:testCase.Length]
+		assert.Equal(t, expected, hash)
+	}
+
+}