llogtail

type BlockingBuffer ¶

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

BlockingBuffer is a thread-safe buffer, with blocking api. It block itself when data is full

func NewBlockingBuffer(bufferSize int) *BlockingBuffer

func (b *BlockingBuffer) Close()

func (b *BlockingBuffer) Fetch() []byte

func (b *BlockingBuffer) IfFullThenWait()

func (b *BlockingBuffer) ReadLinesFrom(reader *os.File, lineSep string) (int, error)

type Event ¶

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

func (e Event) String() string

type FileMeta ¶

type FileMeta struct {
	Inode uint64   `josn:"inode"`
	Dev   uint64   `json:"dev"`
	Hash  [16]byte `json:"hash"`
	// contains filtered or unexported fields
}

type FileSink ¶

type FileSink struct {
	Sink
	// contains filtered or unexported fields
}

func (s *FileSink) Close() error

func (s *FileSink) Open(path string) error

func (s *FileSink) Push(fpath string, content []byte) error

type FileSinkConf ¶

type FileSinkConf struct {
	Dst string `json:"dst"`
}

type LogCollector ¶

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

func NewLogCollector() *LogCollector

func (lc *LogCollector) Close() error

func (lc *LogCollector) Init(conf LogConf) error

func (lc *LogCollector) Join()

func (lc *LogCollector) Run() error

type LogConf ¶

type LogConf struct {
	Dir     string           `json:"dir"`
	Pattern string           `json:"pattern"`
	LineSep string           `json:"lineSeperator"`
	Sink    SinkConf         `json:"sink"`
	Watcher LogWatcherOption `json:"watcher"`
}

func ReadLogCollectorConf(path string) (*LogConf, error)

type LogFileEvent ¶

type LogFileEvent uint64

func (e LogFileEvent) String() string

type LogMeta ¶

type LogMeta struct {
	Dir     string
	Pattern string

	LogInfo os.FileInfo
	// contains filtered or unexported fields
}

type LogOption ¶

type LogOption struct {
	Verbose bool
	Level   logging.Level
}

type LogReader ¶

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

type LogWatchOption ¶

type LogWatchOption struct {
	FilterInterval time.Duration
	PollerInterval time.Duration
}

type LogWatcher ¶

type LogWatcher struct {
	EventC chan *Event
	// contains filtered or unexported fields
}

LogWatcher watch log files based on fsnotify. It mainly consists of a event-watcher and a event transform pattern, transform fsnotify event into LogFileEvent. Basically, it listens inotify event of log files, transforming into LogFileEvent, and pass to upper collector to cope with. The frequency of sending a event is determined by filterInterval and event kind change. It manages two goroutines in event-driven and poller mode. Event-Driven goroutine handle event from fsnotify.Watcher while poller gorotine which is trggered much less frequently, handles operations needing resume after some time, such as removeing a file by mistake.

Commonly speaking, Log File is watched as fellows: Step 1, Register File And add into fsnotify watcher. File Pattern supports glob. Note that log file's path is settled, and any other operation related with log's path will only use the determined path, rather than reseatch files.And a Discover Event is generated for each determined log file. Step 2, Watcher Events. Event is passed to LogWatcher via a chan. LogWatcher will transform raw event into LogFileEvent based on some pattern(linux only at now) or File's status. Specially, when file is removed, we consider it a mistake and put path into poller quere so that it will be rewatcher when poller triggers. Step 3: Send Event.

func NewLogWatcher(option *LogWatchOption) *LogWatcher

func (lw *LogWatcher) Close()

func (lw *LogWatcher) Init() error

func (lw *LogWatcher) RegisterAndWatch(dir, pattern string) error

func (lw *LogWatcher) RunEventHandler()

type LogWatcherOption ¶

type LogWatcherOption struct {
	FilterInterval int `json:"filter"` // in sec
	PollerInterval int `json:"poller"` // in byte
}

type Sink ¶

type Sink interface {
	Open(string) error
	Push(fpath string, content []byte) error
	Close() error
}

func NewFileSink() Sink

type SinkConf ¶

type SinkConf struct {
	Typ  SinkType `json:"type"`
	Conf string   `json:"config"`
}

type SinkType ¶

type SinkType string

const (
	FileSinkType SinkType = "file"
)

type TaskState ¶

type TaskState uint64