Wednesday, July 8, 2020

Analyze Redis Memory usage

We have used Redis as our persistence layer, and were very pleased from its performance.

Everything went fine for several months, until we've found that the production Redis memory consumption go out of control. We are using many GB of RAM on each one of the Redis cluster nodes.

How can we analyze the problem?

We have so many keys, for various components in the system.

How much RAM is used by each component?

What is the RAM distribution among each type of Redis key?

Gladly, I have found that Redis can assist us.
It had supplied the MEMORY USAGE command, which print the memory bytes used by each key.

BUT... It can only run for a single key, and we have millions of keys.

This is when I've decided to create a small GO application to analyze the memory distribution among the various keys.

The general idea is that each component has different prefixes for its keys, for example:

BOOK-book-name
AUTHOR-author-name
CUSTOMER-customer-id

So we can group the memory usage per each key prefix.

Here is the code that handles this.

package main

import (
   "fmt"
   "github.com/go-redis/redis/v7"
   "log"
   "strings"
)

type Memory struct {
   Key  string
   Size uint
}

type Stats struct {
   count uint
   min   uint
   max   uint
   sum   uint
   avg   uint
}

var client *redis.Client

func main() {
   client = redis.NewClient(&redis.Options{
      Addr:     "127.0.0.1:6379",
      Password: "",
      DB:       0,
   })

   memories := getMemories()

   prefixes := make(map[string]Stats)
   prefixes["BOOK"] = Stats{}
   prefixes["AUTHOR"] = Stats{}
   prefixes["CUSTOMER"] = Stats{}

   for _, memory := range memories {
      updateStats(prefixes, memory)
   }

   fmt.Printf("\n\nsummary\n\n")
   fmt.Printf("prefix,totalKeys,totalSize,avgKeySize,minKeySize,maxKeySize\n")
   for prefix, stats := range prefixes {
      if stats.count > 0 {
         stats.avg = stats.sum / stats.count
         fmt.Printf("%v,%+v,%+v,%+v,%+v,%+v\n", prefix, stats.count, stats.sum, stats.avg, stats.min, stats.max)
      }
   }
}

func getMemories() []Memory {
   cmd := client.Keys("*")
   err := cmd.Err()
   if err != nil {
      log.Fatal(err)
   }

   keys, err := cmd.Result()
   if err != nil {
      log.Fatal(err)
   }

   var memories []Memory
   for _, key := range keys {
      cmd := client.MemoryUsage(key)
      err := cmd.Err()
      if err != nil {
         log.Fatal(err)
      }

      value, err := cmd.Result()
      if err != nil {
         log.Fatal(err)
      }
      memory := Memory{
         Key:  key,
         Size: uint(value),
      }
      memories = append(memories, memory)
   }
   return memories
}

func updateStats(prefixes map[string]Stats, memory Memory) {
   for prefix, stats := range prefixes {
      if strings.HasPrefix(memory.Key, prefix) {
         updateByMemory(&stats, memory)
         prefixes[prefix] = stats
         return
      }
   }
   stats := Stats{}
   updateByMemory(&stats, memory)
   prefixes[memory.Key] = stats
}

func updateByMemory(stats *Stats, memory Memory) {
   stats.count++
   if stats.max < memory.Size {
      stats.max = memory.Size
   }
   if stats.min == 0 || stats.min > memory.Size {
      stats.min = memory.Size
   }
   stats.sum += memory.Size
}

The output of this small application is a CSV file:

And can be displayed of course as a chart:

Final Notes

While working on this, I've found what seems to be a bug on Redis.

For more details, check this question at stackoverflow, and the final result is that it was merged into the go-redis library in this pull request.

Saturday, July 4, 2020

Scale your application using HPA on GKE

In this post we will review the required steps to automatically scale an application installed on Google Kubernetes Engine (GKE). We will use a combination of several functionalities:

Metrics-Server

First we need to supply the CPU/memory metrics per pod. This can be done using the metrics-server.

The metric server is:

...a scalable, efficient source of container resource metrics for Kubernetes built-in autoscaling pipelines.

To apply the metrics server we use the following command:

kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/download/v0.3.6/components.yaml

