1 Star 0 Fork 0

zhangjungang/beats

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
filebeat.go 2.49 KB
一键复制 编辑 原始数据 按行查看 历史
urso 提交于 2016-01-20 15:27 . Filebeat async publisher support
package beat
import (
"fmt"
"os"
"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/cfgfile"
"github.com/elastic/beats/libbeat/logp"
cfg "github.com/elastic/beats/filebeat/config"
. "github.com/elastic/beats/filebeat/crawler"
. "github.com/elastic/beats/filebeat/input"
)
// Beater object. Contains all objects needed to run the beat
type Filebeat struct {
FbConfig *cfg.Config
// Channel from harvesters to spooler
publisherChan chan []*FileEvent
Spooler *Spooler
registrar *Registrar
}
func New() *Filebeat {
return &Filebeat{}
}
// Config setups up the filebeat configuration by fetch all additional config files
func (fb *Filebeat) Config(b *beat.Beat) error {
// Load Base config
err := cfgfile.Read(&fb.FbConfig, "")
if err != nil {
return fmt.Errorf("Error reading config file: %v", err)
}
// Check if optional config_dir is set to fetch additional prospector config files
fb.FbConfig.FetchConfigs()
return nil
}
func (fb *Filebeat) Setup(b *beat.Beat) error {
return nil
}
func (fb *Filebeat) Run(b *beat.Beat) error {
defer func() {
p := recover()
if p == nil {
return
}
fmt.Printf("recovered panic: %v", p)
os.Exit(1)
}()
var err error
// Init channels
fb.publisherChan = make(chan []*FileEvent, 1)
// Setup registrar to persist state
fb.registrar, err = NewRegistrar(fb.FbConfig.Filebeat.RegistryFile)
if err != nil {
logp.Err("Could not init registrar: %v", err)
return err
}
crawl := &Crawler{
Registrar: fb.registrar,
}
// Load the previous log file locations now, for use in prospector
fb.registrar.LoadState()
// Init and Start spooler: Harvesters dump events into the spooler.
fb.Spooler = NewSpooler(fb)
err = fb.Spooler.Config()
if err != nil {
logp.Err("Could not init spooler: %v", err)
return err
}
// Start up spooler
go fb.Spooler.Run()
crawl.Start(fb.FbConfig.Filebeat.Prospectors, fb.Spooler.Channel)
// Publishes event to output
pub := newPublisher(fb.FbConfig.Filebeat.PublishAsync,
fb.publisherChan, fb.registrar.Channel, b.Events)
pub.Start()
// registrar records last acknowledged positions in all files.
fb.registrar.Run()
return nil
}
func (fb *Filebeat) Cleanup(b *beat.Beat) error {
return nil
}
// Stop is called on exit for cleanup
func (fb *Filebeat) Stop() {
// Stop harvesters
// Stop prospectors
// Stopping spooler will flush items
fb.Spooler.Stop()
// Stopping registrar will write last state
fb.registrar.Stop()
// Close channels
//close(fb.publisherChan)
}
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/zhangjungang/beats.git
git@gitee.com:zhangjungang/beats.git
zhangjungang
beats
beats
v1.3.1

搜索帮助