go语言并发实战——日志收集系统(九) 基于etcd的代码重构思考与初步实现

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: go语言并发实战——日志收集系统(九) 基于etcd的代码重构思考与初步实现

前言

在之前我们j基于sarama,tail还有go-ini实现了日志收集系统客户端的编写,但是我们梳理一下可以发现,该客户端还存在一些问题:

  • 客户端一次只能读取一个日志文件,无法同时读取多个分区
  • 无法管理日志存放的分区(topic)
    那我们一个如何去解决这个问题呢?在前两篇文章中我们介绍了etcd,它通过可以存储键值对并且通过watch操作来实现对键值对的实时监控,那我们能不能尝试用`etcd``来储存日志文件信息与对应分区信息?这就是我们今天这篇文章所探究的主题.

初步实现的流程

存储数据格式

这里为了存储数据方便,我们利用json格式来存储数据,示例如下:

[
    {
        "path": "G:/goproject/-goroutine-/log-agent/log/log1",
        "topic": "web.log"
    },
    {
        "path": "G:/goproject/-goroutine-/log-agent/log/log2",
        "topic": "s4.log"
    }
]

etcd初始化的编写

在之前有段etcd的博文中,我们已经介绍过etcd的基本使用,这里不做赘述,首先我们在log-agent文件夹下创建etcd文件夹,创建etcd.go文件,编写etcd的初始化:

func Init(address []string) (err error) {
  client, err = clientv3.New(clientv3.Config{
    Endpoints:   address,
    DialTimeout: 5 * time.Second,
  })
  if err != nil {
    logrus.Error("etcd client connect failed,err:%v", err)
    return
  }
  return
}

然后 在main.go文件中调用:

err = etcd.Init(ConfigObj.Etcdaddress.Addr)
  if err != nil {
    logrus.Error("InitEtcd failed, err:%v", err)
    return
  }
  logrus.Infof("InitEtcd success")

通过etcd拉取要收集文件的配置项

在初始化etcd后,我们就要通过etcd来拉取要收集文件的配置项了,首先定义一个结构体来接收信息:

type collectEntry struct {
  Path  string `json:"path"`
  Topic string `json:"topic"`
}

然后创建拉取配置项的函数:

func GetConf(key string) (err error, collectEntryList []collectEntry) {
  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  defer cancel()
  response, err := client.Get(ctx, key)
  if err != nil {
    logrus.Error("get conf from etcd failed,err:%v", err)
    return
  }
  if len(response.Kvs) == 0 {
    logrus.Warningf("get len:0 conf from etcd failed,err:%v", err)
    return
  }
  fmt.Println(response.Kvs[0].Value)                             //此时还是json字符串
  err = json.Unmarshal(response.Kvs[0].Value, &collectEntryList) //把值反序列化到collectEntryList
  if err != nil {
    logrus.Error("json unmarshal failed,err:%v", err)
    return
  }
  return
}

然后在main.go里面调用一下就可以了:

//拉取要收集日志文件的配置项
  err, collectEntryList := etcd.GetConf(ConfigObj.Etcdaddress.Key)
  if err != nil {
    logrus.Error("GetConf failed, err:%v", err)
    return
  }
  fmt.Println(collectEntryList)

尝试用之前的demo设置一下配置文件中的key对应的value:

package main
import (
  "context"
  "fmt"
  clientv3 "go.etcd.io/etcd/client/v3"
  "time"
)
func main() {
  cli, err := clientv3.New(clientv3.Config{
    Endpoints:   []string{"127.0.0.1:2379"}, //服务端通信端口
    DialTimeout: 5 * time.Second,            //连接超时时间
  })
  //put
  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  str := "[{\"path\":\"G:/goproject/-goroutine-/log-agent/log/log1\",\"topic\":\"web.log\"},{\"path\":\"G:/goproject/-goroutine-/log-agent/log/log2\",\"topic\":\"s4.log\"}]"
  _, err = cli.Put(ctx, "collect_log_conf", str)
  cancel()
  if err != nil {
    fmt.Println("put failed,err:%v", err)
    return
  }
}

运行就可以看到我们接收到了配置项了:

涉及改动处的源代码

  • 配置文件
[kafka]
address=127.0.0.1:9092
topic=test1.log
chan_size=100000
[etcd]
address=127.0.0.1:2379
collect_key=collect_log_conf
[collect]
logfile_path:G:\goproject\-goroutine-\log-agent\log\log1
  • main.go
package main
import (
  "fmt"
  "github.com/Shopify/sarama"
  "github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
  "github.com/go-ini/ini"
  "log-agent/Kafka"
  "log-agent/etcd"
  "log-agent/tailFile"
  "strings"
  "time"
)
type Config struct {
  Kafakaddress Kafkaddress `ini:"kafka"`
  LogFilePath  LogFilePath `ini:"collect"`
  Etcdaddress  EtcdAddress `ini:"etcd"`
}
type Kafkaddress struct {
  Addr        []string `ini:"address"`
  Topic       string   `ini:"topic"`
  MessageSize int64    `ini:"chan_size"`
}
type LogFilePath struct {
  Path string `ini:"logfile_path"`
}
type EtcdAddress struct {
  Addr []string `ini:"address"`
  Key  string   `ini:"collect_key"`
}
func run(config *Config) (err error) {
  for {
    line, ok := <-tailFile.TailObj.Lines
    if !ok {
      logrus.Error("read from tail failed,err:", err)
      time.Sleep(2 * time.Second)
      continue
    }
    if len(strings.Trim(line.Text, "\r")) == 0 {
      continue
    }
    msg := &sarama.ProducerMessage{}
    msg.Topic = config.Kafakaddress.Topic
    msg.Value = sarama.StringEncoder(line.Text)
    Kafka.MesChan(msg)
  }
}
func main() {
  //读取配置文件,获取配置信息
  filename := "G:\\goproject\\-goroutine-\\log-agent\\conf\\config.ini"
  ConfigObj := new(Config)
  err := ini.MapTo(ConfigObj, filename)
  if err != nil {
    logrus.Error("%s Load failed,err:", filename, err)
  }
  //初始化Kafka
  err = Kafka.InitKafka(ConfigObj.Kafakaddress.Addr, ConfigObj.Kafakaddress.MessageSize)
  if err != nil {
    logrus.Error("InitKafka failed, err:%v", err)
    return
  }
  logrus.Infof("InitKafka success")
  //初始化etcd
  err = etcd.Init(ConfigObj.Etcdaddress.Addr)
  if err != nil {
    logrus.Error("InitEtcd failed, err:%v", err)
    return
  }
  logrus.Infof("InitEtcd success")
  //拉取要收集日志文件的配置项
  err, collectEntryList := etcd.GetConf(ConfigObj.Etcdaddress.Key)
  if err != nil {
    logrus.Error("GetConf failed, err:%v", err)
    return
  }
  fmt.Println(collectEntryList)
  //初始化tail
  err = tailFile.InitTail(ConfigObj.LogFilePath.Path)
  if err != nil {
    logrus.Error("InitTail failed, err:%v", err)
    return
  }
  logrus.Infof("InitTail success")
  //利用sarama报发送消息到Kafka中
  err = run(ConfigObj)
  if err != nil {
    logrus.Error("run failed, err:%v", err)
    return
  }
}
  • etcd.go
package etcd
import (
  "encoding/json"
  "fmt"
  "github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
  clientv3 "go.etcd.io/etcd/client/v3"
  "golang.org/x/net/context"
  "time"
)
var client *clientv3.Client
type collectEntry struct {
  Path  string `json:"path"`
  Topic string `json:"topic"`
}
func Init(address []string) (err error) {
  client, err = clientv3.New(clientv3.Config{
    Endpoints:   address,
    DialTimeout: 5 * time.Second,
  })
  if err != nil {
    logrus.Error("etcd client connect failed,err:%v", err)
    return
  }
  return
}
func GetConf(key string) (err error, collectEntryList []collectEntry) {
  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
  response, err := client.Get(ctx, key)
  cancel()
  if err != nil {
    logrus.Error("get conf from etcd failed,err:%v", err)
    return
  }
  if len(response.Kvs) == 0 {
    logrus.Warningf("get len:0 conf from etcd failed,err:%v", err)
    return
  }
  fmt.Println(response.Kvs[0].Value)                             //此时还是json字符串
  err = json.Unmarshal(response.Kvs[0].Value, &collectEntryList) //把值反序列化到collectEntryList
  if err != nil {
    logrus.Error("json unmarshal failed,err:%v", err)
    return
  }
  return
}
相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
1天前
|
Ubuntu 编译器 Linux
go语言中SQLite3驱动安装
【11月更文挑战第2天】
16 7
|
1天前
|
关系型数据库 Go 网络安全
go语言中PostgreSQL驱动安装
【11月更文挑战第2天】
15 5
|
1天前
|
SQL 关系型数据库 MySQL
go语言数据库中mysql驱动安装
【11月更文挑战第2天】
13 4
|
1天前
|
存储 设计模式 安全
Go语言中的并发编程:从入门到精通###
本文深入探讨了Go语言中并发编程的核心概念与实践技巧,旨在帮助读者从理论到实战全面掌握Go的并发机制。不同于传统的技术文章摘要,本部分将通过一系列生动的案例和代码示例,直观展示Go语言如何优雅地处理并发任务,提升程序性能与响应速度。无论你是Go语言初学者还是有一定经验的开发者,都能在本文中找到实用的知识与灵感。 ###
|
1天前
|
安全 Go
用 Zap 轻松搞定 Go 语言中的结构化日志
在现代应用程序开发中,日志记录至关重要。Go 语言中有许多日志库,而 Zap 因其高性能和灵活性脱颖而出。本文详细介绍如何在 Go 项目中使用 Zap 进行结构化日志记录,并展示如何定制日志输出,满足生产环境需求。通过基础示例、SugaredLogger 的便捷使用以及自定义日志配置,帮助你在实际开发中高效管理日志。
9 1
|
Go
Go实战(一)-概述
Go实战(一)-概述
112 0
Go实战(一)-概述
|
2天前
|
JavaScript Java Go
探索Go语言在微服务架构中的优势
在微服务架构的浪潮中,Go语言以其简洁、高效和并发处理能力脱颖而出。本文将深入探讨Go语言在构建微服务时的性能优势,包括其在内存管理、网络编程、并发模型以及工具链支持方面的特点。通过对比其他流行语言,我们将揭示Go语言如何成为微服务架构中的一股清流。
|
7天前
|
网络协议 安全 Go
Go语言进行网络编程可以通过**使用TCP/IP协议栈、并发模型、HTTP协议等**方式
【10月更文挑战第28天】Go语言进行网络编程可以通过**使用TCP/IP协议栈、并发模型、HTTP协议等**方式
33 13
|
2天前
|
SQL 关系型数据库 MySQL
go语言中安装数据库驱动
【11月更文挑战第1天】
16 5
|
2天前
|
编译器 Go 开发者
go语言中导入相关包
【11月更文挑战第1天】
11 3