filebeat

README

Easy filebeat

简单的filebeat逻辑实现

harvester

核心逻辑，主要功能点如下

reader

sink

sys

copy from filebeat 项目，主要是获取文件的 I-Node 信息，用来判断文件是不是同一个文件（不受 mv 以及 cp 的影响）

Documentation

Index

• Variables
• type Config
• type Harvester
  • func NewHarvester(cfg Config) (Harvester, error)
• type LineReader
  • func (line *LineReader) Close() error
  • func (line *LineReader) CurFile() *os.File
  • func (line *LineReader) Next() (string, error)
  • func (line *LineReader) Offset() int64
• type Metadata
• type Reader
  • func NewLineReader(name string, offset *int64) (Reader, error)
• type Sink
• type StateOS
  • func GetOSState(info os.FileInfo) StateOS
  • func (fs StateOS) String() string

Variables

var (

	// ErrorRename 文件被重命名错误
	ErrorRename error = errors.New("log already rename")

	// ErrorRemoved 文件被移走错误
	ErrorRemoved error = errors.New("log already removed")

	// ErrorEmpty 空数据错误
	ErrorEmpty error = errors.New("no content")

	// ErrorClosed Reader 已经被关闭了
	ErrorClosed error = errors.New("reader already closed")
)

var (
	EmptyWaitFiles error = errors.New("empty wait files")
)

Types

type Config

type Config struct {
	// Path 监听的文件路径
	// 支持完全匹配以及正则匹配(监听对应规则的文件)
	Path string
	// MetaPath 元数据保存的位置
	MetaPath string
	// Logger 日志输出
	Logger *logrus.Logger
}

Config easy-filebeat 的配置信息

type Harvester

type Harvester interface {
	io.Closer
	// Init 初始化 Harvester
	Init() error
	// RegisterSink 注册一个处理文件的 Sink 处理者
	RegisterSink(sink Sink)
	// Run 执行监听逻辑
	Run(ctx context.Context)
	// OnError 出现异常时的回掉
	OnError(err error)
}

Harvester 监听文件变动

func NewHarvester

func NewHarvester(cfg Config) (Harvester, error)

NewHarvester 创建一个 Harvester 实例

type LineReader

type LineReader struct {
	// contains filtered or unexported fields
}

LineReader 按行读取的 line-reader 实现

func (*LineReader) Close

func (line *LineReader) Close() error

Close

func (*LineReader) CurFile

func (line *LineReader) CurFile() *os.File

CurFile

func (*LineReader) Next

func (line *LineReader) Next() (string, error)

Next

func (*LineReader) Offset

func (line *LineReader) Offset() int64

Offset

type Metadata

type Metadata struct {
	// CurFile 正在处理的文件
	CurFile string
	// CurFileINode 当前处理文件的 INode 信息
	CurFileINode string
	// CurOffset 正在处理文件的当前读取的位点信息
	CurOffset int64
	// PreFileINode 上一个被处理完的文件的 INode 信息
	PreFileINode string
}

Metadata 记录文件处理信息数据

type Reader

type Reader interface {
	io.Closer
	// Offset
	//  @return int64
	Offset() int64
	// CurFile
	//  @return *os.File
	CurFile() *os.File
	// Next
	//  @return Message
	//  @return error
	Next() (string, error)
}

Reader is the interface that wraps the basic Next method for getting a new message. Next returns the message being read or and error. EOF is returned if reader will not return any new message on subsequent calls.

func NewLineReader

func NewLineReader(name string, offset *int64) (Reader, error)

NewLineReader 构造一个 Reader

type Sink

type Sink interface {
	// OnMessage
	OnMessage(msg string)
}

Sink handle log each line

type StateOS

type StateOS struct {
	Inode  uint64 `json:"inode," struct:"inode"`
	Device uint64 `json:"device," struct:"device"`
}

StateOS 文件的 state 信息

func GetOSState

func GetOSState(info os.FileInfo) StateOS

GetOSState returns the FileStateOS for non windows systems

func (StateOS) String

func (fs StateOS) String() string