Add worker label to worker pods to only stream logs from worker pods (#2162)

Author: vishalbollu

pkg/crds/controllers/batch/batchjob_controller_helpers.go

@@ -215,19 +215,21 @@ func (r *BatchJobReconciler) desiredEnqueuerJob(batchJob batch.BatchJob, queueUR
 			Namespace:   batchJob.Namespace,
 			Parallelism: 1,
 			Labels: map[string]string{
-				"apiKind":        userconfig.BatchAPIKind.String(),
-				"apiName":        batchJob.Spec.APIName,
-				"apiID":          batchJob.Spec.APIID,
-				"jobID":          batchJob.Name,
-				"cortex.dev/api": "true",
+				"apiKind":          userconfig.BatchAPIKind.String(),
+				"apiName":          batchJob.Spec.APIName,
+				"apiID":            batchJob.Spec.APIID,
+				"jobID":            batchJob.Name,
+				"cortex.dev/api":   "true",
+				"cortex.dev/batch": "enqueuer",
 			},
 			PodSpec: k8s.PodSpec{
 				Labels: map[string]string{
-					"apiKind":        userconfig.BatchAPIKind.String(),
-					"apiName":        batchJob.Spec.APIName,
-					"apiID":          batchJob.Spec.APIID,
-					"jobID":          batchJob.Name,
-					"cortex.dev/api": "true",
+					"apiKind":          userconfig.BatchAPIKind.String(),
+					"apiName":          batchJob.Spec.APIName,
+					"apiID":            batchJob.Spec.APIID,
+					"jobID":            batchJob.Name,
+					"cortex.dev/api":   "true",
+					"cortex.dev/batch": "enqueuer",
 				},
 				Annotations: map[string]string{
 					"traffic.sidecar.istio.io/excludeOutboundIPRanges": "0.0.0.0/0",
@@ -294,23 +296,25 @@ func (r *BatchJobReconciler) desiredWorkerJob(batchJob batch.BatchJob, apiSpec s
 			Namespace:   batchJob.Namespace,
 			Parallelism: batchJob.Spec.Workers,
 			Labels: map[string]string{
-				"apiKind":        userconfig.BatchAPIKind.String(),
-				"apiName":        batchJob.Spec.APIName,
-				"apiID":          batchJob.Spec.APIID,
-				"specID":         apiSpec.SpecID,
-				"handlerID":      apiSpec.HandlerID,
-				"jobID":          batchJob.Name,
-				"cortex.dev/api": "true",
+				"apiKind":          userconfig.BatchAPIKind.String(),
+				"apiName":          batchJob.Spec.APIName,
+				"apiID":            batchJob.Spec.APIID,
+				"specID":           apiSpec.SpecID,
+				"handlerID":        apiSpec.HandlerID,
+				"jobID":            batchJob.Name,
+				"cortex.dev/api":   "true",
+				"cortex.dev/batch": "worker",
 			},
 			PodSpec: k8s.PodSpec{
 				Labels: map[string]string{
-					"apiKind":        userconfig.BatchAPIKind.String(),
-					"apiName":        batchJob.Spec.APIName,
-					"apiID":          batchJob.Spec.APIID,
-					"specID":         apiSpec.SpecID,
-					"handlerID":      apiSpec.HandlerID,
-					"jobID":          batchJob.Name,
-					"cortex.dev/api": "true",
+					"apiKind":          userconfig.BatchAPIKind.String(),
+					"apiName":          batchJob.Spec.APIName,
+					"apiID":            batchJob.Spec.APIID,
+					"specID":           apiSpec.SpecID,
+					"handlerID":        apiSpec.HandlerID,
+					"jobID":            batchJob.Name,
+					"cortex.dev/api":   "true",
+					"cortex.dev/batch": "worker",
 				},
 				Annotations: map[string]string{
 					"traffic.sidecar.istio.io/excludeOutboundIPRanges": "0.0.0.0/0",

pkg/operator/endpoints/logs_job.go

@@ -52,5 +52,11 @@ func ReadJobLogs(w http.ResponseWriter, r *http.Request) {
 	}
 	defer socket.Close()
 
-	operator.StreamLogsFromRandomPod(map[string]string{"apiName": apiName, "jobID": jobID}, socket)
+	labels := map[string]string{"apiName": apiName, "jobID": jobID}
+
+	if deployedResource.Kind == userconfig.BatchAPIKind {
+		labels["cortex.dev/batch"] = "worker"
+	}
+
+	operator.StreamLogsFromRandomPod(labels, socket)
 }