After a few minutes, the metrics server has collected the statistics for our pods, and we can use the following command to check the pods metrics:

kubectl top pod

HPA

In the kubernetes documentation we find that HPA:

...scales the number of pods in a replication controller, deployment, replica set or stateful set based on observed CPU utilization

Hence, to use HPA, we start by configuring the resources requests in our deployment.

Notice that HPA uses the resources requests, and not the resources limits.

For example, in our deployment, we specify:

spec:
  containers:
    - name: c1
      image: my-image
      resources:
        requests:
          cpu: "0.5"

Next, we create the HPA.

The HPA configures the min and max replicas for the deployment.

It also configures the target averaged CPU usage based on the CPU usage on all of the running pods.

apiVersion: autoscaling/v1
kind: HorizontalPodAutoscaler
metadata:
  name: my-autoscaler
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: my-deployment
  maxReplicas: 10
  minReplicas: 2
  targetCPUUtilizationPercentage: 80

The HPA will create new pods according to the load on the system, but at some stage, there might be too many pods to run on the kubernetes nodes, as the nodes have limited resources, and hence the pods will remain in a pending state. This is where GKE part completes the puzzle.

GKE Cluster Autoscaler

We use GKE Cluster Autoscaler to allocate new kubernetes cluster nodes upon need:

...resizes the number of nodes in a given node pool, based on the demands of your workloads

So once we have pods in a pending state due to insufficient CPU/memory resources, the GKE Cluster Autoscaler adds new nodes.

To configure the GKE Cluster Autoscaler, we update the node pool configuration:

gcloud container clusters update my-k8s-cluster \
    --enable-autoscaling \
    --min-nodes 3 \
    --max-nodes 10 \
    --zone my-zone \
    --node-pool my-pool

Final Notes

In this post we have reviewed the steps to handle kubernetes autoscaling.

Note that you can also configure HPA to use not just memory and CPU, but also custom metrics to scale your application.

Also, we need to use pod anti-affinity rules to avoid all pods from starting on the same node.

Wednesday, July 1, 2020

Using SSL based Ingress on Google Kubernetes Engine

In this post we will review the steps to create a SSL based Ingress on Google Kubernetes Engine (GKE).

An Ingress is an object that exposes external access to services in the kubernetes cluster.

For example we have 2 services: foo-service and bar-service.

We want to expose both of them to the internet, but we want to use a single SSL certificate.

Follow the next steps to create SSL based encryption.

Make sure that all of the services (foo and bar) implement readiness probes.

Add annotation to all of the services to enable routing from the ingress directly to the pod (better performance):

apiVersion: v1
kind: Service
metadata:
  name: foo-service
  annotations:
    cloud.google.com/neg: '{"ingress": true}'

Create a static IP that will be used for the ingress:
```
gcloud compute addresses create my-ip --global
```

Check that the static IP is configured:

gcloud compute addresses describe my-ip --global

Create a SSL certification.

You can either buy a public signed SSL certificate, or create your own self-signed SSL certificate. The self-signed can be used to testing purpose, but in a real world scenario, you would probably need a public signed SSL certificate.
To create a self signed certification, use the following commands:


rm -rf keys
mkdir keys
openssl genrsa -out keys/ingress.key 2048
openssl req -new -key keys/ingress.key -out keys/ingress.csr -subj "/CN=radwarebouncer.com"
openssl x509 -req -days 365 -in keys/ingress.csr -signkey keys/ingress.key -out keys/ingress.crt
kubectl create secret tls bouncer-ingress-secret --cert keys/ingress.crt --key keys/ingress.key

Create Ingress.

Notice that:
- The ingress includes annotation to use the my-ip static IP.
- The ingress includes a specification to use the ingress-secret.

apiVersion: networking.k8s.io/v1beta1
kind: Ingress
metadata:
  name: ingress
  annotations:
    nginx.ingress.kubernetes.io/ssl-redirect: "false"
    kubernetes.io/ingress.global-static-ip-name: "my-ip"
spec:
  tls:
    - secretName: ingress-secret
  rules:
    - host: example.com
      http:
        paths:
          - path: /foo
            backend:
              serviceName: foo-service
              servicePort: 80
          - path: /bar
            backend:
              serviceName: foo-service
              servicePort: 80

