Added listener for foundry inactivity

Dockerfile

@@ -1,4 +1,4 @@
-FROM gradle:8.0-jdk19 AS build
+FROM gradle:8.12.0-jdk21 AS build
 COPY --chown=gradle:gradle . /home/gradle/src
 WORKDIR /home/gradle/src
 RUN gradle buildFatJar --no-daemon

build.gradle.kts

@@ -1,77 +1,49 @@
 import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
-val ktorVersion = "2.3.7"
-val kotlinVersion = "1.9.10"
-val logbackVersion = "1.4.12"
-val koinKtorVersion = "3.5.0"
-
-version = "0.6.0"
 plugins {
-	id("io.ktor.plugin") version "2.3.4"
-	id("org.jetbrains.kotlin.plugin.serialization") version "1.9.20"
-	kotlin("jvm") version "1.9.20"
-}
-
-repositories {
-	mavenCentral()
-	gradlePluginPortal()
+	alias(connectorLibs.plugins.kotlin.jvm)
+	alias(connectorLibs.plugins.kotlin.serialization)
+	alias(libs.plugins.ktor)
 }
 
 application {
 	mainClass.set("org.wagham.kabotapi.KabotApiApplicationKt")
 }
 
 group = "org.wagham"
+version = "0.6.0"
 
 dependencies {
 	implementation(project(":kabot-db-connector"))
-	implementation("org.jetbrains.kotlin:kotlin-reflect")
-	implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
-	implementation(group = "org.jetbrains.kotlinx", name = "kotlinx-coroutines-core", version = "1.7.3")
-	implementation(group = "org.jetbrains.kotlinx", name = "kotlinx-coroutines-reactor", version = "1.7.3")
-	implementation(group = "com.github.ben-manes.caffeine", name = "caffeine", version = "3.1.8")
-	implementation(group = "org.apache.tika", name = "tika-core", version = "2.9.1")
-
-	implementation(group = "io.ktor", name = "ktor-server-core-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-cors-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-content-negotiation-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-serialization-jackson-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-call-logging-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-cio-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-serialization-kotlinx-json", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-serialization-jackson", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-client-cio-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-client-core-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-client-content-negotiation-jvm", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-auth", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-auth-jwt", version = ktorVersion)
-	implementation(group = "io.ktor", name = "ktor-server-status-pages", version = ktorVersion)
-
-	implementation(group = "io.insert-koin", name = "koin-ktor", version = koinKtorVersion)
-	implementation(group = "io.insert-koin", name = "koin-logger-slf4j", version = koinKtorVersion)
-
-	implementation(group = "ch.qos.logback", name = "logback-classic", version = logbackVersion)
-
-	implementation(group = "org.mindrot", name = "jbcrypt", version = "0.4")
-
-	implementation(group="org.litote.kmongo", name="kmongo-coroutine", version = "4.7.0")
-
-	testImplementation(group = "org.junit.jupiter", name = "junit-jupiter", version = "5.10.1")
-	testImplementation(group="io.kotest", name="kotest-assertions-core-jvm", version="5.5.3")
-	testImplementation(group="io.kotest", name="kotest-framework-engine-jvm", version="5.5.3")
-	testImplementation(group = "io.kotest.extensions", name = "kotest-extensions-spring", version = "1.1.3")
+	implementation(libs.kotlin.stdlib)
+	implementation(libs.kotlin.reflect)
+	implementation(libs.bundles.ktor.server)
+	implementation(libs.bundles.ktor.serialization)
+	implementation(libs.bundles.ktor.client)
+	implementation(libs.bundles.koin)
+	implementation(connectorLibs.kotlinx.coroutines.core)
+	implementation(libs.caffeine)
+	implementation(libs.tika)
+	implementation(libs.kfswatch)
+	implementation(libs.logback)
+	implementation(libs.jbcrypt)
+	implementation(connectorLibs.kmongo)
+	implementation(libs.krontab)
+
+	testImplementation(connectorLibs.bundles.kotest)
+	testImplementation(libs.jupyter)
 }
 
 tasks.withType<KotlinCompile> {
 	kotlinOptions {
 		freeCompilerArgs = listOf("-Xjsr305=strict")
-		jvmTarget = "19"
+		jvmTarget = "21"
 	}
 }
 
 java {
-	sourceCompatibility = JavaVersion.VERSION_19
-	targetCompatibility = JavaVersion.VERSION_19
+	sourceCompatibility = JavaVersion.VERSION_21
+	targetCompatibility = JavaVersion.VERSION_21
 }
 
 tasks.withType<Test> {

gradle/libs.versions.toml

@@ -0,0 +1,45 @@
+[versions]
+ktor = "3.1.2"
+logback = "1.4.14"
+kfs = "1.3.0"
+koin = "4.0.4"
+caffeine = "3.1.0"
+tike = "2.9.1"
+jbcrypt = "0.4"
+jupyter = "5.10.1"
+krontab = "2.6.1"
+
+[libraries]
+kotlin-reflect = { module = "org.jetbrains.kotlin:kotlin-reflect" }
+kotlin-stdlib = { module = "org.jetbrains.kotlin:kotlin-stdlib-jdk8" }
+caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version.ref = "caffeine" }
+tika = { module = "org.apache.tika:tika-core", version.ref = "tike" }
+ktor-server-core = { module = "io.ktor:ktor-server-core-jvm", version.ref = "ktor" }
+ktor-server-cors =  { module = "io.ktor:ktor-server-cors-jvm", version.ref = "ktor" }
+ktor-server-content-negotiations = { module = "io.ktor:ktor-server-content-negotiation-jvm", version.ref = "ktor" }
+ktor-server-logging = { module = "io.ktor:ktor-server-call-logging-jvm", version.ref = "ktor" }
+ktor-server-netty = { module = "io.ktor:ktor-server-netty-jvm", version.ref = "ktor" }
+ktor-server-auth = { module = "io.ktor:ktor-server-auth", version.ref = "ktor" }
+ktor-server-auth-jwt = { module = "io.ktor:ktor-server-auth-jwt", version.ref = "ktor" }
+ktor-server-status-pages = { module = "io.ktor:ktor-server-status-pages", version.ref = "ktor" }
+ktor-serialization-kotlinx = { module = "io.ktor:ktor-serialization-kotlinx-json", version.ref = "ktor" }
+ktor-serialization-jackson = { module = "io.ktor:ktor-serialization-jackson", version.ref = "ktor" }
+ktor-client-cio = { module = "io.ktor:ktor-client-cio-jvm", version.ref = "ktor" }
+ktor-client-core = { module = "io.ktor:ktor-client-core-jvm", version.ref = "ktor" }
+ktor-client-content-negotiation = { module = "io.ktor:ktor-client-content-negotiation-jvm", version.ref = "ktor" }
+kfswatch = { module = "io.github.irgaly.kfswatch:kfswatch", version.ref = "kfs" }
+koin-ktor = { module = "io.insert-koin:koin-ktor", version.ref = "koin" }
+koin-logger = { module = "io.insert-koin:koin-logger-slf4j", version.ref = "koin" }
+logback = { module = "ch.qos.logback:logback-classic", version.ref = "logback" }
+jbcrypt = { module = "org.mindrot:jbcrypt", version.ref = "jbcrypt" }
+krontab = { module = "dev.inmo:krontab", version.ref = "krontab" }
+jupyter = { module = "org.junit.jupiter:junit-jupiter", version.ref = "jupyter" }
+
+[bundles]
+ktor-server = ["ktor-server-core", "ktor-server-cors", "ktor-server-content-negotiations", "ktor-server-logging", "ktor-server-netty", "ktor-server-auth", "ktor-server-auth-jwt", "ktor-server-status-pages"]
+ktor-serialization = ["ktor-serialization-kotlinx", "ktor-serialization-jackson"]
+ktor-client = ["ktor-client-cio", "ktor-client-core", "ktor-client-content-negotiation"]
+koin = ["koin-ktor", "koin-logger"]
+
+[plugins]
+ktor = { id = "io.ktor.plugin", version.ref = "ktor" }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.10-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

