From b08e33e2eb2dac32baf490139bcefd3bbb1919a5 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Sun, 19 Jun 2022 10:33:30 +0200 Subject: [PATCH] delete docker kafka logger --- .../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 deletions(-) delete mode 100644 modules/infra-docker-kafka-logger/build.gradle.kts delete mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/api.kt delete mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/logReader.kt delete mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/main.kt delete mode 100644 modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/model.kt delete 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 deleted file mode 100644 index a39813f..0000000 --- a/modules/infra-docker-kafka-logger/build.gradle.kts +++ /dev/null @@ -1,35 +0,0 @@ -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 deleted file mode 100644 index 4abc6e8..0000000 --- a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/api.kt +++ /dev/null @@ -1,138 +0,0 @@ -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 deleted file mode 100644 index 4202b29..0000000 --- a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/logReader.kt +++ /dev/null @@ -1,125 +0,0 @@ -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 deleted file mode 100644 index b5345da..0000000 --- a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/main.kt +++ /dev/null @@ -1,37 +0,0 @@ -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 deleted file mode 100644 index 838a8f5..0000000 --- a/modules/infra-docker-kafka-logger/src/main/kotlin/dev/adamko/dokalo/model.kt +++ /dev/null @@ -1,110 +0,0 @@ -@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 deleted file mode 100644 index 08b319b..0000000 --- a/modules/infra-docker-kafka-logger/src/main/resources/application.conf +++ /dev/null @@ -1,9 +0,0 @@ -ktor { - deployment { - port = 8080 - port = ${?PORT} - } - application { - modules = [ com.example.ApplicationKt.module ] - } -}