janzak-dev · Aug 10, 2026 · Aug 10, 2026 · Aug 10, 2026 · Aug 10, 2026 · Aug 10, 2026
diff --git a/cmd/rod_data_provider/BUILD.bazel b/cmd/rod_data_provider/BUILD.bazel
@@ -27,6 +27,7 @@ compat_go_library(
         "main.go",
         "queue_converters.go",
         "queue_matcher.go",
+        "resource_accumulator.go",
         "server.go",
     ],
     importpath = "xmanager_cloud/cmd/rod_data_provider/main",
@@ -38,6 +39,7 @@ compat_go_library(
         "//cmd/rod_data_provider/reconciler",
         "//internal/compat/google",
         "//internal/configutil",
+        "//internal/domain/workunit",
         "//internal/logging",
         "//internal/storage/spanner/workunit",
         "//internal/validator",
@@ -79,6 +81,7 @@ compat_go_test(
         "config_test.go",
         "jobs_test.go",
         "queue_matcher_test.go",
+        "resource_accumulator_test.go",
         "server_test.go",
     ],
     embed = [":rod_data_provider_lib"],

diff --git a/cmd/rod_data_provider/jobs.go b/cmd/rod_data_provider/jobs.go
@@ -45,7 +45,6 @@ func (s *server) GetRunningAndPendingJobs(ctx context.Context, req *spb.GetRunni
 	workUnits, nextPageToken := extractWorkUnitsResult(result)
 	jobs := mapWorkUnitsToDashboardJobs(ctx, workUnits, time.Now())
 
-	// TODO: Set TotalQueueSize to the total count across all pages once pagination is implemented.
 	return &spb.GetRunningAndPendingJobsResponse{
 		Jobs:           jobs,
 		NextPageToken:  nextPageToken,
@@ -69,7 +68,7 @@ func validateJobsRequest(ctx context.Context, req *spb.GetRunningAndPendingJobsR
 	return nil
 }
 
-func extractWorkUnitsResult(result *workunit.ActiveWorkUnitsResult) ([]*essmsgpb.WorkUnit, string) {
+func extractWorkUnitsResult(result *workunit.ActiveWorkUnitsResult) (workUnits []*essmsgpb.WorkUnit, nextPageToken string) {
 	if result == nil {
 		return nil, ""
 	}
@@ -103,10 +102,10 @@ func mapWorkUnitToDashboardJob(ctx context.Context, wu *essmsgpb.WorkUnit, now t
 		Priority:           int64(wu.GetPriority()),
 		Status:             extractWorkUnitState(wu),
 		TimeInQueueSeconds: calculateQueueDuration(wu, now),
-		// TODO: Populate RequestedResources field using extractRequestedResources once resource accumulator PR is merged.
-		PositionInQueue: extractQueuePosition(wu),
-		CreateTime:      wu.GetCreateTime(),
-		Tags:            wu.GetTags(),
+		RequestedResources: extractRequestedResources(wu),
+		PositionInQueue:    extractQueuePosition(wu),
+		CreateTime:         wu.GetCreateTime(),
+		Tags:               wu.GetTags(),
 	}
 }
 

diff --git a/cmd/rod_data_provider/jobs_test.go b/cmd/rod_data_provider/jobs_test.go
@@ -54,6 +54,26 @@ var (
 				},
 			},
 		},
+		Executable: &essmsgpb.Executable{
+			Type: &essmsgpb.Executable_Jobset{
+				Jobset: &essmsgpb.JobSetExecutable{
+					Jobs: []*essmsgpb.KubernetesJob{
+						{
+							Spec: &essmsgpb.ExecutableSpec{
+								Resources: &essmsgpb.ResourceSet{
+									Resources: &essmsgpb.ResourceSet_CpuResources{
+										CpuResources: &essmsgpb.CpuResources{
+											CpuMillicores: 4000,
+											MemoryBytes:   8192,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
 	}
 )
 
@@ -174,6 +194,10 @@ func TestGetRunningAndPendingJobs(t *testing.T) {
 				Priority:        100,
 				Status:          essmsgpb.WorkUnitStatus_SCHEDULING,
 				PositionInQueue: 1,
+				RequestedResources: []*spb.Resource{
+					{ResourceType: "cpu", Count: 4.0},
+					{ResourceType: "memory", Count: 8192.0},
+				},
 			}
 			if diff := cmp.Diff(wantJob, gotJob, protocmp.Transform(), protocmp.IgnoreFields(&spb.DashboardJob{}, "time_in_queue_seconds", "create_time")); diff != "" {
 				t.Errorf("DashboardJob mismatch (-want +got):\n%s", diff)
@@ -195,3 +219,55 @@ func TestGetRunningAndPendingJobs_ContextCancelled(t *testing.T) {
 		t.Errorf("GetRunningAndPendingJobs status code = %v, want %v", status.Code(err), codes.Canceled)
 	}
 }
+
+func TestMapWorkUnitToDashboardJob_RequestedResources(t *testing.T) {
+	now := time.Now()
+	wu := proto.Modify(baseWorkUnit, func(w *essmsgpb.WorkUnit) {
+		w.Executable = &essmsgpb.Executable{
+			Type: &essmsgpb.Executable_Jobset{
+				Jobset: &essmsgpb.JobSetExecutable{
+					Jobs: []*essmsgpb.KubernetesJob{
+						{
+							Replicas: 2,
+							Spec: &essmsgpb.ExecutableSpec{
+								Resources: &essmsgpb.ResourceSet{
+									Resources: &essmsgpb.ResourceSet_GpuResources{
+										GpuResources: &essmsgpb.GpuResources{
+											Type:  "nvidia-tesla-h100",
+											Count: 4,
+										},
+									},
+								},
+							},
+						},
+						{
+							Replicas: 1,
+							Spec: &essmsgpb.ExecutableSpec{
+								Resources: &essmsgpb.ResourceSet{
+									Resources: &essmsgpb.ResourceSet_GpuResources{
+										GpuResources: &essmsgpb.GpuResources{
+											Type:  "nvidia-tesla-h100",
+											Count: 2,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+	})
+
+	job := mapWorkUnitToDashboardJob(context.Background(), wu, now)
+	if job == nil {
+		t.Fatalf("mapWorkUnitToDashboardJob returned nil")
+	}
+	if len(job.GetRequestedResources()) != 1 {
+		t.Fatalf("RequestedResources len = %d, want 1", len(job.GetRequestedResources()))
+	}
+	res := job.GetRequestedResources()[0]
+	if res.GetResourceType() != "nvidia-tesla-h100" || res.GetCount() != 10.0 {
+		t.Errorf("RequestedResource = %v %v, want nvidia-tesla-h100 10.0", res.GetResourceType(), res.GetCount())
+	}
+}
diff --git a/cmd/rod_data_provider/resource_accumulator.go b/cmd/rod_data_provider/resource_accumulator.go
@@ -0,0 +1,132 @@
+// Copyright 2026 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
+//
+//	http://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 main
+
+import (
+	"strconv"
+	"strings"
+
+	spb "xmanager_cloud/api/protos/rod"
+	essmsgpb "xmanager_cloud/experiment_state_server/proto/messages_go_proto"
+	"xmanager_cloud/internal/domain/workunit"
+)
+
+const (
+	resourceTypeCPU    = "cpu"
+	resourceTypeMemory = "memory"
+	resourceTypeGPU    = "GPU"
+	resourceTypeTPU    = "TPU"
+)
+
+type resourceAccumulator struct {
+	totals map[string]float64
+	order  []string
+}
+
+func (a *resourceAccumulator) add(resType string, count float64) {
+	if count <= 0 {
+		return
+	}
+	if _, exists := a.totals[resType]; !exists {
+		a.order = append(a.order, resType)
+	}
+	a.totals[resType] += count
+}
+
+// extractRequestedResources converts domain ResourceSets from workunit.GetWorkUnitResources
+// into presentation DTOs ([]*spb.Resource) required by the ROD API, aggregating identical resource types.
+func extractRequestedResources(wu *essmsgpb.WorkUnit) []*spb.Resource {
+	resourceSets := workunit.GetWorkUnitResources(wu)
+	if len(resourceSets) == 0 {
+		return nil
+	}
+	totals, order := accumulateResourceTotals(resourceSets)
+	return buildResourceDTOs(totals, order)
+}
+
+func accumulateResourceTotals(resourceSets []*essmsgpb.ResourceSet) (totals map[string]float64, order []string) {
+	acc := &resourceAccumulator{totals: make(map[string]float64, len(resourceSets))}
+	for _, rs := range resourceSets {
+		accumulateResource(rs, acc)
+	}
+	return acc.totals, acc.order
+}
+
+func buildResourceDTOs(totals map[string]float64, order []string) []*spb.Resource {
+	resources := make([]*spb.Resource, 0, len(order))
+	for _, resType := range order {
+		resources = append(resources, &spb.Resource{
+			ResourceType: resType,
+			Count:        totals[resType],
+		})
+	}
+	return resources
+}
+
+func accumulateResource(rs *essmsgpb.ResourceSet, acc *resourceAccumulator) {
+	if rs == nil || acc == nil {
+		return
+	}
+	if cpu := rs.GetCpuResources(); cpu != nil {
+		accumulateCpuResource(cpu, acc)
+	} else if gpu := rs.GetGpuResources(); gpu != nil {
+		accumulateGpuResource(gpu, acc)
+	} else if tpu := rs.GetTpuResources(); tpu != nil {
+		accumulateTpuResource(tpu, acc)
+	}
+}
+
+func accumulateCpuResource(cpu *essmsgpb.CpuResources, acc *resourceAccumulator) {
+	if millicores := cpu.GetCpuMillicores(); millicores > 0 {
+		acc.add(resourceTypeCPU, float64(millicores)/1000.0)
+	}
+	if memoryBytes := cpu.GetMemoryBytes(); memoryBytes > 0 {
+		acc.add(resourceTypeMemory, float64(memoryBytes))
+	}
+}
+
+func accumulateGpuResource(gpu *essmsgpb.GpuResources, acc *resourceAccumulator) {
+	resType := gpu.GetType()
+	if resType == "" {
+		resType = resourceTypeGPU
+	}
+	acc.add(resType, float64(gpu.GetCount()))
+}
+
+func accumulateTpuResource(tpu *essmsgpb.TpuResources, acc *resourceAccumulator) {
+	resType := tpu.GetType()
+	if resType == "" {
+		resType = resourceTypeTPU
+	}
+	chips := parseTotalChips(tpu.GetTopology())
+	acc.add(resType, float64(chips))
+}
+
+func parseTotalChips(topology string) int64 {
+	if topology == "" {
+		return 1
+	}
+	cleaned := strings.TrimSpace(strings.ToLower(topology))
+    	dims := strings.Split(cleaned, "x")
+	totalChips := int64(1)
+	for _, d := range dims {
+		v, err := strconv.ParseInt(d, 10, 64)
+		if err != nil || v <= 0 {
+			return 1
+		}
+		totalChips *= v
+	}
+	return totalChips
+}