目標(biāo)

這里我們實(shí)現(xiàn)一個(gè)簡(jiǎn)單的demo工程，展示auditbeat如何實(shí)現(xiàn)一個(gè)module，并把數(shù)據(jù)發(fā)送到ElasticSearch，工程源碼可以在Github上下載

• 創(chuàng)建一個(gè)auditbeat Module
• auditbeat Module周期性向ElasticSearch發(fā)送含有Host Name的心跳數(shù)據(jù)包

Demo工程基于elastic-beats 8.3.3版本，早期版本原理類似

概念

• Module

  Module就是auditbeat中具體實(shí)現(xiàn)一個(gè)功能的模塊，auditbeat依賴metricbeat和libbeat提供的API，可以很快的開(kāi)發(fā)一個(gè)模塊去采集數(shù)據(jù)并存儲(chǔ)到ElasticSearch中

• MetricSet

  MetricSet是auditbeat的基本單位，一個(gè)MetricSet泛指收集的一種數(shù)據(jù)類型，一個(gè)auditbeatModule可以包含多個(gè)MetricSet，有些簡(jiǎn)單的Module只含有一個(gè)MetricSet。

  例如，auditbeat的systemModule中包含有login, user, socket, process等MetricSet, auditdModule中只有一個(gè)MetricSet

• Fields

  Fields是一個(gè)MetricSet產(chǎn)生的數(shù)據(jù)結(jié)構(gòu)映射，auditbeat會(huì)將Fields通過(guò)auditbeat setup命令將Fields導(dǎo)入到index template中，以便在Kibana中展示

• Event

  Event是auditbeat中產(chǎn)生的包含數(shù)據(jù)的事件，由MetricSet產(chǎn)生，需要符合聲明的Fields映射規(guī)范，并最終存儲(chǔ)到ElasticSearch中

創(chuàng)建Module

1. 初始化Demo Module目錄結(jié)構(gòu)在路徑auditbeat/module/zack_module

    auditbeat/module/zack_module/
    ├── heartbeat
    │   ├── event.go
    │   ├── _meta
    │   │   └── fields.yml
    │   └── metricset.go
    ├── _meta
    │   ├── data.json
    │   └── fields.yml
    ├── zack.go
    └── zack_metric_set.go
   

   _meta路徑很重要，一定要?jiǎng)?chuàng)建，開(kāi)發(fā)elastic-beats的很多工具會(huì)依賴這個(gè)路徑

2. 在auditbeat/module/zack_module/zack.go文件，聲明Module如下

   package zack_module
   
    import (
        "github.com/elastic/beats/v7/metricbeat/mb"
        "github.com/elastic/elastic-agent-libs/logp"
        "github.com/elastic/go-sysinfo"
    )
   
    const moduleName = "zack"
   
    func init() {
        if err := mb.Registry.AddModule(moduleName, NewModule); err != nil {
            panic(err)
        }
    }
   
    type ZackModule struct {
        mb.BaseModule
        hostName string
    }
   
    func NewModule(base mb.BaseModule) (mb.Module, error) {
        log := logp.NewLogger(moduleName)
        var hostName string
        if hostInfo, err := sysinfo.Host(); err != nil {
            log.Errorf("Could not get host info. err=%+v", err)
        } else {
            hostName = hostInfo.Info().Hostname
        }
   
        if hostName == "" {
            log.Warnf("Could not get host name")
        }
   
        return &ZackModule{
            BaseModule: base,
            hostName:   hostName,
        }, nil
     
     
    }
   

   1. auditbeat中要在init方法中將Module通過(guò)mb.Registry.AddModule方法注冊(cè)給metricbeat
   2. Module實(shí)現(xiàn)都要繼承mb.BaseModule

創(chuàng)建MetricSet

1. 文件auditbeat/module/zack_module/zack_metric_set.go中聲明MetricSet

   package zack_module
   
    import "github.com/elastic/beats/v7/metricbeat/mb"
   
    type ZackMetricSet struct {
        mb.BaseMetricSet
        module *ZackModule
    }
   
    func NewZackMetricSet(base mb.BaseMetricSet) ZackMetricSet {
        return ZackMetricSet{
            BaseMetricSet: base,
            module:        base.Module().(*ZackModule),
        }
    }
   
    func (ms *ZackMetricSet) Hostname() string {
        return ms.module.hostName
    }
   
   

   auditbeat的MetricSet實(shí)現(xiàn)需要繼承mb.BaseMetricSet

2. 聲明Event在文件auditbeat/module/zack_module/heartbeat/event.go

   package heartbeat
   
    import (
        "github.com/elastic/beats/v7/metricbeat/mb"
        "github.com/elastic/elastic-agent-libs/mapstr"
    )
   
    type HeartBeatEvent struct {
        HostName string
    }
   
    func (e *HeartBeatEvent) buildHeartBeatEvent() mb.Event {
        data := mapstr.M{
            "host_name": e.HostName,
        }
        return mb.Event{
            RootFields: mapstr.M{
                "heartbeat": data,
            },
        }
    }
   

