Repository: banzaicloud/kafka-operator
Pull request #575 (2021-03-22 13:51:59)

Allow to add sasl listener types
| Q               | A
| --------------- | ---
| Bug fix?        | no|
| New feature?    |yes|
| API breaks?     | no|
| Deprecations?   | no|
| License         | Apache 2.0


### What's in this PR?
<!-- Explain the contents of the PR. Give an overview about the implementation, which decisions were made and why. -->


### Why?
<!-- Which problem does the PR fix? (Please remove this section if you linked an issue above) -->


### Additional context
<!-- Additional information we should know about (eg. edge cases, steps you followed to test the implementation) (Please remove this section if you don't need it) -->


### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [x] Implementation tested
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/developer-guide/blob/master/docs/coding-style/error-handling-guide.md)
- [x] Logging code meets the guideline
- [x] User guide and development docs updated (if needed)

### To Do
- [x] Add tests which covers the two new listener types



Patch:
diff --git a/charts/kafka-operator/templates/crds.yaml b/charts/kafka-operator/templates/crds.yaml
index a6526df4..a7202ea1 100644
--- a/charts/kafka-operator/templates/crds.yaml
+++ b/charts/kafka-operator/templates/crds.yaml
@@ -11067,6 +11067,14 @@ spec:
                             Default value is LoadBalancer
                           type: string
                         type:
+                          description: 'SecurityProtocol is the protocol used to communicate
+                            with brokers. Valid values are: plaintext, ssl, sasl_plaintext,
+                            sasl_ssl.'
+                          enum:
+                          - ssl
+                          - plaintext
+                          - sasl_ssl
+                          - sasl_plaintext
                           type: string
                       required:
                       - containerPort
@@ -11086,6 +11094,14 @@ spec:
                         name:
                           type: string
                         type:
+                          description: 'SecurityProtocol is the protocol used to communicate
+                            with brokers. Valid values are: plaintext, ssl, sasl_plaintext,
+                            sasl_ssl.'
+                          enum:
+                          - ssl
+                          - plaintext
+                          - sasl_ssl
+                          - sasl_plaintext
                           type: string
                         usedForControllerCommunication:
                           type: boolean
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index 22585be4..d9fdb0b0 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -11066,6 +11066,14 @@ spec:
                             Default value is LoadBalancer
                           type: string
                         type:
+                          description: 'SecurityProtocol is the protocol used to communicate
+                            with brokers. Valid values are: plaintext, ssl, sasl_plaintext,
+                            sasl_ssl.'
+                          enum:
+                          - ssl
+                          - plaintext
+                          - sasl_ssl
+                          - sasl_plaintext
                           type: string
                       required:
                       - containerPort
@@ -11085,6 +11093,14 @@ spec:
                         name:
                           type: string
                         type:
+                          description: 'SecurityProtocol is the protocol used to communicate
+                            with brokers. Valid values are: plaintext, ssl, sasl_plaintext,
+                            sasl_ssl.'
+                          enum:
+                          - ssl
+                          - plaintext
+                          - sasl_ssl
+                          - sasl_plaintext
                           type: string
                         usedForControllerCommunication:
                           type: boolean
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index 38a461ed..f7200646 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -228,13 +228,13 @@ func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger,
 				log.Error(errors.New("inter broker listener name already set"), "config error")
 			}
 		}
-		UpperedListenerType := strings.ToUpper(iListener.Type)
+		UpperedListenerType := iListener.Type.ToUpperString()
 		UpperedListenerName := strings.ToUpper(iListener.Name)
 		securityProtocolMapConfig = append(securityProtocolMapConfig, fmt.Sprintf("%s:%s", UpperedListenerName, UpperedListenerType))
 		listenerConfig = append(listenerConfig, fmt.Sprintf("%s://:%d", UpperedListenerName, iListener.ContainerPort))
 	}
 	for _, eListener := range l.ExternalListeners {
-		UpperedListenerType := strings.ToUpper(eListener.Type)
+		UpperedListenerType := eListener.Type.ToUpperString()
 		UpperedListenerName := strings.ToUpper(eListener.Name)
 		securityProtocolMapConfig = append(securityProtocolMapConfig, fmt.Sprintf("%s:%s", UpperedListenerName, UpperedListenerType))
 		listenerConfig = append(listenerConfig, fmt.Sprintf("%s://:%d", UpperedListenerName, eListener.ContainerPort))
diff --git a/pkg/sdk/v1beta1/common_types.go b/pkg/sdk/v1beta1/common_types.go
index 349cf948..432cf60a 100644
--- a/pkg/sdk/v1beta1/common_types.go
+++ b/pkg/sdk/v1beta1/common_types.go
@@ -14,6 +14,8 @@
 
 package v1beta1
 
+import "strings"
+
 // RackAwarenessState stores info about rack awareness status
 type RackAwarenessState string
 
@@ -32,6 +34,10 @@ type ClusterState string
 // ConfigurationState holds info about the configuration state
 type ConfigurationState string
 
+// SecurityProtocol is the protocol used to communicate with brokers.
+// Valid values are: plaintext, ssl, sasl_plaintext, sasl_ssl.
+type SecurityProtocol string
+
 // PerBrokerConfigurationState holds info about the per-broker configuration state
 type PerBrokerConfigurationState string
 
@@ -71,6 +77,31 @@ func (r CruiseControlState) Complete() CruiseControlState {
 	}
 }
 
