committing docker-kafka-logger, so I can delete it next (too much work to implement)

modules/infra-docker-kafka-logger/build.gradle.kts

@@ -0,0 +1,35 @@
+plugins {
+  dev.adamko.kafkatorio.lang.`kotlin-jvm`
+  kotlin("plugin.serialization")
+}
+
+
+dependencies {
+  implementation(platform(projects.modules.versionsPlatform))
+
+//  implementation(gradleKotlinDsl())
+
+  implementation(libs.kotlinx.serialization.core)
+  implementation(libs.kotlinx.serialization.json)
+  implementation(libs.kotlinx.serialization.protobuf)
+
+  implementation(libs.kotlinx.cli)
+
+  implementation(libs.kotlinx.coroutines.core)
+
+  implementation(libs.okio.core)
+
+  implementation(libs.ktor.serverCore)
+  implementation(libs.ktor.serverNetty)
+  implementation(libs.ktor.serverContentNegotiaion)
+  implementation(libs.ktor.serializationKotlinxJson)
+  implementation(libs.ktor.serverResources)
+
+  implementation(libs.kafka.streams)
+  implementation(libs.kafka.kotkaStreams)
+
+  implementation(libs.kotlinx.datetime)
+
+  testImplementation(libs.ktor.test.serverHost)
+  testImplementation(libs.kafka.streamsTestUtils)
+}

modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/api.kt

@@ -0,0 +1,138 @@
+package dev.adamko.dokalo
+
+import dev.adamko.dokalo.Capabilities.Response.readLogs
+import io.ktor.http.ContentType
+import io.ktor.http.HttpStatusCode
+import io.ktor.resources.Resource
+import io.ktor.server.application.Application
+import io.ktor.server.application.call
+import io.ktor.server.resources.post
+import io.ktor.server.response.respond
+import io.ktor.server.response.respondOutputStream
+import io.ktor.server.routing.routing
+import kotlinx.datetime.LocalDateTime
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+
+
+fun Application.configureRouting() {
+
+  routing {
+
+    post<StartLogging.Request> { request ->
+      // Signals to the plugin that a container is starting that the plugin should start receiving logs for.
+      call.respond(StartLogging.Response())
+    }
+
+    post<StopLogging.Request> { request ->
+      // Signals to the plugin to stop collecting logs from the defined file.
+      call.respond(StopLogging.Response())
+    }
+
+    post<Capabilities.Request> {
+      // Defines the capabilities of the log driver.
+      call.respond(Capabilities.Response)
+    }
+
+    post<ReadLogsRequest> { request ->
+      // Reads back logs to the client. This is used when docker logs <container> is called.
+
+      call.respondOutputStream(
+        ContentType.parse("application/x-json-stream"),
+        HttpStatusCode.OK,
+      ) {
+
+      }
+    }
+  }
+}
+
+
+sealed interface StartLogging {
+
+  @Resource("/LogDriver.StartLogging")
+  @Serializable
+  data class Request(
+
+    @SerialName("File")
+    val file: LogFile,
+
+    @SerialName("Info")
+    val info: ContainerInfo,
+  ) : StartLogging
+
+
+  @Serializable
+  data class Response(
+    @SerialName("Err")
+    val error: String? = null,
+  ) : StartLogging
+}
+
+
+sealed interface StopLogging {
+
+  @Resource("/LogDriver.StopLogging")
+  @Serializable
+  data class Request(
+    @SerialName("File")
+    val file: LogFile,
+  ) : StopLogging
+
+
+  @Serializable
+  data class Response(
+    @SerialName("Err")
+    val error: String? = null,
+  ) : StopLogging
+}
+
+
+sealed interface Capabilities {
+
+  @Resource("/LogDriver.Capabilities")
+  @Serializable
+  object Request : Capabilities
+
+  /**
+   * Defines the list of capabilities that a driver can implement. These capabilities are not
+   * required to be a logging driver, however do determine how a logging driver can be used.
+   *
+   * @property[readLogs] Determines if a log driver can read back logs
+   */
+  @Serializable
+  object Response : Capabilities {
+    @SerialName("ReadLogs")
+    val readLogs: Boolean = true
+  }
+}
+
+
+@Resource("/LogDriver.ReadLogs")
+@Serializable
+data class ReadLogsRequest(
+
+  @SerialName("Info")
+  val info: ContainerInfo,
+
+  @SerialName("Config")
+  val config: ReadConfig,
+) {
+
+  /** ReadConfig is the configuration passed into ReadLogs. */
+  @Serializable
+  data class ReadConfig(
+
+    @SerialName("Since")
+    val since: LocalDateTime,
+
+    @SerialName("Until")
+    val until: LocalDateTime,
+
+    @SerialName("Tail")
+    val tail: Int,
+
+    @SerialName("Follow")
+    val follow: Boolean,
+  )
+}

modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/logReader.kt

@@ -0,0 +1,125 @@
+package dev.adamko.dokalo
+
+import kotlin.coroutines.CoroutineContext
+import kotlinx.coroutines.CoroutineExceptionHandler
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.shareIn
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.supervisorScope
+import kotlinx.coroutines.yield
+import okio.BufferedSource
+import okio.FileSystem
+import okio.Path
+import okio.Path.Companion.toPath
+import okio.buffer
+
+
+class LogReaderManager : CoroutineScope {
+
+  override val coroutineContext: CoroutineContext = CoroutineName("LogReaderManager")
+
+  private val commands: MutableSharedFlow<LogReaderCmd> =
+    MutableSharedFlow(0, Int.MAX_VALUE, BufferOverflow.SUSPEND)
+
+  private val readers: MutableStateFlow<Map<LogFile, LogReader>> = MutableStateFlow(emptyMap())
+
+  init {
+    launch {
+      supervisorScope {
+        commands.runningFold(mapOf<LogFile, LogReader>()) { readers, cmd ->
+          when (cmd) {
+            is LogReaderCmd.Start -> {
+              if (cmd.file in readers) {
+                error("can't process 'start' for reader ${cmd.file} it's already registered")
+              } else {
+                readers + (cmd.file to LogReader(cmd.file))
+
+              }
+            }
+
+            is LogReaderCmd.Stop  -> {
+              if (cmd.file !in readers) {
+                error("can't process 'stop' for reader ${cmd.file} - it's not registered")
+              } else {
+                readers - cmd.file
+              }
+            }
+          }
+        }.collect(readers)
+      }
+    }
+  }
+
+  suspend fun start(data: StartLogging.Request) {
+    commands.emit(LogReaderCmd.Start(data.file))
+  }
+
+  suspend fun stop(data: StopLogging.Request) {
+    commands.emit(LogReaderCmd.Stop(data.file))
+  }
+
+
+}
+
+private data class LogReader(
+  val file: LogFile,
+) : CoroutineScope {
+  private val path: Path = file.file.toPath()
+  private val source: BufferedSource = FileSystem.SYSTEM.source(path).buffer()
+
+  override val coroutineContext: CoroutineContext =
+    CoroutineName("LogReader-${file.file.filter { it.isLetterOrDigit() }}") +
+        CoroutineExceptionHandler { coroutineContext, throwable ->
+          source.close()
+        }
+
+  private val _entries: MutableSharedFlow<String> =
+    MutableSharedFlow(0, Int.MAX_VALUE, BufferOverflow.SUSPEND)
+
+
+  init {
+    launch {
+      val f = flow {
+        while (currentCoroutineContext().isActive && source.isOpen) {
+          val entrySize = source.readInt().toLong()
+          val entry = source.readString(entrySize, Charsets.UTF_8)
+          emit(entry)
+          yield()
+        }
+      }
+
+      f.shareIn(this, SharingStarted.Lazily)
+      f.launchIn(this)
+
+
+    }
+  }
+}
+
+fun logReader(file: LogFile) {
+  val reader = LogReader(file)
+
+
+}
+
+
+private sealed interface LogReaderCmd {
+  val file: LogFile
+
+  data class Start(override val file: LogFile) : LogReaderCmd
+
+  data class Stop(override val file: LogFile) : LogReaderCmd
+}
+
+
+private sealed interface LogReaderState {
+  object Active : LogReaderState
+  object Stopped : LogReaderState
+  data class Error(val msg: String, val cause: Exception?) : LogReaderState
+}

modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/main.kt

