feat: add locking to run manager to fix fsm concurrency + add error m…

…essages to status
scc-digitalhub · Sep 19, 2024 · 216d5b2 · 216d5b2
1 parent 4e8a4d3
commit 216d5b2
Show file tree

Hide file tree

Showing 16 changed files with 622 additions and 339 deletions.
diff --git a/application/src/main/java/it/smartcommunitylabdhub/core/components/run/RunManager.java b/application/src/main/java/it/smartcommunitylabdhub/core/components/run/RunManager.java
diff --git a/...n/src/main/java/it/smartcommunitylabdhub/core/components/run/RunTransitionsProcessor.java b/...n/src/main/java/it/smartcommunitylabdhub/core/components/run/RunTransitionsProcessor.java
@@ -15,8 +15,22 @@
 import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
 
-@RunProcessorType(stages = { "onRunning", "onCompleted", "onError", "onStopped", "onDeleted" }, id = K8sProcessor.ID)
+@RunProcessorType(
+    stages = {
+        "onBuilt",
+        "onReady",
+        "onRunning",
+        "onCompleted",
+        "onError",
+        "onStopping",
+        "onStopped",
+        "onDeleting",
+        "onDeleted",
+    },
+    id = K8sProcessor.ID
+)
 @Component(RunTransitionsProcessor.ID)
 @Slf4j
 public class RunTransitionsProcessor implements RunProcessor<RunTransitionsSpec> {
@@ -44,6 +58,7 @@ public RunTransitionsSpec process(Run run, RunRunnable runRunnable, RunBaseStatu
             .builder()
             .event(null) //TODO add explicit event
             .status(status != null ? status.getState() : null)
+            .message(status != null ? status.getMessage() : null)
             .time(OffsetDateTime.ofInstant(Instant.now(), ZoneOffset.UTC))
             .build();
 

diff --git a/.../main/java/it/smartcommunitylabdhub/core/controllers/v1/context/RunContextController.java b/.../main/java/it/smartcommunitylabdhub/core/controllers/v1/context/RunContextController.java
@@ -147,7 +147,7 @@ public Run updateRunById(
 
     @Operation(summary = "Delete a specific run, with optional cascade")
     @DeleteMapping(path = "/{id}")
-    public void deleteRun(
+    public Run deleteRun(
         @PathVariable @Valid @NotNull @Pattern(regexp = Keys.SLUG_PATTERN) String project,
         @PathVariable @Valid @NotNull @Pattern(regexp = Keys.SLUG_PATTERN) String id
     ) throws NoSuchEntityException {
@@ -159,7 +159,7 @@ public void deleteRun(
         }
 
         //delete via manager
-        runManager.delete(run);
+        return runManager.delete(run);
     }
 
     /*

diff --git a/...tion/src/main/java/it/smartcommunitylabdhub/core/models/specs/run/RunTransitionsSpec.java b/...tion/src/main/java/it/smartcommunitylabdhub/core/models/specs/run/RunTransitionsSpec.java
@@ -42,6 +42,7 @@ public static class Transition {
 
         private RunEvent event;
         private String status;
+        private String message;
 
         @JsonFormat(shape = JsonFormat.Shape.STRING)
         protected OffsetDateTime time;

diff --git a/modules/commons/.flattened-pom.xml b/modules/commons/.flattened-pom.xml
@@ -91,5 +91,11 @@
       <version>4.36.0</version>
       <scope>compile</scope>
     </dependency>
+    <dependency>
+      <groupId>net.bytebuddy</groupId>
+      <artifactId>byte-buddy</artifactId>
+      <version>1.14.17</version>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 </project>
diff --git a/...s/commons/src/main/java/it/smartcommunitylabdhub/commons/events/RunnableChangedEvent.java b/...s/commons/src/main/java/it/smartcommunitylabdhub/commons/events/RunnableChangedEvent.java
@@ -16,4 +16,8 @@ public class RunnableChangedEvent<R extends RunRunnable> {
 
     private RunnableMonitorObject runMonitorObject;
     private R runnable;
+
+    public String getId() {
+        return runnable != null ? runnable.getId() : null;
+    }
 }
diff --git a/...ons/src/main/java/it/smartcommunitylabdhub/commons/models/entities/run/RunBaseStatus.java b/...ons/src/main/java/it/smartcommunitylabdhub/commons/models/entities/run/RunBaseStatus.java
@@ -19,11 +19,24 @@
 public class RunBaseStatus extends BaseSpec {
 
     private String state;
+    private String message;
+
+    public RunBaseStatus(String state) {
+        this.state = state;
+    }
 
     @Override
     public void configure(Map<String, Serializable> data) {
-        RunBaseStatus meta = mapper.convertValue(data, RunBaseStatus.class);
+        RunBaseStatus spec = mapper.convertValue(data, RunBaseStatus.class);
+
+        this.state = spec.getState();
+        this.message = spec.getMessage();
+    }
+
+    public static RunBaseStatus with(Map<String, Serializable> data) {
+        RunBaseStatus spec = new RunBaseStatus();
+        spec.configure(data);
 
-        this.state = meta.getState();
+        return spec;
     }
 }
diff --git a/...amework-k8s/src/main/java/it/smartcommunitylabdhub/framework/k8s/base/K8sBaseRuntime.java b/...amework-k8s/src/main/java/it/smartcommunitylabdhub/framework/k8s/base/K8sBaseRuntime.java
@@ -1,11 +1,14 @@
 package it.smartcommunitylabdhub.framework.k8s.base;
 
+import it.smartcommunitylabdhub.commons.infrastructure.RunRunnable;
 import it.smartcommunitylabdhub.commons.models.base.ExecutableBaseSpec;
+import it.smartcommunitylabdhub.commons.models.entities.run.Run;
 import it.smartcommunitylabdhub.commons.models.entities.run.RunBaseSpec;
 import it.smartcommunitylabdhub.commons.models.entities.run.RunBaseStatus;
 import it.smartcommunitylabdhub.commons.runtimes.base.AbstractBaseRuntime;
 import it.smartcommunitylabdhub.framework.k8s.kubernetes.K8sBuilderHelper;
 import it.smartcommunitylabdhub.framework.k8s.runnables.K8sRunnable;
+import jakarta.validation.constraints.NotNull;
 import org.springframework.beans.factory.annotation.Autowired;
 
 public abstract class K8sBaseRuntime<
@@ -23,4 +26,21 @@ protected K8sBaseRuntime(String kind) {
     public void setK8sBuilderHelper(K8sBuilderHelper k8sBuilderHelper) {
         this.k8sBuilderHelper = k8sBuilderHelper;
     }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public Z onError(@NotNull Run run, RunRunnable runnable) {
+        if (runnable != null && runnable instanceof K8sRunnable) {
+            K8sRunnable k8sRunnable = (K8sRunnable) runnable;
+            RunBaseStatus status = RunBaseStatus
+                .baseBuilder()
+                .state(k8sRunnable.getState())
+                .message(k8sRunnable.getError())
+                .build();
+
+            return (Z) status;
+        }
+
+        return null;
+    }
 }
diff --git a/...java/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sCronJobMonitor.java b/...java/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sCronJobMonitor.java
@@ -37,7 +37,9 @@ public K8sCronJobRunnable refresh(K8sCronJobRunnable runnable) {
 
             if (job == null || job.getStatus() == null) {
                 // something is missing, no recovery
+                log.error("Missing or invalid job for {}", runnable.getId());
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Job missing or invalid");
             }
 
             log.info("Job status: {}", job.getStatus().toString());
@@ -88,6 +90,7 @@ public K8sCronJobRunnable refresh(K8sCronJobRunnable runnable) {
         } catch (K8sFrameworkException e) {
             // Set Runnable to ERROR state
             runnable.setState(State.ERROR.name());
+            runnable.setError(e.getClass().getSimpleName() + ":" + String.valueOf(e.getMessage()));
         }
 
         return runnable;

diff --git a/...a/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sDeploymentMonitor.java b/...a/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sDeploymentMonitor.java
@@ -46,6 +46,7 @@ public K8sDeploymentRunnable refresh(K8sDeploymentRunnable runnable) {
                 // something is missing, no recovery
                 log.error("Missing or invalid deployment for {}", runnable.getId());
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Deployment missing or invalid");
             }
 
             log.debug("deployment status: replicas {}", deployment.getStatus().getReadyReplicas());
@@ -94,6 +95,7 @@ public K8sDeploymentRunnable refresh(K8sDeploymentRunnable runnable) {
                     // we observed multiple restarts, stop it
                     log.error("Multiple restarts observed {}", runnable.getId());
                     runnable.setState(State.ERROR.name());
+                    runnable.setError("Multiple pod restarts");
                 }
             }
 
@@ -147,6 +149,7 @@ public K8sDeploymentRunnable refresh(K8sDeploymentRunnable runnable) {
         } catch (K8sFrameworkException e) {
             // Set Runnable to ERROR state
             runnable.setState(State.ERROR.name());
+            runnable.setError(e.getClass().getSimpleName() + ":" + String.valueOf(e.getMessage()));
         }
 
         return runnable;

diff --git a/...ain/java/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sJobMonitor.java b/...ain/java/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sJobMonitor.java
@@ -41,6 +41,7 @@ public K8sJobRunnable refresh(K8sJobRunnable runnable) {
                 // something is missing, no recovery
                 log.error("Missing or invalid job for {}", runnable.getId());
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Job missing or invalid");
             }
 
             if (log.isTraceEnabled()) {
@@ -54,8 +55,9 @@ public K8sJobRunnable refresh(K8sJobRunnable runnable) {
                 runnable.setState(State.COMPLETED.name());
             } else if (job.getStatus().getFailed() != null && job.getStatus().getFailed().intValue() > 0) {
                 // Job has failed delete job and pod
-                log.debug("Job failed succeeded for {}", runnable.getId());
+                log.debug("Job status failed for {}", runnable.getId());
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Job failed: " + job.getStatus().getFailed());
             }
 
             //try to fetch pods
@@ -109,6 +111,7 @@ public K8sJobRunnable refresh(K8sJobRunnable runnable) {
         } catch (K8sFrameworkException e) {
             // Set Runnable to ERROR state
             runnable.setState(State.ERROR.name());
+            runnable.setError(e.getClass().getSimpleName() + ":" + String.valueOf(e.getMessage()));
         }
 
         return runnable;

diff --git a/...n/java/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sServeMonitor.java b/...n/java/it/smartcommunitylabdhub/framework/k8s/infrastructure/monitor/K8sServeMonitor.java
@@ -56,6 +56,7 @@ public K8sServeRunnable refresh(K8sServeRunnable runnable) {
                 // something is missing, no recovery
                 log.error("Missing or invalid deployment for {}", runnable.getId());
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Deployment missing or invalid");
             }
 
             log.debug("deployment status: replicas {}", deployment.getStatus().getReadyReplicas());
@@ -105,6 +106,7 @@ public K8sServeRunnable refresh(K8sServeRunnable runnable) {
                     // we observed multiple restarts, stop it
                     log.error("Multiple restarts observed {}", runnable.getId());
                     runnable.setState(State.ERROR.name());
+                    runnable.setError("Multiple pod restarts");
                 }
             }
 
@@ -160,6 +162,7 @@ public K8sServeRunnable refresh(K8sServeRunnable runnable) {
         } catch (K8sFrameworkException e) {
             // Set Runnable to ERROR state
             runnable.setState(State.ERROR.name());
+            runnable.setError(e.getClass().getSimpleName() + ":" + String.valueOf(e.getMessage()));
         }
 
         return runnable;

diff --git a/...s/src/main/java/it/smartcommunitylabdhub/framework/k8s/listeners/K8sRunnableListener.java b/...s/src/main/java/it/smartcommunitylabdhub/framework/k8s/listeners/K8sRunnableListener.java
@@ -84,6 +84,7 @@ public void process(R runnable) {
             // Set runnable to error state send event
             log.error("Error with k8s for runnable {} {}: {}", clazz.getSimpleName(), runnable.getId(), e.getMessage());
             runnable.setState(State.ERROR.name());
+            runnable.setError(clazz.getSimpleName() + ":" + String.valueOf(e.getMessage()));
 
             try {
                 log.debug("update runnable {} {} in store", clazz.getSimpleName(), runnable.getId());

diff --git a/...ework-k8s/src/main/java/it/smartcommunitylabdhub/framework/k8s/runnables/K8sRunnable.java b/...ework-k8s/src/main/java/it/smartcommunitylabdhub/framework/k8s/runnables/K8sRunnable.java
@@ -80,6 +80,8 @@ public class K8sRunnable implements RunRunnable, SecuredRunnable, CredentialsCon
 
     private String state;
 
+    private String error;
+
     private Map<String, Serializable> results;
 
     @JsonIgnore

diff --git a/...va/it/smartcommunitylabdhub/framework/kaniko/infrastructure/monitor/K8sKanikoMonitor.java b/...va/it/smartcommunitylabdhub/framework/kaniko/infrastructure/monitor/K8sKanikoMonitor.java
@@ -38,7 +38,9 @@ public K8sKanikoRunnable refresh(K8sKanikoRunnable runnable) {
 
             if (job == null || job.getStatus() == null) {
                 // something is missing, no recovery
+                log.error("Missing or invalid job for {}", runnable.getId());
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Job missing or invalid");
             }
 
             log.info("Job status: {}", job.getStatus().toString());
@@ -50,6 +52,7 @@ public K8sKanikoRunnable refresh(K8sKanikoRunnable runnable) {
             } else if (job.getStatus().getFailed() != null && job.getStatus().getFailed().intValue() > 0) {
                 // Job has failed delete job and pod
                 runnable.setState(State.ERROR.name());
+                runnable.setError("Job failed: " + job.getStatus().getFailed());
             }
 
             //try to fetch pods
@@ -91,6 +94,7 @@ public K8sKanikoRunnable refresh(K8sKanikoRunnable runnable) {
         } catch (K8sFrameworkException e) {
             // Set Runnable to ERROR state
             runnable.setState(State.ERROR.name());
+            runnable.setError(e.getClass().getSimpleName() + ":" + String.valueOf(e.getMessage()));
         }
 
         return runnable;

diff --git a/...n/java/it/smartcommunitylabdhub/framework/kaniko/listeners/K8sKanikoRunnableListener.java b/...n/java/it/smartcommunitylabdhub/framework/kaniko/listeners/K8sKanikoRunnableListener.java
@@ -71,6 +71,7 @@ public void listen(K8sKanikoRunnable runnable) {
             // Set runnable to error state send event
             log.error("Error with k8s: {}", e.getMessage());
             runnable.setState(State.ERROR.name());
+            runnable.setError(e.getClass().getSimpleName() + ":" + String.valueOf(e.getMessage()));
 
             try {
                 runnableStore.store(runnable.getId(), runnable);