Merge pull request #477 from liyinan926/master

Fixed mounting of Prometheus configuration ConfigMap

Author: liyinan926

pkg/apis/sparkoperator.k8s.io/v1beta1/types.go

@@ -495,3 +495,25 @@ type PrometheusSpec struct {
 	// Configuration has no effect if ConfigFile is set.
 	Configuration *string `json:"configuration,omitempty"`
 }
+
+// PrometheusMonitoringEnabled returns if Prometheus monitoring is enabled or not.
+func (s *SparkApplication) PrometheusMonitoringEnabled() bool {
+	return s.Spec.Monitoring != nil && s.Spec.Monitoring.Prometheus != nil
+}
+
+// HasPrometheusConfigFile returns if Prometheus monitoring uses a configruation file in the container.
+func (s *SparkApplication) HasPrometheusConfigFile() bool {
+	return s.PrometheusMonitoringEnabled() &&
+		s.Spec.Monitoring.Prometheus.ConfigFile != nil &&
+		*s.Spec.Monitoring.Prometheus.ConfigFile != ""
+}
+
+// ExposeDriverMetrics returns if driver metrics should be exposed.
+func (s *SparkApplication) ExposeDriverMetrics() bool {
+	return s.Spec.Monitoring != nil && s.Spec.Monitoring.ExposeDriverMetrics
+}
+
+// ExposeExecutorMetrics returns if executor metrics should be exposed.
+func (s *SparkApplication) ExposeExecutorMetrics() bool {
+	return s.Spec.Monitoring != nil && s.Spec.Monitoring.ExposeExecutorMetrics
+}

pkg/config/config.go

@@ -51,3 +51,8 @@ func GetExecutorEnvVarConfOptions(app *v1beta1.SparkApplication) []string {
 	}
 	return envVarConfOptions
 }
+
+// GetPrometheusConfigMapName returns the name of the ConfigMap for Prometheus configuration.
+func GetPrometheusConfigMapName(app *v1beta1.SparkApplication) string {
+	return fmt.Sprintf("%s-%s", app.Name, PrometheusConfigMapNameSuffix)
+}

pkg/config/constants.go

@@ -174,11 +174,20 @@ const (
 	HadoopDelegationTokenFileName = "hadoop.token"
 )
 
+const (
+	// PrometheusConfigMapNameSuffix is the name prefix of the Prometheus ConfigMap.
+	PrometheusConfigMapNameSuffix = "prom-conf"
+	// PrometheusConfigMapMountPath is the mount path of the Prometheus ConfigMap.
+	PrometheusConfigMapMountPath = "/etc/metrics/conf"
+)
+
+// DefaultMetricsProperties is the default content of metrics.properties.
 const DefaultMetricsProperties = `
 *.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink
 driver.source.jvm.class=org.apache.spark.metrics.source.JvmSource
 executor.source.jvm.class=org.apache.spark.metrics.source.JvmSource`
 
+// DefaultPrometheusConfiguration is the default content of prometheus.yaml.
 const DefaultPrometheusConfiguration = `
 lowercaseOutputName: true
 attrNameSnakeCase: true
@@ -268,4 +277,6 @@ rules:
       app_id: "$2"
       executor_id: "$3"
 `
+
+// DefaultPrometheusJavaAgentPort is the default port used by the Prometheus JMX exporter.
 const DefaultPrometheusJavaAgentPort int32 = 8090

pkg/controller/sparkapplication/controller.go

