Run Logisland stream within Kubernetes : stage 1¶
This is the begining of a multiple part series of tutorials going through setting up a scalable Apache log indexation to Elasticsearch in kubernetes. This guide will bring you to a fully functionnal Kubernetes logisland setup.
Part 1 - Setting up Elasticsearch Part 2 - Setting up Kibana Part 3 - Setting up a three-node Zookeeper cluster Part 4 - Setting up a three-node Kafka cluster Part 5 - Setting up Logisland
Kafka and Zookeeper can be manually scaled up at any time by altering and re-applying configuration. Kubernetes also provides features for autoscaling, read more about auto scaling Kubernetes Pods should that be a requirement.
sources¶
0 - Pre-requisites & initial setup¶
First of all you’ll need a Kubernetes cluster or a minikube cluster (https://kubernetes.io/docs/tasks/tools/install-minikube/ ). For the first option I would highly recommend to follow the Hello Minikube tutorial for those who don’t have any background with Kubernetes. This will help to get minikube and kubectl commands installed. (Minikube is the local development Kubernetes environment and kubectl is the command line interface used to interact with Kubernetes cluster).
Shaving the Yak!¶
One or two commands that used in this post will be mac or linux specific. Reference this guide to get more up to date and OS specific commands. Once you’ve got the tools all installed, you can now follow along these steps to create a single node Elasticsearch cluster.
If you are using Minikube, make sure that its started properly by running this command
- for mac:
minikube start --vm-driver=hyperkit
- for linux (use virtualbox by default, so you have to install it) :
minikube start
Now set the Minikube context. The context is what determines which cluster kubectl is interacting with.
kubectl config use-context minikube
Verify that kubectl is configured to communicate with your cluster:
kubectl cluster-info
To view the nodes in the cluster, run
kubectl get nodes
Kubernetes Dashboard¶
Minikube includes the kubernetes dashboard as an addon which you can enable.
minikube addons list
returns
- default-storageclass: enabled
- coredns: disabled
- kube-dns: enabled
- ingress: disabled
- registry: disabled
- registry-creds: disabled
- addon-manager: enabled
- dashboard: enabled
- storage-provisioner: enabled
- heapster: disabled
- efk: disabled
You can enable an addon using:
minikube addons enable dashboard
You can then open the dashboard with command
minikube dashboard
Please note that on some virtual environments (like VirtualBox) the minikube VM may start with too few resources (you should allocate at least 4 CPUs and 6Go RAM)
Kubernetes setup¶
The best you can do is to follow the official guides to get the following tools up and running.
The Kubernetes command-line tool, kubectl, allows you to run commands against Kubernetes clusters. You can use kubectl to deploy applications, inspect and manage cluster resources, and view logs. setup kubectl
Minikube, a tool that runs a single-node Kubernetes cluster in a virtual machine on your laptop is the easiest way to start with. setup minikube
Note
Deciding where to run Kubernetes depends on what resources you have available and how much flexibility you need. You can run Kubernetes almost anywhere, from your laptop to VMs on a cloud provider to a rack of bare metal servers. You can also set up a fully-managed cluster by running a single command or craft your own customized cluster on your bare metal servers. setup kubernetes
Namespace¶
In this guide, I use the fictional namespace logisland. You can create this namespace in your cluster or use your own.
Create the file namespace.yml:
apiVersion: v1
kind: Namespace
metadata:
name: logisland
Apply the configuration:
kubectl create -f ./namespace.yml
If you wish to use your own namespace for this Kafka installation, be sure to replace logisland in the configurations below.
Persistent volumes¶
In Kubernetes, managing storage is a distinct problem from managing compute. The PersistentVolume subsystem provides an API for users and administrators that abstracts details of how storage is provided from how it is consumed. To do this we introduce two new API resources: PersistentVolume and PersistentVolumeClaim.
A PersistentVolume (PV) is a piece of storage in the cluster that has been provisioned by an administrator. It is a resource in the cluster just like a node is a cluster resource. PVs are volume plugins like Volumes, but have a lifecycle independent of any individual pod that uses the PV. This API object captures the details of the implementation of the storage,
be that NFS, iSCSI, or a cloud-provider-specific storage system.
A PersistentVolumeClaim (PVC) is a request for storage by a user. It is similar to a pod. Pods consume node resources and PVCs consume PV resources. Pods can request specific levels of resources (CPU and Memory). Claims can request specific size and access modes (e.g., can be mounted once read/write or many times read-only).
Create the local folders where you want to store your files (change this to wherever you want to store data on your nodes) :
mkdir /tmp/data
Create the file pv-volume.yml
kind: PersistentVolume
apiVersion: v1
metadata:
name: datadir
labels:
app: kafka
type: local
namespace: logisland
spec:
storageClassName: manual
capacity:
storage: 10Gi
accessModes:
- ReadWriteOnce
hostPath:
path: "/tmp/data"
Apply the configuration:
kubectl create -f ./pv-volume.yml
Configuration maps¶
We will need a few configuration variables in our setup to bind containers together and define some environment variables. The first config map is specific to loggen tool which is a wrapped python program that sends fake generated apache logs to a given Kafka topic at a specified rate. The second one is a set of settings that will be used by the logisland job in order to configure itself. We’ll go into deeper details in the last section of this post.
Create the file config-maps.yml with the following content
apiVersion: v1
kind: ConfigMap
metadata:
name: special-config
namespace: logisland
data:
loggen.sleep: '0.2'
loggen.num: '0'
loggen.topic: logisland_raw
---
apiVersion: v1
kind: ConfigMap
metadata:
name: logisland-config
namespace: logisland
data:
kafka.brokers: kafka:9092
zk.quorum: zookeeper:2181
es.hosts: elasticsearch:9300
es.cluster.name: es-logisland
Apply the configuration:
kubectl create -f ./config-maps.yml
1 - Setting up Elasticsearch cluster on Kubernetes¶
Single Node Elasticsearch Cluster¶
Create the file elasticsearch-service.yml:
apiVersion: v1
kind: Service
metadata:
name: elasticsearch
namespace: logisland
labels:
component: elasticsearch
spec:
type: ClusterIP
selector:
component: elasticsearch
ports:
- name: http
port: 9200
protocol: TCP
- name: tcp
port: 9300
protocol: TCP
Apply the configuration:
kubectl create -f ./elasticsearch-service.yml
Create the file elasticsearch-deployment.yml:
apiVersion: apps/v1beta2
kind: Deployment
metadata:
name: elasticsearch
namespace: logisland
spec:
selector:
matchLabels:
component: elasticsearch
template:
metadata:
labels:
component: elasticsearch
spec:
containers:
- name: elasticsearch
image: docker.elastic.co/elasticsearch/elasticsearch:5.4.3
env:
- name: discovery.type
value: single-node
- name: cluster.name
value: "es-logisland"
- name: xpack.security.enabled
value: "false"
ports:
- containerPort: 9200
name: http
protocol: TCP
- containerPort: 9300
name: tcp
protocol: TCP
Apply the configuration:
kubectl create -f ./elasticsearch-deployment.yml
Expose the cluster¶
We can verify that the cluster is running by looking at the logs. But, let’s check if elasticsearch api is responding first.
In a seperate shell window, excute the following to start a proxy into Kubernetes cluster.
kubectl -n logisland port-forward svc/elasticsearch 9200:9200
Now, back in the other window, lets execute a curl command to get the response from the pod via the proxy.
curl http://localhost:9200
Outputs:
{
"name" : "19SlwE4",
"cluster_name" : "es-logisland",
"cluster_uuid" : "ef41SIbWRHmSDoDhcFA9WA",
"version" : {
"number" : "5.4.3",
"build_hash" : "eed30a8",
"build_date" : "2017-06-22T00:34:03.743Z",
"build_snapshot" : false,
"lucene_version" : "6.5.1"
},
"tagline" : "You Know, for Search"
}
Great, everything is working.
2 - Setup Kibana¶
Let’s try to setup kibana pointing to our elasticsearch single node cluster.
Create the file kibana-service.yml:
apiVersion: v1
kind: Service
metadata:
name: kibana
namespace: logisland
labels:
component: kibana
spec:
type: NodePort
selector:
component: kibana
ports:
- name: http
port: 5601
targetPort: 5601
nodePort: 30123
protocol: TCP
Apply the configuration:
kubectl create -f ./kibana-service.yml
Create the file kibana-deployment.yml:
apiVersion: apps/v1beta2
kind: Deployment
metadata:
name: kibana
namespace: logisland
spec:
selector:
matchLabels:
component: kibana
template:
metadata:
labels:
component: kibana
spec:
containers:
- name: kibana
image: docker.elastic.co/kibana/kibana:5.4.3
env:
- name: ELASTICSEARCH_URL
value: http://elasticsearch:9200
- name: XPACK_SECURITY_ENABLED
value: "true"
ports:
- containerPort: 5601
name: http
protocol: TCP
Apply the configuration:
kubectl create -f ./kibana-deployment.yml
To access kibana through your localhost forward the port
kubectl -n logisland port-forward svc/kibana 5601:5601
3 - Setting up Zookeeper¶
Kafka requires Zookeeper for maintaining configuration information, naming, providing distributed synchronization, and providing group services to coordinate its nodes.
Zookeeper Headless Service¶
Kubernetes Services are persistent and provide a stable and reliable way to connect to Pods.
Setup a Kubernetes Service named kafka-zookeeper in namespace logisland. The kafka-zookeeper service resolves the domain name kafka-zookeeper to an internal ClusterIP. The automatically assigned ClusterIP uses Kubernetes internal proxy to load balance calls to any Pods found from the configured selector, in this case, app: kafka-zookeeper.
After setting up the kafka-zookeeper Service, a DNS lookup from within the cluster may produce a result similar to the following:
# nslookup kafka-zookeeper
Server: 10.96.0.10
Address: 10.96.0.10#53
Name: kafka-zookeeper.logisland.svc.cluster.local
Address: 10.103.184.71
In the example above, 10.103.184.71 is the internal IP address of the ** kafka-zookeeper* service itself and proxies calls to one of the Zookeeper Pods it finds labeled app: kafka-zookeeper. At this point, no Pods are available until added further down. However, the service finds them when they become active.
Create the file zookeeper-service.yml:
apiVersion: v1
kind: Service
metadata:
name: kafka-zookeeper
namespace: logisland
spec:
ports:
- name: client
port: 2181
protocol: TCP
targetPort: client
selector:
app: kafka-zookeeper
sessionAffinity: None
type: ClusterIP
Apply the configuration:
kubectl create -f ./zookeeper-service.yml
Zookeeper Headless Service¶
A Kubernetes Headless Service does not resolve to a single IP; instead, Headless Services returns the IP addresses of any Pods found by their selector, in this case, Pods labeled app: kafka-zookeeper.
Once Pods labeled app: kafka-zookeeper are running, this Headless Service returns the results of an in-cluster DNS lookup similar to the following:
# nslookup kafka-zookeeper
Server: 10.96.0.10
Address: 10.96.0.10#53
Name: kafka-zookeeper-headless.logisland.svc.cluster.local
Address: 192.168.108.150
Name: kafka-zookeeper-headless.logisland.svc.cluster.local
Address: 192.168.108.181
Name: kafka-zookeeper-headless.logisland.svc.cluster.local
Address: 192.168.108.132
In the example above, the Kubernetes Service kafka-zookeeper-headless returned the internal IP addresses of three individual Pods.
At this point, no Pod IPs can be returned until the Pods are configured in the StatefulSet further down.
Create the file zookeeper-service-headless.yml:
apiVersion: v1
kind: Service
metadata:
name: kafka-zookeeper-headless
namespace: logisland
spec:
#clusterIP: None
ports:
- name: client
port: 2181
protocol: TCP
targetPort: 2181
- name: election
port: 3888
protocol: TCP
targetPort: 3888
- name: server
port: 2888
protocol: TCP
targetPort: 2888
selector:
app: kafka-zookeeper
sessionAffinity: None
type: ClusterIP
Apply the configuration:
kubectl create -f ./zookeeper-service-headless.yml
Zookeeper StatefulSet¶
Kubernetes StatefulSets offer stable and unique network identifiers, persistent storage, ordered deployments, scaling, deletion, termination, and automated rolling updates.
Unique network identifiers and persistent storage are essential for stateful cluster nodes in systems like Zookeeper and Kafka. While it seems strange to have a coordinator like Zookeeper running inside a Kubernetes cluster sitting on its own coordinator Etcd, it makes sense since these systems are built to run independently. Kubernetes supports running services like Zookeeper and Kafka with features like headless services and stateful sets which demonstrates the flexibility of Kubernetes as both a microservices platform and a type of virtual infrastructure.
The following configuration creates three kafka-zookeeper Pods, kafka-zookeeper-0, kafka-zookeeper-1, kafka-zookeeper-2 and can be scaled to as many as desired. Ensure that the number of specified replicas matches the environment variable ZK_REPLICAS specified in the container spec.
Pods in this StatefulSet run the Zookeeper Docker image gcr.io/google_samples/k8szk:v3, which is a sample image provided by Google for testing GKE, it is recommended to use custom and maintained Zookeeper image once you are familiar with this setup.
Create the file zookeeper-statefulset.yml:
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka-zookeeper
namespace: logisland
spec:
podManagementPolicy: OrderedReady
replicas: 3
revisionHistoryLimit: 1
selector:
matchLabels:
app: kafka-zookeeper
serviceName: kafka-zookeeper-headless
template:
metadata:
labels:
app: kafka-zookeeper
spec:
containers:
- command:
- /bin/bash
- -xec
- zkGenConfig.sh && exec zkServer.sh start-foreground
env:
- name: ZK_REPLICAS
value: "3"
- name: JMXAUTH
value: "false"
- name: JMXDISABLE
value: "false"
- name: JMXPORT
value: "1099"
- name: JMXSSL
value: "false"
- name: ZK_CLIENT_PORT
value: "2181"
- name: ZK_ELECTION_PORT
value: "3888"
- name: ZK_HEAP_SIZE
value: 1G
- name: ZK_INIT_LIMIT
value: "5"
- name: ZK_LOG_LEVEL
value: INFO
- name: ZK_MAX_CLIENT_CNXNS
value: "60"
- name: ZK_MAX_SESSION_TIMEOUT
value: "40000"
- name: ZK_MIN_SESSION_TIMEOUT
value: "4000"
- name: ZK_PURGE_INTERVAL
value: "0"
- name: ZK_SERVER_PORT
value: "2888"
- name: ZK_SNAP_RETAIN_COUNT
value: "3"
- name: ZK_SYNC_LIMIT
value: "10"
- name: ZK_TICK_TIME
value: "2000"
image: gcr.io/google_samples/k8szk:v3
imagePullPolicy: IfNotPresent
livenessProbe:
exec:
command:
- zkOk.sh
failureThreshold: 3
initialDelaySeconds: 20
periodSeconds: 10
successThreshold: 1
timeoutSeconds: 1
name: zookeeper
ports:
- containerPort: 2181
name: client
protocol: TCP
- containerPort: 3888
name: election
protocol: TCP
- containerPort: 2888
name: server
protocol: TCP
readinessProbe:
exec:
command:
- zkOk.sh
failureThreshold: 3
initialDelaySeconds: 20
periodSeconds: 10
successThreshold: 1
timeoutSeconds: 1
resources: {}
terminationMessagePath: /dev/termination-log
terminationMessagePolicy: File
volumeMounts:
- mountPath: /var/lib/zookeeper
name: data
dnsPolicy: ClusterFirst
restartPolicy: Always
schedulerName: default-scheduler
securityContext:
fsGroup: 1000
runAsUser: 1000
terminationGracePeriodSeconds: 30
volumes:
- emptyDir: {}
name: data
updateStrategy:
type: OnDelete
Apply the configuration:
kubectl create -f ./zookeeper-statefulset.yml
Zookeeper PodDisruptionBudget¶
PodDisruptionBudget can help keep the Zookeeper service stable during Kubernetes administrative events such as draining a node or updating Pods.
From the official documentation for PDB (PodDisruptionBudget):
A PDB specifies the number of replicas that an application can tolerate having, relative to how many it is intended to have. For example, a Deployment which has a .spec.replicas: 5 is supposed to have 5 pods at any given time. If its PDB allows for there to be 4 at a time, then the Eviction API will allow voluntary disruption of one, but not two pods, at a time.
The configuration below tells Kubernetes that we can only tolerate one of our Zookeeper Pods down at any given time. maxUnavailable may be set to a higher number if we increase the number of Zookeeper Pods in the StatefulSet.
Create the file zookeeper-disruptionbudget.yml:
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
labels:
app: kafka-zookeeper
name: kafka-zookeeper
namespace: logisland
spec:
maxUnavailable: 1
selector:
matchLabels:
app: kafka-zookeeper
Apply the configuration:
kubectl create -f ./zookeeper-disruptionbudget.yml
4 - Setting up Kafka¶
Once Zookeeper is up and running we have satisfied the requirements for Kafka. Kafka is set up in a similar configuration to Zookeeper, utilizing a Service, Headless Service and a StatefulSet.
Kafka Service¶
The following Service provides a persistent internal Cluster IP address that proxies and load balance requests to Kafka Pods found with the label app: kafka and exposing the port 9092.
Create the file kafka-service.yml:
apiVersion: v1 kind: Service metadata:
name: kafka namespace: logisland
- spec:
- ports:
- name: broker port: 9092 protocol: TCP targetPort: kafka
- selector:
- app: kafka
sessionAffinity: None type: ClusterIP
Apply the configuration:
kubectl create -f ./kafka-service.yml
Kafka Headless Service¶
The following Headless Service provides a list of Pods and their internal IPs found with the label app: kafka and exposing the port 9092. The previously created Service: kafka always returns a persistent IP assigned at the creation time of the Service. The following kafka-headless services return the domain names and IP address of individual Pods and are liable to change as Pods are added, removed or updated.
Create the file kafka-service-headless.yml:
apiVersion: v1 kind: Service metadata:
name: kafka-headless namespace: logisland
- spec:
#clusterIP: None ports:
- name: broker port: 9092 protocol: TCP targetPort: 9092
- selector:
- app: kafka
sessionAffinity: None type: ClusterIP
Apply the configuration:
kubectl create -f ./kafka-service-headless.yml
Kafka StatefulSet¶
The following StatefulSet deploys Pods running the confluentinc/cp-kafka:4.1.2-2 Docker image from Confluent.
Each pod is assigned 1Gi of storage using the rook-block storage class. See Rook.io for more information on file, block, and object storage services for cloud-native environments.
Create the file kafka-statefulset.yml:
apiVersion: apps/v1 kind: StatefulSet metadata:
- labels:
- app: kafka
name: kafka namespace: logisland
- spec:
podManagementPolicy: OrderedReady replicas: 3 revisionHistoryLimit: 1 selector:
- matchLabels:
- app: kafka
serviceName: kafka-headless template:
- metadata:
- labels:
- app: kafka
- spec:
- containers:
- command:
sh
-exc
unset KAFKA_PORT && export KAFKA_BROKER_ID=${HOSTNAME##*-} && export KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${POD_IP}:9092 && exec /etc/confluent/docker/run
- env:
name: POD_IP valueFrom:
- fieldRef:
apiVersion: v1 fieldPath: status.podIP
name: KAFKA_HEAP_OPTS value: -Xmx1G -Xms1G
name: KAFKA_ZOOKEEPER_CONNECT value: kafka-zookeeper:2181
# value: 10.105.213.202:2181 # value: ${KAFKA_ZOOKEEPER_SERVICE_HOST}:2181
- name: KAFKA_LOG_DIRS value: /opt/kafka/data/logs
- name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR value: “3”
- name: KAFKA_JMX_PORT value: “5555”
image: confluentinc/cp-kafka:4.1.2-2 imagePullPolicy: IfNotPresent livenessProbe:
- exec:
- command:
- sh
- -ec
- /usr/bin/jps | /bin/grep -q SupportedKafka
failureThreshold: 3 initialDelaySeconds: 30 periodSeconds: 10 successThreshold: 1 timeoutSeconds: 5
name: kafka-broker ports:
- containerPort: 9092 name: kafka protocol: TCP
- readinessProbe:
failureThreshold: 3 initialDelaySeconds: 30 periodSeconds: 10 successThreshold: 1 tcpSocket:
port: kafka
timeoutSeconds: 5
resources: {} terminationMessagePath: /dev/termination-log terminationMessagePolicy: File volumeMounts:
- mountPath: /opt/kafka/data name: datadir-claim
dnsPolicy: ClusterFirst restartPolicy: Always schedulerName: default-scheduler securityContext: {} terminationGracePeriodSeconds: 60
- updateStrategy:
- type: OnDelete
- volumeClaimTemplates:
- metadata:
name: datadir-claim
- spec:
#storageClassName: “standard” # storageClassName: rook-block accessModes:
- ReadWriteOnce
- resources:
- requests:
storage: 1Gi
Apply the configuration:
kubectl create -f ./kafka-statefulset.yml
Kafka Test Pod¶
Add a test Pod to help explore and debug your new Kafka cluster. The Confluent Docker image confluentinc/cp-kafka:4.1.2-2 used for the test Pod is the same as our nodes from the StatefulSet and contain useful command in the /usr/bin/ folder.
Create the file kafka-test.yml:
apiVersion: v1 kind: Pod metadata:
name: kafka-test-client namespace: logisland
- spec:
- containers:
- command:
- sh
- -c
- exec tail -f /dev/null
image: confluentinc/cp-kafka:4.1.2-2 imagePullPolicy: IfNotPresent name: kafka resources: {} terminationMessagePath: /dev/termination-log terminationMessagePolicy: File
Apply the configuration:
kubectl create -f ./kafka-test.yml
5 - Working with Kafka¶
If you have deployed the kafka-test-client pod from the configuration above, the following commands should get you started with some basic operations:
Create Topic¶
kubectl -n logisland exec kafka-test-client -- \
/usr/bin/kafka-topics --zookeeper kafka-zookeeper:2181 \
--topic logisland_raw --create --partitions 3 --replication-factor 1
List Topics¶
kubectl -n logisland exec kafka-test-client -- \
/usr/bin/kafka-topics –zookeeper kafka-zookeeper:2181 –list
Sending logs to Kafka¶
This script generates a boatload of fake apache logs very quickly. Its useful for generating fake workloads for data ingest and/or analytics applications. It can write log lines to console, to log files or directly to gzip files. Or to kafka … It utilizes the excellent Faker library to generate realistic ip’s, URI’s etc.
Create the file loggen-deployment.yml:
apiVersion: v1 kind: Pod metadata:
name: loggen-job namespace: logisland
- spec:
- containers:
name: loggen image: hurence/loggen imagePullPolicy: IfNotPresent env:
name: LOGGEN_SLEEP valueFrom:
- configMapKeyRef:
name: special-config key: loggen.sleep
name: LOGGEN_NUM valueFrom:
- configMapKeyRef:
name: special-config key: loggen.num
name: LOGGEN_KAFKA valueFrom:
- configMapKeyRef:
name: logisland-config key: kafka.brokers
name: LOGGEN_KAFKA_TOPIC valueFrom:
- configMapKeyRef:
name: special-config key: loggen.topic
Apply the configuration:
kubectl create -f ./loggen-deployment.yml
Listen on a Topic¶
make sure some fake apache logs are flowing through kafka topic
kubectl -n logisland exec -ti kafka-test-client -- \
/usr/bin/kafka-console-consumer --bootstrap-server kafka:9092 \
--topic logisland_raw --from-beginning
6 - Setup logisland¶
It’s now time time to dive into log mining. We’ll setup a 3 instances logisland stream that will handle apache logs parsing (coming from loggen script) as a ReplicaSet
Create the file logisland-deployment.yml:
apiVersion: apps/v1beta2
kind: ReplicaSet
metadata:
name: logisland-job
namespace: logisland
spec:
replicas: 3
selector:
matchLabels:
app: logisland-job
template:
metadata:
labels:
app: logisland-job
spec:
containers:
- name: logisland
image: hurence/logisland-job
imagePullPolicy: IfNotPresent
command: ["/opt/logisland/bin/logisland.sh"]
args: ["--standalone", "--conf", "/opt/logisland/conf/index-apache-logs-plainjava.yml"]
env:
- name: ES_CLUSTER_NAME
valueFrom:
configMapKeyRef:
name: logisland-config
key: es.cluster.name
- name: KAFKA_BROKERS
valueFrom:
configMapKeyRef:
name: logisland-config
key: kafka.brokers
- name: ES_HOSTS
valueFrom:
configMapKeyRef:
name: logisland-config
key: es.hosts
Apply the configuration:
kubectl create -f ./logisland-deployment.yml
run the following command to see events parsed by logisland flowing through the output topic
kubectl -n logisland exec -ti kafka-test-client – /usr/bin/kafka-console-consumer –bootstrap-server kafka:9092 –topic logisland_events
check that logs are correctly stored into elasticsearch
kubectl -n logisland exec -ti kafka-test-client – curl http://elasticsearch:9200/logisland.*/_search?pretty=1