golang接受 kafka 日志数据 格式转化后 保存到clickhouse 批量 高速 结构化日志保存
package main
import (
"encoding/json"
"fmt"
"github.com/tidwall/gjson"
"strings"
"time"
)
// FileBeatMsg 从filebeat接收到的原始消息
type FileBeatMsg struct {
Message string `json:"message"` // json字符串格式
Fields struct {
Tag string `json:"tag,omitempty"`
} `json:"fields,omitempty"`
}
// LogMessage 应用程序产生的日志串格式，可能有不同
type LogMessage struct {
Id int64 `db:"id" json:"id"` // uuid
Tag string `db:"tag" json:"tag"` // 非应用程序产生日志字段 filebeat收集额外字段
Ts MyTime `db:"ts" json:"ts"`
Level string `db:"level" json:"level"`
Msg string `db:"msg" json:"msg"`
Host string `db:"host" json:"host,omitempty"`
Trace string `db:"trace" json:"trace,omitempty"`
Logger string `db:"logger" json:"logger,omitempty"`
Thread string `db:"thread" json:"thread,omitempty"`
Version string `db:"version" json:"version,omitempty"` // version版本号
App string `db:"app" json:"app"`
UserId string `db:"userId" json:"userId,omitempty"`
Method string `db:"method" json:"method,omitempty"`
Url string `db:"url" json:"url,omitempty"`
RemoteIp string `db:"remoteIp" json:"remoteIp,omitempty"`
Raw string `db:"raw" json:"raw,omitempty"` // json字符串格式
}
// BuildLogMessageFromFileBeatRawMessage 从fileBeat日志中取出真正的程序日志
// 示例1: {"@timestamp":"2022-12-25T15:45:47.703+08:00","caller":"service.go:77","content":"223311","host":"LAPTOP-SDRI3RH1","level":"info","version":""}
// 示例2: {"@timestamp":"2022-06-02T20:32:56.741+08:00","@version":"1","message":"HikariPool-1","logger_name":"kariDataSource","thread_name":"SpringApplicationShutdownHook","level":"INFO","level_value":"20000"}
// 示例3: {"time":"2022-06-05T15:12:49.61+08:00","version":"1","msg":"HikariPool-1 - Shutdown completed.","logger_name":"com.zaxxer.hikari.HikariDataSource","thread_name":"SpringApplicationShutdownHook","level":"INFO","level_value":20000,"HOSTNAME":"LAPTOP-SDRI3RH1"}
// 示例4: {"ts":"2022-12-18 23:57:34.823","level":"INFO","msg":"ping ok","host":"bb06dfb76f13","version":"221216.14.ac46fe","trace":"lXemAPOHk5lwTi1q","logger":"com.lifangsir.lifangsir.appctrl.controller.aping.HealthPingCtrl","thread":"http-nio-80-exec-8","userId":"","method":"GET","url":"/health/ping","remoteIp":"192.168.0.2","marker":""}
func BuildLogMessageFromFileBeatRawMessage(raw []byte) (logMsg *LogMessage, err error) {
var fileBeatMsg = &FileBeatMsg{}
err = json.Unmarshal(raw, fileBeatMsg)
if err != nil {
return nil, err
}
logMsg = &LogMessage{}
logMsg.Id = NextSnowflakeId()
logMsg.Tag = fileBeatMsg.Fields.Tag
logMsg.Raw = fileBeatMsg.Message
// 异常兜底
defer func() {
if err2 := recover(); err2 != nil {
if err != nil {
println(err2)
println(err)
}
err = nil
}
}()
// 解析日志
jsonFmtMsg := gjson.Parse(fileBeatMsg.Message)
// 通用字段
logMsg.Ts = MyTime(tryParseTimeStr(JsonTryGetVal(jsonFmtMsg, "ts", "time", "@timestamp")))
logMsg.Level = strings.ToLower(JsonTryGetVal(jsonFmtMsg, "level"))
logMsg.Msg = JsonTryGetVal(jsonFmtMsg, "msg", "content", "message")
logMsg.Host = JsonTryGetVal(jsonFmtMsg, "host", "HOSTNAME")
logMsg.Trace = JsonTryGetVal(jsonFmtMsg, "trace")
logMsg.Logger = JsonTryGetVal(jsonFmtMsg, "logger", "logger_name", "caller")
logMsg.Thread = JsonTryGetVal(jsonFmtMsg, "thread", "thread_name")
logMsg.Version = JsonTryGetVal(jsonFmtMsg, "version", "@version")
logMsg.App = JsonTryGetVal(jsonFmtMsg, "app", "app_name", "appName", "AppName")
// 业务字段
logMsg.UserId = JsonTryGetVal(jsonFmtMsg, "userId")
logMsg.Method = JsonTryGetVal(jsonFmtMsg, "method")
logMsg.Url = JsonTryGetVal(jsonFmtMsg, "url")
logMsg.RemoteIp = JsonTryGetVal(jsonFmtMsg, "remoteIp")
// 返回
return logMsg, nil
}
// JsonTryGetVal 通过多个路径尝试获取值
func JsonTryGetVal(json gjson.Result, path ...string) string {
for i := range path {
val := json.Get(path[i])
if val.Str != "" {
return val.Str
}
}
return ""
}
// tryParseTimeStr
func tryParseTimeStr(timeStr string) (ts time.Time) {
defer func() {
if err := recover(); err != nil {
fmt.Printf("time parse error %v --%v", err, timeStr)
ts = time.Now()
}
}()
rfc333 := []string{
time.RFC3339Nano,
time.RFC3339,
}
timeFormats := []string{
"2006-01-02 15:04:05.999",
"2006-01-02 15:04:05",
}
if strings.Contains(timeStr, "T") {
for i := range rfc333 {
parse, err := time.ParseInLocation(rfc333[i], timeStr, time.Local)
if err != nil {
continue
}
return parse
}
} else {
for i := range timeFormats {
parse, err := time.ParseInLocation(timeFormats[i], timeStr, time.Local)
if err != nil {
continue
}
return parse
}
}
return time.Now()
}
// LogMessageRaw 应用程序产生的日志串格式，可能有不同
// {"@timestamp":"2022-12-25T15:45:47.703+08:00","caller":"service.go:77","content":"223311","host":"LAPTOP-SDRI3RH1","level":"info","version":""}
// {"@timestamp":"2022-06-02T20:32:56.741+08:00","@version":"1","message":"HikariPool-1","logger_name":"kariDataSource","thread_name":"SpringApplicationShutdownHook","level":"INFO","level_value":"20000"}
// {"time":"2022-06-05T15:12:49.61+08:00","version":"1","msg":"HikariPool-1 - Shutdown completed.","logger_name":"com.zaxxer.hikari.HikariDataSource","thread_name":"SpringApplicationShutdownHook","level":"INFO","level_value":20000,"HOSTNAME":"LAPTOP-SDRI3RH1"}
// {"ts":"2022-12-18 23:57:34.823","level":"INFO","msg":"ping ok","host":"bb06dfb76f13","version":"221216.14.ac46fe","trace":"lXemAPOHk5lwTi1q","logger":"com.lifangsir.lifangsir.appctrl.controller.aping.HealthPingCtrl","thread":"http-nio-80-exec-8","userId":"","method":"GET","url":"/health/ping","remoteIp":"192.168.0.2","marker":""}
type LogMessageRaw struct {
// time
Ts string `json:"ts,omitempty"`
TimeStamp string `json:"@timestamp,omitempty"`
Time string `json:"time,omitempty"`
// level
Level string `json:"level,omitempty"`
// msg
Msg string `json:"msg,omitempty"`
Content string `json:"content,omitempty"`
Message string `json:"message,omitempty"`
// host
Host string `json:"host,omitempty"`
HostName string `json:"HOSTNAME,omitempty"`
// trace
Trace string `json:"trace,omitempty"`
// logger
Logger string `json:"logger,omitempty"`
LoggerName string `json:"logger_name,omitempty"`
// thread
Thread string `json:"thread,omitempty"`
ThreadName string `json:"thread_name,omitempty"`
// version
Version string `json:"version,omitempty"`
Version2 string `json:"@version,omitempty"`
// ========以下为非通用字段=======================
// userId
UserId string `json:"userId,omitempty"`
// method
Method string `json:"method,omitempty"`
// url
Url string `json:"url,omitempty"`
// remoteIp
RemoteIp string `json:"remoteIp,omitempty"`
}