loadtest-controller: improve styling on the html report

Author: ivchicano

e2e-tests/README.md

@@ -117,7 +117,7 @@ All smoke tests should:
 - "Sessions Created"
 - "Total Participants"
 - "User Connections" (mandatory)
-- Table columns: User, Session, Join date, Disconnect Date, Retry Number
+- Table columns: User, Session, Join Date, Disconnect Date, Retries
 - Two user rows (User1 and User2) present
 
 If validation fails, the result files are kept in the `results/` directory for debugging.

e2e-tests/scripts/validate-default.sh

@@ -96,7 +96,7 @@ if [ -f "$RESULTS_DIR/results.txt" ]; then
 
         if grep -q "User Connections" "$HTML_FILE"; then
             echo "✓ HTML contains 'User Connections'"
-            if grep -q "Join date" "$HTML_FILE" && grep -q "Disconnect Date" "$HTML_FILE" && grep -q "Retry Number" "$HTML_FILE"; then
+            if grep -q "Join Date" "$HTML_FILE" && grep -q "Disconnect Date" "$HTML_FILE" && grep -q "Retries" "$HTML_FILE"; then
                 echo "✓ HTML contains new table columns"
                 if grep -q "User1" "$HTML_FILE" && grep -q "User2" "$HTML_FILE"; then
                     echo "✓ HTML contains two user rows"

loadtest-controller/src/main/java/io/openvidu/loadtest/models/testcase/CreateParticipantResponse.java

@@ -9,12 +9,13 @@ public class CreateParticipantResponse {
     private int participantsInWorker;
     private String userId;
     private String sessionId;
+    private String workerUrl;
 
     public CreateParticipantResponse() {
     }
 
     public CreateParticipantResponse(boolean responseOk, String stopReason, String connectionId,
-            int streamsInWorker, int participantsInWorker, String userId, String sessionId, double workerCpuPct) {
+            int streamsInWorker, int participantsInWorker, String userId, String sessionId, double workerCpuPct, String workerUrl) {
         this.responseOk = responseOk;
         this.stopReason = stopReason;
         this.connectionId = connectionId;
@@ -23,6 +24,7 @@ public CreateParticipantResponse(boolean responseOk, String stopReason, String c
         this.userId = userId;
         this.sessionId = sessionId;
         this.workerCpuPct = workerCpuPct;
+        this.workerUrl = workerUrl;
     }
 
     public boolean isResponseOk() {
@@ -97,12 +99,21 @@ public CreateParticipantResponse setSessionId(String sessionId) {
         return this;
     }
 
+    public String getWorkerUrl() {
+        return workerUrl;
+    }
+
+    public CreateParticipantResponse setWorkerUrl(String workerUrl) {
+        this.workerUrl = workerUrl;
+        return this;
+    }
+
     @Override
     public String toString() {
         return "CreateParticipantResponse [responseOk=" + responseOk + ", stopReason=" + stopReason + ", connectionId="
                 + connectionId + ", workerCpuPct=" + workerCpuPct + ", streamsInWorker=" + streamsInWorker
                 + ", participantsInWorker=" + participantsInWorker + ", userId=" + userId + ", sessionId=" + sessionId
-                + "]";
+                + ", workerUrl=" + workerUrl + "]";
     }
 
 }

loadtest-controller/src/main/java/io/openvidu/loadtest/models/testcase/ResultReport.java

@@ -238,7 +238,8 @@ private void computeAggregates() {
                 errorCounts.merge(resp.getStopReason(), 1, Integer::sum);
             }
         }
-        // Compute average and max CPU across all successful responses
+        // Compute per-worker and global CPU stats
+        Map<String, List<Double>> cpuPerWorker = new TreeMap<>();
         double cpuSum = 0;
         int cpuCount = 0;
         double cpuMax = 0;
@@ -249,13 +250,28 @@ private void computeAggregates() {
                 cpuCount++;
                 if (cpu > cpuMax)
                     cpuMax = cpu;
+                String worker = resp.getWorkerUrl();
+                if (worker == null || worker.isEmpty()) {
+                    worker = "Unknown Worker";
+                }
+                cpuPerWorker.computeIfAbsent(worker, k -> new ArrayList<>()).add(cpu);
             }
         }
         if (cpuCount > 0) {
             Map<String, Double> avgMap = new TreeMap<>();
-            avgMap.put("all_workers", cpuSum / cpuCount);
             Map<String, Double> maxMap = new TreeMap<>();
-            maxMap.put("all_workers", cpuMax);
+            // Add per-worker stats
+            for (Map.Entry<String, List<Double>> entry : cpuPerWorker.entrySet()) {
+                String worker = entry.getKey();
+                List<Double> values = entry.getValue();
+                double workerAvg = values.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
+                double workerMax = values.stream().mapToDouble(Double::doubleValue).max().orElse(0.0);
+                avgMap.put(worker, workerAvg);
+                maxMap.put(worker, workerMax);
+            }
+            // Add global "All Workers" stats
+            avgMap.put("All Workers", cpuSum / cpuCount);
+            maxMap.put("All Workers", cpuMax);
             workerCpuAvg = avgMap;
             workerCpuMax = maxMap;
         }

