big ol' refactor to split up the events-server and events-processors

modules/events-server-syslog/build.gradle.kts

@@ -8,7 +8,7 @@ plugins {
 }
 
 
-//description = "serves Kafkatorio content over the web"
+description = "Consumes Syslog Kafkatorio packets and sends them to Kafka"
 
 val projectId: String by project.extra
 
@@ -20,6 +20,7 @@ dependencies {
   testImplementation(libs.kafka.streamsTestUtils)
 
   implementation(projects.modules.eventsLibrary)
+  implementation(projects.modules.eventsProcessorCore)
 
   implementation(libs.kotlinxSerialization.core)
   implementation(libs.kotlinxSerialization.json)
@@ -62,11 +63,13 @@ dependencies {
 //  implementation(libs.nimbusJoseJwt)
 
   implementation(libs.kafka.kotkaStreams)
+
+  implementation(libs.bundles.hoplite)
 }
 
 
 application {
-  mainClass.set("dev.adamko.kafkatorio.server.EventsServerKt")
+  mainClass.set("dev.adamko.kafkatorio.processor.syslog.SyslogMainKt")
 }
 
 
@@ -77,3 +80,9 @@ tasks.withType<KotlinCompile>().configureEach {
     "-opt-in=kotlinx.serialization.ExperimentalSerializationApi",
   )
 }
+
+
+val runSyslogServer by tasks.registering {
+  group = rootProject.name
+  dependsOn(tasks.run)
+}

modules/events-server-syslog/docker/Dockerfile

@@ -0,0 +1,11 @@
+FROM gradle:7-jdk11 AS build
+COPY --chown=gradle:gradle . /home/gradle/src
+WORKDIR /home/gradle/src
+RUN gradle buildFatJar --no-daemon
+
+
+FROM openjdk:11
+EXPOSE 8080:8080
+RUN mkdir /app
+COPY --from=build /home/gradle/src/build/libs/*.jar /app/ktor-docker-sample.jar
+ENTRYPOINT ["java","-jar","/app/ktor-docker-sample.jar"]

modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/KafkatorioPacketKafkaProducer.kt

@@ -0,0 +1,172 @@
+package dev.adamko.kafkatorio.processor.syslog
+
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.core.Authenticator
+import dev.adamko.kafkatorio.processor.syslog.config.jsonMapper
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG
+import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG_DLQ
+import dev.adamko.kotka.kxs.kafkaSerializer
+import java.util.Properties
+import kotlin.coroutines.coroutineContext
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.job
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.StreamsConfig
+
+
+/**
+ * Consume Syslog messages, verifies each, and forwards them to a Kafka topic.
+ */
+class KafkatorioPacketKafkaProducer(
+  private val appProps: ApplicationProperties,
+  private val syslogServer: SyslogSocketServer,
+  private val authenticator: Authenticator,
+) : AutoCloseable {
+
+  private val props: Properties = appProps.kafkaStreamsConfig.toProperties()
+
+  init {
+    props.updateValue(StreamsConfig.APPLICATION_ID_CONFIG) { "$it.KafkatorioPacketKafkaProducer" }
+  }
+
+  private val producer: KafkaProducer<FactorioServerId, String> = KafkaProducer(
+    props.apply {
+      updateValue(StreamsConfig.APPLICATION_ID_CONFIG) { "$it.KafkatorioPacketKafkaProducer" }
+    },
+    jsonMapper.kafkaSerializer(FactorioServerId.serializer()),
+    Serdes.String().serializer(),
+  )
+
+  private val dlqProducer: KafkaProducer<String, String> = KafkaProducer(
+    props.apply {
+      updateValue(StreamsConfig.APPLICATION_ID_CONFIG) { "$it.KafkatorioPacketKafkaProducer.dlq" }
+    },
+    Serdes.String().serializer(),
+    Serdes.String().serializer(),
+  )
+
+
+  suspend fun launch() {
+    coroutineContext.job.invokeOnCompletion {
+      this@KafkatorioPacketKafkaProducer.close()
+    }
+
+    syslogServer.messages
+      .filter {
+        (it.message ?: "").run {
+          startsWith("KafkatorioPacket:::") || startsWith("KafkatorioPacket encoded:::")
+        }
+      }.onEach { syslog ->
+        runCatching {
+          getFactorioServerId(syslog)
+        }.fold(
+          onSuccess = { serverId -> produceMessage(serverId, syslog) },
+          onFailure = { ex ->
+            produceDlqMessage(syslog, "${ex::class.java.name}: ${ex.message}")
+
+            if (ex is CancellationException) throw ex
+          }
+        )
+      }.collect()
+  }
+
+
+  private fun getFactorioServerId(syslog: SyslogMsg): FactorioServerId {
+    val token = syslog.headerAppName
+    if (token.isNullOrBlank()) {
+      error("no token in syslog message. msg: ${syslog.src}")
+    }
+
+    val serverToken = authenticator.verifyJwt(token)
+    if (serverToken.isNullOrBlank()) {
+      error("could not decode token $token. msg: ${syslog.src}")
+    }
+
+    val serverId = appProps.kafkatorioServers[serverToken]
+    if (serverId.isNullOrBlank()) {
+      error("no server ID for token $serverToken. msg: ${syslog.src}")
+    }
+
+    return serverId
+  }
+
+
+  private fun produceMessage(
+    serverId: FactorioServerId,
+    syslog: SyslogMsg,
+  ) {
+    val kafkatorioMessage = syslog.message?.substringAfter(":::")
+
+    val record = ProducerRecord<FactorioServerId, String>(
+      TOPIC_SRC_SERVER_LOG,
+      serverId,
+      kafkatorioMessage,
+    )
+
+    syslog.copyHeadersTo(record)
+
+    producer.send(record)
+  }
+
+
+  private fun produceDlqMessage(syslog: SyslogMsg, errorMessage: String) {
+    log("error: $errorMessage, $syslog")
+
+    val record = ProducerRecord(
+      TOPIC_SRC_SERVER_LOG_DLQ,
+      "unknown",
+      errorMessage,
+    )
+
+    syslog.copyHeadersTo(record)
+
+    dlqProducer.send(record)
+  }
+
+
+  override fun close() {
+    producer.close()
+  }
+
+
+  companion object {
+    private fun log(msg: String) = println("[KafkatorioPacketKafkaProducer] $msg")
+
+
+    private fun Properties.updateValue(
+      key: String,
+      update: (oldValue: String) -> String
+    ) {
+      val newValue = update(getValue(key) as String)
+      setProperty(key, newValue)
+    }
+
+
+    private fun SyslogMsg.copyHeadersTo(record: ProducerRecord<*, *>) {
+      val headers = SyslogFieldKeys.values()
+        .filter { "HEADER_" in it.name }
+        .mapNotNull { headerKey ->
+          val headerValue = src[headerKey.field]
+          when {
+            headerValue.isNullOrBlank() -> null
+            else                        -> RecordHeader(
+              headerKey.field,
+              headerValue.toByteArray()
+            )
+          }
+        }
+
+      headers.forEach { header ->
+        record.headers().add(header)
+      }
+    }
+  }
+}

modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogMsg.kt

@@ -0,0 +1,51 @@
+package dev.adamko.kafkatorio.processor.syslog
+
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+
+data class SyslogMsg(
+  val src: Map<String, String?>,
+) {
+
+  val message: String?
+    get() = src[SyslogFieldKeys.MESSAGE.field]
+
+  val headerAppName: String?
+    get() = src[SyslogFieldKeys.HEADER_APPNAME.field]
+
+  val headerHostName: String?
+    get() = src[SyslogFieldKeys.HEADER_HOSTNAME.field]
+
+  val headerPri: String?
+    get() = src[SyslogFieldKeys.HEADER_PRI.field]
+
+  val headerPriSeverity: String?
+    get() = src[SyslogFieldKeys.HEADER_PRI_SEVERITY.field]
+
+  val headerPriFacility: String?
+    get() = src[SyslogFieldKeys.HEADER_PRI_FACILITY.field]
+
+  val headerProcId: String?
+    get() = src[SyslogFieldKeys.HEADER_PROCID.field]
+
+  val headerTimestamp: String?
+    get() = src[SyslogFieldKeys.HEADER_TIMESTAMP.field]
+
+  val headerMsgId: String?
+    get() = src[SyslogFieldKeys.HEADER_MSGID.field]
+
+  val headerVersion: String?
+    get() = src[SyslogFieldKeys.HEADER_VERSION.field]
+
+  val structuredBase: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_BASE.field]
+
+  val structuredElementIdFmt: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_FMT.field]
+
+  val structuredElementIdPnameFmt: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_FMT.field]
+
+  val structuredElementIdPnamePattern: String?
+    get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
+
+}

modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/SyslogSocketServer.kt

