一个将日志收集发送到kafka的程序

kafka/kafka.go

@@ -54,7 +54,7 @@ func sendToKafka() {
 			if err != nil {
 				fmt.Println("send message err:", err)
 			}
-			fmt.Printf("pid:%v offset:%v\n", pid, offset)
+			fmt.Printf("pid:%v offset:%v text:%v\n", pid, offset, msg.Value)
 		default:
 			time.Sleep(time.Millisecond * 50)
 		}

put_etcd/etcd_putvalue2.go

@@ -20,7 +20,7 @@ func main() {
 	// put
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	value := `[{"path":"c:/tmp/nginx.log","topic":"web_log"},{"path":"d:/xxx/redis.log","topic":"redis_log"}]`
-	_, err = cli.Put(ctx, "/logagent/collect_log_config", value)
+	_, err = cli.Put(ctx, "/logagent/192.168.0.104/collect_log_config", value)
 	cancel()
 	if err != nil {
 		fmt.Println("put to etcd failed,error:", err)

put_etcd/etcd_putvalue3.go

@@ -20,7 +20,7 @@ func main() {
 	// put
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	value := `[{"path":"c:/tmp/nginx.log","topic":"web_log"},{"path":"d:/xxx/redis.log","topic":"redis_log"},{"path":"d:/xxx/mysql.log","topic":"mysql_log"}]`
-	_, err = cli.Put(ctx, "/logagent/collect_log_config", value)
+	_, err = cli.Put(ctx, "/logagent/192.168.0.104/collect_log_config", value)
 	cancel()
 	if err != nil {
 		fmt.Println("put to etcd failed,error:", err)

taillog/taillog.go

@@ -61,7 +61,7 @@ func (t *TailTask) run() {
 			fmt.Printf("tail task:%v exit\n", t.path+t.topic)
 			return
 		case line := <-t.ReadChan():
-			fmt.Printf("日志已发送kafka，text：,%v\n", line.Text)
+			fmt.Printf("日志已发送channel，text：%v\n", line.Text)
 			//kafka.SendToKafka(t.topic, line.Text) //函数调用函数
 			//先把日志发送到一个通道中
 			kafka.SendToChan(t.topic, line.Text)