loadtest-controller/src/main/java/io/openvidu/loadtest/services/BrowserEmulatorClient.java

@@ -71,6 +71,7 @@ public class BrowserEmulatorClient {
     private ConcurrentHashMap<String, AtomicBoolean> participantConnecting = new ConcurrentHashMap<>();
     private Set<String> participantReconnecting = new CopyOnWriteArraySet<>();
     private ConcurrentHashMap<String, Calendar> userDisconnectTimestamps = new ConcurrentHashMap<>();
+    private Set<String> processingDisconnects = ConcurrentHashMap.newKeySet();
 
     private CreateParticipantResponse lastErrorReconnectingResponse;
 
@@ -102,7 +103,7 @@ public void clean() {
 
     public void addDisconnectTimestamp(String userId, String sessionId) {
         String key = userId + "-" + sessionId;
-        userDisconnectTimestamps.put(key, Calendar.getInstance());
+        userDisconnectTimestamps.putIfAbsent(key, Calendar.getInstance());
     }
 
     public Map<String, Calendar> getPerUserDisconnectTimestamps() {
@@ -182,20 +183,34 @@ public void addClientFailure(String workerUrl, String participant, String sessio
     public void addClientFailure(String workerUrl, String participant, String session, boolean waitForConnection,
             boolean reconnect) {
         if (this.isClean.get()) {
-            // Test finished
             return;
         }
+        String user = participant + "-" + session;
+        ConcurrentHashMap<String, Role> workerRoles = this.clientRoles.get(workerUrl);
+        if (workerRoles == null) {
+            log.debug("Worker {} has no roles map, ignoring disconnect for {}", workerUrl, user);
+            return;
+        }
+        if (!workerRoles.containsKey(user)) {
+            log.debug("Worker {} doesn't have participant {}, ignoring disconnect", workerUrl, user);
+            return;
+        }
+        if (!processingDisconnects.add(user)) {
+            log.debug("Already processing disconnect for {}, ignoring duplicate", user);
+            return;
+        }
+        log.info("Processing disconnect for {} on worker {}", user, workerUrl);
         log.debug("Adding client failure for participant {} in session {}", participant, session);
         log.debug("Wait for connection: {}", waitForConnection);
-        while (waitForConnection && this.participantConnecting.get(participant + "-" + session).get()) {
+        AtomicBoolean isConnecting = this.participantConnecting.get(user);
+        while (waitForConnection && isConnecting != null && isConnecting.get()) {
             if (endOfTest.get()) {
                 return;
             }
             sleeper.sleep(WAIT_S, null);
         }
         ConcurrentHashMap<String, AtomicInteger> failures = this.clientFailures.computeIfAbsent(workerUrl,
                 key -> new ConcurrentHashMap<>());
-        String user = participant + "-" + session;
         AtomicInteger currentFailures = failures.computeIfAbsent(user, key -> new AtomicInteger(0));
         int newFailures = currentFailures.incrementAndGet();
         log.error("Participant {} in session {} failed {} times", participant, session, newFailures);
@@ -212,8 +227,8 @@ public void addClientFailure(String workerUrl, String participant, String sessio
                 log.debug("Stop reconnecting participant {} in session {}", participant, session);
                 this.lastErrorReconnectingResponse = new CreateParticipantResponse()
                         .setResponseOk(false)
-                        .setStopReason("Participant " + participant + " in session " + session + " failed "
-                                + newFailures + " times");
+                        .setStopReason("Participant " + participant + "-" + session + " failed after "
+                                + newFailures + " retries");
             }
         }
     }
@@ -248,34 +263,39 @@ private void reconnect(String workerUrl, String participant, String session) {
     private void afterDisconnect(String workerUrl, String participant, String session) {
         log.debug("After disconnect user {} session {} in {}", participant, session, workerUrl);
         String user = participant + "-" + session;
-        ConcurrentHashMap<String, Role> workerRoles = this.clientRoles.get(workerUrl);
-        if (workerRoles == null) {
-            // The connect request hasn't finished yet, wait for it
-            log.debug("Worker roles is null for {} in session {} in {}. Waiting ...", participant, session, workerUrl);
-            sleeper.sleep(WAIT_S, null);
-            this.afterDisconnect(workerUrl, participant, session);
-            return;
-        }
-        Role role = workerRoles.get(user);
-        // get user number from participant removing prefix
-        int userNumber = Integer.parseInt(participant.replace(loadTestConfig.getUserNamePrefix(), ""));
-        // get session number from session removing prefix
-        int sessionNumber = Integer.parseInt(session.replace(loadTestConfig.getSessionNamePrefix(), ""));
-        CreateParticipantResponse response = null;
-        if (role.equals(Role.PUBLISHER)) {
-            response = this.createPublisher(workerUrl, userNumber, sessionNumber, this.participantTestCases.get(user));
-        } else {
-            response = this.createSubscriber(workerUrl, userNumber, sessionNumber, this.participantTestCases.get(user));
-        }
-        if (response.isResponseOk()) {
-            this.participantReconnecting.remove(user);
-        } else {
-            this.lastErrorReconnectingResponse = response;
-            log.error("Response status is not 200 OK. Exit");
+        try {
+            ConcurrentHashMap<String, Role> workerRoles = this.clientRoles.get(workerUrl);
+            if (workerRoles == null) {
+                log.debug("Worker roles is null for {} in session {} in {}. Waiting ...", participant, session, workerUrl);
+                sleeper.sleep(WAIT_S, null);
+                this.afterDisconnect(workerUrl, participant, session);
+                return;
+            }
+            Role role = workerRoles.get(user);
+            if (role == null) {
+                log.warn("Role is null for {} in session {} in {}. This worker may not have this participant.", participant, session, workerUrl);
+                return;
+            }
+            int userNumber = Integer.parseInt(participant.replace(loadTestConfig.getUserNamePrefix(), ""));
+            int sessionNumber = Integer.parseInt(session.replace(loadTestConfig.getSessionNamePrefix(), ""));
+            CreateParticipantResponse response = null;
+            if (role.equals(Role.PUBLISHER)) {
+                response = this.createPublisher(workerUrl, userNumber, sessionNumber, this.participantTestCases.get(user));
+            } else {
+                response = this.createSubscriber(workerUrl, userNumber, sessionNumber, this.participantTestCases.get(user));
+            }
+            if (response.isResponseOk()) {
+                this.participantReconnecting.remove(user);
+            } else {
+                this.lastErrorReconnectingResponse = response;
+                log.error("Response status is not 200 OK. Exit");
+            }
+        } finally {
+            processingDisconnects.remove(user);
         }
     }
 
-    private HttpResponse<String> disconnectUser(String workerUrl, String participant, String session) {
+    private HttpResponse<String> disconnectUser(String workerUrl, String participant, String session) {
         try {
             log.info("Deleting participant {} from worker {}", participant, workerUrl);
             Map<String, String> headers = new HashMap<>();
@@ -447,20 +467,23 @@ private CreateParticipantResponse createParticipant(String workerUrl, int userNu
                 log.error("Participant {} in session {} failed {} times", userId, sessionId, failures);
                 sleeper.sleep(WAIT_S, null);
                 if (!loadTestConfig.isRetryMode() || isResponseLimitReached(failures) || endOfTest.get()) {
+                    String reason = "Participant " + userId + "-" + sessionId + " failed after "
+                            + failures + " retries";
                     // Set lastErrorReconnectingResponse to trigger test termination
                     this.lastErrorReconnectingResponse = new CreateParticipantResponse()
                             .setResponseOk(false)
-                            .setStopReason("Participant " + userId + " in session " + sessionId + " failed "
-                                    + failures + " times");
-                    return cpr.setResponseOk(false);
+                            .setStopReason(reason);
+                    // Also set stopReason on the returned response to avoid race condition
+                    // where getLastResponse() may return this object before lastErrorReconnectingResponse is visible
+                    return cpr.setResponseOk(false).setStopReason(reason);
                 }
                 log.warn("Retrying");
                 return this.createParticipant(workerUrl, userNumber, sessionNumber, testCase, role);
             } else {
                 this.participantConnecting.get(user).set(false);
                 this.saveParticipantData(workerUrl, testCase.isTeaching() ? Role.PUBLISHER : role);
             }
-            return processResponse(response);
+            return processResponse(response, workerUrl);
         } catch (Exception e) {
             CreateParticipantErrorContext ctx = new CreateParticipantErrorContext(
                     new UserInfo(workerUrl, userNumber, sessionNumber, role, userId, sessionId), testCase, cpr);
@@ -520,7 +543,7 @@ private CreateParticipantResponse createExternalRecordingParticipant(String work
         return okResponse;
     }
 
-    private CreateParticipantResponse processResponse(HttpResponse<String> response) {
+    private CreateParticipantResponse processResponse(HttpResponse<String> response, String workerUrl) {
         CreateParticipantResponse cpr = new CreateParticipantResponse();
         if (response != null && response.statusCode() == HTTP_STATUS_OK) {
             JsonObject jsonResponse = jsonUtils.getJson(response.body());
@@ -538,7 +561,8 @@ private CreateParticipantResponse processResponse(HttpResponse<String> response)
             return cpr.setResponseOk(true).setConnectionId(connectionId)
                     .setUserId(userId).setSessionId(sessionId)
                     .setWorkerCpuPct(workerCpuPct).setStreamsInWorker(streamsInWorker)
-                    .setParticipantsInWorker(participantsInWorker);
+                    .setParticipantsInWorker(participantsInWorker)
+                    .setWorkerUrl(workerUrl);
         }
         if (response == null) {
             log.error("Error. Response is null");

loadtest-controller/src/main/java/io/openvidu/loadtest/services/core/LoadTestService.java

@@ -360,6 +360,8 @@ private void saveResultReport(TestCase testCase, String participantsBySession, C
                 .setTimePerRecordingWorker(shutdownOrchestrator.getRecordingWorkerTimes())
                 .setUserStartTimes(participantOrchestrator.getUserStartTimes())
                 .setUserSuccessTimestamps(userSuccessTimestamps)
+                .setParticipantResponses(participantOrchestrator.getAllParticipantResponses())
+                .setUserRetryCounts(browserEmulatorClient.getPerUserRetryCounts())
                 .build();
 
         io.exportResults(rr);