lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[syslog-server] ignore, but log, non-syslog messages. Minor logging tweaks.

Browse source
This commit is contained in:
Adam 2022-10-17 00:14:27 +02:00
parent 1cc5afa6e2
commit 0216d068c5
2 changed files with 17 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -2,11 +2,11 @@ package dev.adamko.kafkatorio.processor.syslog
import com.github.palindromicity.syslog.dsl.SyslogFieldKeys import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG
import dev.adamko.kafkatorio.processor.config.TOPIC_SRC_SERVER_LOG_DLQ
import dev.adamko.kafkatorio.processor.core.Authenticator import dev.adamko.kafkatorio.processor.core.Authenticator
import dev.adamko.kafkatorio.processor.syslog.config.jsonMapper import dev.adamko.kafkatorio.processor.syslog.config.jsonMapper
import dev.adamko.kafkatorio.schema.common.FactorioServerId 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 dev.adamko.kotka.kxs.kafkaSerializer
import java.util.Properties import java.util.Properties
import kotlin.coroutines.coroutineContext import kotlin.coroutines.coroutineContext
@ -56,6 +56,8 @@ class KafkatorioPacketKafkaProducer(
suspend fun launch() { suspend fun launch() {
coroutineContext.job.invokeOnCompletion { coroutineContext.job.invokeOnCompletion {
val cause = if (it != null) "(cause: ${it::class.java.name} ${it.message})" else ""
log("Closing $cause")
this@KafkatorioPacketKafkaProducer.close() this@KafkatorioPacketKafkaProducer.close()
} }

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

@ -7,6 +7,7 @@ import com.github.palindromicity.syslog.NilPolicy
import com.github.palindromicity.syslog.SyslogParser import com.github.palindromicity.syslog.SyslogParser
import com.github.palindromicity.syslog.SyslogParserBuilder import com.github.palindromicity.syslog.SyslogParserBuilder
import com.github.palindromicity.syslog.SyslogSpecification import com.github.palindromicity.syslog.SyslogSpecification
import com.github.palindromicity.syslog.dsl.ParseException
import dev.adamko.kafkatorio.processor.config.ApplicationProperties import dev.adamko.kafkatorio.processor.config.ApplicationProperties
import io.ktor.network.selector.SelectorManager import io.ktor.network.selector.SelectorManager
import io.ktor.network.sockets.Socket import io.ktor.network.sockets.Socket
@ -83,16 +84,23 @@ class SyslogSocketServer(
"received 'null' line from ${socket.description()}" "received 'null' line from ${socket.description()}"
} }
try {
val syslog = SyslogMsg(syslogParser.parseLine(line)) val syslog = SyslogMsg(syslogParser.parseLine(line))
_messages.emit(syslog) _messages.emit(syslog)
} catch (ex: ParseException) {
log("ignoring non-syslog message: $line")
}
yield() yield()
} }
} catch (e: Throwable) { } catch (e: Throwable) {
if (e is CancellationException) throw e withContext(Dispatchers.IO) { socket.close() }
if (e is CancellationException) {
throw e
} else {
log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}") log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}")
e.printStackTrace() e.printStackTrace()
withContext(Dispatchers.IO) { socket.close() } }
} }
} }