1. 系统介绍

日志收集服务

golang

golnag/lua扩展

流水线作业

2. UParse系统说明

2.1系统结构

2.1 主要模块

pipeline-config

核心调度模块，跟据配置文件启动相应的插件，启动流水线

input

从数据源读取数据，包括kafka_input，TCP_input等

decoder

将数据解析成特定，有json，nginx等

filter

对数据进行聚合，计算，过滤

encoder

对数据重新编码成特定格式

output

输出数据到外部存储或通道

router

根据数据某些字段对filter和output进行路由

2.3 主要特性

2.3.1系统数据传输单元

typePipelinePack struct{

message.Message //处理后的数据，一个类似于hanshmap的多级嵌套数据结构

MsgByte //原始数据， []byte

}

系统中日志数据以PipelinePack数据来在流水线作业之间传递

2.3.2 系统初始化加载方式

启动模式

uparse -config   xxx.toml

uparse -config  config_dir

组件注册

系统初始化时，注册所有的插件到系统(input、output、filter、encoder、decoder)，注册形式为 key：FactoryFunc

插件启动和调度

主goroutine根据配置文件管理类PipelineConfig启动各个插件

首先启动outout插件，然后是filter插件，然后启动router，为filter和output插件建立消息传递通道

2.3.3 插件运行模式

Input、Filter、Output都运行在自己的goroutine，保持高的并发性

各个组件之间使用channel或者文件buffer通信

插件的生命周期由PluginRunners管理

插件与goroutine的紧密耦合保持了并发的完整

组件之间的关联

input与decoder的关系是，属于inout的一个组件，每个input只接收一种格式数据源；

input与filter之间通过router进行交互，多对多的对应关系；

filter与output之间也是使用router进行交互，多对多的关系；

encode和output之间也是从属关系, encode内嵌在output中，在数据输出之前按照encoder进行序列化

2.3.4 消息路由

消息decode完成后使用router将数据传输到相应的filter插件和output插件

filter和output需要指定message matcher

方式：根据某些字段规则进行路由，规则由使用yacc语法解析，支持一下规则

1：Type == “test” && Severity == 6

2：Fields[MyBool] == TRUE

3： Type =~ /test/

message matcher运行在自己的goroutine，很快

2.3.5 SANDBOX-lua

插件可以使用lua实现

lua加载模式为golang使用cgo调用lua脚本

lua可以动态加载，方便插件升级

lua属于性能比较快的脚本语言

lua的常用方式

2.3.6 消息路由缓冲队列

router提供了两种消息队列

golang内部channel机制(内存buffer，线程安全)

使用文件作为缓冲buffer(磁盘buffer)

2.3.7 插件扩展

对所有的插件扩展开发提供了标准规范

golang插件实现特定接口，然后注册到系统

type Input interface {

Run(ir InputRunner, h PluginHelper) (err error)

Stop()

}

type Decoder interface {

Decode(pack *PipelinePack) (packs []*PipelinePack, err error)

}

type Filter interface {

Prepare(fr FilterRunner, h PluginHelper) (err error)

ProcessMessage(pack *PipelinePack) (err error)

CleanUp()

}

type Encoder interface {

Encode(pack *PipelinePack) (output []byte, err error)

}

type Output interface {

Prepare(or OutputRunner, h PluginHelper) (err error)

ProcessMessage(pack *PipelinePack) (err error)

CleanUp()

}

lua插件需要实现特定函数，然后将lua文件放到系统lua脚本文件夹下即可进行加载

lua插件开发规范

3. 系统类图

4 问题