That's it, you are ready to go.

Use curl for your domain name to check the ingress, for example:

curl -k https://example.com/foo

Delete images from a Private Docker Registry

In case you have a private docker, and your builds keep pushing images to the registry, you will eventually run out of space.

To remove old images, you can use the following script.

The script scan all images, and removes any image whose tag is not "latest", and its tag > 950.

Change the skip conditions to match your images removal requirements.

#!/usr/bin/env bash


CheckTag(){
    Name=$1
    Tag=$2

    Skip=0
    if [[ "${Tag}" == "latest" ]]; then
        Skip=1
    fi
    if [[ "${Tag}" -ge "950" ]]; then
        Skip=1
    fi
    if [[ "${Skip}" == "1" ]]; then
        echo "skip ${Name} ${Tag}"
    else
        echo "delete ${Name} ${Tag}"
        Sha=$(curl -v -s -H "Accept: application/vnd.docker.distribution.manifest.v2+json" -X GET http://127.0.0.1:5000/v2/${Name}/manifests/${Tag} 2>&1 | grep Docker-Content-Digest | awk '{print ($3)}')
        Sha="${Sha/$'\r'/}"
        curl -H "Accept: application/vnd.docker.distribution.manifest.v2+json" -X DELETE "http://127.0.0.1:5000/v2/${Name}/manifests/${Sha}"
        return
    fi
}

ScanRepository(){
    Name=$1
    echo "Repository ${Name}"
    curl -s http://127.0.0.1:5000/v2/${Name}/tags/list | jq '.tags[]' |
    while IFS=$"\n" read -r line; do
        line="${line%\"}"
        line="${line#\"}"
        CheckTag $Name $line
    done
}


JqPath=$(which jq)
if [[ "x${JqPath}" == "x" ]]; then
  echo "Couldn't find jq executable."
  exit 2
fi

curl -s http://127.0.0.1:5000/v2/_catalog?n=10000 | jq '.repositories[]' |
while IFS=$"\n" read -r line; do
    line="${line%\"}"
    line="${line#\"}"
    ScanRepository $line
done

Once the cleanup is done, run the docker garbage collector on the docker registry container:

docker exec -it docker-registry bin/registry garbage-collect /etc/docker/registry/config.yml

This would run for several minutes, and then would delete the images from the disk.

Monday, June 29, 2020

Secure connection to Kafka from a GoLang client

In this post we will review how to create a secure kafka connection from GO.

See also the post Kafka Producer and Consumer in GO.

When trying to connect to a secure Apache Kafka server, you will usually receive 2 files:

client.keystore.jks
client.trustsotre.jks

These files should be converted to PEM files.

Use the following script to create the PEM files:

server.cer.pem
client.cer.pem
client.key.pem


keytool -importkeystore \
    -srckeystore input/client.truststore.jks \
    -destkeystore output/server.p12 \
    -deststoretype PKCS12 \
    -srcstorepass "jks-pass" \
    -deststorepass "topsecret"

openssl pkcs12 -in output/server.p12 -nokeys -out output/server.cer.pem -password pass:topsecret

keytool -importkeystore \
    -srckeystore input/client.keystore.jks \
    -destkeystore output/client.p12 \
    -deststoretype PKCS12 \
    -srcstorepass "jks-pass" \
    -deststorepass "topsecret"

openssl pkcs12 -in output/client.p12 -nokeys -out output/client.cer.pem -password pass:topsecret

openssl pkcs12 -in output/client.p12 -nodes -nocerts -out output/client.key.pem -password pass:topsecret

To use a secure connection from a GO client, use the following code:

package consumer

import (
   "crypto/tls"
   "crypto/x509"
   "fmt"
   "github.com/Shopify/sarama"
   "io/ioutil"
)

func connect() {
   config := sarama.NewConfig()
   config.Version = sarama.V1_1_1_0
   config.Consumer.Return.Errors = true
   tlsConfig := newTLSConfig()

   config.Net.TLS.Enable = true
   config.Net.TLS.Config = tlsConfig
   config.Net.SASL.Enable = true
   config.Net.SASL.Mechanism = sarama.SASLTypePlaintext
   config.Net.SASL.User = "my-user"
   config.Net.SASL.Password = "my-password"

   syncProducer, err := sarama.NewSyncProducer([]string{"127.0.0.1:30010"}, nil)
   if err != nil {
      panic(err)
   }

   fmt.Printf("sync producer created: %v", syncProducer)
}

func newTLSConfig() *tls.Config {
   tlsConfig := tls.Config{}

   cert, err := tls.LoadX509KeyPair("client.cer.pem", "client.key.pem")
   if err != nil {
      panic(err)
   }
   tlsConfig.Certificates = []tls.Certificate{cert}

   caCert, err := ioutil.ReadFile("server.cer.pem")
   if err != nil {
      panic(err)
   }
   caCertPool := x509.NewCertPool()
   caCertPool.AppendCertsFromPEM(caCert)
   tlsConfig.RootCAs = caCertPool
   tlsConfig.InsecureSkipVerify = true

   tlsConfig.BuildNameToCertificate()
   return &tlsConfig
}

Kafka Producer and Consumer in GO

In this post we will review how to create an Apache Kafka producer and consumer in GO.

To access kafka we will use Shopify's Sarama library.

See also the post Deploy Apache Kafka on Kubernetes, as well as the post Secure connection to Kafka from a GoLang client.

We will use a simple application that starts a producer and a consumer:

package main

import "kafka/players"

func main() {
   go players.Producer()
   
   consumer:= players.Consumer{}
   consumer.Consume()
}

We will use consts used by both the producer and the consumer:

package players

const KafkaServer = "127.0.0.1:30010"
const KafkaTopic = "my-topic"

The producer code is very strait forward, it sends messages to kafka in loop:

package players

import (
   "github.com/Shopify/sarama"
   "time"
)

func Producer() {
   syncProducer, err := sarama.NewSyncProducer([]string{KafkaServer}, nil)
   if err != nil {
      panic(err)
   }

   for {
      msg := &sarama.ProducerMessage{
         Topic: KafkaTopic,
         Value: sarama.ByteEncoder("Hello World " + time.Now().Format(time.RFC3339)),
      }

      _, _, err = syncProducer.SendMessage(msg)
      if err != nil {
         panic(err)
      }

      time.Sleep(time.Second)
   }
}

The consumer is a bit more complex, as it needs to recover from a broker crash.

package players

import (
   "context"
   "fmt"
   "github.com/Shopify/sarama"
   "time"
)

type Consumer struct {
}

func (c *Consumer) Consume() {
   config := sarama.NewConfig()
   config.Version = sarama.V2_4_0_0
   group, err := sarama.NewConsumerGroup([]string{KafkaServer}, "my-group", config)
   if err != nil {
      panic(err)
   }

   go func() {
      for err := range group.Errors() {
         panic(err)
      }
   }()

   func() {
      ctx := context.Background()
      for {
         topics := []string{KafkaTopic}
         err := group.Consume(ctx, topics, c)
         if err != nil {
            fmt.Printf("kafka consume failed: %v, sleeping and retry in a moment\n", err)
            time.Sleep(time.Second)
         }
      }
   }()
}

func (c *Consumer) Setup(_ sarama.ConsumerGroupSession) error {
   return nil
}

func (c *Consumer) Cleanup(_ sarama.ConsumerGroupSession) error {
   return nil
}

func (c *Consumer) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
   for msg := range claim.Messages() {
      fmt.Printf("consumed a message: %v\n", string(msg.Value))
      sess.MarkMessage(msg, "")
   }
   return nil
}

