[Day 19] 实作 Ktor Request Logging

Request Logging 是非常重要的必备功能，不仅可以用来系统监控、除错，还可以进一步分析使用者行为…等，存在着广泛的应用情境，所以每个 Web 框架都会有解决方案，在此我先简介 Spring 与 Ktor 的作法，再来说明为何我不用 Ktor 的 CallLogging Plugin 就好，需要实作自己的 Plugin。

Request Logging 的作法

Spring CommonsRequestLoggingFilter

以 Spring 框架来说，我们可以自行实作 filter、interceptor、aspect，以不同的方式拦截请求来记录 log，甚至直接使用预设实作 CommonsRequestLoggingFilter 即可。然而上述这些方式，或多或多都存在许多限制及不便，例如 CommonsRequestLoggingFilter只能拿到 HttpServletRequest 物件，所以无法记录 Response 资料。如果改用 Interceptor 实作，则会受到只能读取一次 request input stream 的限制，所以必须要使用 ContentCachingRequestWrapper and ContentCachingResponseWrapper 暂存资料，也就是需要自行处理许多实作细节，读者可以参考 Baeldung 的这篇文章 Spring – Log Incoming Requests。

@Configuration
public class RequestLoggingFilterConfig {

    @Bean
    public CommonsRequestLoggingFilter logFilter() {
        CommonsRequestLoggingFilter filter = new CommonsRequestLoggingFilter();
        filter.setIncludeQueryString(true);
        filter.setIncludePayload(true);
        filter.setMaxPayloadLength(10000);
        filter.setIncludeHeaders(false);
        filter.setAfterMessagePrefix("REQUEST DATA : ");
        return filter;
    }
}

Ktor CallLogging Plugin

Ktor CallLogging Plugin 与 Spring CommonsRequestLoggingFilter 一样，都只能记录 Request 资料，而且也会有不能重复读取 request input stream 的问题，所以需要再安装 Ktor DoubleReceive Plugin 解决。CallLogging Plugin 需要自行实作 filter 与 format 2个 configure function，虽然不像 CommonsRequestLoggingFilter 可以直接呼叫 setIncludeQueryString …之类的方法进行设定，但是可以依自己的需求实作，所以使用上较为弹性。

install(CallLogging) {
    filter { call ->
        call.request.path().startsWith("/api/v1")
    }
    format { call ->
        val status = call.response.status()
        val httpMethod = call.request.httpMethod.value
        val userAgent = call.request.headers["User-Agent"]
        "Status: $status, HTTP method: $httpMethod, User agent: $userAgent"
    }
}

install(DoubleReceive) {
    receiveEntireContent = true
}

实作目标

由於 Ktor CallLogging Plugin 只能记录 Request 资料，而且内部实作是使用 SLF4J Logger，并没有提供 write configure function 让我客制化实作如何写入 log，所以我想自己写 MyCallLogging Plugin 达成以下目标

• 除了记录 Request 资料之外，还要能记录 Response 资料，包括 Response Body 及 HTTP Status Code
• 要能同时记录 Request 与 Response 为一笔 log，不可拆开为2笔记录
• 实作自订 filter 完全控制写 log 的行为，而且可以透过 Ktor 设定档进行设定。
• 不使用字串型态写入 Log，而是定义自己的 RequestLog 资料类别，再根据写入的目的地，转换为特定的资料格式，例如 JSON
• 写入 RequestLog 的目的地不限於档案，可以自行实作 LogWriter 以非同步方式写到任意目的地，例如资料库，或是转为 JSON 传送到 AWS Kinesis Data Firehose。

改写 Ktor CallLogging Plugin 实作 MyCallLogging Plugin

虽然 Ktor CallLogging Plugin 不合我所用，但可以拿它的原始码，当作样版来修改会比较简单。我实作的 MyCallLoggingPlugin 多了 RequestLogConfig 及 LogWriter 物件。RequestLogConfig 是读取 Ktor application.conf 设定档的 logging config，包含了许多 filter 设定，LogWriter 则是根据设定档 writer 设定从 Koin DI 拿到的物件

class MyCallLoggingFeature private constructor(
    private val monitor: ApplicationEvents,
    private val mdcEntries: List<MDCEntry>,
    private val config: RequestLogConfig,
    private val logWriter: LogWriter
)

logging {
    request {
        enabled = true
        destination = "AwsKinesis" # File(default), Database, AwsKinesis
        includeHeaders = false
        includeQueryString = false
        includeResponseBody = false
        includeGetMethod = false
        excludePaths = ["/ops/sysstat/healthCheck"]
        excludeRequestBodyPaths = ["/login", "/myPassword"]
    }
    writer {
        awsKinesis {
            streamName = "logging"
            nettyClient {
                #maxConcurrency = 50 # => aws default value = 50
                #maxPendingConnectionAcquires = 10000 => aws default value = 10000
                #maxIdleConnectionTimeout = 60s => aws default value = 60s
            }
            threadPool {
                fixedPoolSize = 3
            }
        }
    }
}

拦截 Ktor ApplicationSendPipeline.Render PipelinePhase 取得 Response 资料