settings.gradle.kts

@@ -1,3 +1,27 @@
 rootProject.name = "kabot-api"
 
+pluginManagement {
+	repositories {
+		gradlePluginPortal()
+		mavenCentral()
+	}
+}
+
+dependencyResolutionManagement {
+	@Suppress("UnstableApiUsage")
+	repositoriesMode.set(RepositoriesMode.PREFER_SETTINGS)
+	@Suppress("UnstableApiUsage")
+	repositories {
+		mavenLocal()
+		mavenCentral()
+	}
+
+	versionCatalogs {
+		create("connectorLibs") {
+			from(files("./kabot-db-connector/libs.versions.toml"))
+		}
+	}
+
+}
+
 include(":kabot-db-connector")

src/main/kotlin/org/wagham/kabotapi/KabotApiApplication.kt

@@ -6,7 +6,9 @@ import org.wagham.kabotapi.configuration.configureHTTP
 import org.wagham.kabotapi.configuration.configureKoin
 import org.wagham.kabotapi.configuration.configureRouting
 
-fun main(args: Array<String>) = io.ktor.server.cio.EngineMain.main(args)
+fun main(args: Array<String>) {
+	io.ktor.server.netty.EngineMain.main(args)
+}
 
 @Suppress("unused")
 fun Application.module() {

src/main/kotlin/org/wagham/kabotapi/components/InstanceConfigManager.kt

@@ -0,0 +1,76 @@
+package org.wagham.kabotapi.components
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import io.github.irgaly.kfswatch.KfsDirectoryWatcher
+import io.github.irgaly.kfswatch.KfsEvent
+import io.ktor.util.logging.*
+import kotlinx.coroutines.launch
+import kotlinx.serialization.json.Json
+import org.wagham.kabotapi.model.foundry.FoundryOptions
+import java.io.File
+
+class InstanceConfigManager(
+	private val baseFolder: String
+) {
+
+	companion object {
+		private const val OPTIONS_FILENAME = "options.json"
+	}
+
+	private val logger = KtorSimpleLogger(this.javaClass.simpleName)
+	private val scope = CoroutineScope(Dispatchers.IO)
+	private val watcher = KfsDirectoryWatcher(scope)
+	private val instancesByUrl = mutableMapOf<String, InstanceInfo>()
+
+	fun startWatching() {
+		logger.info("Starting instance manager")
+		val dirsToWatch = File(baseFolder).walkTopDown().filter {
+			it.isFile && it.name == OPTIONS_FILENAME
+		}.map {
+			updateInstancesWith(it)
+			it.parent
+		}.toList()
+		scope.launch {
+			watcher.add(*dirsToWatch.toTypedArray())
+			watcher.onEventFlow.collect {
+				try {
+					val file = File("${it.targetDirectory}/${it.path}")
+					if (file.isFile && file.name == OPTIONS_FILENAME) {
+						when (it.event) {
+							KfsEvent.Create, KfsEvent.Modify -> {
+								updateInstancesWith(file)
+							}
+							else -> {
+								logger.info("Deleted ${file.absolutePath}")
+							}
+						}
+					}
+				} catch (e: Exception) {
+					logger.error(e.message)
+				}
+			}
+		}
+	}
+
+	fun getInfoByUrl(url: String): InstanceInfo? = instancesByUrl[url]
+
+	private fun updateInstancesWith(optionsFile: File) {
+		val options = Json.decodeFromString<FoundryOptions>(optionsFile.readText())
+		val info = InstanceInfo(
+			id = options.dataPath.split("/").last(),
+			url = options.routePrefix,
+			name = options.masterName ?: "unknown",
+		)
+		instancesByUrl[info.url] = info.also {
+			logger.info("Updating ${info.url} with $it")
+		}
+	}
+
+	data class InstanceInfo(
+		val id: String,
+		val url: String,
+		val name: String
+	)
+
+}

src/main/kotlin/org/wagham/kabotapi/components/InstanceInactivityManager.kt

@@ -0,0 +1,93 @@
+package org.wagham.kabotapi.components
+
+import com.github.benmanes.caffeine.cache.Caffeine
+import com.github.benmanes.caffeine.cache.RemovalCause
+import dev.inmo.krontab.doInfinity
+import io.ktor.util.logging.*
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import org.wagham.kabotapi.components.socket.CommandComponent
+import org.wagham.kabotapi.components.socket.NginxLogsListener
+import org.wagham.kabotapi.components.socket.Pm2ListCommand
+import org.wagham.kabotapi.components.socket.Pm2StopCommand
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.hours
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.toJavaDuration
+
+class InstanceInactivityManager(
+	private val instanceTtl: Duration,
+	private val commandComponent: CommandComponent,
+	private val nginxLogsListener: NginxLogsListener,
+	private val instanceConfigManager: InstanceConfigManager,
+	private val excludedInstances: Set<String>
+) {
+
+	private val urlExtractingRegex = Regex(".* \"https://fnd\\.kaironbot\\.net/([^/]+).*")
+	private val managerScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+	private val logger = KtorSimpleLogger(this.javaClass.simpleName)
+	private val instanceActivity = Caffeine.newBuilder()
+		.expireAfterWrite(instanceTtl.toJavaDuration())
+		.evictionListener { target: String?, _: Long?, _: RemovalCause ->
+			if (target != null) {
+				managerScope.launch {
+					commandComponent.sendSocketCommand(Pm2StopCommand(target))
+				}
+			}
+		}.build<String, Long>()
+
+	fun startListening() {
+		logger.info("Starting inactivity manager")
+		listenForZombies()
+		listenForLogs()
+	}
+
+	private fun listenForZombies() = managerScope.launch {
+		delay(instanceTtl)
+		doInfinity("0 0 * * * *") {
+			try {
+				commandComponent.sendSocketCommand(Pm2ListCommand()).filter {
+					it.name !in excludedInstances
+				}.forEach {
+					if ((System.currentTimeMillis() - it.pm2Env.uptime).milliseconds > 1.hours) {
+						val lastActivity = instanceActivity.getIfPresent(it.name)
+						if (lastActivity == null) {
+							commandComponent.sendSocketCommand(Pm2StopCommand(it.name))
+						}
+					}
+				}
+			} catch (e: Exception) {
+				logger.error("Error while getting zombies", e)
+			}
+		}
+	}
+
+	private fun listenForLogs() = managerScope.launch {
+		try {
+			nginxLogsListener.subscribe().collect {
+				try {
+					val info = getInstanceFromLog(it)
+					if (info != null) {
+						instanceActivity.put(info.id, System.currentTimeMillis())
+					}
+				} catch (e: Exception) {
+					logger.error("Error parsing log $it", e)
+				}
+			}
+		} catch (e: Exception) {
+			logger.error("Log stream interrupted", e)
+		}
+	}
+
+
+
+	private fun getInstanceFromLog(log: String): InstanceConfigManager.InstanceInfo? =
+		urlExtractingRegex.find(log)?.groupValues?.get(1)?.let {
+			instanceConfigManager.getInfoByUrl(it)
+		}?.takeIf {
+			it.id !in excludedInstances
+		}
+}