Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #935 (2020-05-29 11:18:37)

Fixed issue (#934)
The issue is related to [#934](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/issues/934). It allows to properly exposing the prometheus exporter to a SeriveMonitor or to Prometheus to scrape metrics. Up to now, we can only see the metrics generated locally but not exposed outside the pod running the JMX exporter.


Patch:
diff --git a/docs/who-is-using.md b/docs/who-is-using.md
index 321983c..56a6544 100644
--- a/docs/who-is-using.md
+++ b/docs/who-is-using.md
@@ -20,3 +20,4 @@
 | Cyren | @avnerl | Evaluation | Data pipelines
 | Shell (Agile Hub) | @TomLous | Production | Data pipelines
 | Nielsen Identity Engine | @roitvt | Evaluation | Data pipelines
+| [Data Mechanics](https://www.datamechanics.co)  | @jrj-d | Production | Managed Spark Platform |
diff --git a/pkg/config/constants.go b/pkg/config/constants.go
index 46116d6..e96998f 100644
--- a/pkg/config/constants.go
+++ b/pkg/config/constants.go
@@ -270,6 +270,9 @@ rules:
 // DefaultPrometheusJavaAgentPort is the default port used by the Prometheus JMX exporter.
 const DefaultPrometheusJavaAgentPort int32 = 8090
 
+// DefaultPrometheusPortProtocol is the default protocol used by the Prometheus JMX exporter.
+const DefaultPrometheusPortProtocol string = "TCP"
+
 const (
 	// SparkDriverContainerName is name of driver container in spark driver pod
 	SparkDriverContainerName = "spark-kubernetes-driver"
diff --git a/pkg/webhook/patch.go b/pkg/webhook/patch.go
index b45cc7b..1f3ad82 100644
--- a/pkg/webhook/patch.go
+++ b/pkg/webhook/patch.go
@@ -53,7 +53,7 @@ func patchSparkPod(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOperat
 	patchOps = append(patchOps, addGeneralConfigMaps(pod, app)...)
 	patchOps = append(patchOps, addSparkConfigMap(pod, app)...)
 	patchOps = append(patchOps, addHadoopConfigMap(pod, app)...)
-	patchOps = append(patchOps, addPrometheusConfigMap(pod, app)...)
+	patchOps = append(patchOps, getPrometheusConfigPatches(pod, app)...)
 	patchOps = append(patchOps, addTolerations(pod, app)...)
 	patchOps = append(patchOps, addSidecarContainers(pod, app)...)
 	patchOps = append(patchOps, addInitContainers(pod, app)...)
@@ -342,7 +342,7 @@ func addGeneralConfigMaps(pod *corev1.Pod, app *v1beta2.SparkApplication) []patc
 	return patchOps
 }
 
-func addPrometheusConfigMap(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOperation {
+func getPrometheusConfigPatches(pod *corev1.Pod, app *v1beta2.SparkApplication) []patchOperation {
 	// Skip if Prometheus Monitoring is not enabled or an in-container ConfigFile is used,
 	// in which cases a Prometheus ConfigMap won't be created.
 	if !app.PrometheusMonitoringEnabled() || (app.HasMetricsPropertiesFile() && app.HasPrometheusConfigFile()) {
@@ -360,15 +360,51 @@ func addPrometheusConfigMap(pod *corev1.Pod, app *v1beta2.SparkApplication) []pa
 	name := config.GetPrometheusConfigMapName(app)
 	volumeName := name + "-vol"
 	mountPath := config.PrometheusConfigMapMountPath
+	port := config.DefaultPrometheusJavaAgentPort
+	if app.Spec.Monitoring.Prometheus.Port != nil {
+		port = *app.Spec.Monitoring.Prometheus.Port
+	}
+	protocol := config.DefaultPrometheusPortProtocol
 	patchOps = append(patchOps, addConfigMapVolume(pod, name, volumeName))
 	vmPatchOp := addConfigMapVolumeMount(pod, volumeName, mountPath)
 	if vmPatchOp == nil {
+		glog.Warningf("could not mount volume %s in path %s", volumeName, mountPath)
 		return nil
 	}
 	patchOps = append(patchOps, *vmPatchOp)
+	portPatchOp := addContainerPort(pod, port, protocol)
+	if portPatchOp == nil {
+		glog.Warningf("could not expose port %d to scrape metrics outside the pod", port)
+		return nil
+	}
+	patchOps = append(patchOps, *portPatchOp)
+
 	return patchOps
 }
 
+func addContainerPort(pod *corev1.Pod, port int32, protocol string) *patchOperation {
+	i := findContainer(pod)
+	if i < 0 {
+		glog.Warningf("not able to add containerPort %d as Spark container was not found in pod %s", port, pod.Name)
+		return nil
+	}
+
+	path := fmt.Sprintf("/spec/containers/%d/ports", i)
+	containerPort := corev1.ContainerPort{
+		ContainerPort: port,
+		Protocol:      corev1.Protocol(protocol),
+	}
+	var value interface{}
+	if len(pod.Spec.Containers[i].Ports) == 0 {
+		value = []corev1.ContainerPort{containerPort}
+	} else {
+		path += "/-"
+		value = containerPort
+	}
+	pod.Spec.Containers[i].Ports = append(pod.Spec.Containers[i].Ports, containerPort)
+	return &patchOperation{Op: "add", Path: path, Value: value}
+}
+
 func addConfigMapVolume(pod *corev1.Pod, configMapName string, configMapVolumeName string) patchOperation {
 	volume := corev1.Volume{
 		Name: configMapVolumeName,


Test patch:
diff --git a/pkg/webhook/patch_test.go b/pkg/webhook/patch_test.go
index ecd88f9..6184ece 100644
--- a/pkg/webhook/patch_test.go
+++ b/pkg/webhook/patch_test.go
@@ -498,6 +498,7 @@ func TestPatchSparkPod_HadoopConfigMap(t *testing.T) {
 }
 
 func TestPatchSparkPod_PrometheusConfigMaps(t *testing.T) {
+	var appPort int32 = 9999
 	app := &v1beta2.SparkApplication{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "spark-test",
@@ -505,7 +506,12 @@ func TestPatchSparkPod_PrometheusConfigMaps(t *testing.T) {
 		},
 		Spec: v1beta2.SparkApplicationSpec{
 			Monitoring: &v1beta2.MonitoringSpec{
-				Prometheus:          &v1beta2.PrometheusSpec{},
+				Prometheus: &v1beta2.PrometheusSpec{
+					JmxExporterJar: "",
+					Port:           &appPort,
+					ConfigFile:     nil,
+					Configuration:  nil,
+				},
 				ExposeDriverMetrics: true,
 			},
 		},
@@ -536,6 +542,7 @@ func TestPatchSparkPod_PrometheusConfigMaps(t *testing.T) {
 
 	expectedConfigMapName := config.GetPrometheusConfigMapName(app)
 	expectedVolumeName := expectedConfigMapName + "-vol"
+	expectedContainerPort := *app.Spec.Monitoring.Prometheus.Port
 	assert.Equal(t, 1, len(modifiedPod.Spec.Volumes))
 	assert.Equal(t, expectedVolumeName, modifiedPod.Spec.Volumes[0].Name)
 	assert.True(t, modifiedPod.Spec.Volumes[0].ConfigMap != nil)
@@ -543,6 +550,8 @@ func TestPatchSparkPod_PrometheusConfigMaps(t *testing.T) {
 	assert.Equal(t, 1, len(modifiedPod.Spec.Containers[0].VolumeMounts))
 	assert.Equal(t, expectedVolumeName, modifiedPod.Spec.Containers[0].VolumeMounts[0].Name)
 	assert.Equal(t, config.PrometheusConfigMapMountPath, modifiedPod.Spec.Containers[0].VolumeMounts[0].MountPath)
+	assert.Equal(t, expectedContainerPort, modifiedPod.Spec.Containers[0].Ports[0].ContainerPort)
+	assert.Equal(t, corev1.Protocol(config.DefaultPrometheusPortProtocol), modifiedPod.Spec.Containers[0].Ports[0].Protocol)
 }
 
 func TestPatchSparkPod_Tolerations(t *testing.T) {