And that's all, the output from the application is:

consumed a message: Hello World 2020-06-30T08:28:10+03:00
consumed a message: Hello World 2020-06-30T08:28:18+03:00
consumed a message: Hello World 2020-06-30T08:28:14+03:00
consumed a message: Hello World 2020-06-30T08:28:19+03:00
consumed a message: Hello World 2020-06-30T08:28:16+03:00
consumed a message: Hello World 2020-06-30T08:28:15+03:00
consumed a message: Hello World 2020-06-30T08:28:20+03:00

Final Notes

In this post we have created a simple kafka producer & consumer application in GO.

Notice that the message sent from the producer is a simple text, but it could also be a JSON based on marshal of a structure.

Wednesday, June 24, 2020

Deploy Apache Kafka on Kubernetes

In this post we will review the steps required to deploy Apache Kafka on kubernetes.

From the Apache Kafka site:

Kafka® is used for building real-time data pipelines and streaming apps. It is horizontally scalable, fault-tolerant, wicked fast, and runs in production in thousands of companies.

Notice:

To use Kafka on kubernetes, start by deploy Apache ZooKeeper, which is used by Kafka to manage the cluster brokers, and the leader election. See my previous post: Deploy Apache Zookeeper on Kubernetes.

Once the Zookeeper is deployed, we will create the following kubernetes resources:

