Merge pull request #47 from newrelic-experimental/feat/minor-spark-me…

…tric-updates-and-docs

feat: add job/stage durations, expose stage/task ids, add spark docs

internal/spark/receiver.go

@@ -337,6 +337,26 @@ func collectSparkAppJobMetrics(
 
 		// Write all the things.
 
+		if job.CompletionTime != "" {
+			jobDuration, err := calcDateDifferenceMillis(
+				job.SubmissionTime,
+				job.CompletionTime,
+			)
+			if err != nil {
+				log.Warnf("could not calculate job duration: %v", err)
+
+				continue
+			}
+
+			writeGauge(
+				metricPrefix,
+				"app.job.duration",
+				jobDuration,
+				attrs,
+				writer,
+			)
+		}
+
 		writeGauge(
 			metricPrefix,
 			"app.job.indices.completed",
@@ -355,7 +375,7 @@ func collectSparkAppJobMetrics(
 			writer,
 		)
 
-		attrs["sparkAppStageStatus"] = "completed"
+		attrs["sparkAppStageStatus"] = "complete"
 
 		writeGauge(
 			metricPrefix,
@@ -397,7 +417,7 @@ func collectSparkAppJobMetrics(
 			writer,
 		)
 
-		attrs["sparkAppTaskStatus"] = "completed"
+		attrs["sparkAppTaskStatus"] = "complete"
 
 		writeGauge(
 			metricPrefix,
@@ -521,8 +541,8 @@ func collectSparkAppStageMetrics(
 		attrs["sparkAppStageStatus"] = stageStatus
 		// @TODO: The attributes below may cause high cardinality. Further
 		// investigation is needed.
-		//attrs["sparkAppStageId"] = stage.StageId
-		//attrs["sparkAppStageAttemptId"] = stage.AttemptId
+		attrs["sparkAppStageId"] = stage.StageId
+		attrs["sparkAppStageAttemptId"] = stage.AttemptId
 		//attrs["sparkAppStageSchedulingPool"] = stage.SchedulingPool
 		//attrs["sparkAppStageResourceProfileId"] = stage.ResourceProfileId
 
@@ -540,6 +560,26 @@ func collectSparkAppStageMetrics(
 
 		// Write all the things.
 
+		if stage.CompletionTime != "" {
+			stageDuration, err := calcDateDifferenceMillis(
+				stage.FirstTaskLaunchedTime,
+				stage.CompletionTime,
+			)
+			if err != nil {
+				log.Warnf("could not calculate stage duration: %v", err)
+
+				continue
+			}
+
+			writeGauge(
+				metricPrefix,
+				"app.stage.duration",
+				stageDuration,
+				attrs,
+				writer,
+			)
+		}
+
 		writeGauge(
 			metricPrefix,
 			"app.stage.peakNettyDirectMemory",
@@ -831,8 +871,8 @@ func collectSparkAppStageMetrics(
 		}
 
 		writePeakMemoryMetrics(
-			metricPrefix + "app.stage.memory.peak.",
-			&stage.PeakMemoryMetrics,
+			metricPrefix + "app.stage.executor.memory.peak.",
+			&stage.PeakExecutorMetrics,
 			attrs,
 			writer,
 		)
@@ -972,8 +1012,8 @@ func writeStageTaskMetrics(
 	taskMetricAttrs["sparkAppTaskSpeculative"] = task.Speculative
 	// @TODO: The attributes below may cause high cardinality. Further
 	// investigation is needed.
-	//attrs["sparkAppTaskId"] = task.TaskId
-	//attrs["sparkAppTaskAttempt"] = task.Attempt
+	taskMetricAttrs["sparkAppTaskId"] = task.TaskId
+	taskMetricAttrs["sparkAppTaskAttempt"] = task.Attempt
 	//attrs["sparkAppTaskPartitionId"] = task.PartitionId
 
 	writeGauge(

internal/spark/spark.go

@@ -250,7 +250,7 @@ type SparkStage struct {
 	Tasks								map[string]SparkTask `json:"tasks"`
 	ExecutorSummary						map[string]SparkExecutorSummary `json:"executorSummary"`
 	ResourceProfileId   				int 			`json:"resourceProfileId"`
-	PeakMemoryMetrics SparkExecutorPeakMemoryMetrics 	`json:"peakMemoryMetrics"`
+	PeakExecutorMetrics SparkExecutorPeakMemoryMetrics 	`json:"peakExecutorMetrics"`
 	ShuffleMergersCount  				int  			`json:"shuffleMergersCount"`
 }
 

internal/spark/util.go

@@ -0,0 +1,26 @@
+package spark
+
+import (
+	"fmt"
+	"time"
+)
+
+const (
+	RFC3339Milli = "2006-01-02T15:04:05.000GMT"
+)
+
+func calcDateDifferenceMillis(startDate string, endDate string) (
+	int64, error,
+) {
+	start, err := time.Parse(RFC3339Milli, startDate)
+	if err != nil {
+		return 0, fmt.Errorf("invalid start date: %s", startDate)
+	}
+
+	end, err := time.Parse(RFC3339Milli, endDate)
+	if err != nil {
+		return 0, fmt.Errorf("invalid end date: %s", endDate)
+	}
+
+	return end.Sub(start).Milliseconds(), nil
+}