From 3a42f02fc0dff54e611cc66800ce9a0307312611 Mon Sep 17 00:00:00 2001 From: Minutis Date: Tue, 5 Sep 2023 20:15:12 +0300 Subject: [PATCH] Parametrise interval for scheduling of new sessions Parametrise interval for tracking running sessions Use java.time.Duration for session timeout Closes #607 --- docs/configuration.md | 6 ++-- .../application/sessions/SessionHandler.java | 12 +++---- .../configuration/AppConfiguration.java | 32 +++++++++++++++---- server/src/main/resources/application.yml | 4 ++- .../sessions/SessionHandlerTest.groovy | 4 +-- .../configuration/AppConfigurationTest.groovy | 2 +- .../exacaster/lighter/test/Factories.groovy | 6 ++-- 7 files changed, 45 insertions(+), 21 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 328204a1..c2ef2edf 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -12,8 +12,10 @@ Lighter can be configured by using environment variables. Currently, Lighter sup | LIGHTER_EXTERNAL_LOGS_URL_TEMPLATE | Template for link to external logs (Grafana, Graylog, etc.) used on the Lighter UI. Allowed placeholders: `{{id}}`, `{{appId}}`, `{{createdTs}}` | | | LIGHTER_PY_GATEWAY_PORT | Port for live Spark session communication | 25333 | | LIGHTER_URL | URL which can be used to access Lighter form Spark Job | http://lighter.spark:8080 | -| LIGHTER_SESSION_TIMEOUT_MINUTES | Session lifetime in minutes (from last statement creation). Use negative value to disable | 90 | -| LIGHTER_SESSION_TIMEOUT_ACTIVE | Should Lighter kill sessions with waiting statements (obsolete when `LIGHTER_SESSION_TIMEOUT_MINUTES` is negative) | false | +| LIGHTER_SESSION_TIMEOUT_INTERVAL | `java.time.Duration` representing session lifetime (from last statement creation). Use `0m` value to disable | 90m | +| LIGHTER_SESSION_TIMEOUT_ACTIVE | Should Lighter kill sessions with waiting statements (obsolete when `LIGHTER_SESSION_TIMEOUT_INTERVAL` is `0m`) | false | +| LIGHTER_SESSION_SCHEDULE_INTERVAL | `java.time.Duration` representing the interval at which a task is triggered to initiate scheduled sessions | 1m | +| LIGHTER_SESSION_TRACK_RUNNING_INTERVAL | `java.time.Duration` representing the interval at which a task is triggered to process and update running session state | 2m | | LIGHTER_STORAGE_JDBC_URL | JDBC url for lighter storage | jdbc:h2:mem:lighter | | LIGHTER_STORAGE_JDBC_USERNAME | JDBC username | sa | | LIGHTER_STORAGE_JDBC_PASSWORD | JDBC password | | diff --git a/server/src/main/java/com/exacaster/lighter/application/sessions/SessionHandler.java b/server/src/main/java/com/exacaster/lighter/application/sessions/SessionHandler.java index 757f14cf..7dc7d5f7 100644 --- a/server/src/main/java/com/exacaster/lighter/application/sessions/SessionHandler.java +++ b/server/src/main/java/com/exacaster/lighter/application/sessions/SessionHandler.java @@ -74,7 +74,7 @@ public void keepPermanentSessions() throws InterruptedException { } @SchedulerLock(name = "processScheduledSessions") - @Scheduled(fixedRate = "1m") + @Scheduled(fixedRate = "${lighter.session.schedule-interval}") public void processScheduledSessions() throws InterruptedException { assertLocked(); var waitables = sessionService.fetchByState(ApplicationState.NOT_STARTED, SortOrder.ASC, 10).stream() @@ -93,7 +93,7 @@ private Waitable launchSession(Application session) { } @SchedulerLock(name = "trackRunningSessions", lockAtMostFor = "1m") - @Scheduled(fixedRate = "2m") + @Scheduled(fixedRate = "${lighter.session.track-running-interval}") public void trackRunning() { assertLocked(); var running = sessionService.fetchRunning(); @@ -110,14 +110,14 @@ public void trackRunning() { public void handleTimeout() { assertLocked(); var sessionConfiguration = appConfiguration.getSessionConfiguration(); - var timeout = sessionConfiguration.getTimeoutMinutes(); - if (timeout != null && timeout > 0) { + var timeoutInterval = sessionConfiguration.getTimeoutInterval(); + if (timeoutInterval != null && !timeoutInterval.isZero()) { sessionService.fetchRunning() .stream() .filter(s -> isNotPermanent(sessionConfiguration, s)) .filter(s -> sessionConfiguration.shouldTimeoutActive() || !sessionService.isActive(s)) - .filter(s -> sessionService.lastUsed(s.getId()).isBefore(LocalDateTime.now().minusMinutes(timeout))) - .peek(s -> LOG.info("Killing because of timeout {}, session: {}", timeout, s)) + .filter(s -> sessionService.lastUsed(s.getId()).isBefore(LocalDateTime.now().minus(timeoutInterval))) + .peek(s -> LOG.info("Killing because of timeout {}, session: {}", timeoutInterval, s)) .forEach(sessionService::killOne); } diff --git a/server/src/main/java/com/exacaster/lighter/configuration/AppConfiguration.java b/server/src/main/java/com/exacaster/lighter/configuration/AppConfiguration.java index 0fef8090..018cf47f 100644 --- a/server/src/main/java/com/exacaster/lighter/configuration/AppConfiguration.java +++ b/server/src/main/java/com/exacaster/lighter/configuration/AppConfiguration.java @@ -14,6 +14,8 @@ import io.micronaut.core.annotation.Introspected; import io.micronaut.core.annotation.Nullable; import io.micronaut.core.convert.format.MapFormat; + +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.StringJoiner; @@ -140,21 +142,27 @@ public String toString() { @ConfigurationProperties("session") public static class SessionConfiguration { - private final Integer timeoutMinutes; + private final Duration timeoutInterval; private final Boolean timeoutActive; private final List permanentSessions; + private final Duration scheduleInterval; + private final Duration trackRunningInterval; @ConfigurationInject - public SessionConfiguration(@Nullable Integer timeoutMinutes, + public SessionConfiguration(@Nullable Duration timeoutInterval, Boolean timeoutActive, - List permanentSessions) { - this.timeoutMinutes = timeoutMinutes; + List permanentSessions, + Duration scheduleInterval, + Duration trackRunningInterval) { + this.timeoutInterval = timeoutInterval; this.timeoutActive = timeoutActive; this.permanentSessions = permanentSessions; + this.scheduleInterval = scheduleInterval; + this.trackRunningInterval = trackRunningInterval; } - public Integer getTimeoutMinutes() { - return timeoutMinutes; + public Duration getTimeoutInterval() { + return timeoutInterval; } public boolean shouldTimeoutActive() { @@ -165,11 +173,21 @@ public List getPermanentSessions() { return permanentSessions; } + public Duration getScheduleInterval() { + return scheduleInterval; + } + + public Duration getTrackRunningInterval() { + return trackRunningInterval; + } + @Override public String toString() { return new StringJoiner(", ", SessionConfiguration.class.getSimpleName() + "[", "]") - .add("timeoutMinutes=" + timeoutMinutes) + .add("timeoutMinutes=" + timeoutInterval) .add("permanentSessions=" + permanentSessions) + .add("scheduleIntervalSeconds=" + scheduleInterval) + .add("trackRunningInterval=" + trackRunningInterval) .toString(); } } diff --git a/server/src/main/resources/application.yml b/server/src/main/resources/application.yml index 85d86840..3a7e07a9 100644 --- a/server/src/main/resources/application.yml +++ b/server/src/main/resources/application.yml @@ -6,9 +6,11 @@ lighter: py-gateway-port: 25333 url: http://lighter.spark:8080 session: - timeout-minutes: 90 + timeout-interval: 90m timeout-active: false permanent-sessions: [] + schedule-interval: 1m + track-running-interval: 2m kubernetes: enabled: false master: k8s://kubernetes.default.svc.cluster.local:443 diff --git a/server/src/test/groovy/com/exacaster/lighter/application/sessions/SessionHandlerTest.groovy b/server/src/test/groovy/com/exacaster/lighter/application/sessions/SessionHandlerTest.groovy index 388ea5ff..1e5db9e5 100644 --- a/server/src/test/groovy/com/exacaster/lighter/application/sessions/SessionHandlerTest.groovy +++ b/server/src/test/groovy/com/exacaster/lighter/application/sessions/SessionHandlerTest.groovy @@ -35,7 +35,7 @@ class SessionHandlerTest extends Specification { def "kills timeouted sessions"() { given: def oldSession = newSession() - service.lastUsed(oldSession.id) >> LocalDateTime.now().minusMinutes(conf.sessionConfiguration.timeoutMinutes + 1) + service.lastUsed(oldSession.id) >> LocalDateTime.now() - conf.sessionConfiguration.timeoutInterval.plusMinutes(1) def newSession = app() service.lastUsed(newSession.id) >> newSession.createdAt @@ -62,7 +62,7 @@ class SessionHandlerTest extends Specification { def "preserves active timeouted sessions"() { given: def oldSession = newSession() - service.lastUsed(oldSession.id) >> LocalDateTime.now().minusMinutes(conf.sessionConfiguration.timeoutMinutes + 1) + service.lastUsed(oldSession.id) >> LocalDateTime.now() - conf.sessionConfiguration.timeoutInterval.plusMinutes(1) service.isActive(oldSession) >> true 1 * service.fetchRunning() >> [ diff --git a/server/src/test/groovy/com/exacaster/lighter/configuration/AppConfigurationTest.groovy b/server/src/test/groovy/com/exacaster/lighter/configuration/AppConfigurationTest.groovy index 782d2072..fd6b6503 100644 --- a/server/src/test/groovy/com/exacaster/lighter/configuration/AppConfigurationTest.groovy +++ b/server/src/test/groovy/com/exacaster/lighter/configuration/AppConfigurationTest.groovy @@ -17,7 +17,7 @@ class AppConfigurationTest extends Specification { def "binds properties form yaml"() { expect: appConfiguration.maxRunningJobs == 5 - appConfiguration.sessionConfiguration.timeoutMinutes == 90 + appConfiguration.sessionConfiguration.timeoutInterval.toMinutes() == 90 appConfiguration.sessionConfiguration.permanentSessions.size() == 1 appConfiguration.sessionConfiguration.permanentSessions.get(0).id == "permanentId1" appConfiguration.sessionConfiguration.permanentSessions.get(0).submitParams.conf == [ diff --git a/server/src/test/groovy/com/exacaster/lighter/test/Factories.groovy b/server/src/test/groovy/com/exacaster/lighter/test/Factories.groovy index 71e551d0..77d346d0 100644 --- a/server/src/test/groovy/com/exacaster/lighter/test/Factories.groovy +++ b/server/src/test/groovy/com/exacaster/lighter/test/Factories.groovy @@ -9,6 +9,7 @@ import com.exacaster.lighter.backend.kubernetes.KubernetesProperties import com.exacaster.lighter.configuration.AppConfiguration import com.exacaster.lighter.application.SubmitParams +import java.time.Duration import java.time.LocalDateTime class Factories { @@ -62,8 +63,9 @@ class Factories { null, 5432, "http://lighter:8080", - new AppConfiguration.SessionConfiguration(20, false, - [new AppConfiguration.PermanentSession("permanentSessionId", submitParams())]), + new AppConfiguration.SessionConfiguration(Duration.ofMinutes(20), false, + [new AppConfiguration.PermanentSession("permanentSessionId", submitParams())] + , Duration.ofMinutes(1), Duration.ofMinutes(2)), ["spark.kubernetes.driverEnv.TEST": "test"], ["spark.kubernetes.driverEnv.TEST": "test"] )