Ktor CallLogging 原本只在 ApplicationCallPipeline.Monitoring 插入 loggingPhase，当 proceed() 执行完请求後，呼叫 logSuccess() 写入 log。为了拿到 Response 资料，我在 ApplicationSendPipeline.Render 之後插入 PipelinePhase，如果 message 是 OutgoingContent 的子类别 TextContent 物件，则把它储存在 ApplicationCall 的 attribute 里面。未来 logSuccess() 呼叫 logWriter 写入 log 时，就可以透过 call.attributes.getOrNull(ATTRIBUTE_KEY_RSP_BODY) 拿到 response body，把 request 与 response 资料写到同一笔 log。

pipeline.insertPhaseBefore(ApplicationCallPipeline.Monitoring, loggingPhase)
pipeline.intercept(loggingPhase) {
    proceed()
    feature.logSuccess(call)
}

if (requestConfig.includeResponseBody) {
    pipeline.sendPipeline.insertPhaseAfter(ApplicationSendPipeline.Render, afterRenderResponsePhase)
    pipeline.sendPipeline.intercept(afterRenderResponsePhase) { message ->
        if (message is TextContent) {
            call.attributes.put(ATTRIBUTE_KEY_RSP_BODY, message.text)
        }
    }
}

使用自己的 LogWriter 替换 Ktor CallLogging Plugin 预设的 SLF4j Logger

原本 Ktor CallLogging Plugin 的 logSuccess 方法的内部实作是使用 SLF4j Logger

private val log: Logger // SLF4j Logger

private fun logSuccess(call: ApplicationCall) {
    if (filters.isEmpty() || filters.any { it(call) }) {
        log(formatCall(call))
    }
}
private fun log(message: String) = when (level) {
    Level.ERROR -> log.error(message)
    Level.WARN -> log.warn(message)
    Level.INFO -> log.info(message)
    Level.DEBUG -> log.debug(message)
    Level.TRACE -> log.trace(message)
}

改写为使用 LogWriter 写入由 buildLogMessage(call) 方法建立的 RequestLog 物件

private fun logSuccess(call: ApplicationCall) {
    if (config.enabled) {
        if (filter(call)) {
            logWriter.write(buildLogMessage(call))
        }
    }
}

定义 RequestLog 资料类别结构化 Log 资料

RequestLog 类别除了一般 Request Logging 记录的 http request 资料之外，我还多记录了许多自定义栏位。此外，还要标注 @Serializable，kotlinx.serialization library 才能序列化为 JSON。

• reqId => 使用 Ktor CallId Plugin 产生的乱数作为请求 id，方便 trace & debug
• project => 在多专案架构下，记录这个 request 是属於那个子专案
• tag => 处理请求的时候，视需要可以填入字串，未来作为搜寻 log 的条件
• source, principal => 客户端验证资讯
• clientId, clientVersion, ip => 客户端装置资讯

@Serializable
data class RequestLog(
    val reqId: String,
    @Serializable(with = InstantSerializer::class) val reqAt: Instant,
    val api: String,
    val headers: String? = null,
    val querystring: String? = null,
    val reqBody: String?,
    val project: String,
    val function: String,
    val tag: String? = null,
    val source: PrincipalSource,
    val tenantId: TenantId? = null,
    val principal: String,
    val runAs: Boolean,
    val clientId: String? = null,
    val clientVersion: String? = null,
    val ip: String? = null,
    @Serializable(with = InstantSerializer::class) val rspAt: Instant,
    val rspTime: Long,
    val rspStatus: Int,
    val rspBody: String?
) : LogMessage() 

实作各种 LogWriter 写入 RequestLog 到不同目的地

目前 RequestLog 可以使用以下3个 LogWriter 写到不同目的地

• RequestLogDBWriter => 使用 Exposed ORM 写到资料库
• FileLogWriter => 使用 KotlinLogging 函式库写到档案
• AwsKinesisLogWriter => 转为 JSON 传送到 AWS Kinesis Stream

interface LogWriter {

    fun write(message: LogMessage)
}

class RequestLogDBWriter : LogWriter {

    override fun write(message: LogMessage) {
        val dto = message as RequestLog
        transaction {
            RequestLogTable.insert {
                it[reqId] = dto.reqId
                it[reqAt] = dto.reqAt
                it[api] = dto.api
                it[headers] = dto.headers
                it[querystring] = dto.querystring
                it[reqBody] = dto.reqBody
                it[project] = dto.project
                it[function] = dto.function
                it[tag] = dto.tag
                it[sourceId] = dto.source
                it[tenantId] = dto.tenantId?.value
                it[principal] = dto.principal
                it[runAs] = dto.runAs
                it[clientId] = dto.clientId
                it[clientVersion] = dto.clientVersion
                it[ip] = dto.ip
                it[rspAt] = dto.rspAt
                it[rspTime] = dto.rspTime
                it[rspStatus] = dto.rspStatus
                it[rspBody] = dto.rspBody
            }
        }
    }
}

结语

今天说明如何实作自己的 MyCallLogging Plugin 取代 Ktor CallLogging Plugin，满足许多更复杂进阶的需求，但是系统的 log 绝对不只有 Request Log 而已，还有系统错误的 Error Log, 使用者登入的 Login Log…等，所以明天的主题是如何建立系统的 Logging 机制，以非同步方式写入各种类型的 log。