@@ -0,0 +1,37 @@
+package dev.adamko.dokalo
+
+import com.typesafe.config.ConfigFactory
+import io.ktor.server.application.install
+import io.ktor.server.config.HoconApplicationConfig
+import io.ktor.server.engine.applicationEngineEnvironment
+import io.ktor.server.engine.connector
+import io.ktor.server.engine.embeddedServer
+import io.ktor.server.netty.Netty
+import io.ktor.server.resources.Resources
+
+
+fun main() {
+
+  embeddedServer(Netty, port = 8080) {
+
+    install(Resources)
+
+    configureRouting()
+  }.start(wait = true)
+
+
+//  embeddedServer(Netty, environment = applicationEngineEnvironment {
+//
+//    config = HoconApplicationConfig(ConfigFactory.load())
+//
+//    module {
+//      main()
+//    }
+//
+//    connector {
+//      port = 8080
+//      host = "127.0.0.1"
+//    }
+//  }).start(wait = true)
+
+}

modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/model.kt

@@ -0,0 +1,110 @@
+@file:OptIn(ExperimentalSerializationApi::class)
+
+package dev.adamko.dokalo
+
+import kotlinx.datetime.LocalDateTime
+import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.protobuf.ProtoNumber
+
+
+@Serializable
+@JvmInline
+value class ContainerId(private val id: String)
+
+
+@Serializable
+@JvmInline
+value class LogFile(val file: String)
+
+
+/** Info provides enough information for a logging driver to do its function. */
+@Serializable
+data class ContainerInfo(
+  @SerialName("ContainerID")
+  val containerId: ContainerId,
+
+  @SerialName("Config")
+  val config: Map<String, String>? = null,
+
+  @SerialName("ContainerName")
+  val containerName: String? = null,
+
+  @SerialName("ContainerEntrypoint")
+  val containerEntrypoint: String? = null,
+
+  @SerialName("ContainerArgs")
+  val containerArgs: List<String>? = null,
+
+  @SerialName("ContainerImageID")
+  val containerImageId: String? = null,
+
+  @SerialName("ContainerImageName")
+  val containerImageName: String? = null,
+
+  @SerialName("ContainerCreated")
+  val containerCreated: LocalDateTime? = null,
+
+  @SerialName("ContainerEnv")
+  val containerEnv: List<String>? = null,
+
+  @SerialName("ContainerLabels")
+  val containerLabels: Map<String, String>? = null,
+
+  @SerialName("LogPath")
+  val logPath: String? = null,
+
+  @SerialName("DaemonName")
+  val daemonName: String? = null,
+)
+
+
+@Serializable
+data class LogEntry(
+  @ProtoNumber(1)
+  val source: String,
+  @ProtoNumber(2)
+  val time_nano: Long,
+  @ProtoNumber(3)
+  val line: ByteArray,
+  @ProtoNumber(4)
+  val partial: Boolean,
+  @ProtoNumber(5)
+  val partial_log_metadata: PartialLogEntryMetadata,
+) {
+  override fun equals(other: Any?): Boolean {
+    if (this === other) return true
+    if (javaClass != other?.javaClass) return false
+
+    other as LogEntry
+
+    if (source != other.source) return false
+    if (time_nano != other.time_nano) return false
+    if (!line.contentEquals(other.line)) return false
+    if (partial != other.partial) return false
+    if (partial_log_metadata != other.partial_log_metadata) return false
+
+    return true
+  }
+
+  override fun hashCode(): Int {
+    var result = source.hashCode()
+    result = 31 * result + time_nano.hashCode()
+    result = 31 * result + line.contentHashCode()
+    result = 31 * result + partial.hashCode()
+    result = 31 * result + partial_log_metadata.hashCode()
+    return result
+  }
+}
+
+
+@Serializable
+data class PartialLogEntryMetadata(
+  @ProtoNumber(1)
+  val last: Boolean,
+  @ProtoNumber(2)
+  val id: String,
+  @ProtoNumber(3)
+  val ordinal: Int,
+)

modules/infra-docker-kafka-logger/src/main/resources/application.conf

@@ -0,0 +1,9 @@
+ktor {
+    deployment {
+        port = 8080
+        port = ${?PORT}
+    }
+    application {
+        modules = [ com.example.ApplicationKt.module ]
+    }
+}