From dda48bac78eb05e87dd011b59c967b677deb2497 Mon Sep 17 00:00:00 2001 From: Adam <897017+aSemy@users.noreply.github.com> Date: Tue, 18 Oct 2022 21:27:16 +0200 Subject: [PATCH] refactor the KafkatorioPacket tag that the mod logs, and the syslog-server parses --- .../lua/main/emitting/PacketEmitter.lua | 12 ++- .../typescript/main/emitting/PacketEmitter.ts | 16 +++- .../topology/factorioServerPackets.kt | 79 ++++++++++++++----- modules/infra-kafka-pipe/src/kafka-pipe.sh | 2 +- 4 files changed, 84 insertions(+), 25 deletions(-) diff --git a/modules/events-mod/src/generated/lua/main/emitting/PacketEmitter.lua b/modules/events-mod/src/generated/lua/main/emitting/PacketEmitter.lua index bbc2fb6..2ec7a8b 100644 --- a/modules/events-mod/src/generated/lua/main/emitting/PacketEmitter.lua +++ b/modules/events-mod/src/generated/lua/main/emitting/PacketEmitter.lua @@ -1,5 +1,6 @@ local ____lualib = require("lualib_bundle") local __TS__Class = ____lualib.__TS__Class +local __TS__StringTrim = ____lualib.__TS__StringTrim local __TS__New = ____lualib.__TS__New local ____exports = {} ____exports.KafkatorioPacketEmitter = __TS__Class() @@ -15,9 +16,16 @@ function KafkatorioPacketEmitter.prototype.emitKeyedPacket(self, data) end function KafkatorioPacketEmitter.emitPacket(self, packet) local data = game.table_to_json(packet) + if #__TS__StringTrim(data) <= 0 then + print("[error] table_to_json returned empty string for packet:" .. tostring(packet)) + return + end local encodedData = game.encode_string(data) - if encodedData ~= nil then - print("KafkatorioPacket encoded::: " .. encodedData) + if encodedData == nil or #__TS__StringTrim(encodedData) <= 0 then + print("[error] could not encode packet") + print("KafkatorioPacket:::" .. data) + else + print("KafkatorioPacket:::encoded:" .. encodedData) end end local PacketEmitter = __TS__New(____exports.KafkatorioPacketEmitter) diff --git a/modules/events-mod/src/main/typescript/main/emitting/PacketEmitter.ts b/modules/events-mod/src/main/typescript/main/emitting/PacketEmitter.ts index eb7b6b5..69e36a9 100644 --- a/modules/events-mod/src/main/typescript/main/emitting/PacketEmitter.ts +++ b/modules/events-mod/src/main/typescript/main/emitting/PacketEmitter.ts @@ -27,11 +27,19 @@ export class KafkatorioPacketEmitter { /** Emit a serialised KafkatorioPacket */ private static emitPacket(packet: T) { const data = game.table_to_json(packet) - // print(`KafkatorioPacket::: ${data}`) + + if (data.trim().length <= 0) { + print(`[error] table_to_json returned empty string for packet:${packet}`) + return + } + const encodedData = game.encode_string(data) - if (encodedData != null) { - print(`KafkatorioPacket encoded::: ${encodedData}`) - // rcon.print(`KafkatorioPacket: ${data}`) + + if (encodedData == null || encodedData.trim().length <= 0) { + print(`[error] could not encode packet`) + print(`KafkatorioPacket:::${data}`) + } else { + print(`KafkatorioPacket:::encoded:${encodedData}`) // this.appendToFile(encodedData) } } diff --git a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt index 42999ab..69872ae 100644 --- a/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt +++ b/modules/events-processor-core/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt @@ -1,12 +1,12 @@ package dev.adamko.kafkatorio.server.processor.topology +import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG +import dev.adamko.kafkatorio.processor.config.jsonMapper +import dev.adamko.kafkatorio.processor.config.topicName 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.KafkatorioPacketDataError -import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG -import dev.adamko.kafkatorio.processor.config.jsonMapper -import dev.adamko.kafkatorio.processor.config.topicName import dev.adamko.kotka.extensions.component1 import dev.adamko.kotka.extensions.component2 import dev.adamko.kotka.extensions.consumedAs @@ -26,6 +26,10 @@ import org.apache.kafka.streams.StreamsBuilder import org.apache.kafka.streams.kstream.KStream +private const val PACKET_TAG = "KafkatorioPacket:::" +private const val PACKET_ENCODED_TAG = "${PACKET_TAG}encoded:" + + fun factorioServerPacketStream( builder: StreamsBuilder = StreamsBuilder(), ): KStream { @@ -39,21 +43,7 @@ fun factorioServerPacketStream( ) ).mapValues("decode-packets") { _: FactorioServerId, value: String? -> - val packet = try { - val decoded = decodeFactorioEncodedString(value!!) - jsonMapper.decodeFromString(KafkatorioPacket.serializer(), decoded) - } catch (e: Exception) { - if (e is CancellationException) throw e - - KafkatorioPacket( - modVersion = "unknown", - tick = Tick(0u), - data = KafkatorioPacketDataError( - exception = e, - rawValue = value, - ) - ) - } + val packet = decodeKafkatorioPacket(value) if (packet.data is KafkatorioPacketDataError) { println("error parsing $TOPIC_SRC_SERVER_LOG message: $packet") @@ -81,6 +71,31 @@ fun splitFactorioServerPacketStream( } +private fun decodeKafkatorioPacket(value: String?): KafkatorioPacket = try { + when { + value.isNullOrBlank() -> + createErrorPacket(message = "null/blank message", rawValue = value) + + value.substringAfter(PACKET_TAG).isBlank() -> + createErrorPacket(message = "missing '$PACKET_TAG'", rawValue = value) + + else -> { + val packetJson = if (PACKET_ENCODED_TAG in value) { + val encodedValue = value.substringAfter(PACKET_ENCODED_TAG) + decodeFactorioEncodedString(encodedValue) + } else { + value.substringAfter(PACKET_TAG) + } + + jsonMapper.decodeFromString(KafkatorioPacket.serializer(), packetJson) + } + } +} catch (ex: Exception) { + if (ex is CancellationException) throw ex + errorPacket(exception = ex, rawValue = value) +} + + /** * Factorio `game.encode_string` zlib encodes a string, then base64's it. Here we do the reverse. */ @@ -94,3 +109,31 @@ private fun decodeFactorioEncodedString( .buffer() .readString(Charsets.UTF_8) } + + +private fun createErrorPacket( + message: String, + rawValue: String?, +): KafkatorioPacket = + KafkatorioPacket( + modVersion = "unknown", + tick = Tick(0u), + data = KafkatorioPacketDataError( + message = message, + rawValue = rawValue, + ) + ) + + +private fun errorPacket( + exception: Exception, + rawValue: String?, +): KafkatorioPacket { + val message = """ +${exception::class.qualifiedName} ${exception.message} + +${exception.stackTraceToString()} +""".trimIndent() + + return createErrorPacket(message, rawValue) +} diff --git a/modules/infra-kafka-pipe/src/kafka-pipe.sh b/modules/infra-kafka-pipe/src/kafka-pipe.sh index e0922be..3c16fe8 100644 --- a/modules/infra-kafka-pipe/src/kafka-pipe.sh +++ b/modules/infra-kafka-pipe/src/kafka-pipe.sh @@ -17,7 +17,7 @@ KEY="${INSTANCE_ID:-none}" while :; do docker logs --tail 0 -f "$FACTORIO_SERVER_CONTAINER_NAME" | - sed -n -e 's/^KafkatorioPacket encoded::: //p' | + sed -n -e 's/^KafkatorioPacket::://p' | kcat -P \ -b "$KAFKA_HOST" \ -t kafkatorio.src.server-log \