@@ -545,16 +545,14 @@ func hasRetryIntervalPassed(retryInterval *int64, attemptsDone int32, lastEventT
 
 // submitSparkApplication creates a new submission for the given SparkApplication and submits it using spark-submit.
 func (c *Controller) submitSparkApplication(app *v1beta1.SparkApplication) *v1beta1.SparkApplication {
-	// Make a copy since configPrometheusMonitoring may update app.Spec which causes an onUpdate callback.
-	appToSubmit := app.DeepCopy()
-	if appToSubmit.Spec.Monitoring != nil && appToSubmit.Spec.Monitoring.Prometheus != nil {
-		if err := configPrometheusMonitoring(appToSubmit, c.kubeClient); err != nil {
+	if app.PrometheusMonitoringEnabled() {
+		if err := configPrometheusMonitoring(app, c.kubeClient); err != nil {
 			glog.Error(err)
 		}
 	}
 
 	submissionID := uuid.New().String()
-	submissionCmdArgs, err := buildSubmissionCommandArgs(appToSubmit, submissionID)
+	submissionCmdArgs, err := buildSubmissionCommandArgs(app, submissionID)
 	if err != nil {
 		app.Status = v1beta1.SparkApplicationStatus{
 			AppState: v1beta1.ApplicationState{
@@ -568,7 +566,7 @@ func (c *Controller) submitSparkApplication(app *v1beta1.SparkApplication) *v1be
 	}
 
 	// Try submitting the application by running spark-submit.
-	submitted, err := runSparkSubmit(newSubmission(submissionCmdArgs, appToSubmit))
+	submitted, err := runSparkSubmit(newSubmission(submissionCmdArgs, app))
 	if err != nil {
 		app.Status = v1beta1.SparkApplicationStatus{
 			AppState: v1beta1.ApplicationState{

pkg/controller/sparkapplication/monitoring_config.go

@@ -18,6 +18,7 @@ package sparkapplication
 
 import (
 	"fmt"
+
 	"github.com/golang/glog"
 	corev1 "k8s.io/api/core/v1"
 	apiErrors "k8s.io/apimachinery/pkg/api/errors"
@@ -30,13 +31,11 @@ import (
 )
 
 const (
-	metricsPropertiesKey          = "metrics.properties"
-	prometheusConfigKey           = "prometheus.yaml"
-	prometheusConfigMapNameSuffix = "prom-conf"
-	prometheusConfigMapMountPath  = "/etc/metrics/conf"
-	prometheusScrapeAnnotation    = "prometheus.io/scrape"
-	prometheusPortAnnotation      = "prometheus.io/port"
-	prometheusPathAnnotation      = "prometheus.io/path"
+	metricsPropertiesKey       = "metrics.properties"
+	prometheusConfigKey        = "prometheus.yaml"
+	prometheusScrapeAnnotation = "prometheus.io/scrape"
+	prometheusPortAnnotation   = "prometheus.io/port"
+	prometheusPathAnnotation   = "prometheus.io/path"
 )
 
 func configPrometheusMonitoring(app *v1beta1.SparkApplication, kubeClient clientset.Interface) error {
@@ -45,22 +44,18 @@ func configPrometheusMonitoring(app *v1beta1.SparkApplication, kubeClient client
 		port = *app.Spec.Monitoring.Prometheus.Port
 	}
 
-	var configFile string
-	if app.Spec.Monitoring.Prometheus.ConfigFile != nil {
-		configFile = *app.Spec.Monitoring.Prometheus.ConfigFile
-	}
 	var javaOption string
-	if configFile != "" {
+	if app.HasPrometheusConfigFile() {
+		configFile := *app.Spec.Monitoring.Prometheus.ConfigFile
 		glog.V(2).Infof("Overriding the default Prometheus configuration with config file %s in the Spark image.", configFile)
 		javaOption = fmt.Sprintf("-javaagent:%s=%d:%s", app.Spec.Monitoring.Prometheus.JmxExporterJar,
 			port, configFile)
 	} else {
 		glog.V(2).Infof("Using the default Prometheus configuration.")
-		prometheusConfigMapName := fmt.Sprintf("%s-%s", app.Name, prometheusConfigMapNameSuffix)
-		configMap := buildPrometheusConfigMap(app, prometheusConfigMapName)
+		configMapName := config.GetPrometheusConfigMapName(app)
+		configMap := buildPrometheusConfigMap(app, configMapName)
 		retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-			cm, err := kubeClient.CoreV1().ConfigMaps(app.Namespace).Get(prometheusConfigMapName, metav1.GetOptions{})
-
+			cm, err := kubeClient.CoreV1().ConfigMaps(app.Namespace).Get(configMapName, metav1.GetOptions{})
 			if apiErrors.IsNotFound(err) {
 				_, createErr := kubeClient.CoreV1().ConfigMaps(app.Namespace).Create(configMap)
 				return createErr
@@ -75,29 +70,20 @@ func configPrometheusMonitoring(app *v1beta1.SparkApplication, kubeClient client
 		})
 
 		if retryErr != nil {
-			return fmt.Errorf("failed to apply %s in namespace %s: %v", prometheusConfigMapName, app.Namespace, retryErr)
+			return fmt.Errorf("failed to apply %s in namespace %s: %v", configMapName, app.Namespace, retryErr)
 		}
 
-		javaOption = fmt.Sprintf("-javaagent:%s=%d:%s/%s", app.Spec.Monitoring.Prometheus.JmxExporterJar,
-			port, prometheusConfigMapMountPath, prometheusConfigKey)
-
-		if app.Spec.Monitoring.ExposeDriverMetrics {
-			app.Spec.Driver.ConfigMaps = append(app.Spec.Driver.ConfigMaps, v1beta1.NamePath{
-				Name: prometheusConfigMapName,
-				Path: prometheusConfigMapMountPath,
-			})
-		}
-		if app.Spec.Monitoring.ExposeExecutorMetrics {
-			app.Spec.Executor.ConfigMaps = append(app.Spec.Executor.ConfigMaps, v1beta1.NamePath{
-				Name: prometheusConfigMapName,
-				Path: prometheusConfigMapMountPath,
-			})
-		}
+		javaOption = fmt.Sprintf(
+			"-javaagent:%s=%d:%s/%s",
+			app.Spec.Monitoring.Prometheus.JmxExporterJar,
+			port,
+			config.PrometheusConfigMapMountPath,
+			prometheusConfigKey)
 	}
 
 	/* work around for push gateway issue: https://github.com/prometheus/pushgateway/issues/97 */
 	metricNamespace := fmt.Sprintf("%s.%s", app.Namespace, app.Name)
-	metricConf := fmt.Sprintf("%s/%s", prometheusConfigMapMountPath, metricsPropertiesKey)
+	metricConf := fmt.Sprintf("%s/%s", config.PrometheusConfigMapMountPath, metricsPropertiesKey)
 	if app.Spec.SparkConf == nil {
 		app.Spec.SparkConf = make(map[string]string)
 	}

pkg/controller/sparkapplication/monitoring_config_test.go

@@ -44,7 +44,7 @@ func TestConfigPrometheusMonitoring(t *testing.T) {
 			t.Errorf("failed to configure Prometheus monitoring: %v", err)
 		}
 
-		configMapName := fmt.Sprintf("%s-%s", test.app.Name, prometheusConfigMapNameSuffix)
+		configMapName := config.GetPrometheusConfigMapName(test.app)
 		configMap, err := fakeClient.CoreV1().ConfigMaps(test.app.Namespace).Get(configMapName, metav1.GetOptions{})
 		if err != nil {
 			t.Errorf("failed to get ConfigMap %s: %v", configMapName, err)
@@ -67,10 +67,6 @@ func TestConfigPrometheusMonitoring(t *testing.T) {
 		}
 
 		if test.app.Spec.Monitoring.ExposeDriverMetrics {
-			if len(test.app.Spec.Driver.ConfigMaps) != 1 {
-				t.Errorf("expected %d driver ConfigMaps got %d", 1, len(test.app.Spec.Driver.ConfigMaps))
-			}
-
 			if len(test.app.Spec.Driver.Annotations) != 3 {
 				t.Errorf("expected %d driver annotations got %d", 3, len(test.app.Spec.Driver.Annotations))
 			}
@@ -84,10 +80,6 @@ func TestConfigPrometheusMonitoring(t *testing.T) {
 		}
 
 		if test.app.Spec.Monitoring.ExposeExecutorMetrics {
-			if len(test.app.Spec.Executor.ConfigMaps) != 1 {
-				t.Errorf("expected %d driver ConfigMaps got %d", 1, len(test.app.Spec.Executor.ConfigMaps))
-			}
-
 			if len(test.app.Spec.Executor.Annotations) != 3 {
 				t.Errorf("expected %d driver annotations got %d", 3, len(test.app.Spec.Executor.Annotations))
 			}

pkg/webhook/patch.go

@@ -52,6 +52,7 @@ func patchSparkPod(pod *corev1.Pod, app *v1beta1.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, addTolerations(pod, app)...)
 
 	if pod.Spec.SchedulerName == "" {
@@ -67,6 +68,7 @@ func patchSparkPod(pod *corev1.Pod, app *v1beta1.SparkApplication) []patchOperat
 			patchOps = append(patchOps, *op)
 		}
 	}
+
 	if pod.Spec.SecurityContext == nil {
 		op := addSecurityContext(pod, app)
 		if op != nil {
@@ -219,6 +221,29 @@ func addGeneralConfigMaps(pod *corev1.Pod, app *v1beta1.SparkApplication) []patc
 	return patchOps
 }
 
+func addPrometheusConfigMap(pod *corev1.Pod, app *v1beta1.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.HasPrometheusConfigFile() {
+		return nil
+	}
+
+	if util.IsDriverPod(pod) && !app.ExposeDriverMetrics() {
+		return nil
+	}
+	if util.IsExecutorPod(pod) && !app.ExposeExecutorMetrics() {
+		return nil
+	}
+
+	var patchOps []patchOperation
+	name := config.GetPrometheusConfigMapName(app)
+	volumeName := name + "-vol"
+	mountPath := config.PrometheusConfigMapMountPath
+	patchOps = append(patchOps, addConfigMapVolume(pod, name, volumeName))
+	patchOps = append(patchOps, addConfigMapVolumeMount(pod, volumeName, mountPath))
+	return patchOps
+}
+
 func addConfigMapVolume(pod *corev1.Pod, configMapName string, configMapVolumeName string) patchOperation {
 	volume := corev1.Volume{
 		Name: configMapVolumeName,

pkg/webhook/patch_test.go

@@ -339,6 +339,54 @@ func TestPatchSparkPod_HadoopConfigMap(t *testing.T) {
 	assert.Equal(t, config.DefaultHadoopConfDir, modifiedPod.Spec.Containers[0].Env[0].Value)
 }
 
+func TestPatchSparkPod_PrometheusConfigMaps(t *testing.T) {
+	app := &v1beta1.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-test",
+			UID:  "spark-test-1",
+		},
+		Spec: v1beta1.SparkApplicationSpec{
+			Monitoring: &v1beta1.MonitoringSpec{
+				Prometheus:          &v1beta1.PrometheusSpec{},
+				ExposeDriverMetrics: true,
+			},
+		},
+	}
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-driver",
+			Labels: map[string]string{
+				config.SparkRoleLabel:               config.SparkDriverRole,
+				config.LaunchedBySparkOperatorLabel: "true",
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  sparkDriverContainerName,
+					Image: "spark-driver:latest",
+				},
+			},
+		},
+	}
+
+	modifiedPod, err := getModifiedPod(pod, app)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedConfigMapName := config.GetPrometheusConfigMapName(app)
+	expectedVolumeName := expectedConfigMapName + "-vol"
+	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)
+	assert.Equal(t, expectedConfigMapName, modifiedPod.Spec.Volumes[0].ConfigMap.Name)
+	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)
+}
+
 func TestPatchSparkPod_Tolerations(t *testing.T) {
 	app := &v1beta1.SparkApplication{
 		ObjectMeta: metav1.ObjectMeta{