delete docker kafka logger
This commit is contained in:
Adam
parent
87ff50ff32
commit
b08e33e2eb
6 changed files with 0 additions and 454 deletions
|
|
@ -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)
|
||||
}
|
||||
|
|
@ -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<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,
|
||||
)
|
||||
}
|
||||
|
|
@ -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<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
|
||||
}
|
||||
|
|
@ -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)
|
||||
|
||||
}
|
||||
|
|
@ -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<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,
|
||||
)
|
||||
|
|
@ -1,9 +0,0 @@
|
|||
ktor {
|
||||
deployment {
|
||||
port = 8080
|
||||
port = ${?PORT}
|
||||
}
|
||||
application {
|
||||
modules = [ com.example.ApplicationKt.module ]
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue