better grouping of Apache Spark applications managed by spark-operator

auditor/jvm.go

@@ -21,10 +21,14 @@ func (a *appAuditor) jvm() {
 
 	for _, i := range a.app.Instances {
 		obsolete := i.IsObsolete()
+		succeeded := false
+		if i.Pod != nil {
+			succeeded = i.Pod.IsSucceeded()
+		}
 		for name, j := range i.Jvms {
 			fullName := name + "@" + i.Name
 
-			if !obsolete && !j.IsUp() {
+			if !obsolete && !succeeded && !j.IsUp() {
 				availabilityCheck.AddItem(fullName)
 			}
 			if !obsolete && j.SafepointTime.Last() > safepointCheck.Threshold {
@@ -45,7 +49,7 @@ func (a *appAuditor) jvm() {
 				safepointChart.AddSeries(fullName, j.SafepointTime)
 			}
 
-			if !obsolete && table != nil {
+			if !obsolete && !succeeded && table != nil {
 				name := model.NewTableCell(fullName)
 				status := model.NewTableCell().SetStatus(model.OK, "up")
 				if !j.IsUp() {

constructor/k8s.go

@@ -12,7 +12,7 @@ import (
 )
 
 var (
-	argoWorkflowNameSuffixRe = regexp.MustCompile(`-([a-z0-9]{5}|\d{10,})$`)
+	jobSuffixRe = regexp.MustCompile(`-([a-z0-9]{5}|\d{10,})$`)
 )
 
 func loadKubernetesMetadata(w *model.World, metrics map[string][]model.MetricValues, servicesByClusterIP map[string]*model.Service) {
@@ -83,12 +83,14 @@ func loadApplications(w *model.World, metrics map[string][]model.MetricValues) {
 
 func podInfo(w *model.World, metrics []model.MetricValues) map[string]*model.Instance {
 	pods := map[string]*model.Instance{}
+	podOwners := map[podId]model.ApplicationId{}
+	var podsOwnedByPods []*model.Instance
 	for _, m := range metrics {
 		w.IntegrationStatus.KubeStateMetrics.Installed = true
 		pod := m.Labels["pod"]
 		ns := m.Labels["namespace"]
 		ownerName := m.Labels["created_by_name"]
-		ownerKind := m.Labels["created_by_kind"]
+		ownerKind := model.ApplicationKind(m.Labels["created_by_kind"])
 		nodeName := m.Labels["node"]
 		uid := m.Labels["uid"]
 		if uid == "" {
@@ -101,24 +103,21 @@ func podInfo(w *model.World, metrics []model.MetricValues) map[string]*model.Ins
 		switch {
 		case ownerKind == "" || ownerKind == "<none>" || ownerKind == "Node":
 			appId = model.NewApplicationId(ns, model.ApplicationKindStaticPods, strings.TrimSuffix(pod, "-"+nodeName))
-		case model.ApplicationKind(ownerKind) == model.ApplicationKindArgoWorkflow:
-			appId = model.NewApplicationId(
-				ns,
-				model.ApplicationKindArgoWorkflow,
-				argoWorkflowNameSuffixRe.ReplaceAllString(ownerName, ""),
-			)
+		case ownerKind == model.ApplicationKindSparkApplication || ownerKind == model.ApplicationKindArgoWorkflow:
+			appId = model.NewApplicationId(ns, ownerKind, jobSuffixRe.ReplaceAllString(ownerName, ""))
 		case ownerName != "":
-			appId = model.NewApplicationId(ns, model.ApplicationKind(ownerKind), ownerName)
+			appId = model.NewApplicationId(ns, ownerKind, ownerName)
 		default:
 			continue
 		}
+		podOwners[podId{name: pod, ns: ns}] = appId
 		instance := pods[uid]
 		if instance == nil {
 			instance = w.GetOrCreateApplication(appId).GetOrCreateInstance(pod, node)
 			if instance.Pod == nil {
 				instance.Pod = &model.Pod{}
 			}
-			if model.ApplicationKind(ownerKind) == model.ApplicationKindReplicaSet {
+			if ownerKind == model.ApplicationKindReplicaSet {
 				instance.Pod.ReplicaSet = ownerName
 			}
 			pods[uid] = instance
@@ -137,6 +136,19 @@ func podInfo(w *model.World, metrics []model.MetricValues) map[string]*model.Ins
 				instance.TcpListens[model.Listen{IP: podIp, Port: "0", Proxied: false}] = isActive
 			}
 		}
+		if appId.Kind == model.ApplicationKindPod {
+			podsOwnedByPods = append(podsOwnedByPods, instance)
+		}
+	}
+	for _, instance := range podsOwnedByPods {
+		id := podId{name: instance.OwnerId.Name, ns: instance.OwnerId.Namespace}
+		if ownerOfOwner, ok := podOwners[id]; ok {
+			if app := w.GetApplication(ownerOfOwner); app != nil {
+				delete(w.Applications, instance.OwnerId)
+				instance.OwnerId = ownerOfOwner
+				app.Instances = append(app.Instances, instance)
+			}
+		}
 	}
 	return pods
 }

model/kubernetes.go

@@ -23,6 +23,7 @@ const (
 	ApplicationKindElasticacheCluster ApplicationKind = "ElasticacheCluster"
 	ApplicationKindNomadJobGroup      ApplicationKind = "NomadJobGroup"
 	ApplicationKindArgoWorkflow       ApplicationKind = "Workflow"
+	ApplicationKindSparkApplication   ApplicationKind = "SparkApplication"
 )
 
 type Job struct{}

model/pod.go

@@ -35,3 +35,7 @@ func (pod *Pod) IsFailed() bool {
 func (pod *Pod) IsReady() bool {
 	return pod.Ready.Last() > 0
 }
+
+func (pod *Pod) IsSucceeded() bool {
+	return pod.Phase == "Succeeded"
+}