ConfigMap
Headless Service
Exposed Service
StatefulSet
Init container

1. The ConfigMap

The ConfigMap includes two files:

The logger configuration
The Kafka server.properties

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config
data:
  log4j.properties: |-
    # Unspecified loggers and loggers with additivity=true output to server.log and stdout
    # Note that INFO only applies to unspecified loggers, the log level of the child logger is used otherwise
    log4j.rootLogger=INFO, stdout
    
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    log4j.appender.kafkaAppender=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
    log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
    log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
    log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
    log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
    log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
    log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
    log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
    log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
    log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
    log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
    log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
    log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
    
    # Change the two lines below to adjust ZK client logging
    log4j.logger.org.I0Itec.zkclient.ZkClient=INFO
    log4j.logger.org.apache.zookeeper=INFO
    
    # Change the two lines below to adjust the general broker logging level (output to server.log and stdout)
    log4j.logger.kafka=INFO
    log4j.logger.org.apache.kafka=INFO
    
    # Change to DEBUG or TRACE to enable request logging
    log4j.logger.kafka.request.logger=WARN, requestAppender
    log4j.additivity.kafka.request.logger=false
    
    # Uncomment the lines below and change log4j.logger.kafka.network.RequestChannel$ to TRACE for additional output
    # related to the handling of requests
    #log4j.logger.kafka.network.Processor=TRACE, requestAppender
    #log4j.logger.kafka.server.KafkaApis=TRACE, requestAppender
    #log4j.additivity.kafka.server.KafkaApis=false
    log4j.logger.kafka.network.RequestChannel$=WARN, requestAppender
    log4j.additivity.kafka.network.RequestChannel$=false
    
    log4j.logger.kafka.controller=TRACE, controllerAppender
    log4j.additivity.kafka.controller=false
    
    log4j.logger.kafka.log.LogCleaner=INFO, cleanerAppender
    log4j.additivity.kafka.log.LogCleaner=false
    
    log4j.logger.state.change.logger=TRACE, stateChangeAppender
    log4j.additivity.state.change.logger=false
    
    # Change to DEBUG to enable audit log for the authorizer
    log4j.logger.kafka.authorizer.logger=WARN, authorizerAppender
    log4j.additivity.kafka.authorizer.logger=false
    
  server.properties: |-
    log.dirs=/var/lib/kafka/data/topics
    num.partitions=12
    default.replication.factor=3
    min.insync.replicas=2
    auto.create.topics.enable=false
    broker.rack=rack1
    listeners=PLAINTEXT://:9092,OUTSIDE://:9094
    listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,OUTSIDE:PLAINTEXT
    inter.broker.listener.name=PLAINTEXT
    offsets.retention.minutes=10080
    log.retention.hours=-1
    zookeeper.connect=zookeeper-exposed-service:80

2. The Headless Service

The Kafka headless service is used by the init container to update the published FQDN of the broker.

apiVersion: v1
kind: Service
metadata:
  name: kafka-internal-service
spec:
  selector:
    configid: kafka-container
  type: ClusterIP
  clusterIP: None
  publishNotReadyAddresses: true
  ports:
      - port: 9092

3. The Exposed Service

The Kafka service exposes the API for the clients.

The client connection is as follows:

The client connects to the exposed service
The client reaches randomly (by kubernetes service) to one of the Kafka brokers
The broker returns the published FQDN of the selected Kafka broker. This uses the following FQDN: <POD_NAME>.<HEADLESS SERVICE NAME>
The client directly connects to the selected broker

