Couchbase追踪包日志转为field=value格式的最优方案？

一、事件总线订阅（你的初始思路，是最优方案之一） #

直接通过Couchbase的事件总线拦截追踪事件，在应用层提前格式化为field=value输出，完全避免Splunk解析JSON的开销，还能精准控制要输出的字段，是性能最优的方案。

具体实现代码可以参考下面的示例：

clusterEnvironment.eventBus().subscribe(event -> {
    if (event instanceof OverThresholdRequestsRecordedEvent) {
        OverThresholdRequestsRecordedEvent traceEvent = (OverThresholdRequestsRecordedEvent) event;
        // 遍历每条超标请求记录
        traceEvent.kvTopRequests().forEach(request -> {
            StringBuilder logLine = new StringBuilder();
            // 按field=value格式拼接字段，注意特殊字符转义
            logLine.append("operation_name=").append(escapeValue(request.operationName())).append(" ");
            logLine.append("last_dispatch_duration_us=").append(request.lastDispatchDurationUs()).append(" ");
            logLine.append("last_remote_socket=").append(escapeValue(request.lastRemoteSocket())).append(" ");
            logLine.append("last_local_id=").append(escapeValue(request.lastLocalId())).append(" ");
            logLine.append("last_local_socket=").append(escapeValue(request.lastLocalSocket())).append(" ");
            logLine.append("total_dispatch_duration_us=").append(request.totalDispatchDurationUs()).append(" ");
            logLine.append("total_server_duration_us=").append(request.totalServerDurationUs()).append(" ");
            logLine.append("operation_id=").append(escapeValue(request.operationId())).append(" ");
            logLine.append("timeout_ms=").append(request.timeoutMs()).append(" ");
            logLine.append("last_server_duration_us=").append(request.lastServerDurationUs()).append(" ");
            logLine.append("total_duration_us=").append(request.totalDurationUs()).append(" ");
            logLine.append("total_count=").append(traceEvent.kvTotalCount());
            
            // 通过应用日志框架输出到指定文件，供Splunk采集
            LoggerFactory.getLogger("CouchbaseTraceLogger").info(logLine.toString());
        });
    }
});

// 辅助方法：处理字段值中的空格、引号等特殊字符，用双引号包裹
private String escapeValue(String value) {
    if (value == null) return "\"\"";
    return "\"" + value.replace("\"", "\\\"") + "\"";
}

注意事项：

• 配置日志框架（如SLF4J+Logback），将CouchbaseTraceLogger的输出定向到Splunk监控的日志文件路径
• 必须处理字段值中的特殊字符（比如空格、双引号），避免Splunk解析kv格式时出错

二、自定义日志格式化器（无代码侵入方案） #

如果不想修改业务代码，可以通过日志框架的自定义转换器，直接将Couchbase输出的JSON日志转成field=value格式。以Logback为例：

1. 编写自定义转换器类：

public class JsonToKeyValueConverter extends ClassicConverter {
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    @Override
    public String convert(ILoggingEvent event) {
        try {
            JsonNode rootNode = OBJECT_MAPPER.readTree(event.getFormattedMessage());
            StringBuilder sb = new StringBuilder();
            // 解析kv节点下的内容
            JsonNode kvNode = rootNode.get("kv");
            if (kvNode != null) {
                long totalCount = kvNode.get("total_count").asLong();
                JsonNode topRequests = kvNode.get("top_requests");
                if (topRequests != null && topRequests.isArray()) {
                    for (JsonNode request : topRequests) {
                        // 拼接每个请求的字段
                        sb.append("operation_name=").append(escapeValue(request.get("operation_name").asText())).append(" ");
                        sb.append("last_dispatch_duration_us=").append(request.get("last_dispatch_duration_us").asLong()).append(" ");
                        // 其他字段同理...
                        sb.append("total_count=").append(totalCount);
                        sb.append("\n");
                    }
                }
            }
            return sb.toString().trim();
        } catch (Exception e) {
            // 解析失败时返回原JSON，避免日志丢失
            return event.getFormattedMessage();
        }
    }

    private String escapeValue(String value) {
        if (value == null) return "\"\"";
        return "\"" + value.replace("\"", "\\\"") + "\"";
    }
}

2. 在Logback配置文件中添加转换器和日志定向：

<configuration>
    <!-- 注册自定义转换器 -->
    <conversionRule conversionWord="jsonToKeyValue" converterClass="com.yourcompany.JsonToKeyValueConverter"/>

    <!-- 配置Splunk采集的appender -->
    <appender name="COUCHBASE_TRACE" class="ch.qos.logback.core.FileAppender">
        <file>/var/log/couchbase/trace.log</file>
        <encoder>
            <!-- 使用自定义转换器格式化日志 -->
            <pattern>%jsonToKeyValue(%msg)%n</pattern>
        </encoder>
    </appender>

    <!-- 定向Couchbase追踪日志到指定appender -->
    <logger name="com.couchbase.tracing" level="INFO" additivity="false">
        <appender-ref ref="COUCHBASE_TRACE"/>
    </logger>
</configuration>

这个方案不需要修改业务代码，适合对代码侵入有要求的场景，性能略逊于事件总线订阅，但比Splunk端解析高效得多。

三、Splunk采集层预处理（备选方案） #

如果上面两个方案都无法实施，可以在Splunk的采集阶段做预处理，将JSON转成kv格式。虽然你提到会增加开销，但采集时处理比查询时解析的开销小很多。

在Splunk的props.conf中添加：

[source::/var/log/couchbase/trace.log]
TRANSFORMS-json_to_kv = couchbase_json_transform

在transforms.conf中添加：

[couchbase_json_transform]
REGEX = "([^"]+)":\s*(?:"([^"]+)"|(\d+))
FORMAT = $1::$2$3
DEST_KEY = _raw

注意：这个正则只能处理简单的键值对，嵌套的JSON结构（比如top_requests数组）需要更复杂的规则，处理起来容易遗漏字段，所以仅作为备选。

内容的提问来源于stack exchange，提问作者Yeikel