Stream events (#405)

onyxia-api/pom.xml

@@ -36,6 +36,10 @@
             <artifactId>spring-restdocs-mockmvc</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springdoc</groupId>
             <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>

onyxia-api/src/main/java/fr/insee/onyxia/api/controller/api/mylab/MyLabController.java

@@ -1,5 +1,6 @@
 package fr.insee.onyxia.api.controller.api.mylab;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import fr.insee.onyxia.api.configuration.CatalogWrapper;
 import fr.insee.onyxia.api.configuration.Catalogs;
 import fr.insee.onyxia.api.configuration.NotFoundException;
@@ -14,16 +15,21 @@
 import fr.insee.onyxia.model.region.Region;
 import fr.insee.onyxia.model.service.Service;
 import fr.insee.onyxia.model.service.UninstallService;
+import io.fabric8.kubernetes.api.model.Event;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.enums.ParameterIn;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.security.SecurityRequirement;
 import io.swagger.v3.oas.annotations.tags.Tag;
+import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 @Tag(name = "My lab", description = "My services")
 @RequestMapping("/my-lab")
@@ -36,12 +42,18 @@ public class MyLabController {
 
     private final CatalogService catalogService;
 
+    private ObjectMapper objectMapper;
+
     @Autowired
     public MyLabController(
-            AppsService helmAppsService, UserProvider userProvider, CatalogService catalogService) {
+            AppsService helmAppsService,
+            UserProvider userProvider,
+            CatalogService catalogService,
+            ObjectMapper objectMapper) {
         this.helmAppsService = helmAppsService;
         this.userProvider = userProvider;
         this.catalogService = catalogService;
+        this.objectMapper = objectMapper;
     }
 
     @Operation(
@@ -171,6 +183,52 @@ public String getLogs(
         return null;
     }
 
+    @Operation(
+            summary = "Stream events for the entire user namespace",
+            description = "Stream events for the entire user namespace.",
+            parameters = {
+                @Parameter(
+                        name = "ONYXIA-PROJECT",
+                        description =
+                                "Project associated with the namespace, defaults to user project.",
+                        in = ParameterIn.HEADER,
+                        schema =
+                                @Schema(
+                                        name = "ONYXIA-PROJECT",
+                                        type = "string",
+                                        description = "Generated project id.",
+                                        example = "project-id-example"))
+            })
+    @GetMapping("/events")
+    public SseEmitter getEvents(
+            @Parameter(hidden = true) Region region, @Parameter(hidden = true) Project project)
+            throws Exception {
+
+        if (Service.ServiceType.KUBERNETES.equals(region.getServices().getType())) {
+            final SseEmitter emitter = new SseEmitter();
+
+            Watcher<Event> watcher =
+                    new Watcher<Event>() {
+                        @Override
+                        public void eventReceived(Action action, Event event) {
+                            try {
+                                emitter.send(objectMapper.writeValueAsString(event));
+                            } catch (IOException e) {
+                                // Response is already commited
+                            }
+                        }
+
+                        @Override
+                        public void onClose(WatcherException e) {
+                            emitter.complete();
+                        }
+                    };
+            helmAppsService.getEvents(region, project, userProvider.getUser(region), watcher);
+            return emitter;
+        }
+        return null;
+    }
+
     @Operation(
             summary = "Delete an installed service(s) launched through Onyxia.",
             description =

onyxia-api/src/main/java/fr/insee/onyxia/api/services/AppsService.java

@@ -8,7 +8,11 @@
 import fr.insee.onyxia.model.region.Region;
 import fr.insee.onyxia.model.service.Service;
 import fr.insee.onyxia.model.service.UninstallService;
+import io.fabric8.kubernetes.api.model.Event;
+import io.fabric8.kubernetes.client.Watcher;
+import io.github.inseefrlab.helmwrapper.service.HelmInstallService;
 import java.io.IOException;
+import java.text.ParseException;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -44,4 +48,7 @@ UninstallService destroyService(
             Region region, Project project, User user, String path, boolean bulk) throws Exception;
 
     String getLogs(Region region, Project project, User user, String serviceId, String taskId);
+
+    void getEvents(Region region, Project project, User user, Watcher<Event> watcher)
+            throws HelmInstallService.MultipleServiceFound, ParseException;
 }

onyxia-api/src/main/java/fr/insee/onyxia/api/services/impl/HelmAppsService.java

@@ -28,6 +28,7 @@
 import fr.insee.onyxia.model.service.*;
 import io.fabric8.kubernetes.api.model.EventList;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.Watcher;
 import io.github.inseefrlab.helmwrapper.configuration.HelmConfiguration;
 import io.github.inseefrlab.helmwrapper.model.HelmInstaller;
 import io.github.inseefrlab.helmwrapper.model.HelmLs;
@@ -290,6 +291,21 @@ public String getLogs(
                 .getLog();
     }
 
+    @Override
+    public void getEvents(
+            Region region,
+            Project project,
+            User user,
+            Watcher<io.fabric8.kubernetes.api.model.Event> watcher) {
+        KubernetesClient client = kubernetesClientProvider.getUserClient(region, user);
+        client.v1()
+                .events()
+                .inNamespace(
+                        kubernetesService.determineNamespaceAndCreateIfNeeded(
+                                region, project, user))
+                .watch(watcher);
+    }
+
     @Override
     public Service getUserService(Region region, Project project, User user, String serviceId)
             throws MultipleServiceFound, ParseException {

onyxia-api/src/main/resources/application.properties

@@ -28,4 +28,6 @@ springdoc.swagger-ui.oauth.clientId=
 # useless because it's javascript
 springdoc.swagger-ui.oauth.clientSecret=
 # Enable events logging
-event.logging.enabled=true
+event.logging.enabled=true
+# Response stream configuration
+spring.mvc.async.request-timeout: 600000