Refactored how status update is logged (kubeflow#859)

pkg/controller/sparkapplication/controller.go

@@ -587,7 +587,6 @@ func (c *Controller) syncSparkApplication(key string) error {
 	}
 
 	if appToUpdate != nil {
-		glog.V(2).Infof("Trying to update SparkApplication %s/%s, from: [%v] to [%v]", app.Namespace, app.Name, app.Status, appToUpdate.Status)
 		err = c.updateStatusAndExportMetrics(app, appToUpdate)
 		if err != nil {
 			glog.Errorf("failed to update SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
@@ -760,20 +759,33 @@ func (c *Controller) updateApplicationStatusWithRetries(
 // updateStatusAndExportMetrics updates the status of the SparkApplication and export the metrics.
 func (c *Controller) updateStatusAndExportMetrics(oldApp, newApp *v1beta2.SparkApplication) error {
 	// Skip update if nothing changed.
-	if equality.Semantic.DeepEqual(oldApp, newApp) {
+	if equality.Semantic.DeepEqual(oldApp.Status, newApp.Status) {
 		return nil
 	}
 
+	oldStatusJSON, err := printStatus(&oldApp.Status)
+	if err != nil {
+		return err
+	}
+	newStatusJSON, err := printStatus(&newApp.Status)
+	if err != nil {
+		return err
+	}
+
+	glog.V(2).Infof("Update the status of SparkApplication %s/%s from:\n%s\nto:\n%s", newApp.Namespace, newApp.Name, oldStatusJSON, newStatusJSON)
 	updatedApp, err := c.updateApplicationStatusWithRetries(oldApp, func(status *v1beta2.SparkApplicationStatus) {
 		*status = newApp.Status
 	})
+	if err != nil {
+		return err
+	}
 
 	// Export metrics if the update was successful.
-	if err == nil && c.metrics != nil {
+	if c.metrics != nil {
 		c.metrics.exportMetrics(oldApp, updatedApp)
 	}
 
-	return err
+	return nil
 }
 
 func (c *Controller) getSparkApplication(namespace string, name string) (*v1beta2.SparkApplication, error) {

pkg/controller/sparkapplication/sparkapp_metrics.go

@@ -164,8 +164,6 @@ func (sm *sparkAppMetrics) registerMetrics() {
 
 func (sm *sparkAppMetrics) exportMetrics(oldApp, newApp *v1beta2.SparkApplication) {
 	metricLabels := fetchMetricLabels(newApp, sm.labels)
-	glog.V(2).Infof("Exporting metrics for %s; old status: %v new status: %v", newApp.Name,
-		oldApp.Status, newApp.Status)
 
 	oldState := oldApp.Status.AppState.State
 	newState := newApp.Status.AppState.State
@@ -183,7 +181,6 @@ func (sm *sparkAppMetrics) exportMetrics(oldApp, newApp *v1beta2.SparkApplicatio
 		case v1beta2.SucceedingState:
 			if !newApp.Status.LastSubmissionAttemptTime.Time.IsZero() && !newApp.Status.TerminationTime.Time.IsZero() {
 				d := newApp.Status.TerminationTime.Time.Sub(newApp.Status.LastSubmissionAttemptTime.Time)
-
 				if m, err := sm.sparkAppSuccessExecutionTime.GetMetricWith(metricLabels); err != nil {
 					glog.Errorf("Error while exporting metrics: %v", err)
 				} else {
@@ -232,19 +229,20 @@ func (sm *sparkAppMetrics) exportMetrics(oldApp, newApp *v1beta2.SparkApplicatio
 		}
 	}
 
+	oldExecutorStates := oldApp.Status.ExecutorState
 	// Potential Executor status updates
 	for executor, newExecState := range newApp.Status.ExecutorState {
 		switch newExecState {
 		case v1beta2.ExecutorRunningState:
-			if oldApp.Status.ExecutorState[executor] != newExecState {
+			if oldExecutorStates[executor] != newExecState {
 				glog.V(2).Infof("Exporting Metrics for Executor %s. OldState: %v NewState: %v", executor,
-					oldApp.Status.ExecutorState[executor], newExecState)
+					oldExecutorStates[executor], newExecState)
 				sm.sparkAppExecutorRunningCount.Inc(metricLabels)
 			}
 		case v1beta2.ExecutorCompletedState:
-			if oldApp.Status.ExecutorState[executor] != newExecState {
+			if oldExecutorStates[executor] != newExecState {
 				glog.V(2).Infof("Exporting Metrics for Executor %s. OldState: %v NewState: %v", executor,
-					oldApp.Status.ExecutorState[executor], newExecState)
+					oldExecutorStates[executor], newExecState)
 				sm.sparkAppExecutorRunningCount.Dec(metricLabels)
 				if m, err := sm.sparkAppExecutorSuccessCount.GetMetricWith(metricLabels); err != nil {
 					glog.Errorf("Error while exporting metrics: %v", err)
@@ -253,9 +251,9 @@ func (sm *sparkAppMetrics) exportMetrics(oldApp, newApp *v1beta2.SparkApplicatio
 				}
 			}
 		case v1beta2.ExecutorFailedState:
-			if oldApp.Status.ExecutorState[executor] != newExecState {
+			if oldExecutorStates[executor] != newExecState {
 				glog.V(2).Infof("Exporting Metrics for Executor %s. OldState: %v NewState: %v", executor,
-					oldApp.Status.ExecutorState[executor], newExecState)
+					oldExecutorStates[executor], newExecState)
 				sm.sparkAppExecutorRunningCount.Dec(metricLabels)
 				if m, err := sm.sparkAppExecutorFailureCount.GetMetricWith(metricLabels); err != nil {
 					glog.Errorf("Error while exporting metrics: %v", err)
@@ -286,20 +284,18 @@ func (sm *sparkAppMetrics) exportJobStartLatencyMetrics(app *v1beta2.SparkApplic
 }
 
 func fetchMetricLabels(app *v1beta2.SparkApplication, labels []string) map[string]string {
-	specLabels := app.Labels
-
-	// Transform spec labels since our labels names might be not same as specLabels if we removed invalid characters.
-	validSpecLabels := make(map[string]string)
-	for labelKey, v := range specLabels {
+	// Convert app labels into ones that can be used as metric labels.
+	validLabels := make(map[string]string)
+	for labelKey, v := range app.Labels {
 		newKey := util.CreateValidMetricNameLabel("", labelKey)
-		validSpecLabels[newKey] = v
+		validLabels[newKey] = v
 	}
 
 	metricLabels := make(map[string]string)
 	for _, label := range labels {
-		if value, ok := validSpecLabels[label]; ok {
+		if value, ok := validLabels[label]; ok {
 			metricLabels[label] = value
-		} else if label == "namespace" { // if the "namespace" label is in the metrics config, use it
+		} else if label == "namespace" { // If the "namespace" label is in the metrics config, use it.
 			metricLabels[label] = app.Namespace
 		} else {
 			metricLabels[label] = "Unknown"

pkg/controller/sparkapplication/sparkapp_util.go

@@ -17,7 +17,9 @@ limitations under the License.
 package sparkapplication
 
 import (
+	"encoding/json"
 	"fmt"
+
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/kubernetes/pkg/apis/policy"
 
@@ -133,3 +135,11 @@ func getVolumeFSType(v v1.Volume) (policy.FSType, error) {
 
 	return "", fmt.Errorf("unknown volume type for volume: %#v", v)
 }
+
+func printStatus(status *v1beta2.SparkApplicationStatus) (string, error) {
+	marshalled, err := json.MarshalIndent(status, "", "  ")
+	if err != nil {
+		return "", err
+	}
+	return string(marshalled), nil
+}

pkg/controller/sparkapplication/sparkapp_util_test.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2020 Google LLC
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package sparkapplication
+
+import (
+	"testing"
+
+	"github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2"
+)
+
+var expectedStatusString = `{
+  "sparkApplicationId": "test-app",
+  "submissionID": "test-app-submission",
+  "lastSubmissionAttemptTime": null,
+  "terminationTime": null,
+  "driverInfo": {},
+  "applicationState": {
+    "state": "COMPLETED"
+  },
+  "executorState": {
+    "executor-1": "COMPLETED"
+  }
+}`
+
+func TestPrintStatus(t *testing.T) {
+	status := &v1beta2.SparkApplicationStatus{
+		SparkApplicationID: "test-app",
+		SubmissionID:       "test-app-submission",
+		AppState: v1beta2.ApplicationState{
+			State: v1beta2.CompletedState,
+		},
+		ExecutorState: map[string]v1beta2.ExecutorState{
+			"executor-1": v1beta2.ExecutorCompletedState,
+		},
+	}
+
+	statusString, err := printStatus(status)
+	if err != nil {
+		t.Fail()
+	}
+
+	if statusString != expectedStatusString {
+		t.Errorf("status string\n %s is different from expected status string\n %s", statusString, expectedStatusString)
+	}
+}