Compute alive worker metrics in WorkerService#updateMetrics scheduled job (#739)

jbern0rd · web-flow · commit 40f55de28768 · 2025-03-04T15:06:30.000+01:00
These metrics are then used in `MetricService#getPlatformMetrics` to build responses
to queries on `/metrics` endpoint. Those queries no more trigger reads on MongoDB.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -13,6 +13,7 @@ All notable changes to this project will be documented in this file.
 - Configuration server is now optional by default. (#728)
 - Improve switch statements after Java 17 migration. (#729)
 - Remove redundant blockchain calls to diminish pressure on Ethereum JSON-RPC API. (#734)
+- Compute alive workers metrics in `WorkerService#updateMetrics` scheduled job. (#739)
 
 ### Breaking API changes
 
@@ -22,6 +23,7 @@ All notable changes to this project will be documented in this file.
 - Move `TaskAbortCause` from `iexec-commons-poco` to `iexec-core-library`. (#736)
 - Remove deprecated methods in `iexec-core-library`. (#737)
 - Remove unused `ContributionUtils` class. (#738)
+- Rework metrics to expose count of computing CPUs or GPUs instead of available ones. (#739)
 
 ### Dependency Upgrades
 
diff --git a/iexec-task-api/src/main/java/com/iexec/core/metric/PlatformMetric.java b/iexec-task-api/src/main/java/com/iexec/core/metric/PlatformMetric.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2023 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 IEXEC BLOCKCHAIN TECH
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -30,10 +30,10 @@
 @JsonDeserialize(builder = PlatformMetric.PlatformMetricBuilder.class)
 public class PlatformMetric {
     int aliveWorkers;
-    int aliveTotalCpu;
-    int aliveAvailableCpu;
-    int aliveTotalGpu;
-    int aliveAvailableGpu;
+    int aliveComputingCpu;
+    int aliveRegisteredCpu;
+    int aliveComputingGpu;
+    int aliveRegisteredGpu;
     LinkedHashMap<TaskStatus, Long> currentTaskStatusesCount;
     long dealEventsCount;
     long dealsCount;
diff --git a/iexec-task-api/src/test/java/com/iexec/core/metric/PlatformMetricTests.java b/iexec-task-api/src/test/java/com/iexec/core/metric/PlatformMetricTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2023-2023 IEXEC BLOCKCHAIN TECH
+ * Copyright 2023-2025 IEXEC BLOCKCHAIN TECH
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,10 +33,10 @@ class PlatformMetricTests {
     void shouldSerializeAndDeserialize() throws JsonProcessingException {
         final PlatformMetric platformMetric = PlatformMetric.builder()
                 .aliveWorkers(4)
-                .aliveTotalCpu(12)
-                .aliveAvailableCpu(7)
-                .aliveTotalGpu(0)
-                .aliveAvailableGpu(0)
+                .aliveComputingCpu(5)
+                .aliveRegisteredCpu(12)
+                .aliveComputingGpu(0)
+                .aliveRegisteredGpu(0)
                 .currentTaskStatusesCount(createCurrentTaskStatusesCount())
                 .dealEventsCount(3000)
                 .dealsCount(1100)
diff --git a/src/main/java/com/iexec/core/metric/MetricService.java b/src/main/java/com/iexec/core/metric/MetricService.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 IEXEC BLOCKCHAIN TECH
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
 import com.iexec.core.chain.DealWatcherService;
 import com.iexec.core.task.TaskStatus;
 import com.iexec.core.task.event.TaskStatusesCountUpdatedEvent;
+import com.iexec.core.worker.AliveWorkerMetrics;
 import com.iexec.core.worker.WorkerService;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Service;
@@ -31,21 +32,22 @@ public class MetricService {
     private final WorkerService workerService;
     private LinkedHashMap<TaskStatus, Long> currentTaskStatusesCount;
 
-    public MetricService(DealWatcherService dealWatcherService,
-                         WorkerService workerService) {
+    public MetricService(final DealWatcherService dealWatcherService,
+                         final WorkerService workerService) {
         this.dealWatcherService = dealWatcherService;
         this.workerService = workerService;
 
         this.currentTaskStatusesCount = new LinkedHashMap<>();
     }
 
     public PlatformMetric getPlatformMetrics() {
+        final AliveWorkerMetrics aliveWorkerMetrics = workerService.getAliveWorkerMetrics();
         return PlatformMetric.builder()
-                .aliveWorkers(workerService.getAliveWorkers().size())
-                .aliveTotalCpu(workerService.getAliveTotalCpu())
-                .aliveAvailableCpu(workerService.getAliveAvailableCpu())
-                .aliveTotalGpu(workerService.getAliveTotalGpu())
-                .aliveAvailableGpu(workerService.getAliveAvailableGpu())
+                .aliveWorkers(aliveWorkerMetrics.aliveWorkers())
+                .aliveComputingCpu(aliveWorkerMetrics.aliveComputingCpu())
+                .aliveRegisteredCpu(aliveWorkerMetrics.aliveRegisteredCpu())
+                .aliveComputingGpu(aliveWorkerMetrics.aliveComputingGpu())
+                .aliveRegisteredGpu(aliveWorkerMetrics.aliveRegisteredGpu())
                 .currentTaskStatusesCount(currentTaskStatusesCount)
                 .dealEventsCount(dealWatcherService.getDealEventsCount())
                 .dealsCount(dealWatcherService.getDealsCount())
@@ -55,7 +57,7 @@ public PlatformMetric getPlatformMetrics() {
     }
 
     @EventListener
-    void onTaskStatusesCountUpdateEvent(TaskStatusesCountUpdatedEvent event) {
+    void onTaskStatusesCountUpdateEvent(final TaskStatusesCountUpdatedEvent event) {
         this.currentTaskStatusesCount = event.getCurrentTaskStatusesCount();
     }
 }
diff --git a/src/main/java/com/iexec/core/worker/AliveWorkerMetrics.java b/src/main/java/com/iexec/core/worker/AliveWorkerMetrics.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2025 IEXEC BLOCKCHAIN TECH
+ *
+ * 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 com.iexec.core.worker;
+
+import lombok.Builder;
+
+@Builder
+public record AliveWorkerMetrics(int aliveWorkers,
+                                 int aliveComputingCpu,
+                                 int aliveRegisteredCpu,
+                                 int aliveComputingGpu,
+                                 int aliveRegisteredGpu) {
+}
diff --git a/src/main/java/com/iexec/core/worker/WorkerService.java b/src/main/java/com/iexec/core/worker/WorkerService.java
@@ -20,7 +20,6 @@
 import com.iexec.core.configuration.WorkerConfiguration;
 import com.mongodb.client.result.UpdateResult;
 import io.micrometer.core.instrument.Metrics;
-import jakarta.annotation.PostConstruct;
 import lombok.Data;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -50,15 +49,19 @@ public class WorkerService {
 
     private static final String WALLET_ADDRESS_FIELD = "walletAddress";
     public static final String METRIC_WORKERS_GAUGE = "iexec.core.workers";
-    public static final String METRIC_CPU_TOTAL_GAUGE = "iexec.core.cpu.total";
-    public static final String METRIC_CPU_AVAILABLE_GAUGE = "iexec.core.cpu.available";
+    public static final String METRIC_CPU_COMPUTING_GAUGE = "iexec.core.cpu.computing";
+    public static final String METRIC_CPU_REGISTERED_GAUGE = "iexec.core.cpu.registered";
+    public static final String METRIC_GPU_COMPUTING_GAUGE = "iexec.core.gpu.computing";
+    public static final String METRIC_GPU_REGISTERED_GAUGE = "iexec.core.gpu.registered";
     private final MongoTemplate mongoTemplate;
     private final WorkerRepository workerRepository;
     private final WorkerConfiguration workerConfiguration;
     private final ContextualLockRunner<String> contextualLockRunner;
-    private AtomicInteger aliveWorkersGauge;
-    private AtomicInteger aliveTotalCpuGauge;
-    private AtomicInteger aliveAvailableCpuGauge;
+    private final AtomicInteger aliveWorkersGauge;
+    private final AtomicInteger aliveComputingCpuGauge;
+    private final AtomicInteger aliveRegisteredCpuGauge;
+    private final AtomicInteger aliveComputingGpuGauge;
+    private final AtomicInteger aliveRegisteredGpuGauge;
     @Getter
     private final ConcurrentHashMap<String, WorkerStats> workerStatsMap = new ConcurrentHashMap<>();
 
@@ -80,38 +83,52 @@ public WorkerService(MongoTemplate mongoTemplate,
         this.workerRepository = workerRepository;
         this.workerConfiguration = workerConfiguration;
         this.contextualLockRunner = new ContextualLockRunner<>();
-    }
 
-    @PostConstruct
-    void init() {
-        aliveWorkersGauge = Metrics.gauge(METRIC_WORKERS_GAUGE, new AtomicInteger(getAliveWorkers().size()));
-        aliveTotalCpuGauge = Metrics.gauge(METRIC_CPU_TOTAL_GAUGE, new AtomicInteger(getAliveTotalCpu()));
-        aliveAvailableCpuGauge = Metrics.gauge(METRIC_CPU_AVAILABLE_GAUGE, new AtomicInteger(getAliveAvailableCpu()));
+        this.aliveWorkersGauge = Metrics.gauge(METRIC_WORKERS_GAUGE, new AtomicInteger(0));
+        this.aliveComputingCpuGauge = Metrics.gauge(METRIC_CPU_COMPUTING_GAUGE, new AtomicInteger(0));
+        this.aliveRegisteredCpuGauge = Metrics.gauge(METRIC_CPU_REGISTERED_GAUGE, new AtomicInteger(0));
+        this.aliveComputingGpuGauge = Metrics.gauge(METRIC_GPU_COMPUTING_GAUGE, new AtomicInteger(0));
+        this.aliveRegisteredGpuGauge = Metrics.gauge(METRIC_GPU_REGISTERED_GAUGE, new AtomicInteger(0));
     }
 
     /**
      * updateMetrics is used to update all workers metrics
      */
-    @Scheduled(fixedDelayString = "${cron.metrics.refresh.period}", initialDelayString = "${cron.metrics.refresh.period}")
+    @Scheduled(fixedDelayString = "${cron.metrics.refresh.period}")
     void updateMetrics() {
         // Fusion of methods getAliveTotalCpu and getAliveAvailableCpu to prevent making 3 calls to getAliveWorkers
-        int availableCpus = 0;
-        int totalCpus = 0;
-        List<Worker> workers = getAliveWorkers();
-        for (Worker worker : workers) {
+        int computingCpus = 0;
+        int registeredCpus = 0;
+        int computingGpus = 0;
+        int registeredGpus = 0;
+        final List<Worker> workers = getAliveWorkers();
+        for (final Worker worker : workers) {
             if (worker.isGpuEnabled()) {
-                continue;
+                registeredGpus++;
+                if (!worker.getComputingChainTaskIds().isEmpty()) {
+                    computingGpus++;
+                }
+            } else {
+                registeredCpus += worker.getCpuNb();
+                computingCpus += worker.getComputingChainTaskIds().size();
             }
-            int workerCpuNb = worker.getCpuNb();
-            int computingReplicateNb = worker.getComputingChainTaskIds().size();
-            int availableCpu = workerCpuNb - computingReplicateNb;
-            totalCpus += workerCpuNb;
-            availableCpus += availableCpu;
         }
 
         aliveWorkersGauge.set(workers.size());
-        aliveTotalCpuGauge.set(totalCpus);
-        aliveAvailableCpuGauge.set(availableCpus);
+        aliveComputingCpuGauge.set(computingCpus);
+        aliveRegisteredCpuGauge.set(registeredCpus);
+        aliveComputingGpuGauge.set(computingGpus);
+        aliveRegisteredGpuGauge.set(registeredGpus);
+    }
+
+    public AliveWorkerMetrics getAliveWorkerMetrics() {
+        return AliveWorkerMetrics.builder()
+                .aliveWorkers(aliveWorkersGauge.get())
+                .aliveComputingCpu(aliveComputingCpuGauge.get())
+                .aliveRegisteredCpu(aliveRegisteredCpuGauge.get())
+                .aliveComputingGpu(aliveComputingGpuGauge.get())
+                .aliveRegisteredGpu(aliveRegisteredGpuGauge.get())
+                .build();
     }
 
     // region Read methods
@@ -120,12 +137,9 @@ public Optional<Worker> getWorker(String walletAddress) {
     }
 
     public boolean isAllowedToJoin(String workerAddress) {
-        List<String> whitelist = workerConfiguration.getWhitelist();
+        final List<String> whitelist = workerConfiguration.getWhitelist();
         // if the whitelist is empty, there is no restriction on the workers
-        if (whitelist.isEmpty()) {
-            return true;
-        }
-        return whitelist.contains(workerAddress);
+        return whitelist.isEmpty() || whitelist.contains(workerAddress);
     }
 
     public boolean isWorkerAllowedToAskReplicate(String walletAddress) {
@@ -190,56 +204,6 @@ public boolean canAcceptMoreWorks(Worker worker) {
 
         return true;
     }
-
-    public int getAliveAvailableCpu() {
-        int availableCpus = 0;
-        for (Worker worker : getAliveWorkers()) {
-            if (worker.isGpuEnabled()) {
-                continue;
-            }
-
-            int workerCpuNb = worker.getCpuNb();
-            int computingReplicateNb = worker.getComputingChainTaskIds().size();
-            int availableCpu = workerCpuNb - computingReplicateNb;
-            availableCpus += availableCpu;
-        }
-        return availableCpus;
-    }
-
-    public int getAliveTotalCpu() {
-        int totalCpus = 0;
-        for (Worker worker : getAliveWorkers()) {
-            if (worker.isGpuEnabled()) {
-                continue;
-            }
-            totalCpus += worker.getCpuNb();
-        }
-        return totalCpus;
-    }
-
-    // We suppose for now that 1 Gpu enabled worker has only one GPU
-    public int getAliveTotalGpu() {
-        int totalGpus = 0;
-        for (Worker worker : getAliveWorkers()) {
-            if (worker.isGpuEnabled()) {
-                totalGpus++;
-            }
-        }
-        return totalGpus;
-    }
-
-    public int getAliveAvailableGpu() {
-        int availableGpus = getAliveTotalGpu();
-        for (Worker worker : getAliveWorkers()) {
-            if (worker.isGpuEnabled()) {
-                boolean isWorking = !worker.getComputingChainTaskIds().isEmpty();
-                if (isWorking) {
-                    availableGpus = availableGpus - 1;
-                }
-            }
-        }
-        return availableGpus;
-    }
     // endregion
 
     // region Read-and-write methods
diff --git a/src/test/java/com/iexec/core/metric/MetricControllerTests.java b/src/test/java/com/iexec/core/metric/MetricControllerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 IEXEC BLOCKCHAIN TECH
+ * Copyright 2020-2025 IEXEC BLOCKCHAIN TECH
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,16 +16,17 @@
 
 package com.iexec.core.metric;
 
-import org.assertj.core.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
+import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.http.HttpStatus;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.when;
 
+@ExtendWith(MockitoExtension.class)
 class MetricControllerTests {
 
     @Mock
@@ -34,23 +35,16 @@ class MetricControllerTests {
     @InjectMocks
     private MetricController metricController;
 
-    @BeforeEach
-    void init() {
-        MockitoAnnotations.openMocks(this);
-    }
-
     @Test
     void shouldGetMetrics() {
         PlatformMetric metric = PlatformMetric.builder()
-                .aliveAvailableCpu(1)
-                .aliveAvailableGpu(1)
+                .aliveRegisteredCpu(1)
+                .aliveRegisteredCpu(1)
                 .build();
         when(metricService.getPlatformMetrics()).thenReturn(metric);
-        Assertions.assertThat(
-                metricController.getPlatformMetric().getStatusCode())
-        .isEqualTo(HttpStatus.OK);
-        Assertions.assertThat(
-                metricController.getPlatformMetric().getBody())
-        .isEqualTo(metric);
+        assertThat(metricController.getPlatformMetric().getStatusCode())
+                .isEqualTo(HttpStatus.OK);
+        assertThat(metricController.getPlatformMetric().getBody())
+                .isEqualTo(metric);
     }
 }
diff --git a/src/test/java/com/iexec/core/metric/MetricServiceTests.java b/src/test/java/com/iexec/core/metric/MetricServiceTests.java
diff --git a/src/test/java/com/iexec/core/worker/WorkerServiceTests.java b/src/test/java/com/iexec/core/worker/WorkerServiceTests.java
