[feature] StreamInfo Thread sleep time changed (3min -> 10min)

- active streamer gauge connected the wrong variable.

build.gradle.kts

@@ -29,6 +29,7 @@ repositories {
 dependencies {
     // https://mvnrepository.com/artifact/ch.qos.logback/logback-classic
     implementation("ch.qos.logback:logback-classic:1.5.13")
+    implementation("com.github.loki4j:loki-logback-appender:2.0.0")
 
     // https://mvnrepository.com/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.9.0")

chatbot/src/main/kotlin/space/mori/chzzk_bot/chatbot/chzzk/ChzzkHandler.kt

@@ -9,6 +9,7 @@ import space.mori.chzzk_bot.chatbot.chzzk.Connector.getChannel
 import space.mori.chzzk_bot.chatbot.discord.Discord
 import space.mori.chzzk_bot.chatbot.utils.refreshAccessToken
 import space.mori.chzzk_bot.common.events.*
+import space.mori.chzzk_bot.common.metrics.Metrics
 import space.mori.chzzk_bot.common.models.User
 import space.mori.chzzk_bot.common.services.LiveStatusService
 import space.mori.chzzk_bot.common.services.TimerConfigService
@@ -27,6 +28,7 @@ import java.lang.Thread
 import java.net.SocketTimeoutException
 import java.nio.charset.Charset
 import java.time.LocalDateTime
+import java.util.concurrent.ConcurrentHashMap
 
 object ChzzkHandler {
     private val handlers = mutableListOf<UserHandler>()
@@ -35,6 +37,9 @@ object ChzzkHandler {
     @Volatile private var running: Boolean = false
     private val dispatcher: CoroutinesEventBus by inject(CoroutinesEventBus::class.java)
 
+    private val lastRunMap = ConcurrentHashMap<String, Long>()
+    private val requiredWait = 300_000L
+
     fun addUser(chzzkChannel: ChzzkChannel, user: User) {
         handlers.add(UserHandler(chzzkChannel, logger, user, streamStartTime = LocalDateTime.now()))
     }
@@ -127,6 +132,7 @@ object ChzzkHandler {
                     } catch (e: Exception) {
                         logger.info("Thread 1 Exception: ${it.channel.channelName} / ${e.stackTraceToString()}")
                     } finally {
+                        lastRunMap[it.channel.channelId] = System.currentTimeMillis()
                         Thread.sleep(5000)
                     }
                 }
@@ -135,12 +141,21 @@ object ChzzkHandler {
         }
 
         val threadRunner2 = Runnable {
-            logger.info("Thread 2 started!")
             logger.info("Thread 2 started!")
             while (running) {
                 handlers.forEach {
                     if (!running) return@forEach
                     try {
+                        val now = System.currentTimeMillis()
+                        val lastRun = lastRunMap[it.channel.channelId] ?: 0L
+                        val waitedTime = now - lastRun
+
+                        if(waitedTime < requiredWait) {
+                            val sleepTime = requiredWait - waitedTime
+                            logger.info("Thread 2 Sleep: ${it.channel.channelName} / ${sleepTime}ms")
+                            Thread.sleep(sleepTime)
+                        }
+
                         val streamInfo = Connector.getLive(it.channel.channelId)
                         if (streamInfo?.isOnline == true && !it.isActive) {
                             try {
@@ -203,10 +218,11 @@ class UserHandler(
     val logger: Logger,
     private var user: User,
     var streamStartTime: LocalDateTime?,
+    val chatLogger: Logger = LoggerFactory.getLogger("${channel.channelName}-chat"),
 ) {
     lateinit var client: ChzzkClient
     lateinit var chatChannelId: String
-    lateinit var listener: ChzzkUserSession
+    var listener: ChzzkUserSession? = null
     lateinit var messageHandler: MessageHandler
 
     private val dispatcher: CoroutinesEventBus by inject(CoroutinesEventBus::class.java)
@@ -233,14 +249,12 @@ class UserHandler(
 
         client.loginAsync().await()
         listener = ChzzkSessionBuilder(client).buildUserSession()
-        listener.createAndConnectAsync().await()
-        listener.subscribeAsync(ChzzkSessionSubscriptionType.CHAT)?.await()
 
         delay(5000L)
 
         messageHandler = MessageHandler(this@UserHandler)
         logger.info("${user.username} message handler init.")
-        listener.on(SessionChatMessageEvent::class.java) {
+        listener?.on(SessionChatMessageEvent::class.java) {
             messageHandler.handle(it.message, user)
         }
         logger.info("${user.username} is connected.")
@@ -264,8 +278,8 @@ class UserHandler(
     }
 
     internal suspend fun disable() {
-        listener.unsubscribeAsync(ChzzkSessionSubscriptionType.CHAT)?.await()
+        listener?.unsubscribeAsync(ChzzkSessionSubscriptionType.CHAT)?.await()
-        listener.disconnectAsync()?.await()
+        listener?.disconnectAsync()?.await()
 
         _isActive = false
     }
@@ -284,14 +298,18 @@ class UserHandler(
     internal fun isActive(value: Boolean, status: ChzzkLiveDetail) {
         if(value) {
             CoroutineScope(Dispatchers.Default).launch {
-                connect()
+                if(listener == null)
+                    connect()
 
+                listener!!.createAndConnectAsync()?.await()
+                listener!!.subscribeAsync(ChzzkSessionSubscriptionType.CHAT)?.await()
                 logger.info("${user.username} is live.")
 
                 reloadUser(UserService.getUser(user.id.value)!!)
 
                 logger.info("ChzzkChat connecting... ${channel.channelName} - ${channel.channelId}")
                 streamStartTime = LocalDateTime.now()
+                Metrics.increaseStreaming()
 
                 if(!_isActive) {
                     _isActive = true
@@ -325,9 +343,10 @@ class UserHandler(
         } else {
             logger.info("${user.username} is offline.")
             streamStartTime = null
-            listener.unsubscribeAsync(ChzzkSessionSubscriptionType.CHAT)?.join()
+            listener?.unsubscribeAsync(ChzzkSessionSubscriptionType.CHAT)?.join()
-            listener.disconnectAsync()?.join()
+            listener?.disconnectAsync()?.join()
             _isActive = false
+            Metrics.decreaseStreaming()
 
             CoroutineScope(Dispatchers.Default).launch {
                 val events = listOf(
@@ -366,6 +385,7 @@ class UserHandler(
         GlobalScope.launch {
             delay(100L)
             client.sendChatToLoggedInChannel(msg.limitUtf8Length(100))
+            chatLogger.info("[SEND]${channel.channelName}: ${msg.limitUtf8Length(100)}{${msg.length} / 100}")
         }
     }
 }

chatbot/src/main/kotlin/space/mori/chzzk_bot/chatbot/chzzk/MessageHandler.kt

@@ -31,6 +31,7 @@ class MessageHandler(
 
     private val channel = handler.channel
     private val logger = handler.logger
+    private val chatLogger = handler.chatLogger
     private val listener = handler.listener
 
     private val dispatcher: CoroutinesEventBus by inject(CoroutinesEventBus::class.java)
@@ -372,7 +373,7 @@ class MessageHandler(
 
     internal fun handle(msg: SessionChatMessage, user: User) {
         if(msg.senderChannelId == ChzzkHandler.botUid) return
-
+        chatLogger.info("[RECV]${channel.channelName}: ${msg.content}{${msg.content.length} / 100}")
         val commandKey = msg.content.split(' ')[0]
         commands[commandKey.lowercase()]?.let { it(msg, user) }
     }

common/build.gradle.kts

@@ -37,6 +37,11 @@ dependencies {
     // https://mvnrepository.com/artifact/com.google.code.gson/gson
     implementation("com.google.code.gson:gson:2.11.0")
 
+    api("io.micrometer:micrometer-registry-prometheus:1.15.1")
+
+    // https://mvnrepository.com/artifact/io.insert-koin/koin-core
+    api("io.insert-koin:koin-core:4.0.0")
+
     testImplementation(kotlin("test"))
 }
 

common/src/main/kotlin/space/mori/chzzk_bot/common/metrics/Metrics.kt

@@ -0,0 +1,43 @@
+package space.mori.chzzk_bot.common.metrics
+
+import io.micrometer.core.instrument.Gauge
+import io.micrometer.prometheusmetrics.PrometheusConfig
+import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
+import space.mori.chzzk_bot.common.services.UserService
+
+object Metrics {
+    val registry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
+
+    var streamer =  0.0
+    val streamerGauge: Gauge = Gauge.builder("streamer_gauge", this) { streamer }
+        .description("Current All Streamer Count")
+        .register(registry)
+
+    var activeStreamer = 0.0
+    val activateGauge: Gauge = Gauge.builder("active_streamer_gauge", this) { activeStreamer }
+        .description("Current Active Streamer Count")
+        .register(registry)
+
+    var streaming: Double = 0.0
+    val streamingGauge: Gauge = Gauge.builder("streaming_gauge", this) { streaming }
+        .description("Current Streaming Streamer Count")
+        .register(registry)
+
+    fun refreshStreamerMetrics() {
+        val streamers = UserService.getAllUsers()
+
+        streamer = streamers.size.toDouble()
+        activeStreamer = streamers.filter { !it.isDisabled }.size.toDouble()
+    }
+
+    fun increaseStreaming(inc: Int = 1) {
+        streaming += inc
+    }
+    fun decreaseStreaming(dec: Int = 1) {
+        streaming -= dec
+    }
+
+    init {
+        refreshStreamerMetrics()
+    }
+}

src/main/kotlin/space/mori/chzzk_bot/Main.kt

@@ -12,6 +12,7 @@ import space.mori.chzzk_bot.chatbot.discord.Discord
 import space.mori.chzzk_bot.chatbot.chzzk.Connector as ChzzkConnector
 import space.mori.chzzk_bot.common.Connector
 import space.mori.chzzk_bot.common.events.CoroutinesEventBus
+import space.mori.chzzk_bot.common.metrics.Metrics
 import space.mori.chzzk_bot.webserver.start
 import space.mori.chzzk_bot.webserver.stop
 import java.util.concurrent.TimeUnit
@@ -25,6 +26,7 @@ val logger: Logger = LoggerFactory.getLogger("main")
 fun main(args: Array<String>) {
     val dispatcher = module {
         single { CoroutinesEventBus() }
+        single { Metrics.registry }
     }
     startKoin {
         modules(dispatcher)

src/main/resources/logback.xml

@@ -1,5 +1,5 @@
 <configuration>
-
+    <property name="lokiUrl" value="${LOKI_ENDPOINT}"/>
     <!-- 콘솔에 출력하는 기본 로그 설정 -->
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
@@ -10,6 +10,15 @@
         </encoder>
     </appender>
 
+    <appender name="LOKI" class="com.github.loki4j.logback.Loki4jAppender">
+        <http>
+            <url>${lokiUrl}</url>
+        </http>
+        <labels>
+            app = chibot-chzzk-bot
+        </labels>
+    </appender>
+
     <!-- HikariCP 로그 레벨 설정 -->
     <logger name="com.zaxxer.hikari" level="INFO" />
     <logger name="o.m.jdbc.client.impl.StandardClient" level="INFO" />
@@ -31,5 +40,8 @@
     <root level="DEBUG">
         <appender-ref ref="STDOUT" />
     </root>
+    <root level="INFO">
+        <appender-ref ref="LOKI" />
+    </root>
 
 </configuration>

webserver/build.gradle.kts

@@ -34,16 +34,16 @@ dependencies {
     // https://mvnrepository.com/artifact/org.jetbrains.kotlin/kotlin-reflect
     implementation("org.jetbrains.kotlin:kotlin-reflect:2.0.21")
 
-    // https://mvnrepository.com/artifact/io.insert-koin/koin-core
-    implementation("io.insert-koin:koin-core:4.0.0")
-
     // https://mvnrepository.com/artifact/ch.qos.logback/logback-classic
     implementation("ch.qos.logback:logback-classic:1.5.12")
 
     // https://mvnrepository.com/artifact/io.github.cdimascio/dotenv-kotlin
     implementation("io.github.cdimascio:dotenv-kotlin:6.4.2")
 
+    implementation("io.ktor:ktor-server-metrics-micrometer:$ktorVersion")
+
     implementation(project(":common"))
+    implementation("io.ktor:ktor-server-metrics-micrometer:3.1.3")
 
     testImplementation(kotlin("test"))
 }

webserver/src/main/kotlin/space/mori/chzzk_bot/webserver/Main.kt

@@ -11,6 +11,7 @@ import io.ktor.serialization.kotlinx.json.*
 import io.ktor.server.application.*
 import io.ktor.server.auth.*
 import io.ktor.server.engine.*
+import io.ktor.server.metrics.micrometer.MicrometerMetrics
 import io.ktor.server.netty.*
 import io.ktor.server.plugins.contentnegotiation.*
 import io.ktor.server.plugins.cors.routing.*
@@ -19,9 +20,16 @@ import io.ktor.server.response.*
 import io.ktor.server.routing.*
 import io.ktor.server.sessions.*
 import io.ktor.server.websocket.*
+import io.micrometer.core.instrument.binder.jvm.JvmGcMetrics
+import io.micrometer.core.instrument.binder.jvm.JvmMemoryMetrics
+import io.micrometer.core.instrument.binder.system.ProcessorMetrics
+import io.micrometer.prometheusmetrics.PrometheusConfig
+import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
 import kotlinx.coroutines.delay
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.Json
+import org.koin.core.context.startKoin
+import org.koin.dsl.module
 import org.koin.java.KoinJavaComponent.inject
 import space.mori.chzzk_bot.common.events.CoroutinesEventBus
 import space.mori.chzzk_bot.common.events.UserRegisterEvent
@@ -232,6 +240,18 @@ val server = embeddedServer(Netty, port = 8080, ) {
             }
         }
 
+        val appMicrometerRegistry: PrometheusMeterRegistry by inject(PrometheusMeterRegistry::class.java)
+
+        install(MicrometerMetrics) {
+            registry = appMicrometerRegistry
+
+            meterBinders = listOf(
+                JvmMemoryMetrics(),
+                JvmGcMetrics(),
+                ProcessorMetrics()
+            )
+        }
+
         apiRoutes()
         apiSongRoutes()
         apiCommandRoutes()
@@ -242,6 +262,8 @@ val server = embeddedServer(Netty, port = 8080, ) {
         wsSongRoutes()
         wsSongListRoutes()
 
+        metricRoutes()
+
         swaggerUI("swagger-ui/index.html", "openapi/documentation.yaml") {
             options {
                 version = "1.2.0"

webserver/src/main/kotlin/space/mori/chzzk_bot/webserver/routes/ApiRoutes.kt

@@ -1,6 +1,7 @@
 package space.mori.chzzk_bot.webserver.routes
 
 import io.ktor.http.*
+import io.ktor.server.request.receive
 import io.ktor.server.response.*
 import io.ktor.server.routing.*
 import io.ktor.server.sessions.*
@@ -15,6 +16,7 @@ import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.withTimeoutOrNull
 import space.mori.chzzk_bot.common.events.ChzzkUserFindEvent
 import space.mori.chzzk_bot.common.events.ChzzkUserReceiveEvent
+import space.mori.chzzk_bot.common.metrics.Metrics
 
 @Serializable
 data class GetUserDTO(
@@ -146,4 +148,50 @@ fun Routing.apiRoutes() {
             call.respond(HttpStatusCode.OK, returnUsers)
         }
     }
-}
+
+    route("/settings") {
+        get {
+            val session = call.sessions.get<UserSession>()
+            if(session == null) {
+                call.respondText("No session found", status = HttpStatusCode.Unauthorized)
+                return@get
+            }
+            val user = UserService.getUser(session.id)
+            if(user == null) {
+                call.respondText("No user found", status = HttpStatusCode.NotFound)
+                return@get
+            }
+
+            call.respond(HttpStatusCode.OK, IUserSettingsDTO(
+                user.isDisabled,
+                user.isDisableStartupMsg
+            ))
+        }
+
+        post {
+            val session = call.sessions.get<UserSession>()
+            val body: IUserSettingsDTO = call.receive()
+            if(session == null) {
+                call.respondText("No session found", status = HttpStatusCode.Unauthorized)
+                return@post
+            }
+            val user = UserService.getUser(session.id)
+            if(user == null) {
+                call.respondText("No user found", status = HttpStatusCode.NotFound)
+                return@post
+            }
+
+            UserService.setIsDisabled(user, body.isBotDisabled)
+            UserService.setIsStartupDisabled(user, body.isBotMsgDisabled)
+
+            Metrics.refreshStreamerMetrics()
+
+            call.respond(HttpStatusCode.OK, body)
+        }
+    }
+}
+
+data class IUserSettingsDTO(
+    val isBotDisabled: Boolean,
+    val isBotMsgDisabled: Boolean
+)

webserver/src/main/kotlin/space/mori/chzzk_bot/webserver/routes/MetricRoutes.kt

@@ -0,0 +1,36 @@
+package space.mori.chzzk_bot.webserver.routes
+
+import io.ktor.server.application.ApplicationStopped
+import io.ktor.server.response.respondText
+import io.ktor.server.routing.Routing
+import io.ktor.server.routing.get
+import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import org.koin.java.KoinJavaComponent.inject
+import space.mori.chzzk_bot.common.events.CoroutinesEventBus
+import space.mori.chzzk_bot.common.events.UserRegisterEvent
+import space.mori.chzzk_bot.common.metrics.Metrics
+import kotlin.getValue
+
+
+val metricScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+fun Routing.metricRoutes() {
+    environment.monitor.subscribe(ApplicationStopped) {
+        metricScope.cancel()
+    }
+
+    val dispatcher: CoroutinesEventBus by inject(CoroutinesEventBus::class.java)
+    val registry: PrometheusMeterRegistry by inject(PrometheusMeterRegistry::class.java)
+
+    dispatcher.subscribe(UserRegisterEvent::class) {
+        Metrics.refreshStreamerMetrics()
+    }
+
+    get("/metrics") {
+        call.respondText(registry.scrape())
+    }
+
+}