Repository: banzaicloud/kafka-operator
Pull request #221 (2019-11-30 23:51:47)

Add external hostname to SSL certificate DNS names - allow for optional override of hostname with alias
| Q               | A
| --------------- | ---
| Bug fix?        | kinda
| New feature?    | yes
| API breaks?     | no
| Deprecations?   | no
| License         | Apache 2.0


### What's in this PR?

This PR addresses an issue where the load-balancer IP on clusters with an external SSL listener was not included in the DNS names for the server certificate. In addition to that fix, you can also override this value (e.g. with a CNAME or other alias) using `spec.listenersConfig.externalListeners.*.hostnameOverride`. 

There is also some general tidying up and a couple new unit tests.


### Why?

SSL verification would fail when connecting to the external listener. Also, for some users, the load balancer IP/address may not be static, and they would like to be able to alias to it and still use SSL verification. 


### Checklist

- [X] Implementation tested
- [X] Error handling code meets the [guideline]



Patch:
diff --git a/api/v1beta1/kafkacluster_types.go b/api/v1beta1/kafkacluster_types.go
index bd557151..b0cd3425 100644
--- a/api/v1beta1/kafkacluster_types.go
+++ b/api/v1beta1/kafkacluster_types.go
@@ -176,6 +176,7 @@ type ExternalListenerConfig struct {
 	Name                 string `json:"name"`
 	ExternalStartingPort int32  `json:"externalStartingPort"`
 	ContainerPort        int32  `json:"containerPort"`
+	HostnameOverride     string `json:"hostnameOverride,omitempty"`
 }
 
 // InternalListenerConfig defines the internal listener config for Kafka
diff --git a/charts/kafka-operator/templates/operator-kafka-crd.yaml b/charts/kafka-operator/templates/operator-kafka-crd.yaml
index 3d0f7533..1ec5d3f0 100644
--- a/charts/kafka-operator/templates/operator-kafka-crd.yaml
+++ b/charts/kafka-operator/templates/operator-kafka-crd.yaml
@@ -1152,6 +1152,8 @@ spec:
                         type: string
                       type:
                         type: string
+                      hostnameOverride:
+                        type: string
                     required:
                       - containerPort
                       - externalStartingPort
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index 4335f832..0cb452a7 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -1144,6 +1144,8 @@ spec:
                       externalStartingPort:
                         format: int32
                         type: integer
+                      hostnameOverride:
+                        type: string
                       name:
                         type: string
                       type:
diff --git a/pkg/pki/certmanagerpki/certmanager.go b/pkg/pki/certmanagerpki/certmanager.go
index a72a27c5..39982bd4 100644
--- a/pkg/pki/certmanagerpki/certmanager.go
+++ b/pkg/pki/certmanagerpki/certmanager.go
@@ -18,7 +18,7 @@ import (
 	"flag"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
-	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
+	"github.com/banzaicloud/kafka-operator/pkg/util/pki"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -29,7 +29,7 @@ func init() {
 }
 
 type CertManager interface {
-	pkicommon.PKIManager
+	pki.Manager
 }
 
 // certManager implements a PKIManager using cert-manager as the backend
diff --git a/pkg/pki/certmanagerpki/certmanager_pki.go b/pkg/pki/certmanagerpki/certmanager_pki.go
index fd0ff203..0a9ce833 100644
--- a/pkg/pki/certmanagerpki/certmanager_pki.go
+++ b/pkg/pki/certmanagerpki/certmanager_pki.go
@@ -85,10 +85,10 @@ func (c *certManager) FinalizePKI(ctx context.Context, logger logr.Logger) error
 	return nil
 }
 
-func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme) (err error) {
+func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames []string) (err error) {
 	logger.Info("Reconciling cert-manager PKI")
 
-	resources, err := c.kafkapki(ctx, scheme)
+	resources, err := c.kafkapki(ctx, scheme, externalHostnames)
 	if err != nil {
 		return err
 	}
@@ -109,14 +109,14 @@ func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, sche
 	return nil
 }
 
-func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme) ([]runtime.Object, error) {
+func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme, externalHostnames []string) ([]runtime.Object, error) {
 	if c.cluster.Spec.ListenersConfig.SSLSecrets.Create {
-		return fullPKI(c.cluster, scheme), nil
+		return fullPKI(c.cluster, scheme, externalHostnames), nil
 	}
-	return userProvidedPKI(ctx, c.client, c.cluster, scheme)
+	return userProvidedPKI(ctx, c.client, c.cluster, scheme, externalHostnames)
 }
 
-func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme) []runtime.Object {
+func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames []string) []runtime.Object {
 	return []runtime.Object{
 		// A self-signer for the CA Certificate
 		selfSignerForCluster(cluster, scheme),
@@ -126,13 +126,13 @@ func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme) []runtime.Ob
 		// for producers/consumers in other namespaces
 		mainIssuerForCluster(cluster, scheme),
 		// Broker "user"
-		pkicommon.BrokerUserForCluster(cluster),
+		pkicommon.BrokerUserForCluster(cluster, externalHostnames),
 		// Operator user
 		pkicommon.ControllerUserForCluster(cluster),
 	}
 }
 
-func userProvidedPKI(ctx context.Context, client client.Client, cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme) ([]runtime.Object, error) {
+func userProvidedPKI(ctx context.Context, client client.Client, cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames []string) ([]runtime.Object, error) {
 	// If we aren't creating the secrets we need a cluster issuer made from the provided secret
 	caSecret, err := caSecretForProvidedCert(ctx, client, cluster, scheme)
 	if err != nil {
@@ -147,7 +147,7 @@ func userProvidedPKI(ctx context.Context, client client.Client, cluster *v1beta1
 		//
 		// TODO: (tinyzimmer) - Would it be better to allow the KafkaUser to take a user-provided cert/key combination?
 		// It would have to be validated first as signed by whatever the CA is - probably via a webhook.
-		pkicommon.BrokerUserForCluster(cluster),
+		pkicommon.BrokerUserForCluster(cluster, externalHostnames),
 		pkicommon.ControllerUserForCluster(cluster),
 	}, nil
 }
diff --git a/pkg/pki/pki_manager.go b/pkg/pki/pki_manager.go
index ca07d61d..0a40c256 100644
--- a/pkg/pki/pki_manager.go
+++ b/pkg/pki/pki_manager.go
@@ -22,7 +22,7 @@ import (
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/pki/certmanagerpki"
 	"github.com/banzaicloud/kafka-operator/pkg/pki/vaultpki"
-	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
+	"github.com/banzaicloud/kafka-operator/pkg/util/pki"
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -32,7 +32,7 @@ import (
 var MockBackend = v1beta1.PKIBackend("mock")
 
 // GetPKIManager returns a PKI/User manager interface for a given cluster
-func GetPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pkicommon.PKIManager {
+func GetPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pki.Manager {
 	switch cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend {
 
 	// Use cert-manager for pki backend
@@ -57,16 +57,16 @@ func GetPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pkicommo
 // Mock types and functions
 
 type mockPKIManager struct {
-	pkicommon.PKIManager
+	pki.Manager
 	client  client.Client
 	cluster *v1beta1.KafkaCluster
 }
 
-func newMockPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pkicommon.PKIManager {
+func newMockPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pki.Manager {
 	return &mockPKIManager{client: client, cluster: cluster}
 }
 
-func (m *mockPKIManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme) error {
+func (m *mockPKIManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames []string) error {
 	return nil
 }
 
@@ -74,8 +74,8 @@ func (m *mockPKIManager) FinalizePKI(ctx context.Context, logger logr.Logger) er
 	return nil
 }
 
-func (m *mockPKIManager) ReconcileUserCertificate(ctx context.Context, user *v1alpha1.KafkaUser, scheme *runtime.Scheme) (*pkicommon.UserCertificate, error) {
-	return &pkicommon.UserCertificate{}, nil
+func (m *mockPKIManager) ReconcileUserCertificate(ctx context.Context, user *v1alpha1.KafkaUser, scheme *runtime.Scheme) (*pki.UserCertificate, error) {
+	return &pki.UserCertificate{}, nil
 }
 
 func (m *mockPKIManager) FinalizeUserCertificate(ctx context.Context, user *v1alpha1.KafkaUser) error {
diff --git a/pkg/pki/vaultpki/vault.go b/pkg/pki/vaultpki/vault.go
index 713e8558..e6699305 100644
--- a/pkg/pki/vaultpki/vault.go
+++ b/pkg/pki/vaultpki/vault.go
@@ -16,14 +16,14 @@ package vaultpki
 
 import (
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
-	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
+	"github.com/banzaicloud/kafka-operator/pkg/util/pki"
 	vaultapi "github.com/hashicorp/vault/api"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 // VaultPKI implements a PKIManager using vault as the backend
 type VaultPKI interface {
-	pkicommon.PKIManager
+	pki.Manager
 }
 
 type vaultPKI struct {
diff --git a/pkg/pki/vaultpki/vault_pki.go b/pkg/pki/vaultpki/vault_pki.go
index 80ad9ab7..a1728011 100644
--- a/pkg/pki/vaultpki/vault_pki.go
+++ b/pkg/pki/vaultpki/vault_pki.go
@@ -55,7 +55,7 @@ func (v *vaultPKI) FinalizePKI(ctx context.Context, logger logr.Logger) error {
 }
 
 // ReconcilePKI will use the user-provided paths to ensure broker/operator users for a new KafkaCluster
-func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme) (err error) {
+func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames []string) (err error) {
 	log := logger.WithName("vault_pki")
 
 	log.Info("Retrieving vault client")
@@ -65,7 +65,7 @@ func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme
 	}
 
 	log.Info("Reconciling broker user in vault")
-	brokerCert, err := v.reconcileBrokerCert(ctx, vault)
+	brokerCert, err := v.reconcileBrokerCert(ctx, vault, externalHostnames)
 	if err != nil {
 		return
 	}
@@ -149,8 +149,8 @@ func (v *vaultPKI) reconcileBootstrapSecrets(ctx context.Context, scheme *runtim
 	return
 }
 
-func (v *vaultPKI) reconcileBrokerCert(ctx context.Context, vault *vaultapi.Client) (*pkicommon.UserCertificate, error) {
-	return v.reconcileStartupUser(ctx, vault, pkicommon.BrokerUserForCluster(v.cluster))
+func (v *vaultPKI) reconcileBrokerCert(ctx context.Context, vault *vaultapi.Client, externalHostnames []string) (*pkicommon.UserCertificate, error) {
+	return v.reconcileStartupUser(ctx, vault, pkicommon.BrokerUserForCluster(v.cluster, externalHostnames))
 }
 
 func (v *vaultPKI) reconcileControllerCert(ctx context.Context, vault *vaultapi.Client) (*pkicommon.UserCertificate, error) {
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index 7209f2c2..22545c39 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -71,7 +71,7 @@ log.dirs={{ .StorageConfig }}
 super.users={{ .SuperUsers }}
 `
 
-func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, loadBalancerIP, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
+func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, loadBalancerIPs []string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
 	var out bytes.Buffer
 	t := template.Must(template.New("bConfig-config").Parse(kafkaConfigTemplate))
 	if err := t.Execute(&out, map[string]interface{}{
@@ -82,7 +82,7 @@ func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, lo
 		"ZookeeperConnectString":             strings.Join(r.KafkaCluster.Spec.ZKAddresses, ","),
 		"CruiseControlBootstrapServers":      getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
 		"StorageConfig":                      generateStorageConfig(bConfig.StorageConfigs),
-		"AdvertisedListenersConfig":          generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, loadBalancerIP, r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
+		"AdvertisedListenersConfig":          generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, loadBalancerIPs, r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
 		"SuperUsers":                         strings.Join(generateSuperUsers(superUsers), ";"),
 		"ServerKeystorePath":                 serverKeystorePath,
 		"ClientKeystorePath":                 clientKeystorePath,
@@ -104,18 +104,19 @@ func generateSuperUsers(users []string) (suStrings []string) {
 	return
 }
 
-func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIP, serverPass, clientPass string, superUsers []string, log logr.Logger) runtime.Object {
+func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs []string, serverPass, clientPass string, superUsers []string, log logr.Logger) runtime.Object {
 	return &corev1.ConfigMap{
 		ObjectMeta: templates.ObjectMeta(fmt.Sprintf(brokerConfigTemplate+"-%d", r.KafkaCluster.Name, id), util.MergeLabels(labelsForKafka(r.KafkaCluster.Name), map[string]string{"brokerId": fmt.Sprintf("%d", id)}), r.KafkaCluster),
-		Data:       map[string]string{"broker-config": r.generateBrokerConfig(id, brokerConfig, loadBalancerIP, serverPass, clientPass, superUsers, log)},
+		Data:       map[string]string{"broker-config": r.generateBrokerConfig(id, brokerConfig, loadBalancerIPs, serverPass, clientPass, superUsers, log)},
 	}
 }
 
-func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadBalancerIP, namespace, crName string, headlessServiceEnabled bool) string {
+func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadBalancerIPs []string, namespace, crName string, headlessServiceEnabled bool) string {
 	advertisedListenerConfig := []string{}
 	for _, eListener := range l.ExternalListeners {
+		// use first element of loadBalancerIPs slice for external listener name
 		advertisedListenerConfig = append(advertisedListenerConfig,
-			fmt.Sprintf("%s://%s:%d", strings.ToUpper(eListener.Name), loadBalancerIP, eListener.ExternalStartingPort+id))
+			fmt.Sprintf("%s://%s:%d", strings.ToUpper(eListener.Name), loadBalancerIPs[0], eListener.ExternalStartingPort+id))
 	}
 	for _, iListener := range l.InternalListeners {
 		if headlessServiceEnabled {
@@ -196,7 +197,7 @@ func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32,
 	return internalListener
 }
 
-func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIP, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
+func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs []string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
 	parsedReadOnlyClusterConfig := util.ParsePropertiesFormat(r.KafkaCluster.Spec.ReadOnlyConfig)
 	var parsedReadOnlyBrokerConfig = map[string]string{}
 
@@ -214,7 +215,7 @@ func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerC
 	//Generate the Complete Configuration for the Broker
 	completeConfigMap := map[string]string{}
 
-	if err := mergo.Merge(&completeConfigMap, util.ParsePropertiesFormat(r.getConfigString(brokerConfig, id, loadBalancerIP, serverPass, clientPass, superUsers, log))); err != nil {
+	if err := mergo.Merge(&completeConfigMap, util.ParsePropertiesFormat(r.getConfigString(brokerConfig, id, loadBalancerIPs, serverPass, clientPass, superUsers, log))); err != nil {
 		log.Error(err, "error occurred during merging operator generated configs")
 	}
 
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index ddc3f077..43fb23c3 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -203,26 +203,36 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 			}
 		}
 	}
-	lBIp := ""
+
+	lbIPs := make([]string, 0)
 
 	if r.KafkaCluster.Spec.ListenersConfig.ExternalListeners != nil {
-		lBIp, err = getLoadBalancerIP(r.Client, r.KafkaCluster.Namespace, log)
+		// TODO: This is a hack that needs to be banished when the time is right.
+		// Currently we only support one external listener but this will be fixed
+		// sometime in the future
+		if r.KafkaCluster.Spec.ListenersConfig.ExternalListeners[0].HostnameOverride != "" {
+			// first element of slice will be used for external advertised listener
+			lbIPs = append(lbIPs, r.KafkaCluster.Spec.ListenersConfig.ExternalListeners[0].HostnameOverride)
+		}
+		lbIP, err := getLoadBalancerIP(r.Client, r.KafkaCluster.Namespace, log)
 		if err != nil {
 			return err
 		}
+		lbIPs = append(lbIPs, lbIP)
 	}
 	//TODO remove after testing
 	//lBIp := "192.168.0.1"
 
-	// We need to grab names for servers and client in case user is enabling ACLs
-	// That way we can continue to manage topics and users
+	// Setup the PKI if using SSL
 	if r.KafkaCluster.Spec.ListenersConfig.SSLSecrets != nil {
 		// reconcile the PKI
-		if err := pki.GetPKIManager(r.Client, r.KafkaCluster).ReconcilePKI(context.TODO(), log, r.Scheme); err != nil {
+		if err := pki.GetPKIManager(r.Client, r.KafkaCluster).ReconcilePKI(context.TODO(), log, r.Scheme, lbIPs); err != nil {
 			return err
 		}
 	}
 
+	// We need to grab names for servers and client in case user is enabling ACLs
+	// That way we can continue to manage topics and users
 	serverPass, clientPass, superUsers, err := r.getServerAndClientDetails()
 	if err != nil {
 		return err
@@ -242,7 +252,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 		}
 		if r.KafkaCluster.Spec.RackAwareness == nil {
-			o := r.configMap(broker.Id, brokerConfig, lBIp, serverPass, clientPass, superUsers, log)
+			o := r.configMap(broker.Id, brokerConfig, lbIPs, serverPass, clientPass, superUsers, log)
 			err := k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
 			if err != nil {
 				return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resource", o.GetObjectKind().GroupVersionKind())
@@ -250,7 +260,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		} else {
 			if brokerState, ok := r.KafkaCluster.Status.BrokersState[strconv.Itoa(int(broker.Id))]; ok {
 				if brokerState.RackAwarenessState == v1beta1.Configured {
-					o := r.configMap(broker.Id, brokerConfig, lBIp, serverPass, clientPass, superUsers, log)
+					o := r.configMap(broker.Id, brokerConfig, lbIPs, serverPass, clientPass, superUsers, log)
 					err := k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
 					if err != nil {
 						return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resource", o.GetObjectKind().GroupVersionKind())
diff --git a/pkg/util/pki/common.go b/pkg/util/pki/common.go
index 59a533ce..7403be13 100644
--- a/pkg/util/pki/common.go
+++ b/pkg/util/pki/common.go
@@ -39,17 +39,21 @@ const (
 	BrokerIssuerTemplate = "%s-issuer"
 	// BrokerControllerTemplate is the template used for operator certificate resources
 	BrokerControllerTemplate = "%s-controller"
+	// BrokerControllerFQDNTemplate is combined with the above and cluster namespace
+	// to create a 'fake' full-name for the controller user
+	BrokerControllerFQDNTemplate = "%s.%s.mgt.cluster.local"
 	// CAIntermediateTemplate is the template used for intermediate CA resources
 	CAIntermediateTemplate = "%s-intermediate.%s.cluster.local"
 	// CAFQDNTemplate is the template used for the FQDN of a CA
 	CAFQDNTemplate = "%s-ca.%s.cluster.local"
 )
 
-// PKIManager is the main interface for performing PKI operations
-type PKIManager interface {
+// Manager is the main interface for objects performing PKI operations
+type Manager interface {
 	// ReconcilePKI ensures a PKI for a kafka cluster - should be idempotent.
 	// This method should at least setup any issuer needed for user certificates
-	ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme) error
+	// as well as broker/cruise-control secrets
+	ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames []string) error
 
 	// FinalizePKI performs any cleanup steps necessary for a PKI backend
 	FinalizePKI(ctx context.Context, logger logr.Logger) error
@@ -86,8 +90,8 @@ func (u *UserCertificate) DN() string {
 	return cert.Subject.String()
 }
 
-// GetDNSNames returns all potential DNS names for a kafka cluster - including brokers
-func GetDNSNames(cluster *v1beta1.KafkaCluster) (dnsNames []string) {
+// GetInternalDNSNames returns all potential DNS names for a kafka cluster - including brokers
+func GetInternalDNSNames(cluster *v1beta1.KafkaCluster) (dnsNames []string) {
 	dnsNames = make([]string, 0)
 	dnsNames = append(dnsNames, clusterDNSNames(cluster)...)
 	return
@@ -154,12 +158,12 @@ func LabelsForKafkaPKI(name string) map[string]string {
 }
 
 // BrokerUserForCluster returns a KafkaUser CR for the broker certificates in a KafkaCluster
-func BrokerUserForCluster(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaUser {
+func BrokerUserForCluster(cluster *v1beta1.KafkaCluster, additionalHostnames []string) *v1alpha1.KafkaUser {
 	return &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(GetCommonName(cluster), LabelsForKafkaPKI(cluster.Name), cluster),
 		Spec: v1alpha1.KafkaUserSpec{
 			SecretName: fmt.Sprintf(BrokerServerCertTemplate, cluster.Name),
-			DNSNames:   GetDNSNames(cluster),
+			DNSNames:   append(GetInternalDNSNames(cluster), additionalHostnames...),
 			IncludeJKS: true,
 			ClusterRef: v1alpha1.ClusterReference{
 				Name:      cluster.Name,
@@ -173,7 +177,7 @@ func BrokerUserForCluster(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaUser {
 func ControllerUserForCluster(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaUser {
 	return &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(
-			fmt.Sprintf("%s.%s.mgt.cluster.local", fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace),
+			fmt.Sprintf(BrokerControllerFQDNTemplate, fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace),
 			LabelsForKafkaPKI(cluster.Name), cluster,
 		),
 		Spec: v1alpha1.KafkaUserSpec{


Test patch:
diff --git a/pkg/pki/certmanagerpki/certmanager_pki_test.go b/pkg/pki/certmanagerpki/certmanager_pki_test.go
index 61995f7d..90648d88 100644
--- a/pkg/pki/certmanagerpki/certmanager_pki_test.go
+++ b/pkg/pki/certmanagerpki/certmanager_pki_test.go
@@ -95,40 +95,40 @@ func TestReconcilePKI(t *testing.T) {
 	manager := newMock(cluster)
 	ctx := context.Background()
 
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme); err == nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err == nil {
 		t.Error("Expected error got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 	}
 
 	manager.client.Create(ctx, newServerSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme); err == nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err == nil {
 		t.Error("Expected error got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 	}
 
 	manager.client.Create(ctx, newControllerSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme); err == nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err == nil {
 		t.Error("Expected error got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 	}
 
 	manager.client.Create(ctx, newCASecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
 		t.Error("Expected successful reconcile, got:", err)
 	}
 
 	cluster.Spec.ListenersConfig.SSLSecrets.Create = false
 	manager = newMock(cluster)
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme); err == nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err == nil {
 		t.Error("Expected error got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 	}
 	manager.client.Create(ctx, newPreCreatedSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
 		t.Error("Expected successful reconcile, got:", err)
 	}
 }
diff --git a/pkg/pki/pki_manager_test.go b/pkg/pki/pki_manager_test.go
index 31810d53..0b472757 100644
--- a/pkg/pki/pki_manager_test.go
+++ b/pkg/pki/pki_manager_test.go
@@ -57,7 +57,7 @@ func TestGetPKIManager(t *testing.T) {
 
 	// Test mock functions
 	var err error
-	if err = mock.ReconcilePKI(ctx, log, scheme.Scheme); err != nil {
+	if err = mock.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
 		t.Error("Expected nil error got:", err)
 	}
 
diff --git a/pkg/pki/vaultpki/vault_test.go b/pkg/pki/vaultpki/vault_test.go
index 89b04853..0c4d8a5b 100644
--- a/pkg/pki/vaultpki/vault_test.go
+++ b/pkg/pki/vaultpki/vault_test.go
@@ -161,7 +161,7 @@ func TestAll(t *testing.T) {
 		t.Fatal("Failed to convert test cert to JKS")
 	}
 
-	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme); err == nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err == nil {
 		t.Error("Expected resource not ready, got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected resource not ready, got:", err)
@@ -185,7 +185,7 @@ func TestAll(t *testing.T) {
 	}))
 
 	// Should be safe to do multiple times
-	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme); err != nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
 		t.Error("Expected no error, got:", err)
 	}
 
@@ -194,7 +194,7 @@ func TestAll(t *testing.T) {
 		t.Error("Expected no error, got:", err)
 	}
 
-	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme); err != nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
 		t.Error("Expected no error, got:", err)
 	}
 
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index 2cdcf9de..5a404eca 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -140,7 +140,7 @@ zookeeper.connect=example.zk:2181`,
 					},
 				},
 			}
-			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, "", "", "", []string{}, nil)
+			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, []string{}, "", "", []string{}, nil)
 
 			if generatedConfig != test.expectedConfig {
 				t.Errorf("the expected config is %s, received: %s", test.expectedConfig, generatedConfig)
diff --git a/pkg/util/pki/pki_common_test.go b/pkg/util/pki/pki_common_test.go
index f181b40a..f0962ffb 100644
--- a/pkg/util/pki/pki_common_test.go
+++ b/pkg/util/pki/pki_common_test.go
@@ -19,10 +19,21 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
 	certutil "github.com/banzaicloud/kafka-operator/pkg/util/cert"
 )
 
+func testCluster(t *testing.T) *v1beta1.KafkaCluster {
+	t.Helper()
+	cluster := &v1beta1.KafkaCluster{}
+	cluster.Name = "test-cluster"
+	cluster.Namespace = "test-namespace"
+	cluster.Spec = v1beta1.KafkaClusterSpec{}
+	return cluster
+}
+
 func TestDN(t *testing.T) {
 	cert, _, expected, err := certutil.GenerateTestCert()
 	if err != nil {
@@ -68,18 +79,11 @@ func TestLabelsForKafkaPKI(t *testing.T) {
 	}
 }
 
-func TestGetDNSNames(t *testing.T) {
-	cluster := &v1beta1.KafkaCluster{}
-	cluster.Name = "test-cluster"
-	cluster.Namespace = "test-namespace"
-
-	cluster.Spec = v1beta1.KafkaClusterSpec{}
-	cluster.Spec.Brokers = []v1beta1.Broker{
-		{Id: 0},
-	}
+func TestGetInternalDNSNames(t *testing.T) {
+	cluster := testCluster(t)
 
 	cluster.Spec.HeadlessServiceEnabled = true
-	headlessNames := GetDNSNames(cluster)
+	headlessNames := GetInternalDNSNames(cluster)
 	expected := []string{
 		"*.test-cluster-headless.test-namespace.svc.cluster.local",
 		"test-cluster-headless.test-namespace.svc.cluster.local",
@@ -94,7 +98,7 @@ func TestGetDNSNames(t *testing.T) {
 	}
 
 	cluster.Spec.HeadlessServiceEnabled = false
-	allBrokerNames := GetDNSNames(cluster)
+	allBrokerNames := GetInternalDNSNames(cluster)
 	expected = []string{
 		"*.test-cluster-all-broker.test-namespace.svc.cluster.local",
 		"test-cluster-all-broker.test-namespace.svc.cluster.local",
@@ -108,3 +112,49 @@ func TestGetDNSNames(t *testing.T) {
 		t.Error("Expected:", expected, "got:", allBrokerNames)
 	}
 }
+
+func TestBrokerUserForCluster(t *testing.T) {
+	cluster := testCluster(t)
+	user := BrokerUserForCluster(cluster, []string{})
+
+	expected := &v1alpha1.KafkaUser{
+		ObjectMeta: templates.ObjectMeta(GetCommonName(cluster), LabelsForKafkaPKI(cluster.Name), cluster),
+		Spec: v1alpha1.KafkaUserSpec{
+			SecretName: fmt.Sprintf(BrokerServerCertTemplate, cluster.Name),
+			DNSNames:   GetInternalDNSNames(cluster),
+			IncludeJKS: true,
+			ClusterRef: v1alpha1.ClusterReference{
+				Name:      cluster.Name,
+				Namespace: cluster.Namespace,
+			},
+		},
+	}
+
+	if !reflect.DeepEqual(user, expected) {
+		t.Errorf("Expected %+v\nGot %+v", expected, user)
+	}
+}
+
+func TestControllerUserForCluster(t *testing.T) {
+	cluster := testCluster(t)
+	user := ControllerUserForCluster(cluster)
+
+	expected := &v1alpha1.KafkaUser{
+		ObjectMeta: templates.ObjectMeta(
+			fmt.Sprintf(BrokerControllerFQDNTemplate, fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace),
+			LabelsForKafkaPKI(cluster.Name), cluster,
+		),
+		Spec: v1alpha1.KafkaUserSpec{
+			SecretName: fmt.Sprintf(BrokerControllerTemplate, cluster.Name),
+			IncludeJKS: true,
+			ClusterRef: v1alpha1.ClusterReference{
+				Name:      cluster.Name,
+				Namespace: cluster.Namespace,
+			},
+		},
+	}
+
+	if !reflect.DeepEqual(user, expected) {
+		t.Errorf("Expected %+v\nGot %+v", expected, user)
+	}
+}