Go语言开发logrus日志送到elasticsearch构成elk体系

写在前面

这篇文章我们来讲讲怎么把logrus日志送到es。

使用的日志库是 github.com/sirupsen/logrus，由于这个包中的日志对象是可以接入很多个hook的，所以我们可以使用hook来接入 elasticsearch 来操作 。

hook 就是钩子，当设置hook在某个点之后，hook会执行这个点之后异步进行。
比如让我们把hook设置到log日志的地方，当我们log日志的时候，就会异步执行hook。

1. logrus对象创建

先定义一个logrus对象

var LogrusObj *logrus.Logger

初始化这个logrus

func InitLog() {
	if LogrusObj != nil {
		src, _ := setOutputFile()	// 设置输出
		LogrusObj.Out = src
		return
	}
	logger := logrus.New() 	// 实例化
	src, _ := setOutputFile()
	logger.Out = src	// 设置输出
	logger.SetLevel(logrus.DebugLevel)	// 设置日志级别
	// 设置日志格式
	logger.SetFormatter(&logrus.JSONFormatter{
		TimestampFormat: "2006-01-02 15:04:05",
	})
	LogrusObj = logger
}

上面函数方法中设置的src是日志输出的文件路径，也就是相当于类似下面这种日志文件。

我们用天数来进行日志文件的区分。日志文件如果不存在就创建，存在就进行添加写入日志。

func setOutputFile() (*os.File, error) {
	now := time.Now()
	logFilePath := ""
	if dir, err := os.Getwd(); err == nil {
		logFilePath = dir   "/logs/" // 日志文件存储路径
	}
	_, err := os.Stat(logFilePath)
	if os.IsNotExist(err) {
		if err := os.MkdirAll(logFilePath, 0777); err != nil {
			log.Println(err.Error())
			return nil, err
		}
	}
	logFileName := now.Format("2006-01-02")   ".log"
	// 日志文件
	fileName := path.Join(logFilePath, logFileName)
	if _, err := os.Stat(fileName); err != nil {
		if _, err := os.Create(fileName); err != nil {
			log.Println(err.Error())
			return nil, err
		}
	}
	// 写入文件
	src, err := os.OpenFile(fileName, os.O_APPEND|os.O_WRONLY, os.ModeAppend)
	if err != nil {
		log.Println(err)
		return nil, err
	}
	return src, nil
}

至此我们的logrus对象就添加成功了，我们来测试一下。

func TestLogrus(t *testing.T) {
	InitLog()
	LogrusObj.Infoln("测试日志文件")
}

2. logrus hook

2.1 初始化ES

我们这里用的 es 包是 elastic "github.com/elastic/go-elasticsearch" ，
定义一个es client对象。

var EsClient *elastic.Client

初始化es client对象

// InitEs 初始化es
func InitEs() {
	esConn := fmt.Sprintf("http://%s:%s", "localhost", "9200")
	cfg := elastic.Config{
		Addresses: []string{esConn},
	}
	client, err := elastic.NewClient(cfg)
	if err != nil {
		log.Panic(err)
	}
	EsClient = client
}

2.2 初始化hook

我们这里用的hook是 github.com/CocaineCong/eslogrus 这是我之前封装的一个 logrus 到 es 的小hook。

func EsHookLog() *eslogrus.ElasticHook {
	hook, err := eslogrus.NewElasticHook(EsClient, "localhost", logrus.DebugLevel, "index")
	if err != nil {
		log.Panic(err)
	}
	return hook
}

调用 eslogrus 包中的 NewElasticHook 方法，传入 es client 对象，es的host，以及日志的输出等级和日志所存储的index。

然后在创建logrus的init函数中，也就是InitLog函数的最后，加入hook即可

hook := es.EsHookLog()
logger.AddHook(hook)

2.3 效果展示

先使用docker启动es，新建docker-compose.yml文件

version: '3.7'

services:
  elasticsearch:
    image: elasticsearch:8.4.2
    container_name: elasticsearch
    environment:
      bootstrap.memory_lock: "true"
      ES_JAVA_OPTS: "-Xms512m -Xmx512m"
      discovery.type: single-node
      ingest.geoip.downloader.enabled: "false"
      TZ: Asia/Shanghai
      xpack.security.enabled: "false"
    healthcheck:
      test: ["CMD-SHELL", "curl -sf http://localhost:9200/_cluster/health || exit 1"] #⼼跳检测，成功之后不再执⾏后⾯的退出
      interval: 60s #⼼跳检测间隔周期
      timeout: 10s
      retries: 3
      start_period: 60s #⾸次检测延迟时间
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - /usr/local/elasticsearch/data:/usr/local/elasticsearch/data
      - /usr/local/elasticsearch/config/es/config:/usr/local/elasticsearch/config
    ports:
      - "9200:9200"
    restart: always

