log

Documentation

Overview

Package log harvests different inputs for new information. Currently two harvester types exist:

• log

• stdin

  The log harvester reads a file line by line. In case the end of a file is found with an incomplete line, the line pointer stays at the beginning of the incomplete line. As soon as the line is completed, it is read and returned.

  The stdin harvesters reads data from stdin.

Index

• Constants
• Variables
• func NewProspector(cfg *common.Config, outlet channel.Factory, context prospector.Context) (prospector.Prospectorer, error)
• type File
  • func (File) Continuable() bool
  • func (File) HasState() bool
• type FileSortInfo
• type Harvester
  • func NewHarvester(config *common.Config, state file.State, states *file.States, ...) (*Harvester, error)
  • func (h *Harvester) ID() uuid.UUID
  • func (h *Harvester) Run() error
  • func (h *Harvester) SendStateUpdate()
  • func (h *Harvester) Setup() error
  • func (h *Harvester) Stop()
• type Log
  • func NewLog(fs harvester.Source, config LogConfig) (*Log, error)
  • func (f *Log) Close()
  • func (f *Log) Read(buf []byte) (int, error)
• type LogConfig
• type Pipe
  • func (p Pipe) Close() error
  • func (p Pipe) Continuable() bool
  • func (p Pipe) HasState() bool
  • func (p Pipe) Name() string
  • func (p Pipe) Read(b []byte) (int, error)
  • func (p Pipe) Stat() (os.FileInfo, error)
• type Prospector
  • func (p *Prospector) Run()
  • func (p *Prospector) Stop()
  • func (p *Prospector) Wait()

Constants

const (
	ScanOrderAsc     = "asc"
	ScanOrderDesc    = "desc"
	ScanSortNone     = ""
	ScanSortModtime  = "modtime"
	ScanSortFilename = "filename"
)

Contains available scan options

Variables

var (
	ErrFileTruncate = errors.New("detected file being truncated")
	ErrRenamed      = errors.New("file was renamed")
	ErrRemoved      = errors.New("file was removed")
	ErrInactive     = errors.New("file inactive")
	ErrClosed       = errors.New("reader closed")
)

var ValidScanOrder = map[string]struct{}{
	ScanOrderAsc:  {},
	ScanOrderDesc: {},
}

ValidScanOrder of valid scan orders

var ValidScanSort = map[string]struct{}{
	ScanSortNone:     {},
	ScanSortModtime:  {},
	ScanSortFilename: {},
}

ValidScanOrder of valid scan orders

Functions

func NewProspector

func NewProspector(
	cfg *common.Config,
	outlet channel.Factory,
	context prospector.Context,
) (prospector.Prospectorer, error)

NewProspector instantiates a new Log

Types

type File

type File struct {
	*os.File
}

func (File) Continuable

func (File) Continuable() bool

func (File) HasState

func (File) HasState() bool

type FileSortInfo

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

type Harvester

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

Harvester contains all harvester related data

func NewHarvester

func NewHarvester(
	config *common.Config,
	state file.State,
	states *file.States,
	publishState func(*util.Data) bool,
	outlet channel.Outleter,
) (*Harvester, error)

NewHarvester creates a new harvester

func (*Harvester) ID

func (h *Harvester) ID() uuid.UUID

ID returns the unique harvester identifier

func (*Harvester) Run

func (h *Harvester) Run() error

Run start the harvester and reads files line by line and sends events to the defined output

func (*Harvester) SendStateUpdate

func (h *Harvester) SendStateUpdate()

SendStateUpdate send an empty event with the current state to update the registry close_timeout does not apply here to make sure a harvester is closed properly. In case the output is blocked the harvester will stay open to make sure no new harvester is started. As soon as the output becomes available again, the finished state is written and processing can continue.

func (*Harvester) Setup

func (h *Harvester) Setup() error

Setup opens the file handler and creates the reader for the harvester

func (*Harvester) Stop

func (h *Harvester) Stop()

Stop stops harvester and waits for completion

type Log

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

Log contains all log related data

func NewLog

func NewLog(
	fs harvester.Source,
	config LogConfig,
) (*Log, error)

NewLog creates a new log instance to read log sources

func (*Log) Close

func (f *Log) Close()

Close closes the done channel but no th the file handler

func (*Log) Read

func (f *Log) Read(buf []byte) (int, error)

Read reads from the reader and updates the offset The total number of bytes read is returned.

type LogConfig

type LogConfig struct {
	Backoff       time.Duration `config:"backoff" validate:"min=0,nonzero"`
	BackoffFactor int           `config:"backoff_factor" validate:"min=1"`
	MaxBackoff    time.Duration `config:"max_backoff" validate:"min=0,nonzero"`
	CloseInactive time.Duration `config:"close_inactive"`
	CloseRemoved  bool          `config:"close_removed"`
	CloseRenamed  bool          `config:"close_renamed"`
	CloseEOF      bool          `config:"close_eof"`
	CloseTimeout  time.Duration `config:"close_timeout" validate:"min=0"`
}

type Pipe

type Pipe struct {
	File *os.File
}

restrict file to minimal interface of FileSource to prevent possible casts to additional interfaces supported by underlying file

func (Pipe) Close

func (p Pipe) Close() error

func (Pipe) Continuable

func (p Pipe) Continuable() bool

func (Pipe) HasState

func (p Pipe) HasState() bool

func (Pipe) Name

func (p Pipe) Name() string

func (Pipe) Read

func (p Pipe) Read(b []byte) (int, error)

func (Pipe) Stat

func (p Pipe) Stat() (os.FileInfo, error)

type Prospector

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

Prospector contains the prospector and its config

func (*Prospector) Run

func (p *Prospector) Run()

Run runs the prospector

func (*Prospector) Stop

func (p *Prospector) Stop()

Stop stops all harvesters and then stops the prospector

func (*Prospector) Wait

func (p *Prospector) Wait()

Wait waits for the all harvesters to complete and only then call stop