Add monitoring docs and small fixes

Author: baluchicken

docs/monitoring.md

@@ -0,0 +1,88 @@
+This documentation is intended to guide you how to enable custom monitoring on operator installed Kafka cluster
+
+#### Using Helm for Prometheus:
+
+By default operator installs Kafka Pods with the following annotations, also it opens port 9020 in all brokers to enable scraping.
+
+```
+    "prometheus.io/scrape": "true"
+    "prometheus.io/port":   "9020"
+```
+
+Prometheus must be configured to recognize these annotations. The following example contains the required config.
+
+```
+# Example scrape config for pods
+#
+# The relabeling allows the actual pod scrape endpoint to be configured via the
+# following annotations:
+#
+# * `prometheus.io/scrape`: Only scrape pods that have a value of `true`
+# * `prometheus.io/path`: If the metrics path is not `/metrics` override this.
+# * `prometheus.io/port`: Scrape the pod on the indicated port instead of the default of `9102`.
+- job_name: 'kubernetes-pods'
+
+  kubernetes_sd_configs:
+    - role: pod
+
+  relabel_configs:
+    - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
+      action: keep
+      regex: true
+    - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path]
+      action: replace
+      target_label: __metrics_path__
+      regex: (.+)
+    - source_labels: [__address__, __meta_kubernetes_pod_annotation_prometheus_io_port]
+      action: replace
+      regex: ([^:]+)(?::\d+)?;(\d+)
+      replacement: $1:$2
+      target_label: __address__
+```
+
+Using the provided [CR](https://github.com/banzaicloud/kafka-operator/blob/master/config/samples/banzaicloud_v1alpha1_kafkacluster.yaml), the operator installs the official [jmx exporter](https://github.com/prometheus/jmx_exporter) for Prometheus.
+
+To change this behavior please modify the following lines in the end of the CR.
+
+```
+monitoringConfig:
+   jmxImage describes the used prometheus jmx exporter agent container
+    jmxImage: "banzaicloud/jmx-javaagent:0.12.0"
+   pathToJar describes the path to the jar file in the given image
+    pathToJar: "/opt/jmx_exporter/jmx_prometheus_javaagent-0.12.0.jar"
+   kafkaJMXExporterConfig describes jmx exporter config for Kafka
+    kafkaJMXExporterConfig: |
+     lowercaseOutputName: true
+```
+
+#### Using the ServiceMonitors:
+
+To use ServiceMonitors, we recommend to use kafka with unique service/broker instead of headless service.
+
+Configure the CR in a following way:
+
+```
+  # Specify if the cluster should use headlessService for Kafka or individual services
+  # using service/broker may come in handy in case of service mesh
+  headlessServiceEnabled: false
+```
+
+Disabling Headless service means the operator will set up Kafka with unique services per broker.
+
+Once you have a cluster up and running create as many ServiceMonitors as brokers.
+
+```
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  name: kafka-0
+spec:
+  selector:
+    matchLabels:
+      app: kafka
+      brokerId: "0"
+      kafka_cr: kafka
+  endpoints:
+  - port: metrics
+    interval: 10s
+```

pkg/resources/cruisecontrol/cruisecontrol.go

@@ -34,6 +34,7 @@ const (
 	pemFilesVolume         = "pem-files"
 	jmxVolumePath          = "/opt/jmx-exporter/"
 	jmxVolumeName          = "jmx-jar-data"
+	metricsPort            = 9020
 )
 
 var labelSelector = map[string]string{

pkg/resources/cruisecontrol/deployment.go

@@ -56,7 +56,7 @@ func (r *Reconciler) deployment(log logr.Logger) runtime.Object {
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels:      labelSelector,
-					Annotations: util.MonitoringAnnotations(),
+					Annotations: util.MonitoringAnnotations(metricsPort),
 				},
 				Spec: corev1.PodSpec{
 					TerminationGracePeriodSeconds: util.Int64Pointer(30),

pkg/resources/cruisecontrol/service.go

@@ -33,6 +33,12 @@ func (r *Reconciler) service(log logr.Logger) runtime.Object {
 					TargetPort: intstr.FromInt(8090),
 					Protocol:   corev1.ProtocolTCP,
 				},
+				{
+					Name:       "metrics",
+					Port:       metricsPort,
+					TargetPort: intstr.FromInt(metricsPort),
+					Protocol:   corev1.ProtocolTCP,
+				},
 			},
 		},
 	}

