Programmatically create permanent session

server/src/main/java/com/exacaster/lighter/application/Application.java

@@ -1,6 +1,7 @@
 package com.exacaster.lighter.application;
 
 import com.exacaster.lighter.storage.Entity;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.micronaut.core.annotation.Introspected;
 
 import java.time.LocalDateTime;
@@ -22,9 +23,11 @@ public class Application implements Entity {
     private final LocalDateTime createdAt;
     private final LocalDateTime contactedAt;
 
+    private final boolean deleted;
+
     public Application(String id, ApplicationType type, ApplicationState state, String appId, String appInfo,
             SubmitParams submitParams,
-            LocalDateTime createdAt, LocalDateTime contactedAt) {
+            LocalDateTime createdAt, LocalDateTime contactedAt, boolean deleted) {
         this.id = id;
         this.type = type;
         this.state = state;
@@ -34,6 +37,7 @@ public Application(String id, ApplicationType type, ApplicationState state, Stri
         this.createdAt = createdAt;
         this.contactedAt = contactedAt;
         this.kind = "pyspark";
+        this.deleted = deleted;
     }
 
     @Override
@@ -56,7 +60,7 @@ public String getAppId() {
     public String getAppInfo() {
         return appInfo;
     }
-    
+
     public List<String> getLog() {
         return log;
     }
@@ -65,6 +69,16 @@ public SubmitParams getSubmitParams() {
         return submitParams;
     }
 
+    @JsonIgnore
+    public boolean isDeleted() {
+        return deleted;
+    }
+
+    @JsonIgnore
+    public boolean isNotDeleted() {
+        return !isDeleted();
+    }
+
     @Override
     public LocalDateTime getCreatedAt() {
         return createdAt;
@@ -89,6 +103,7 @@ public String toString() {
                 .add("submitParams=" + submitParams)
                 .add("createdAt=" + createdAt)
                 .add("contactedAt=" + contactedAt)
+                .add("deleted=" + deleted)
                 .toString();
     }
 }

server/src/main/java/com/exacaster/lighter/application/ApplicationBuilder.java

@@ -12,6 +12,7 @@ public class ApplicationBuilder {
     private SubmitParams submitParams;
     private LocalDateTime createdAt;
     private LocalDateTime contactedAt;
+    private boolean deleted;
 
     private ApplicationBuilder() {
 
@@ -75,7 +76,12 @@ public ApplicationBuilder setContactedAt(LocalDateTime contactedAt) {
         return this;
     }
 
+    public ApplicationBuilder setDeleted(boolean deleted) {
+        this.deleted = deleted;
+        return this;
+    }
+
     public Application build() {
-        return new Application(id, type, state, appId, appInfo, submitParams, createdAt, contactedAt);
+        return new Application(id, type, state, appId, appInfo, submitParams, createdAt, contactedAt, deleted);
     }
 }

server/src/main/java/com/exacaster/lighter/application/ApplicationType.java

@@ -2,5 +2,6 @@
 
 public enum ApplicationType {
     BATCH,
-    SESSION
+    SESSION,
+    PERMANENT_SESSION,
 }

server/src/main/java/com/exacaster/lighter/application/SubmitParams.java

@@ -1,16 +1,17 @@
 package com.exacaster.lighter.application;
 
-import static io.micronaut.core.convert.format.MapFormat.MapTransformation.FLAT;
-import static io.micronaut.core.naming.conventions.StringConvention.RAW;
-import static java.util.Optional.ofNullable;
-
 import io.micronaut.core.annotation.Introspected;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.core.convert.format.MapFormat;
+
 import java.util.List;
 import java.util.Map;
 import java.util.StringJoiner;
 
+import static io.micronaut.core.convert.format.MapFormat.MapTransformation.FLAT;
+import static io.micronaut.core.naming.conventions.StringConvention.RAW;
+import static java.util.Optional.ofNullable;
+
 @Introspected
 public class SubmitParams {
 
@@ -148,4 +149,5 @@ public String toString() {
                 .add("conf=" + conf)
                 .toString();
     }
+
 }

server/src/main/java/com/exacaster/lighter/application/batch/BatchService.java

@@ -4,15 +4,17 @@
 import com.exacaster.lighter.application.ApplicationBuilder;
 import com.exacaster.lighter.application.ApplicationState;
 import com.exacaster.lighter.application.ApplicationType;
-import com.exacaster.lighter.backend.Backend;
 import com.exacaster.lighter.application.SubmitParams;
+import com.exacaster.lighter.backend.Backend;
 import com.exacaster.lighter.storage.ApplicationStorage;
 import com.exacaster.lighter.storage.SortOrder;
+import jakarta.inject.Singleton;
+
 import java.time.LocalDateTime;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
-import jakarta.inject.Singleton;
 
 @Singleton
 public class BatchService {
@@ -26,7 +28,7 @@ public BatchService(ApplicationStorage applicationStorage, Backend backend) {
     }
 
     public List<Application> fetch(Integer from, Integer size) {
-        return applicationStorage.findApplications(ApplicationType.BATCH, from, size);
+        return applicationStorage.findApplications(EnumSet.of(ApplicationType.BATCH), from, size);
     }
 
     public Application create(SubmitParams batch) {

server/src/main/java/com/exacaster/lighter/application/sessions/SessionHandler.java

@@ -1,26 +1,31 @@
 package com.exacaster.lighter.application.sessions;
 
-import static java.util.Optional.ofNullable;
-import static net.javacrumbs.shedlock.core.LockAssert.assertLocked;
-import static org.slf4j.LoggerFactory.getLogger;
-
 import com.exacaster.lighter.application.Application;
 import com.exacaster.lighter.application.ApplicationInfo;
 import com.exacaster.lighter.application.ApplicationState;
 import com.exacaster.lighter.application.ApplicationStatusHandler;
+import com.exacaster.lighter.application.SubmitParams;
 import com.exacaster.lighter.application.sessions.processors.StatementHandler;
 import com.exacaster.lighter.backend.Backend;
 import com.exacaster.lighter.concurrency.Waitable;
 import com.exacaster.lighter.configuration.AppConfiguration;
 import com.exacaster.lighter.storage.SortOrder;
 import io.micronaut.scheduling.annotation.Scheduled;
 import jakarta.inject.Singleton;
+import net.javacrumbs.shedlock.micronaut.SchedulerLock;
+import org.apache.hadoop.thirdparty.com.google.common.collect.Sets;
+import org.slf4j.Logger;
+
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.stream.Collectors;
-import net.javacrumbs.shedlock.micronaut.SchedulerLock;
-import org.slf4j.Logger;
+import java.util.stream.Stream;
+
+import static java.util.Optional.ofNullable;
+import static net.javacrumbs.shedlock.core.LockAssert.assertLocked;
+import static org.slf4j.LoggerFactory.getLogger;
 
 @Singleton
 public class SessionHandler {
@@ -51,28 +56,53 @@ public Waitable launch(Application application, Consumer<Throwable> errorHandler
     }
 
     @SchedulerLock(name = "keepPermanentSession", lockAtLeastFor = "1m")
-    @Scheduled(fixedRate = "1m")
+    @Scheduled(fixedRate = "1m", initialDelay = "2s")
     public void keepPermanentSessions() throws InterruptedException {
         assertLocked();
         LOG.info("Start provisioning permanent sessions.");
-        for (var sessionConf : appConfiguration.getSessionConfiguration().getPermanentSessions()) {
-            var session = sessionService.fetchOne(sessionConf.getId());
+
+        final var allPermanentSessions = getPermanentSessionToCheck();
+
+        for (var perm : allPermanentSessions) {
+            var session = sessionService.fetchOne(perm.getSessionId());
             if (session.map(Application::getState).filter(this::running).isEmpty() ||
                     session.flatMap(backend::getInfo).map(ApplicationInfo::getState).filter(this::running).isEmpty()) {
-                LOG.info("Permanent session {} needs to be (re)started.", sessionConf.getId());
-                var sessionToLaunch = sessionService.createSession(
-                        sessionConf.getSubmitParams(),
-                        sessionConf.getId()
+                LOG.info("Permanent session {} needs to be (re)started.", perm.getSessionId());
+                sessionService.deletePermanentSession(perm.sessionId);
+                var sessionToLaunch = sessionService.createPermanentSession(
+                        perm.getSessionId(),
+                        perm.getSubmitParams()
                 );
 
-                sessionService.deleteOne(sessionToLaunch);
                 launchSession(sessionToLaunch).waitCompletion();
-                LOG.info("Permanent session {} (re)started.", sessionConf.getId());
+                LOG.info("Permanent session {} (re)started.", perm.getSessionId());
             }
         }
         LOG.info("End provisioning permanent sessions.");
     }
 
+    private List<PermanentSessionParam> getPermanentSessionToCheck() {
+        final var dbPermanentSessions = sessionService.fetchAllPermanentSessions();
+
+        final var configurationPermanentSessions = appConfiguration.getSessionConfiguration().getPermanentSessions().stream().collect(
+                Collectors.toMap(permanentSession -> permanentSession.getId(), Function.identity()));
+
+        final var fromYamlOnly = Sets.difference(configurationPermanentSessions.keySet(), dbPermanentSessions.keySet()).stream().map(
+                id -> new PermanentSessionParam(id, configurationPermanentSessions.get(id).getSubmitParams())
+        );
+
+        final var intersection = Sets.intersection(configurationPermanentSessions.keySet(), dbPermanentSessions.keySet()).stream()
+                .filter(id -> dbPermanentSessions.get(id).isNotDeleted())
+                .map(id -> new PermanentSessionParam(id, dbPermanentSessions.get(id).getSubmitParams()));
+
+        final var fromStorageOnly = Sets.difference(dbPermanentSessions.keySet(), configurationPermanentSessions.keySet()).stream().map(
+                id -> new PermanentSessionParam(id, dbPermanentSessions.get(id).getSubmitParams())
+        );
+
+        return Stream.concat(fromStorageOnly, Stream.concat(fromYamlOnly, intersection)).collect(Collectors.toList());
+
+    }
+
     @SchedulerLock(name = "processScheduledSessions")
     @Scheduled(fixedRate = "${lighter.session.schedule-interval}")
     public void processScheduledSessions() throws InterruptedException {
@@ -96,7 +126,7 @@ private Waitable launchSession(Application session) {
     @Scheduled(fixedRate = "${lighter.session.track-running-interval}")
     public void trackRunning() {
         assertLocked();
-        var running = sessionService.fetchRunning();
+        var running = sessionService.fetchRunningSession();
 
         var idleAndRunning = running.stream()
                 .collect(Collectors.groupingBy(statementStatusChecker::hasWaitingStatement));
@@ -112,9 +142,8 @@ public void handleTimeout() {
         var sessionConfiguration = appConfiguration.getSessionConfiguration();
         var timeoutInterval = sessionConfiguration.getTimeoutInterval();
         if (timeoutInterval != null && !timeoutInterval.isZero()) {
-            sessionService.fetchRunning()
+            sessionService.fetchRunningSession()
                     .stream()
-                    .filter(s -> isNotPermanent(sessionConfiguration, s))
                     .filter(s -> sessionConfiguration.shouldTimeoutActive() || !sessionService.isActive(s))
                     .filter(s -> sessionService.lastUsed(s.getId()).isBefore(LocalDateTime.now().minus(timeoutInterval)))
                     .peek(s -> LOG.info("Killing because of timeout {}, session: {}", timeoutInterval, s))
@@ -135,4 +164,22 @@ private <T> List<T> selfOrEmpty(List<T> list) {
     private boolean running(ApplicationState state) {
         return !state.isComplete();
     }
+
+    private static class PermanentSessionParam {
+        private final String sessionId;
+        private final SubmitParams submitParams;
+
+        public PermanentSessionParam(String sessionId, SubmitParams submitParams) {
+            this.sessionId = sessionId;
+            this.submitParams = submitParams;
+        }
+
+        public String getSessionId() {
+            return sessionId;
+        }
+
+        public SubmitParams getSubmitParams() {
+            return submitParams;
+        }
+    }
 }