根目录下执行

docker-compose -f docker-compose.yml up -d

结果

看到这个 status 是 up 的状态即可。

然后我们浏览器访问 http://localhost:9200/ 出现以下信息即可

然后我们再执行一次即可，注意我们的初始化一定要先初始化ES再初始化log。因为log里面用到了es。

这也提到了go中的init函数，在实际开发中，我们不要使用init函数，因为这个init函数是不可控的，不可控意味着这个顺序是不可控的，最好自己手动init。

func TestLogrus(t *testing.T) {
	InitEs()
	InitLog()
	LogrusObj.Infoln("测试日志文件")
}

结果如下

验证一下

curl --location 'http://localhost:9200/index/_search'\?pretty

我们请求一下这个es的查询接口，以及把我们的日志信息放进去了。 index 就是我们的索引，_search就是搜索功能，pretty只是让展示好看一点而已。

下面我们来解析一下eslogrus包中所做的东西，看看我们的log是如何hook到es的！

3. eslogrus

接下来，我们来讲一下eslogrus包是如何 hook 到 es 的

定义一个ES的hook

type ElasticHook struct {
	client    *elastic.Client    // es的客户端
	host      string             // es 的 host
	index     IndexNameFunc      // 获取索引的名字
	levels    []logrus.Level     // 日志的级别 info，error
	ctx       context.Context    // 上下文
	ctxCancel context.CancelFunc // 上下文cancel的函数，
	fireFunc  fireFunc           // 需要实现这个
}

定义发送到es的一个消息结构体

// 发送到es的信息结构
type message struct {
	Host      string
	Timestamp string `json:"@timestamp"`
	Message   string
	Data      logrus.Fields
	Level     string
}

新建es hook对象的函数

// NewElasticHook 新建一个es hook对象
func NewElasticHook(client *elastic.Client, host string, level logrus.Level, index string) (*ElasticHook, error) {
	return NewElasticHookWithFunc(client, host, level, func() string { return index })
}

func NewElasticHookWithFunc(client *elastic.Client, host string, level logrus.Level, indexFunc IndexNameFunc) (*ElasticHook, error) {
	return newHookFuncAndFireFunc(client, host, level, indexFunc, syncFireFunc)
}

新建一个hook

func newHookFuncAndFireFunc(client *elastic.Client, host string, level logrus.Level, indexFunc IndexNameFunc, fireFunc fireFunc) (*ElasticHook, error) {
	var levels []logrus.Level
	for _, l := range []logrus.Level{
		logrus.PanicLevel,
		logrus.FatalLevel,
		logrus.ErrorLevel,
		logrus.WarnLevel,
		logrus.InfoLevel,
		logrus.DebugLevel,
	} {
		if l <= level {
			levels = append(levels, l)
		}
	}

	ctx, cancel := context.WithCancel(context.TODO())

	return &ElasticHook{
		client:    client,
		host:      host,
		index:     indexFunc,
		levels:    levels,
		ctx:       ctx,
		ctxCancel: cancel,
		fireFunc:  fireFunc,
	}, nil
}

创建消息

// createMessage 创建信息
func createMessage(entry *logrus.Entry, hook *ElasticHook) *message {
	level := entry.Level.String()

	if e, ok := entry.Data[logrus.ErrorKey]; ok && e != nil {
		if err, ok := e.(error); ok {
			entry.Data[logrus.ErrorKey] = err.Error()
		}
	}

	return &message{
		hook.host,
		entry.Time.UTC().Format(time.RFC3339Nano),
		entry.Message,
		entry.Data,
		strings.ToUpper(level),
	}
}

异步发送到es

func syncFireFunc(entry *logrus.Entry, hook *ElasticHook) error {
	data, err := json.Marshal(createMessage(entry, hook))

	req := esapi.IndexRequest{
		Index:   hook.index(),
		Body:    bytes.NewReader(data),
		Refresh: "true",
	}

	res, err := req.Do(hook.ctx, hook.client)
	if err != nil {
		log.Fatalf("Error getting response: %s", err)
	}
	var r map[string]interface{}
	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
		log.Printf("Error parsing the response body: %s", err)
	} else {
		// Print the response status and indexed document version.
		log.Printf("[%s] %s; version=%d", res.Status(), r["result"], int(r["_version"].(float64)))
	}
	return err
}

这两个是必须要实现的

// Fire 实现 logrus hook 必须要的接口函数
func (hook *ElasticHook) Fire(entry *logrus.Entry) error {
	return hook.fireFunc(entry, hook)
}

// Levels 实现 logrus hook 必须要的接口函数
func (hook *ElasticHook) Levels() []logrus.Level {
	return hook.levels
}

这篇好文章是转载于：学新通技术网