pkg/resources/kafka/kafka.go

@@ -52,6 +52,7 @@ const (
 	pemFilesVolume                = "pem-files"
 	jmxVolumePath                 = "/opt/jmx-exporter/"
 	jmxVolumeName                 = "jmx-jar-data"
+	metricsPort                   = 9020
 
 	//jaasConfig  = "jaas-config"
 	//scramSecret = "scram-secret"

pkg/resources/kafka/pod.go

@@ -68,7 +68,7 @@ func (r *Reconciler) pod(broker banzaicloudv1alpha1.BrokerConfig, pvcs []corev1.
 	}
 
 	pod := &corev1.Pod{
-		ObjectMeta: templates.ObjectMetaWithGeneratedNameAndAnnotations(r.KafkaCluster.Name, util.MergeLabels(labelsForKafka(r.KafkaCluster.Name), map[string]string{"brokerId": fmt.Sprintf("%d", broker.Id)}), util.MonitoringAnnotations(), r.KafkaCluster),
+		ObjectMeta: templates.ObjectMetaWithGeneratedNameAndAnnotations(r.KafkaCluster.Name, util.MergeLabels(labelsForKafka(r.KafkaCluster.Name), map[string]string{"brokerId": fmt.Sprintf("%d", broker.Id)}), util.MonitoringAnnotations(metricsPort), r.KafkaCluster),
 		Spec: corev1.PodSpec{
 			InitContainers: append(initContainers, []corev1.Container{
 				{

pkg/resources/kafka/service.go

@@ -40,9 +40,15 @@ func (r *Reconciler) service(broker banzaicloudv1alpha1.BrokerConfig, log logr.L
 			Protocol:   corev1.ProtocolTCP,
 		})
 	}
+	usedPorts = append(usedPorts, corev1.ServicePort{
+		Name:       "metrics",
+		Port:       metricsPort,
+		TargetPort: intstr.FromInt(metricsPort),
+		Protocol:   corev1.ProtocolTCP,
+	})
 
 	return &corev1.Service{
-		ObjectMeta: templates.ObjectMeta(fmt.Sprintf("%s-%d", r.KafkaCluster.Name, broker.Id), labelsForKafka(r.KafkaCluster.Name), r.KafkaCluster),
+		ObjectMeta: templates.ObjectMeta(fmt.Sprintf("%s-%d", r.KafkaCluster.Name, broker.Id), util.MergeLabels(labelsForKafka(r.KafkaCluster.Name), map[string]string{"brokerId": fmt.Sprintf("%d", broker.Id)}), r.KafkaCluster),
 		Spec: corev1.ServiceSpec{
 			Type:            corev1.ServiceTypeClusterIP,
 			SessionAffinity: corev1.ServiceAffinityNone,

pkg/resources/kafkamonitoring/configmap.go

@@ -26,7 +26,7 @@ import (
 func (r *Reconciler) configMap() runtime.Object {
 	configMap := &corev1.ConfigMap{
 		ObjectMeta: templates.ObjectMeta(fmt.Sprintf(BrokerJmxTemplate, r.KafkaCluster.Name), labelsForJmx(r.KafkaCluster.Name), r.KafkaCluster),
-		Data:       map[string]string{"config.yaml": r.KafkaCluster.Spec.MonitoringConfig.KafkaJMXExporterConfig},
+		Data:       map[string]string{"config.yaml": r.KafkaCluster.Spec.MonitoringConfig.GetKafkaJMXExporterConfig()},
 	}
 	return configMap
 }

pkg/util/util.go

@@ -60,10 +60,10 @@ func MergeLabels(l map[string]string, l2 map[string]string) map[string]string {
 }
 
 // MonitoringAnnotations returns specific prometheus annotations
-func MonitoringAnnotations() map[string]string {
+func MonitoringAnnotations(port int) map[string]string {
 	return map[string]string{
 		"prometheus.io/scrape": "true",
-		"prometheus.io/port":   "9020",
+		"prometheus.io/port":   string(port),
 	}
 }