#32 Provide endpoint for engine status and update analysis status when disconnected engine reconnects

datanode/pom.xml

@@ -86,7 +86,7 @@
         <dependency>
             <groupId>com.odysseusinc.arachne</groupId>
             <artifactId>execution-engine-commons</artifactId>
-            <version>2.0.0</version>
+            <version>2.1.1</version>
         </dependency>
         <dependency>
             <artifactId>arachne-sys-settings</artifactId>

datanode/src/main/java/com/odysseusinc/arachne/datanode/controller/admin/AdminController.java

@@ -16,33 +16,30 @@
 package com.odysseusinc.arachne.datanode.controller.admin;
 
 import com.odysseusinc.arachne.commons.api.v1.dto.util.JsonResult;
-import com.odysseusinc.arachne.datanode.Constants;
 import com.odysseusinc.arachne.datanode.controller.BaseController;
 import com.odysseusinc.arachne.datanode.dto.converters.AnalysisToSubmissionDTOConverter;
 import com.odysseusinc.arachne.datanode.dto.submission.SubmissionDTO;
 import com.odysseusinc.arachne.datanode.dto.user.UserDTO;
-import com.odysseusinc.arachne.datanode.exception.AuthException;
+import com.odysseusinc.arachne.datanode.engine.EngineStatusDTO;
+import com.odysseusinc.arachne.datanode.engine.EngineStatusService;
 import com.odysseusinc.arachne.datanode.exception.PermissionDeniedException;
 import com.odysseusinc.arachne.datanode.model.analysis.Analysis;
 import com.odysseusinc.arachne.datanode.model.user.User;
 import com.odysseusinc.arachne.datanode.repository.AnalysisRepository;
-import com.odysseusinc.arachne.datanode.service.AnalysisService;
-import com.odysseusinc.arachne.datanode.service.DataNodeService;
-import com.odysseusinc.arachne.datanode.service.ExecutionEngineIntegrationService;
-import com.odysseusinc.arachne.datanode.service.ExecutionEngineStatus;
 import com.odysseusinc.arachne.datanode.service.UserService;
 import io.swagger.annotations.ApiOperation;
+import lombok.Getter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.convert.support.GenericConversionService;
 import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.web.PageableDefault;
 import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
