Golang实战之海量日志收集系统（二）收集应用程序日志到Kafka中

package main

import (
	"fmt"
	"time"

	"github.com/Shopify/sarama"
)

func main() {
	
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll         // 发送完数据需要leader和follow都确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner// 新选出⼀个partition
	config.Producer.Return.Successes = true                  // 成功交付的消息将在success channel返回

	// 连接kafka
	client, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config) 
	if err != nil {
		fmt.Println("producer close, err:", err)
		return
	}

	defer client.Close()
	for {
		// 构造⼀个消息
		msg := &sarama.ProducerMessage{}
		msg.Topic = "nginx_log"
		msg.Value = sarama.StringEncoder("this is a good test, my message is good")
		// 发送消息
		pid, offset, err := client.SendMessage(msg)
		if err != nil {
			fmt.Println("send message failed,", err)
			return
		}

		fmt.Printf("pid:%v offset:%vn", pid, offset)
		time.Sleep(10 * time.Millisecond)
	}
}

连接Kafka 成功,并且成功生产出数据并在本地消费

接下来我们还需要知道一个工具的使用tailf

tailf

我们的agent需要读日志目录下的日志文件，而日志文件是不停的增加并且切换文件的，所以我们就需要借助于tailf这个包来读文件，当然这里的tailf和linux里的tail -f命令虽然不同，但是效果是差不多的，都是为了获取日志文件新增加的内容。

而我们的客户端非常重要的一个地方就是要读日志文件并且将读到的日志文件推送到kafka

这里需要我们下载一个包：go get github.com/hpcloud/tail

我们通过下面一个例子对这个包进行一个基本的使用，更详细的api说明看：https://godoc.org/github.com/hpcloud/tail

package main

import (
	"fmt"
	"github.com/hpcloud/tail"
	"time"
)

// tail demo
func main() {
	fileName := "E:\Go\logagent\tailf\main\my.log"
	config := tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true}
	tails, err := tail.TailFile(fileName, config)
	if err != nil {
		fmt.Println("tail file failed, err:", err)
		return
	}
	var (
		msg *tail.Line
		ok  bool
	)
	for {
		msg, ok = <-tails.Lines
		if !ok {
			fmt.Printf("tail file close reopen, filename:%sn", tails.Filename)
			time.Sleep(time.Second)
			continue
		}
		fmt.Println("msg:", msg.Text)
	}
}

日志库的使用

这里是通过beego的日志库实现的，beego的日志库是可以单独拿出来用

package main

import (
	"fmt"
	"github.com/astaxie/beego/config"

)

func main() {
	conf, err := config.NewConfig("ini", "E:\Go\logagent\conf\logagent.conf")
	if err != nil {
		fmt.Println("new config failed, err:", err)
		return
	}

	port, err := conf.Int("server::port")
	if err != nil {
		fmt.Println("read server:port failed, err:", err)
		return
	}

	fmt.Println("Port:", port)
	log_level := conf.String("logs::log_level")
	if len(log_level) == 0 {
		log_level = "debug"
	}
	fmt.Println("log_level:", log_level)

	log_path := conf.String("logs::log_path")
	fmt.Println("log_path:", log_path)
}