Use kubectl and RBAC to look up zone for broker.rack

01rbac.yml

@@ -0,0 +1,33 @@
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: kafka
+  namespace: kafka
+automountServiceAccountToken: true
+---
+# Sufficient rights to look up self's pod description and pod's node description
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1beta1
+metadata:
+  name: node-reader
+rules:
+  - apiGroups:
+      - ""
+    resources:
+      - nodes
+    verbs:
+      - get
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1beta1
+metadata:
+  name: kafka-node-reader
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: node-reader
+subjects:
+- kind: ServiceAccount
+  name: kafka
+  namespace: kafka

10broker-config.yml

@@ -11,19 +11,27 @@ data:
     export KAFKA_BROKER_ID=${HOSTNAME##*-}
     sed -i "s/\${KAFKA_BROKER_ID}/$KAFKA_BROKER_ID/" /etc/kafka/server.properties
 
-    PODNAME=$HOSTNAME
-    NAMESPACE=$(cat /var/run/secrets/kubernetes.io/serviceaccount/namespace)
-
     # todo add curl to kafka image, switch to a curl image for init or write the whole lookup in java
     hash curl 2>/dev/null || { apt-get update; DEBIAN_FRONTEND=noninteractive apt-get install curl -y --no-install-recommends; }
 
-    API=https://$KUBERNETES_SERVICE_HOST:$KUBERNETES_SERVICE_PORT/api
-    AUTH="--cacert /run/secrets/kubernetes.io/serviceaccount/ca.crt --header \"Authorization: Bearer $(cat /run/secrets/kubernetes.io/serviceaccount/token)\""
+    echo "Service Account is $MY_POD_SERVICE_ACCOUNT"
 
-    curl -s $AUTH $API/namespaces/kafka/pods/$PODNAME -I --fail-early || {
-      echo "Access problems. Could be RBAC."
+    # curl auth was with system:anonymous so trying kubectl instead
+    hash kubectl 2>/dev/null || {
+      curl -sLS -o k.tar.gz -k https://dl.k8s.io/v1.7.2/kubernetes-client-linux-amd64.tar.gz
+      echo "9c2363710d61a12a28df2d8a4688543b785156369973d33144ab1f2c1d5c7b53  k.tar.gz" | sha256sum -c
+      tar xvf k.tar.gz -C /usr/local/bin/ --strip-components=3 kubernetes/client/bin/kubectl
+      rm k.tar.gz
     }
 
+    ZONE=$(kubectl get node $MY_NODE_NAME -o=go-template='{{index .metadata.labels "failure-domain.beta.kubernetes.io/zone"}}')
+    [ -z "$ZONE" ] &&
+    if [ ! -z "$ZONE" ]; then
+      sed -i "s/#broker.rack=$/broker.rack=$ZONE/" /etc/kafka/server.properties
+    else
+      sed -i "s/#broker.rack=$# No zone label found/" /etc/kafka/server.properties
+    fi
+
   server.properties: |-
     # Licensed to the Apache Software Foundation (ASF) under one or more
     # contributor license agreements.  See the NOTICE file distributed with
@@ -47,7 +55,7 @@ data:
     # The id of the broker. This must be set to a unique integer for each broker.
     broker.id=${KAFKA_BROKER_ID}
 
-    #broker.rack=${KAFKA_BROKER_RACK}
+    #broker.rack=
 
     # Switch to enable topic deletion or not, default value is false
     delete.topic.enable=true

50kafka.yml

@@ -13,10 +13,20 @@ spec:
       annotations:
     spec:
       terminationGracePeriodSeconds: 30
+      serviceAccountName: kafka
       initContainers:
       - name: init-config
         image: solsson/kafka:0.11.0.0@sha256:b27560de08d30ebf96d12e74f80afcaca503ad4ca3103e63b1fd43a2e4c976ce
         command: ['/bin/bash', '/etc/kafka/init.sh']
+        env:
+        - name: MY_NODE_NAME
+          valueFrom:
+            fieldRef:
+              fieldPath: spec.nodeName
+        - name: MY_POD_SERVICE_ACCOUNT
+          valueFrom:
+            fieldRef:
+              fieldPath: spec.serviceAccountName
         volumeMounts:
         - name: config
           mountPath: /etc/kafka