3. 在MetricSet auditbeat/module/zack_module/heartbeat/metricset.go路徑中，添加MetricSet實(shí)現(xiàn)

   package heartbeat
   
    import (
        "github.com/elastic/beats/v7/auditbeat/module/zack_module"
        "github.com/elastic/beats/v7/metricbeat/mb"
        "github.com/elastic/beats/v7/metricbeat/mb/parse"
        "github.com/elastic/elastic-agent-libs/logp"
    )
   
    const (
        moduleName    = "zack"
        metricSetName = "heartbeat"
        namespace     = "."
    )
   
    type MetricSet struct {
        zack_module.ZackMetricSet
        // We can add config or other attibutes implementation here for HeartBeat MetricBeat
        log *logp.Logger
    }
   
    func init() {
        mb.Registry.MustAddMetricSet(moduleName, metricSetName, New,
            mb.DefaultMetricSet(), mb.WithHostParser(parse.EmptyHostParser), mb.WithNamespace(namespace))
    }
   
    func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
        return &MetricSet{
            ZackMetricSet: zack_module.NewZackMetricSet(base),
            log:           logp.NewLogger(metricSetName),
        }, nil
    }
   
    func (ms *MetricSet) Fetch(reporter mb.ReporterV2) {
        event := HeartBeatEvent{
            HostName: ms.ZackMetricSet.Hostname(),
        }
        ms.log.Info("sending heartbeat event")
        reporter.Event(event.buildHeartBeatEvent())
    }
   
   

   這里我們實(shí)現(xiàn)了Fetch接口來(lái)周期性獲取數(shù)據(jù)，默認(rèn)internal為10s，我們還可以使用Run接口來(lái)一次性觸發(fā)任務(wù)

注冊(cè)Zack Module到auditbeat中

1. 通過(guò)工具/dev-tools/cmd/module_include_list/module_include_list.go注冊(cè)我們開(kāi)發(fā)的Module和MetricSet，VSCode launch.json示例如下

   {
        "name": "[Dev] Module Include",
        "type": "go",
        "request": "launch",
        "mode": "auto",
        "cwd": "${workspaceFolder}",
        "program": "${workspaceFolder}/dev-tools/cmd/module_include_list/module_include_list.go",
        "console": "integratedTerminal",
        "args": [
            "-moduleDir=auditbeat/module",
            "-out=auditbeat/include/list.go"
        ]
    }
   

   運(yùn)行后查看auditbeat/include/list.go是否引入了我們最新開(kāi)發(fā)的Module和MetricSet
   如果沒(méi)有引入，請(qǐng)檢查Module和MetricSet路徑是否包含_meta路徑

2. 編輯fields.yml，聲明heatbeat.host_name映射為keyword

   - key: zack
    title: "Zack"
    decription: >
        These are fields generated by the insight module.
    fields:
        - name: heartbeat
        type: group
        fields:
            - name: host_name
            type: keyword
   
   

3. 通過(guò)工具/dev-tools/cmd/asset/asset.go生成Fields映射，生成auditbeat/module/zack_module/fields.go文件，VSCode launch.json示例如下：

   {
        "name": "[Dev] Generate Fields",
        "type": "go",
        "request": "launch",
        "mode": "auto",
        "cwd": "${workspaceFolder}",
        "program": "${workspaceFolder}/dev-tools/cmd/asset/asset.go",
        "console": "integratedTerminal",
        "args": [
            "-in=auditbeat/module/zack_module/_meta/fields.yml",
            "-out=auditbeat/module/zack_module/fields.go",
            "-name=zack",
            "-priority=asset.ModuleFieldsPri",
            "-pkg=zack_module",
            "auditbeat"
        ]
    }
   

4. 啟用Module和MetriSet，編輯auditbeat.yml如下：

   auditbeat.modules:
    - module: zack
    metricsets:
        - heartbeat
   

驗(yàn)證

1. 通過(guò)auditbeat setup命令更新index template，VSCode launch.json如下

   {
        "name": "AuditBeat Setup",
        "type": "go",
        "request": "launch",
        "mode": "auto",
        "program": "${workspaceFolder}/auditbeat/main.go",
        "cwd": "${workspaceFolder}",
        "console": "integratedTerminal",
        "asRoot": true,
        "args": [
            "setup",
            "-c",
            "path to auditbeat.yml",
            "-e"
        ]
    }
   

2. Kinana上驗(yàn)證Fields成功導(dǎo)入如下

   
   
   index_template.png

3. 啟動(dòng)auditbeat，VSCode launch.json如下

   {
       "name": "AuditBeat",
       "type": "go",
       "request": "launch",
       "mode": "auto",
       "program": "${workspaceFolder}/auditbeat/main.go",
       "console": "integratedTerminal",
       "asRoot": true,
       "args": [
           "-c",
           "<path to auditbeat.yml>",
           "-e"
       ]
   }
   

4. 驗(yàn)證Module數(shù)據(jù)成功導(dǎo)入ElaticSearch

   
   
   data.png