@@ -62,17 +59,13 @@ public class AdminController extends BaseController {
     public static final int DEFAULT_PAGE_SIZE = 10;
     private final Map<String, Consumer<List<String>>> propertiesMap = new HashMap<>();
     @Autowired
-    private AnalysisToSubmissionDTOConverter analysisToSubmissionDTO;
-    @Autowired
-    private  GenericConversionService conversionService;
-    @Autowired
-    private  AnalysisRepository analysisRepository;
+    private EngineStatusService engine;
     @Autowired
-    private  AnalysisService analysisService;
+    private AnalysisToSubmissionDTOConverter analysisToSubmissionDTO;
     @Autowired
-    private  DataNodeService dataNodeService;
+    private GenericConversionService conversionService;
     @Autowired
-    private  ExecutionEngineIntegrationService executionEngineIntegrationService;
+    private AnalysisRepository analysisRepository;
 
     public AdminController(UserService userService) {
         super(userService);
@@ -125,28 +118,12 @@ public JsonResult<?> removeAdmin(@PathVariable String username) {
         return new JsonResult<>(JsonResult.ErrorCode.NO_ERROR);
     }
 
-    @ApiOperation(value = "Invalidate all unfinished analyses")
-    @PostMapping(Constants.Api.Analysis.INVALIDATE_ALL_UNFINISHED)
-    public Integer invalidateAllUnfinishedAnalyses(final Principal principal) throws PermissionDeniedException {
-
-        if (principal == null) {
-            throw new AuthException("user not found");
-        }
-
-        return analysisService.invalidateAllUnfinishedAnalyses(getUser(principal));
-    }
-
     @ApiOperation(value = "list submissions")
     @GetMapping("/api/v1/admin/submissions")
     public Page<SubmissionDTO> list(@PageableDefault(value = DEFAULT_PAGE_SIZE, sort = "id",
             direction = Sort.Direction.DESC) Pageable pageable) {
 
-        Pageable p;
-        if (!isCustomSort(pageable)) {
-            p = pageable;
-        } else {
-            p = buildPageRequest(pageable);
-        }
+        Pageable p = isCustomSort(pageable) ? buildPageRequest(pageable) : pageable;
         Page<Analysis> analyses;
         if (isFinishedSort(pageable)) {
             analyses = analysisRepository.findAllPagedOrderByFinished(p);
@@ -157,21 +134,12 @@ public Page<SubmissionDTO> list(@PageableDefault(value = DEFAULT_PAGE_SIZE, sort
         } else {
             analyses = analysisRepository.findAll(p);
         }
-        return analyses.map(analysis -> analysisToSubmissionDTO.convert(analysis));
-    }
-
-    @ApiOperation(value = "get execution engine status")
-    @GetMapping("/api/v1/admin/execution-engine/status")
-    public EngineStatusResponse getExecutionEngineStatus() {
-        return new EngineStatusResponse(executionEngineIntegrationService.getExecutionEngineStatus());
+        Page<SubmissionDTO> items = analyses.map(analysis -> analysisToSubmissionDTO.convert(analysis));
+        EngineStatusDTO engineStatus = engine.getStatusInfo();
+        return new PageWithStatus(items, engineStatus);
     }
 
     protected Pageable buildPageRequest(Pageable pageable) {
-
-        if (pageable.getSort() == null) {
-            return pageable;
-        }
-        PageRequest result;
         List<String> properties = new LinkedList<>();
         Sort.Direction direction = Sort.Direction.ASC;
         for (Sort.Order order : pageable.getSort()) {
@@ -183,10 +151,9 @@ protected Pageable buildPageRequest(Pageable pageable) {
                 properties.add(property);
             }
         }
-        result = PageRequest.of(pageable.getPageNumber(), pageable.getPageSize(), direction,
-                properties.toArray(new String[properties.size()]));
 
-        return result;
+        return PageRequest.of(pageable.getPageNumber(), pageable.getPageSize(), direction,
+                properties.toArray(new String[properties.size()]));
     }
 
     private boolean isCustomSort(final Pageable pageable) {
@@ -237,10 +204,13 @@ protected void initProps() {
         propertiesMap.put("status", p -> p.add("journal.state"));
     }
 
-    private class EngineStatusResponse {
-        public EngineStatusResponse(final ExecutionEngineStatus status) {
-            this.status = status;
+    @Getter
+    public static class PageWithStatus extends PageImpl<SubmissionDTO> {
+        private EngineStatusDTO engine;
+
+        public PageWithStatus(Page<SubmissionDTO> page, EngineStatusDTO engine) {
+            super(page.getContent(), page.getPageable(), page.getTotalElements());
+            this.engine = engine;
         }
-        public ExecutionEngineStatus status;
     }
 }

datanode/src/main/java/com/odysseusinc/arachne/datanode/controller/analysis/AnalysisController.java

@@ -27,9 +27,8 @@
 import com.odysseusinc.arachne.datanode.model.analysis.AnalysisFile;
 import com.odysseusinc.arachne.datanode.model.user.User;
 import com.odysseusinc.arachne.datanode.service.AnalysisResultsService;
+import com.odysseusinc.arachne.datanode.service.AnalysisService;
 import com.odysseusinc.arachne.datanode.service.UserService;
-import com.odysseusinc.arachne.datanode.service.impl.AnalysisResultsServiceImpl;
-import com.odysseusinc.arachne.datanode.service.impl.AnalysisServiceImpl;
 import com.odysseusinc.arachne.datanode.util.AddToZipFileVisitor;
 import com.odysseusinc.arachne.execution_engine_common.util.CommonFileUtils;
 import lombok.extern.slf4j.Slf4j;
@@ -76,7 +75,7 @@
 public class AnalysisController {
 
     @Autowired
-    private AnalysisServiceImpl analysisService;
+    private AnalysisService analysisService;
     @Autowired
     private AnalysisResultsService analysisResultsService;
     @Autowired
@@ -167,7 +166,7 @@ public void downloadResults(@PathVariable("id") Long analysisId, HttpServletResp
         String code = types().filter(t -> Objects.equals(t.name(), type)).findFirst().map(CommonAnalysisType::getCode).orElse(type);
         String filename = MessageFormat.format("{0}-a{1,number,#}-results", code, analysis.getId());
 
-        if (AnalysisResultsServiceImpl.isListOfArchive(resultFiles)) {
+        if (AnalysisResultsService.isListOfArchive(resultFiles)) {
             AnalysisFile headFile = resultFiles.stream().filter(f -> f.getLink().matches(".*\\.zip")).findFirst().orElseThrow(() -> {
                 log.error("Head file not found in multi-volume archive for results [{}]", analysisId);
                 return new IllegalOperationException(MessageFormat.format("No head file of multi-volume archvie for results [{0}]", analysisId));

datanode/src/main/java/com/odysseusinc/arachne/datanode/dto/converters/AnalysisToSubmissionDTOConverter.java

@@ -47,6 +47,8 @@ public SubmissionDTO convert(Analysis analysis) {
         dto.setId(analysis.getId());
         dto.setOrigin(analysis.getOrigin());
         dto.setStudy(analysis.getStudyTitle());
+        dto.setStage(analysis.getStage());
+        dto.setError(analysis.getError());
         DataSource dataSource = analysis.getDataSource();
         if (dataSource != null && conversionService.canConvert(dataSource.getClass(), DataSourceDTO.class)) {
             dto.setDataSource(conversionService.convert(dataSource, DataSourceDTO.class));

datanode/src/main/java/com/odysseusinc/arachne/datanode/dto/submission/SubmissionDTO.java

@@ -17,10 +17,11 @@
 import com.odysseusinc.arachne.datanode.dto.datasource.DataSourceDTO;
 import com.odysseusinc.arachne.datanode.model.analysis.AnalysisAuthor;
 import com.odysseusinc.arachne.datanode.model.analysis.AnalysisOrigin;
-import java.util.Date;
 import lombok.Getter;
 import lombok.Setter;
 
+import java.util.Date;
+
 @Getter
 @Setter
 public class SubmissionDTO {
@@ -34,4 +35,6 @@ public class SubmissionDTO {
     private Date submitted;
     private Date finished;
     private String environment;
+    private String stage;
+    private String error;
 }

datanode/src/main/java/com/odysseusinc/arachne/datanode/service/ExecutionEngineStatus.java → datanode/src/main/java/com/odysseusinc/arachne/datanode/engine/EngineStatusDTO.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021, 2023 Odysseus Data Services, Inc.
+ * Copyright 2024 Odysseus Data Services, Inc.
  * 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
@@ -12,9 +12,18 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.odysseusinc.arachne.datanode.service;
+package com.odysseusinc.arachne.datanode.engine;
 
-public enum ExecutionEngineStatus {
-    ONLINE,
-    OFFLINE
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.time.Instant;
+
+@Getter
+@RequiredArgsConstructor(staticName = "of")
+public class EngineStatusDTO {
+    private final String status;
+    private final Instant since;
+    private final Instant seenLast;
+    private final String error;
 }

datanode/src/main/java/com/odysseusinc/arachne/datanode/engine/EngineStatusService.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2024 Odysseus Data Services, Inc.
+ * 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.odysseusinc.arachne.datanode.engine;
+
+import com.odysseusinc.arachne.datanode.util.JpaSugar;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.persistence.EntityManager;
+import javax.persistence.PersistenceContext;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+/**
+ * Some statistics on the execution engine
+ */
+@Slf4j
+@Service
+public class EngineStatusService {
+    @PersistenceContext
+    private EntityManager em;
+
+    @Transactional
+    public void update(Instant timestamp, String error, Runnable onChange) {
+        getStatus().filter(status ->
+                Objects.equals(status.getError(), error)
+        ).orElseGet(() -> {
+            ExecutionEngineStatus status = new ExecutionEngineStatus();
+            status.setId(UUID.randomUUID());
+            status.setSince(timestamp);
+            status.setError(error);
+            em.persist(status);
+            onChange.run();
+            return status;
+        }).setSeenLast(timestamp);
+    }
+
+    @Transactional
+    public EngineStatusDTO getStatusInfo() {
+        return getStatus().map(status ->
+                EngineStatusDTO.of(status.getError() == null ? "OK" : "ERROR", status.getSince(), status.getSince(), status.getError())
+        ).orElseGet(() ->
+                EngineStatusDTO.of("UNKNOWN", null, null, null)
+        );
+    }
+
+    private Optional<ExecutionEngineStatus> getStatus() {
+        return JpaSugar.select(em, ExecutionEngineStatus.class, (cb, query) -> root ->
+            query.orderBy(cb.desc(root.get(ExecutionEngineStatus_.since)))
+        ).getResultStream().findFirst();
+    }
+
+}

datanode/src/main/java/com/odysseusinc/arachne/datanode/engine/ExecutionEngineStatus.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2024 Odysseus Data Services, Inc.
+ * 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.odysseusinc.arachne.datanode.engine;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.time.Instant;
+import java.util.UUID;
+
+@Getter
+@Setter
+@Entity
+@Table(name = "engine_status")
+@NoArgsConstructor
+public class ExecutionEngineStatus {
+    @Id
+    @Column(name = "id")
+    private UUID id;
+
+    /**
+     * Moment when the status was observed for the first time
+     */
+    @Column(name = "since")
+    private Instant since;
+
+    /**
+     * Moment when the status was observed last time.
+     */
+    @Column(name = "seen_last")
+    private Instant seenLast;
+
+    /**
+     * Error observed. Null if the record marks interval when EE was connected without errors
+     */
+    @Column(name = "error")
+    private String error;
+}