set up syslog docker parser (remove need for Kafka Pipe script)

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

@@ -1,6 +1,9 @@
 package dev.adamko.kafkatorio.processor
 
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.tileserver.SyslogSocketServer
 import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
 import dev.adamko.kafkatorio.processor.topology.broadcastToWebsocket
 import dev.adamko.kafkatorio.processor.topology.colourMapChunks
@@ -17,10 +20,17 @@ import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
+import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerConfig
+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.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.StreamsConfig
@@ -29,12 +39,14 @@ import org.apache.kafka.streams.TopologyDescription
 
 
 internal class KafkatorioTopology(
-  private val websocketServer: WebsocketServer,
   private val appProps: ApplicationProperties,
+  private val websocketServer: WebsocketServer,
+  private val syslogServer: SyslogSocketServer,
 ) {
 
   private val builderScope = CoroutineScope(CoroutineName("KafkaTopology"))
 
+
   suspend fun start() = builderScope.launch {
     splitPackets()
 
@@ -42,8 +54,11 @@ internal class KafkatorioTopology(
     saveTiles()
 
     websocketBroadcaster()
+
+    syslogProducer()
   }
 
+
   private suspend fun splitPackets() {
     val builder = StreamsBuilder()
     val packets = factorioServerPacketStream(builder)
@@ -54,18 +69,21 @@ internal class KafkatorioTopology(
     launchTopology("splitPackets", topology)
   }
 
+
   private suspend fun groupTilesMapChunks() {
     val builder = StreamsBuilder()
     val topology = colourMapChunks(builder)
     launchTopology("groupTilesMapChunks", topology)
   }
 
+
   private suspend fun saveTiles() {
     val builder = StreamsBuilder()
     val topology = saveMapTiles(builder, appProps.tileDir)
     launchTopology("saveTiles", topology)
   }
 
+
   private suspend fun websocketBroadcaster() {
     val builder = StreamsBuilder()
     broadcastToWebsocket(websocketServer, builder)
@@ -75,11 +93,65 @@ internal class KafkatorioTopology(
     launchTopology(
       "websocketBroadcaster", topology, mapOf(
         StreamsConfig.COMMIT_INTERVAL_MS_CONFIG to "${1.seconds.inWholeMilliseconds}",
-        StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${10.milliseconds.inWholeMilliseconds}"
+        StreamsConfig.producerPrefix(ProducerConfig.LINGER_MS_CONFIG) to "${1.milliseconds.inWholeMilliseconds}"
       )
     )
   }
 
+
+  private fun syslogProducer(
+    id: String = "syslog-producer"
+  ) {
+    val props: Properties = appProps.kafkaStreamsConfig.toProperties()
+
+    val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String
+    props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
+
+
+    val producer: KafkaProducer<String, String> = KafkaProducer(
+      props,
+      Serdes.String().serializer(),
+      Serdes.String().serializer(),
+    )
+
+    builderScope.coroutineContext.job.invokeOnCompletion {
+      producer.close()
+    }
+
+    syslogServer.messages
+      .filter { it.message?.startsWith("KafkatorioPacket:") == true }
+      .onEach { syslogMsg ->
+
+        val kafkatorioMessage = syslogMsg.message?.substringAfter("KafkatorioPacket:")
+
+        val record = ProducerRecord<String, String>(
+          TOPIC_SRC_SERVER_LOG,
+          "syslog-test",
+          kafkatorioMessage,
+        )
+
+        val headers = SyslogFieldKeys.values()
+          .filter { "HEADER_" in it.name }
+          .mapNotNull { headerKey ->
+            val headerValue = syslogMsg.src[headerKey.field]
+            when {
+              headerValue.isNullOrBlank() -> null
+              else                        -> RecordHeader(
+                headerKey.field,
+                headerValue.toByteArray()
+              )
+            }
+          }
+
+        headers.forEach { header ->
+          record.headers().add(header)
+        }
+
+        producer.send(record)
+      }.launchIn(builderScope)
+  }
+
+
   private suspend fun launchTopology(
     id: String,
     topology: Topology,
@@ -105,11 +177,8 @@ internal class KafkatorioTopology(
         streams.setStateListener { newState: KafkaStreams.State, _ ->
           when (newState) {
             KafkaStreams.State.NOT_RUNNING,
-//            KafkaStreams.State.ERROR,
-                 -> if (cont.isActive) {
-              cont.resume(newState)
-            }
-            else -> Unit // do nothing
+            KafkaStreams.State.ERROR -> if (cont.isActive) cont.resume(newState)
+            else                     -> Unit // do nothing
           }
         }
 
@@ -127,6 +196,7 @@ internal class KafkatorioTopology(
     }
   }
 
+
   private fun printTopologyDescription(appId: String, topology: Topology) {
     val description: TopologyDescription = topology.describe()
 
@@ -146,6 +216,7 @@ internal class KafkatorioTopology(
     )
   }
 
+
   companion object {
     private fun Map<String, String>.toProperties(): Properties =
       entries.fold(Properties()) { props, (k, v) ->

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/admin/kafkatorioTopics.kt

@@ -48,14 +48,15 @@ fun allTopics(): Set<String> = buildSet {
 
 val KafkatorioPacketData.topicName: String
   get() = when (this) {
-    is ConfigurationUpdate  -> packetTopicNames.getValue(ConfigurationUpdate::class)
-    is ConsoleChatUpdate    -> packetTopicNames.getValue(ConsoleChatUpdate::class)
-    is ConsoleCommandUpdate -> packetTopicNames.getValue(ConsoleCommandUpdate::class)
-    is PrototypesUpdate     -> packetTopicNames.getValue(PrototypesUpdate::class)
-    is SurfaceUpdate        -> packetTopicNames.getValue(SurfaceUpdate::class)
-    is EntityUpdate         -> packetTopicNames.getValue(EntityUpdate::class)
-    is MapChunkUpdate       -> packetTopicNames.getValue(MapChunkUpdate::class)
-    is PlayerUpdate         -> packetTopicNames.getValue(PlayerUpdate::class)
+    is ConfigurationUpdate        -> packetTopicNames.getValue(ConfigurationUpdate::class)
+    is ConsoleChatUpdate          -> packetTopicNames.getValue(ConsoleChatUpdate::class)
+    is ConsoleCommandUpdate       -> packetTopicNames.getValue(ConsoleCommandUpdate::class)
+    is PrototypesUpdate           -> packetTopicNames.getValue(PrototypesUpdate::class)
+    is SurfaceUpdate              -> packetTopicNames.getValue(SurfaceUpdate::class)
+    is EntityUpdate               -> packetTopicNames.getValue(EntityUpdate::class)
+    is MapChunkUpdate             -> packetTopicNames.getValue(MapChunkUpdate::class)
+    is PlayerUpdate               -> packetTopicNames.getValue(PlayerUpdate::class)
+    is KafkatorioPacketData.Error -> packetTopicNames.getValue(KafkatorioPacketData.Error::class)
   }
 
 
@@ -68,6 +69,7 @@ private val packetTopicNames: Map<KClass<out KafkatorioPacketData>, String> = ma
   EntityUpdate::class to "entity",
   MapChunkUpdate::class to "map-chunk",
   PlayerUpdate::class to "player",
+  KafkatorioPacketData.Error::class to "error",
 ).mapValues { (_, v) ->
   "$DOMAIN.packet.$v"
 }

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

@@ -12,6 +12,9 @@ data class ApplicationProperties(
   val tileDir: Path,
   val kafkaAdminConfig: Map<String, String>,
   val kafkaStreamsConfig: Map<String, String>,
+
+  val socketServerHost: String,
+  val socketServerPort: Int,
 ) {
 
   companion object {

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

@@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.processor
 
 import dev.adamko.kafkatorio.processor.admin.KafkatorioKafkaAdmin
 import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.tileserver.SyslogSocketServer
 import dev.adamko.kafkatorio.processor.tileserver.WebMapTileServer
 import dev.adamko.kafkatorio.processor.tileserver.WebsocketServer
 import dev.adamko.kafkatorio.processor.tileserver.webServer
@@ -11,7 +12,6 @@ import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
-import kotlinx.serialization.Serializable
 import org.http4k.server.Http4kServer
 
 
@@ -22,12 +22,18 @@ suspend fun main() {
   val admin = KafkatorioKafkaAdmin(appProps)
   admin.createKafkatorioTopics()
 
+
+  val syslogServer = SyslogSocketServer(appProps)
   val wsServer = WebsocketServer()
   val tileServer = WebMapTileServer(appProps)
+  val topology = KafkatorioTopology(appProps, wsServer, syslogServer)
 
   coroutineScope {
     launch {
-      val topology = KafkatorioTopology(wsServer, appProps)
+      syslogServer.start()
+    }
+
+    launch {
       topology.start()
 
       println("launched KafkatorioTopology")

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

@@ -0,0 +1,168 @@
+package dev.adamko.kafkatorio.processor.tileserver
+
+import com.github.palindromicity.syslog.AllowableDeviations
+import com.github.palindromicity.syslog.SyslogParser
+import com.github.palindromicity.syslog.SyslogParserBuilder
+import com.github.palindromicity.syslog.SyslogSpecification
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+import dev.adamko.kafkatorio.processor.config.ApplicationProperties
+import dev.adamko.kafkatorio.processor.tileserver.SyslogMsg.Companion.toSyslogMsg
+import io.ktor.network.selector.SelectorManager
+import io.ktor.network.sockets.aSocket
+import io.ktor.network.sockets.awaitClosed
+import io.ktor.network.sockets.openReadChannel
+import io.ktor.utils.io.readUTF8Line
+import java.util.EnumSet
+import kotlinx.coroutines.CoroutineDispatcher
+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.yield
+
+
+class SyslogSocketServer(
+  appProps: ApplicationProperties,
+  dispatcher: CoroutineDispatcher = Dispatchers.IO,
+) {
+
+  private val selectorManager: SelectorManager = SelectorManager(dispatcher)
+
+
+  private val serverSocketTcp = aSocket(selectorManager)
+    .tcp()
+    .bind(
+      appProps.socketServerHost,
+      appProps.socketServerPortTcp,
+    )
+
+
+//  private val serverSocketUdp: BoundDatagramSocket = aSocket(selectorManager)
+//    .udp()
+//    .bind(
+//      localAddress = InetSocketAddress(
+//        appProps.socketServerHost,
+//        appProps.socketServerPortUdp,
+//      ),
+//    )
+
+
+  private val syslogParser: SyslogParser = SyslogParserBuilder()
+    .withDeviations(EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION))
+    .forSpecification(SyslogSpecification.RFC_5424)
+    .build()
+
+
+  private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
+  val messages: SharedFlow<SyslogMsg>
+    get() = _messages.asSharedFlow()
+
+
+  suspend fun start() {
+//    udp()
+    tcp()
+  }
+
+//  private suspend fun udp() {
+//    println("[SyslogSocketServer] started listening: ${serverSocketUdp.localAddress}")
+//
+//    serverSocketUdp.incoming
+//      .receiveAsFlow()
+//      .map {udpPacket: Datagram ->
+//        val msg = udpPacket.packet.readText()
+//        println("[SyslogSocketServer] received msg: $msg")
+//        syslogParser.parseLine(msg).toSyslogMsg()
+//      }.collect(_messages)
+//  }
+
+  private suspend fun tcp() = coroutineScope {
+    println("[SyslogSocketServer] started listening: ${serverSocketTcp.localAddress}")
+
+    while (isActive) {
+      val socket = serverSocketTcp.accept()
+
+      launch {
+        val channel = socket.openReadChannel()
+        try {
+          while (socket.socketContext.isActive) {
+            val line = channel.readUTF8Line()
+            println("[SyslogSocketServer] received: $line")
+
+            val syslog = syslogParser.parseLine(line).toSyslogMsg()
+
+            _messages.emit(syslog)
+
+            yield()
+          }
+        } catch (e: Exception) {
+          socket.awaitClosed()
+        }
+      }
+    }
+
+    yield()
+  }
+}
+
+
+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]
+
+  companion object {
+    fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
+      val map = entries.mapNotNull { (k, v) ->
+        when (v) {
+          is String -> k to v
+          else      -> null
+        }
+      }.toMap()
+      return SyslogMsg(map)
+    }
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/processor/topology/factorioServerPackets.kt

@@ -5,7 +5,9 @@ import dev.adamko.kafkatorio.processor.admin.TOPIC_SRC_SERVER_LOG
 import dev.adamko.kafkatorio.processor.admin.topicName
 import dev.adamko.kafkatorio.processor.serdes.jsonMapper
 import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.Tick
 import dev.adamko.kafkatorio.schema.packets.KafkatorioPacket
+import dev.adamko.kafkatorio.schema.packets.KafkatorioPacketData
 import dev.adamko.kotka.extensions.component1
 import dev.adamko.kotka.extensions.component2
 import dev.adamko.kotka.extensions.consumedAs
@@ -31,8 +33,26 @@ fun factorioServerPacketStream(
       Serdes.String(),
     )
   ).map("decode-packets") { serverId: String, value: String ->
-    val packet = jsonMapper.decodeFromString<KafkatorioPacket>(value)
+    val packet = runCatching {
+      jsonMapper.decodeFromString<KafkatorioPacket>(value)
+    }.getOrElse { e ->
+     KafkatorioPacket(
+        modVersion = "unknown",
+        tick = Tick(0u),
+        data = KafkatorioPacketData.Error(
+          message = e.message,
+          rawValue = value,
+        )
+      )
+    }
+
+    if (packet.data is KafkatorioPacketData.Error) {
+      println("error parsing $TOPIC_SRC_SERVER_LOG message: ${packet.data}")
+    }
+
     FactorioServerId(serverId) to packet
+  }.filter { _, value ->
+    value.data !is KafkatorioPacketData.Error
   }
 }
 

modules/events-server/src/main/resources/config.yml

@@ -3,10 +3,16 @@
 web-port: ${SERVER_PORT:-12080}
 tile-dir: ${TILE_DIR:-./tile-dir}
 
+
+socket-server-host: ${SOCKET_SERVER_HOST:-127.0.0.1}
+socket-server-port: ${SOCKET_SERVER_PORT:-9002}
+
+
 kafka-admin-config:
   client.id: kafkatorio.events-server-admin
   bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:-http://localhost:9092}
 
+
 kafka-streams-config:
   application.id: kafkatorio-events-processor
   bootstrap.servers: ${KAFKA_BOOTSTRAP_SERVERS:-http://localhost:9092}