apiVersion: v1
kind: Service
metadata:
  name: kafka-service
spec:
  selector:
    configid: kafka-container
  ports:
      - port: 9092

3. The StatefulSet

The StatefulSet includes the configuration of the Kafka broker nodes.

It includes an init container to update the kafka configuration.

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka-statefulset
spec:
  serviceName: kafka-internal-service
  selector:
    matchLabels:
      configid: kafka-container
  replicas: 3
  template:
    metadata:
      labels:
        configid: kafka-container        
    spec:
      terminationGracePeriodSeconds: 30
      initContainers:
        - name: init
          imagePullPolicy: IfNotPresent
          image: my-registry/kafka-init/dev:latest
          volumeMounts:
            - name: configmap
              mountPath: /etc/kafka-configmap
            - name: config
              mountPath: /etc/kafka
            - name: extensions
              mountPath: /opt/kafka/libs/extensions
      containers:
        - name: broker
          image: solsson/kafka:2.4.1@sha256:79761e15919b4fe9857ec00313c9df799918ad0340b684c0163ab7035907bb5a
          env:
            - name: CLASSPATH
              value: /opt/kafka/libs/extensions/*
            - name: KAFKA_LOG4J_OPTS
              value: -Dlog4j.configuration=file:/etc/kafka/log4j.properties
            - name: JMX_PORT
              value: "5555"
          command:
            - ./bin/kafka-server-start.sh
            - /etc/kafka/server.properties
          lifecycle:
            preStop:
              exec:
                command: ["sh", "-ce", "kill -s TERM 1; while $(kill -0 1 2>/dev/null); do sleep 1; done"]
          readinessProbe:
            tcpSocket:
              port: 9092
            timeoutSeconds: 1
          volumeMounts:
            - name: config
              mountPath: /etc/kafka
            - name: data
              mountPath: /var/lib/kafka/data
            - name: extensions
              mountPath: /opt/kafka/libs/extensions
      volumes:
        - name: configmap
          configMap:
            name: kafka-config
        - name: config
          emptyDir: {}
        - name: extensions
          emptyDir: {}
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes: [ "ReadWriteOnce" ]
        storageClassName: "hostpath"
        resources:
          requests:
            storage: 500Mi

The Init Container

The init container updates the server.properties file with the FQDN of the pod.

The Dockerfile is:

FROM ubuntu:18.04
COPY files /
ENTRYPOINT /entrypoint.sh

and the entrypoint script is:

#!/bin/bash

cp /etc/kafka-configmap/* /etc/kafka/

KAFKA_BROKER_ID=${HOSTNAME##*-}

serverName="kafka-statefulset-${KAFKA_BROKER_ID}.kafka-internal-service"
sed -i "s/#init#broker.id/broker.id=${KAFKA_BROKER_ID}/" /etc/kafka/server.properties
sed -i "s/#init#advertised.listeners/advertised.listeners=PLAINTEXT:\\/\\/${serverName}:9092/" /etc/kafka/server.properties

Final Notes

In this post we have reviewed Kafka deployment on kubernetes.

Notice that we did not get into configuration the Kafka itself for your application need.

You will probably need to update the server.properties for you needs.

For example, to run a single replica of Kafka, you will need to update the server properties with:

default.replication.factor=1
min.insync.replicas=1
auto.create.topics.enable=true

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

Liked this post? Leave a comment...

Full Blog TOC

Full Blog Table Of Content with Keywords Available HERE

Wednesday, July 8, 2020

Analyze Redis Memory usage

Final Notes

Saturday, July 4, 2020

Scale your application using HPA on GKE

Metrics-Server

HPA

GKE Cluster Autoscaler

Final Notes

Wednesday, July 1, 2020

Using SSL based Ingress on Google Kubernetes Engine

Delete images from a Private Docker Registry

Monday, June 29, 2020

Secure connection to Kafka from a GoLang client

Kafka Producer and Consumer in GO

Final Notes

Wednesday, June 24, 2020

Deploy Apache Kafka on Kubernetes

1. The ConfigMap

2. The Headless Service

3. The Exposed Service

3. The StatefulSet

The Init Container

Final Notes