From 87ff50ff3247fa55016e512bf3a596b60f07a214 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 19 Jun 2022 10:33:07 +0200 Subject: [PATCH] committing docker-kafka-logger, so I can delete it next (too much work to implement) --- .../build.gradle.kts | 35 +++++ .../src/main/kotlin/dev/adamko/dokalo/api.kt | 138 ++++++++++++++++++ .../kotlin/dev/adamko/dokalo/logReader.kt | 125 ++++++++++++++++ .../src/main/kotlin/dev/adamko/dokalo/main.kt | 37 +++++ .../main/kotlin/dev/adamko/dokalo/model.kt | 110 ++++++++++++++ .../src/main/resources/application.conf | 9 ++ 6 files changed, 454 insertions(+) create mode 100644 modules/infra-docker-kafka-logger/build.gradle.kts create mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/api.kt create mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/logReader.kt create mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/main.kt create mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/model.kt create mode 100644 modules/infra-docker-kafka-logger/src/main/resources/application.conf diff --git a/modules/infra-docker-kafka-logger/build.gradle.kts b/modules/infra-docker-kafka-logger/build.gradle.kts new file mode 100644 index 0000000..a39813f --- /dev/null +++ b/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) +} diff --git a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/api.kt b/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/api.kt new file mode 100644 index 0000000..4abc6e8 --- /dev/null +++ b/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 { request -> + // Signals to the plugin that a container is starting that the plugin should start receiving logs for. + call.respond(StartLogging.Response()) + } + + post { request -> + // Signals to the plugin to stop collecting logs from the defined file. + call.respond(StopLogging.Response()) + } + + post { + // Defines the capabilities of the log driver. + call.respond(Capabilities.Response) + } + + post { request -> + // Reads back logs to the client. This is used when docker logs 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, + ) +} diff --git a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/logReader.kt b/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/logReader.kt new file mode 100644 index 0000000..4202b29 --- /dev/null +++ b/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 = + MutableSharedFlow(0, Int.MAX_VALUE, BufferOverflow.SUSPEND) + + private val readers: MutableStateFlow> = MutableStateFlow(emptyMap()) + + init { + launch { + supervisorScope { + commands.runningFold(mapOf()) { 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 = + 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 +} diff --git a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/main.kt b/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/main.kt new file mode 100644 index 0000000..b5345da --- /dev/null +++ b/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) + +} diff --git a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/model.kt b/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/model.kt new file mode 100644 index 0000000..838a8f5 --- /dev/null +++ b/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? = null, + + @SerialName("ContainerName") + val containerName: String? = null, + + @SerialName("ContainerEntrypoint") + val containerEntrypoint: String? = null, + + @SerialName("ContainerArgs") + val containerArgs: List? = null, + + @SerialName("ContainerImageID") + val containerImageId: String? = null, + + @SerialName("ContainerImageName") + val containerImageName: String? = null, + + @SerialName("ContainerCreated") + val containerCreated: LocalDateTime? = null, + + @SerialName("ContainerEnv") + val containerEnv: List? = null, + + @SerialName("ContainerLabels") + val containerLabels: Map? = 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, +) diff --git a/modules/infra-docker-kafka-logger/src/main/resources/application.conf b/modules/infra-docker-kafka-logger/src/main/resources/application.conf new file mode 100644 index 0000000..08b319b --- /dev/null +++ b/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 ] + } +}