log collect

.idea/.gitignore

@@ -0,0 +1,8 @@
+# 默认忽略的文件
+/shelf/
+/workspace.xml
+# 基于编辑器的 HTTP 客户端请求
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/logagent.iml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="WEB_MODULE" version="4">
+  <component name="Go" enabled="true" />
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/logagent.iml" filepath="$PROJECT_DIR$/.idea/logagent.iml" />
+    </modules>
+  </component>
+</project>

go.mod

@@ -0,0 +1,18 @@
+module logagent
+
+go 1.22
+
+require (
+	github.com/Shopify/sarama v1.19.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/eapache/go-resiliency v1.7.0 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
+	github.com/eapache/queue v1.1.0 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/hpcloud/tail v1.0.0 // indirect
+	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	golang.org/x/sys v0.22.0 // indirect
+	gopkg.in/fsnotify.v1 v1.4.7 // indirect
+	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
+)

go.sum

@@ -0,0 +1,24 @@
+github.com/Shopify/sarama v1.19.0 h1:9oksLxC6uxVPHPVYUmq6xhr1BOF/hHobWH2UzO67z1s=
+github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eapache/go-resiliency v1.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA=
+github.com/eapache/go-resiliency v1.7.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho=
+github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4ALJ04o5Qqpdz8XLIpNA3WM/iSIXqxtqo7UGVws=
+github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
+github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
+github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM=
+github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
+github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
+golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI=
+golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=

kafka/kafka.go

@@ -0,0 +1,43 @@
+package kafka
+
+import (
+	"fmt"
+	"github.com/Shopify/sarama"
+)
+
+//专门往kafka写日志的模块
+
+// 定义全局的连接
+var (
+	client sarama.SyncProducer //声明一个连接全局的kafka生产者client
+
+)
+
+// 初始化client
+func Init(address []string) (err error) {
+	config := sarama.NewConfig()
+	config.Producer.RequiredAcks = sarama.WaitForAll          //发送完包需要 leader和follower都确认
+	config.Producer.Partitioner = sarama.NewRandomPartitioner //新选出一个partition
+	config.Producer.Return.Successes = true
+	client, err = sarama.NewSyncProducer(address, config)
+	if err != nil {
+		fmt.Println("producer closed, err:", err)
+		return
+	}
+	fmt.Printf("连接kafka成功！\n")
+	return
+}
+
+func SendToKafka(topic string, data string) {
+	//构造一个消息
+	msg := &sarama.ProducerMessage{}
+	msg.Topic = topic
+	msg.Value = sarama.StringEncoder(data)
+
+	//发送kafka
+	pid, offset, err := client.SendMessage(msg)
+	if err != nil {
+		fmt.Println("send message err:", err)
+	}
+	fmt.Printf("pid:%v offset:%v\n", pid, offset)
+}

main.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"fmt"
+	"logagent/kafka"
+	"logagent/taillog"
+)
+
+//logagent的入口程序
+
+func run() {
+	//1.读取日志
+	//2.发送给kafka
+}
+
+func main() {
+	//1.初始化kafka连接
+	err := kafka.Init([]string{"152.136.226.203:9093"})
+	if err != nil {
+		fmt.Printf("init kafka failed,error:%v\n", err)
+		return
+	}
+	//2.打开日志文件准备收集日志
+	err = taillog.Init("./mylog.log")
+	if err != nil {
+		fmt.Printf("taillog init failed,error:%v\n", err)
+		return
+	}
+}

taillog/taillog.go

@@ -0,0 +1,27 @@
+package taillog
+
+import (
+	"fmt"
+	"github.com/hpcloud/tail"
+)
+
+var (
+	tailChan *tail.Tail
+)
+
+//专门从日志文件收集日志的模块
+
+func Init(fileName string) (err error) {
+	config := tail.Config{
+		Location:  &tail.SeekInfo{Offset: 0, Whence: 2}, //从文件的哪个地方开始读 filebeat记录了文件断点的位置
+		ReOpen:    true,                                 //重新打开，切分用
+		MustExist: false,                                //文件不存在不报错
+		Poll:      true,
+		Follow:    true, //是否跟随
+	}
+	tailChan, err = tail.TailFile(fileName, config)
+	if err != nil {
+		fmt.Printf("tail file error:%s\n", err)
+	}
+	return
+}