Add pool to JobRunningEvent and JobPendingEvent API messages (#4695)

Expose the pool of a job run in the JobRunningEvent and JobPendingEvent
API messages, matching the existing behaviour of JobLeasedEvent. The
pool is read from the pod's armadaproject.io/pool annotation by the
executor and threaded through the internal JobRunRunning and
JobRunAssigned events to the API conversion layer.

Signed-off-by: Maurice Yap <mauriceyap@hotmail.co.uk>

Author: mauriceyap

internal/executor/reporter/event.go

@@ -31,6 +31,7 @@ func CreateEventForCurrentState(pod *v1.Pod, clusterId string) (*armadaevents.Ev
 				JobRunAssigned: &armadaevents.JobRunAssigned{
 					RunId: runId,
 					JobId: jobId,
+					Pool:  util.ExtractPool(pod),
 					ResourceInfos: []*armadaevents.KubernetesResourceInfo{
 						{
 							ObjectMeta: &armadaevents.ObjectMeta{
@@ -57,6 +58,7 @@ func CreateEventForCurrentState(pod *v1.Pod, clusterId string) (*armadaevents.Ev
 				JobRunRunning: &armadaevents.JobRunRunning{
 					RunId: runId,
 					JobId: jobId,
+					Pool:  util.ExtractPool(pod),
 					ResourceInfos: []*armadaevents.KubernetesResourceInfo{
 						{
 							ObjectMeta: &armadaevents.ObjectMeta{

internal/executor/reporter/event_sender_test.go

@@ -15,9 +15,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 
-	"github.com/armadaproject/armada/internal/executor/domain"
-
+	"github.com/armadaproject/armada/internal/common/constants"
 	"github.com/armadaproject/armada/internal/common/util"
+	"github.com/armadaproject/armada/internal/executor/domain"
 	"github.com/armadaproject/armada/pkg/executorapi"
 )
 
@@ -103,7 +103,8 @@ func makeTestPod(phase v1.PodPhase) *v1.Pod {
 				domain.JobRunId: uuid.New().String(),
 			},
 			Annotations: map[string]string{
-				domain.JobSetId: "job-set-id-1",
+				domain.JobSetId:          "job-set-id-1",
+				constants.PoolAnnotation: "test-pool",
 			},
 			CreationTimestamp: metav1.Time{time.Now().Add(-10 * time.Minute)},
 			UID:               k8sTypes.UID(util.NewULID()),

internal/executor/reporter/event_test.go

@@ -17,8 +17,9 @@ func TestCreateEventForCurrentState_WhenPodPending(t *testing.T) {
 	assert.Nil(t, err)
 
 	assert.Len(t, result.Events, 1)
-	_, ok := result.Events[0].Event.(*armadaevents.EventSequence_Event_JobRunAssigned)
+	assigned, ok := result.Events[0].Event.(*armadaevents.EventSequence_Event_JobRunAssigned)
 	assert.True(t, ok)
+	assert.Equal(t, "test-pool", assigned.JobRunAssigned.Pool)
 }
 
 func TestCreateEventForCurrentState_WhenPodRunning(t *testing.T) {
@@ -28,8 +29,9 @@ func TestCreateEventForCurrentState_WhenPodRunning(t *testing.T) {
 	assert.Nil(t, err)
 
 	assert.Len(t, result.Events, 1)
-	_, ok := result.Events[0].Event.(*armadaevents.EventSequence_Event_JobRunRunning)
+	running, ok := result.Events[0].Event.(*armadaevents.EventSequence_Event_JobRunRunning)
 	assert.True(t, ok)
+	assert.Equal(t, "test-pool", running.JobRunRunning.Pool)
 }
 
 func TestCreateEventForCurrentState_WhenPodFailed(t *testing.T) {

internal/server/event/conversion/conversions.go

@@ -382,6 +382,7 @@ func FromInternalJobRunRunning(queueName string, jobSetName string, time time.Ti
 		JobSetId: jobSetName,
 		Queue:    queueName,
 		Created:  protoutil.ToTimestamp(time),
+		Pool:     e.Pool,
 	}
 
 	if len(e.ResourceInfos) > 0 {
@@ -409,6 +410,7 @@ func FromInternalJobRunAssigned(queueName string, jobSetName string, time time.T
 		JobSetId: jobSetName,
 		Queue:    queueName,
 		Created:  protoutil.ToTimestamp(time),
+		Pool:     e.Pool,
 	}
 
 	if len(e.ResourceInfos) > 0 {

internal/server/event/conversion/conversions_test.go

@@ -574,6 +574,7 @@ func TestConvertJobRunning(t *testing.T) {
 			JobRunRunning: &armadaevents.JobRunRunning{
 				RunId: runId,
 				JobId: jobId,
+				Pool:  "test-pool",
 				ResourceInfos: []*armadaevents.KubernetesResourceInfo{
 					{
 						ObjectMeta: &armadaevents.ObjectMeta{
@@ -608,6 +609,7 @@ func TestConvertJobRunning(t *testing.T) {
 					PodNumber:    podNumber,
 					PodName:      podName,
 					PodNamespace: namespace,
+					Pool:         "test-pool",
 				},
 			},
 		},
@@ -669,6 +671,7 @@ func TestConvertJobAssigned(t *testing.T) {
 			JobRunAssigned: &armadaevents.JobRunAssigned{
 				RunId: runId,
 				JobId: jobId,
+				Pool:  "test-pool",
 				ResourceInfos: []*armadaevents.KubernetesResourceInfo{
 					{
 						ObjectMeta: &armadaevents.ObjectMeta{
@@ -702,6 +705,7 @@ func TestConvertJobAssigned(t *testing.T) {
 					PodNumber:    podNumber,
 					PodName:      podName,
 					PodNamespace: namespace,
+					Pool:         "test-pool",
 				},
 			},
 		},

pkg/api/api.swagger.go

@@ -1430,6 +1430,9 @@ func SwaggerJsonTemplate() string {
 		"          \"type\": \"integer\",\n" +
 		"          \"format\": \"int32\"\n" +
 		"        },\n" +
+		"        \"pool\": {\n" +
+		"          \"type\": \"string\"\n" +
+		"        },\n" +
 		"        \"queue\": {\n" +
 		"          \"type\": \"string\"\n" +
 		"        }\n" +
@@ -1727,6 +1730,9 @@ func SwaggerJsonTemplate() string {
 		"          \"type\": \"integer\",\n" +
 		"          \"format\": \"int32\"\n" +
 		"        },\n" +
+		"        \"pool\": {\n" +
+		"          \"type\": \"string\"\n" +
+		"        },\n" +
 		"        \"queue\": {\n" +
 		"          \"type\": \"string\"\n" +
 		"        }\n" +

pkg/api/api.swagger.json

@@ -1419,6 +1419,9 @@
           "type": "integer",
           "format": "int32"
         },
+        "pool": {
+          "type": "string"
+        },
         "queue": {
           "type": "string"
         }
@@ -1716,6 +1719,9 @@
           "type": "integer",
           "format": "int32"
         },
+        "pool": {
+          "type": "string"
+        },
         "queue": {
           "type": "string"
         }