@@ -0,0 +1,111 @@
+package dev.adamko.kafkatorio.processor.syslog
+
+import com.github.palindromicity.syslog.AllowableDeviations
+import com.github.palindromicity.syslog.DefaultKeyProvider
+import com.github.palindromicity.syslog.Flat5424MessageHandler
+import com.github.palindromicity.syslog.NilPolicy
+import com.github.palindromicity.syslog.SyslogParser
+import com.github.palindromicity.syslog.SyslogParserBuilder
+import com.github.palindromicity.syslog.SyslogSpecification
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import io.ktor.network.selector.SelectorManager
+import io.ktor.network.sockets.Socket
+import io.ktor.network.sockets.aSocket
+import io.ktor.network.sockets.openReadChannel
+import io.ktor.utils.io.readUTF8Line
+import java.util.EnumSet
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.asSharedFlow
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.yield
+
+
+class SyslogSocketServer(
+  appProps: ApplicationProperties,
+) {
+
+  /** Incoming messages */
+  private val serverSocketTcp = aSocket(SelectorManager(Dispatchers.IO))
+    .tcp()
+    .bind(
+      appProps.socketServerHost,
+      appProps.socketServerPort,
+    )
+
+
+  private val syslogParser: SyslogParser<Map<String, String>> =
+    SyslogParserBuilder<Map<String, String>>()
+      .forSpecification(SyslogSpecification.RFC_5424)
+      .withSyslogBuilder(
+        Flat5424MessageHandler(
+          DefaultKeyProvider(),
+          NilPolicy.OMIT,
+          EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION),
+        )
+      ).build()
+
+
+  private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
+  val messages: SharedFlow<SyslogMsg>
+    get() = _messages.asSharedFlow()
+
+
+  suspend fun start(): Unit = coroutineScope {
+    log("started listening: ${serverSocketTcp.localAddress}")
+
+    while (isActive) {
+      val socket: Socket = serverSocketTcp.accept()
+      log("new socket ${socket.description()}")
+
+      launch {
+        val channel = socket.openReadChannel()
+        try {
+          while (socket.socketContext.isActive && !channel.isClosedForRead) {
+            val line = runCatching {
+              channel.readUTF8Line()
+            }.fold(
+              onSuccess = { line ->
+                if (line == null) {
+                  log("received 'null' from ${socket.description()} - closing connection")
+                  socket.close()
+                }
+                // log("received '$line' from ${socket.description()}")
+                line
+              },
+              onFailure = { ex ->
+                if (ex is CancellationException) throw ex
+
+                log("reading line from socket failed ${socket.description()}, cause: $ex")
+                null
+              },
+            )
+            if (line != null) {
+              val syslog = SyslogMsg(syslogParser.parseLine(line))
+              _messages.emit(syslog)
+            }
+            yield()
+          }
+        } catch (e: Exception) {
+          if (e is CancellationException) throw e
+          log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}")
+          e.printStackTrace()
+          withContext(Dispatchers.IO) { socket.close() }
+        }
+      }
+
+      yield()
+    }
+  }
+
+  companion object {
+    private fun log(msg: String) = println("[SyslogSocketServer] $msg")
+
+    private fun Socket.description(): String = "Socket[local:$localAddress, remote:$remoteAddress]"
+  }
+}

modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/config/kxsJson.kt

@@ -0,0 +1,12 @@
+package dev.adamko.kafkatorio.processor.syslog.config
+
+import dev.adamko.kafkatorio.library.jsonMapperKafkatorio
+import kotlinx.serialization.json.Json
+
+
+@PublishedApi
+internal val jsonMapper: Json = Json {
+  prettyPrint = true
+  prettyPrintIndent = "  "
+  serializersModule = jsonMapperKafkatorio.serializersModule
+}

modules/events-server-syslog/src/main/kotlin/dev/adamko/kafkatorio/processor/syslog/syslogMain.kt

@@ -0,0 +1,20 @@
+package dev.adamko.kafkatorio.processor.syslog
+
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.core.Authenticator
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.launch
+
+
+/** Receive log messages from the syslog socket, and send them to a Kafka topic */
+suspend fun main(): Unit = coroutineScope {
+
+  val appProps = ApplicationProperties.load()
+  val syslogServer = SyslogSocketServer(appProps)
+  val authenticator = Authenticator(appProps)
+
+  val producer = KafkatorioPacketKafkaProducer(appProps, syslogServer, authenticator)
+
+  launch { syslogServer.start() }
+  launch { producer.launch() }
+}