+// IsSSL determines if the receiver is using SSL
+func (r SecurityProtocol) IsSSL() bool {
+	return r.Equal(SecurityProtocolSaslSSL) || r.Equal(SecurityProtocolSSL)
+}
+
+// IsSasl determines if the receiver is using Sasl
+func (r SecurityProtocol) IsSasl() bool {
+	return r.Equal(SecurityProtocolSaslSSL) || r.Equal(SecurityProtocolSaslPlaintext)
+}
+
+// IsPlaintext determines if the receiver is using plaintext
+func (r SecurityProtocol) IsPlaintext() bool {
+	return r.Equal(SecurityProtocolPlaintext) || r.Equal(SecurityProtocolSaslPlaintext)
+}
+
+// ToUpperString converts SecurityProtocol to an upper string
+func (r SecurityProtocol) ToUpperString() string {
+	return strings.ToUpper(string(r))
+}
+
+// Equal checks the equality between two SecurityProtocols
+func (r SecurityProtocol) Equal(s SecurityProtocol) bool {
+	return r.ToUpperString() == s.ToUpperString()
+}
+
 const (
 	// PKIBackendCertManager invokes cert-manager for user certificate management
 	PKIBackendCertManager PKIBackend = "cert-manager"
@@ -182,4 +213,13 @@ const (
 	PerBrokerConfigOutOfSync PerBrokerConfigurationState = "PerBrokerConfigOutOfSync"
 	// PerBrokerConfigError states that the generated per-broker brokerConfig can not be set in the Broker
 	PerBrokerConfigError PerBrokerConfigurationState = "PerBrokerConfigError"
+
+	// SecurityProtocolSSL
+	SecurityProtocolSSL SecurityProtocol = "ssl"
+	// SecurityProtocolPlaintext
+	SecurityProtocolPlaintext SecurityProtocol = "plaintext"
+	// SecurityProtocolSaslSSL
+	SecurityProtocolSaslSSL SecurityProtocol = "sasl_ssl"
+	// SecurityProtocolSaslPlaintext
+	SecurityProtocolSaslPlaintext SecurityProtocol = "sasl_plaintext"
 )
diff --git a/pkg/sdk/v1beta1/kafkacluster_types.go b/pkg/sdk/v1beta1/kafkacluster_types.go
index 157d8154..92ab5a78 100644
--- a/pkg/sdk/v1beta1/kafkacluster_types.go
+++ b/pkg/sdk/v1beta1/kafkacluster_types.go
@@ -397,9 +397,10 @@ type InternalListenerConfig struct {
 
 // CommonListenerSpec defines the common building block for Listener type
 type CommonListenerSpec struct {
-	Type          string `json:"type"`
-	Name          string `json:"name"`
-	ContainerPort int32  `json:"containerPort"`
+	// +kubebuilder:validation:Enum=ssl;plaintext;sasl_ssl;sasl_plaintext
+	Type          SecurityProtocol `json:"type"`
+	Name          string           `json:"name"`
+	ContainerPort int32            `json:"containerPort"`
 }
 
 // ListenerStatuses holds information about the statuses of the configured listeners.
diff --git a/pkg/util/client/common.go b/pkg/util/client/common.go
index 1cf41163..0a59b205 100644
--- a/pkg/util/client/common.go
+++ b/pkg/util/client/common.go
@@ -22,7 +22,7 @@ import (
 )
 
 func UseSSL(cluster *v1beta1.KafkaCluster) bool {
-	return cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].Type != "plaintext"
+	return cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].Type.IsSSL()
 }
 
 func determineInternalListenerForInnerCom(internalListeners []v1beta1.InternalListenerConfig) int {
diff --git a/pkg/util/util.go b/pkg/util/util.go
index 4a588f25..5c4d733a 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -126,7 +126,7 @@ func ConvertStringToInt32(s string) int32 {
 func IsSSLEnabledForInternalCommunication(l []v1beta1.InternalListenerConfig) (enabled bool) {
 
 	for _, listener := range l {
-		if strings.ToLower(listener.Type) == "ssl" {
+		if listener.Type.IsSSL() {
 			enabled = true
 			break
 		}


Test patch:
diff --git a/controllers/tests/kafkacluster_controller_externalnodeport_test.go b/controllers/tests/kafkacluster_controller_externalnodeport_test.go
index 1a2d16b0..5fc2e230 100644
--- a/controllers/tests/kafkacluster_controller_externalnodeport_test.go
+++ b/controllers/tests/kafkacluster_controller_externalnodeport_test.go
@@ -101,6 +101,7 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", func() {
 					CommonListenerSpec: v1beta1.CommonListenerSpec{
 						Name:          "test",
 						ContainerPort: 9733,
+						Type:          "plaintext",
 					},
 					ExternalStartingPort: 31123,
 					AccessMethod:         corev1.ServiceTypeNodePort,
@@ -232,6 +233,7 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", func() {
 					CommonListenerSpec: v1beta1.CommonListenerSpec{
 						Name:          "test",
 						ContainerPort: 9733,
+						Type:          "plaintext",
 					},
 					ExternalStartingPort: 30100,
 					IngressServiceSettings: v1beta1.IngressServiceSettings{
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index 78b0b5ed..ee99a76f 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -17,11 +17,12 @@ package kafka
 import (
 	"testing"
 
-	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
+
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/resources"
 )
@@ -37,6 +38,7 @@ func TestGenerateBrokerConfig(t *testing.T) {
 		perBrokerReadOnlyConfig   string
 		perBrokerConfig           string
 		advertisedListenerAddress string
+		listenerType              string
 		expectedConfig            string
 		perBrokerStorageConfig    []v1beta1.StorageConfig
 	}{
@@ -50,6 +52,7 @@ func TestGenerateBrokerConfig(t *testing.T) {
 			perBrokerConfig:           ``,
 			perBrokerReadOnlyConfig:   ``,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
@@ -70,6 +73,7 @@ zookeeper.connect=example.zk:2181/`,
 			perBrokerConfig:           ``,
 			perBrokerReadOnlyConfig:   ``,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
@@ -90,6 +94,7 @@ zookeeper.connect=example.zk:2181/kafka`,
 			perBrokerConfig:           ``,
 			perBrokerReadOnlyConfig:   ``,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
@@ -110,6 +115,7 @@ zookeeper.connect=example.zk:2181/`,
 			perBrokerConfig:           ``,
 			perBrokerReadOnlyConfig:   ``,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
@@ -135,6 +141,7 @@ zookeeper.connect=example.zk:2181,example.zk-1:2181/kafka`,
 				},
 			},
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
@@ -156,6 +163,7 @@ zookeeper.connect=example.zk:2181/`,
 			perBrokerConfig:           ``,
 			perBrokerReadOnlyConfig:   ``,
 			advertisedListenerAddress: `kafka-0.kafka.svc.foo.bar:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.foo.bar:9092
 broker.id=0
 cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.foo.bar:9092
@@ -184,6 +192,7 @@ compression.type=snappy
 auto.create.topics.enable=true
 `,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "plaintext",
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 auto.create.topics.enable=true
 broker.id=0
@@ -194,6 +203,27 @@ inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
 listeners=INTERNAL://:9092
 metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
+zookeeper.connect=example.zk:2181/`,
+		},
+		{
+			testName:                  "configWithSasl",
+			readOnlyConfig:            ``,
+			zkAddresses:               []string{"example.zk:2181"},
+			zkPath:                    ``,
+			kubernetesClusterDomain:   ``,
+			clusterWideConfig:         ``,
+			perBrokerConfig:           ``,
+			perBrokerReadOnlyConfig:   ``,
+			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
+			listenerType:              "sasl_plaintext",
+			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+broker.id=0
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.kubernetes.mode=true
+inter.broker.listener.name=INTERNAL
+listener.security.protocol.map=INTERNAL:SASL_PLAINTEXT
+listeners=INTERNAL://:9092
+metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
 zookeeper.connect=example.zk:2181/`,
 		},
 	}
@@ -218,7 +248,7 @@ zookeeper.connect=example.zk:2181/`,
 							ListenersConfig: v1beta1.ListenersConfig{
 								InternalListeners: []v1beta1.InternalListenerConfig{{
 									CommonListenerSpec: v1beta1.CommonListenerSpec{
-										Type:          "plaintext",
+										Type:          v1beta1.SecurityProtocol(test.listenerType),
 										Name:          "internal",
 										ContainerPort: 9092,
 									},