fileconsumer

Documentation

Index

• Constants
• Variables
• type Config
  • func NewConfig() *Config
  • func (c Config) Build(logger *zap.SugaredLogger, emit EmitFunc) (*Manager, error)
  • func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit EmitFunc, splitFunc bufio.SplitFunc) (*Manager, error)
• type EmitFunc
• type FileAttributes
  • func (f *FileAttributes) HeaderAttributesCopy() map[string]any
• type Finder
  • func (f Finder) FindFiles() []string
• type Fingerprint
  • func NewFingerprint(file *os.File, size int) (*Fingerprint, error)
  • func (f Fingerprint) Copy() *Fingerprint
  • func (f Fingerprint) StartsWith(old *Fingerprint) bool
• type HeaderConfig
• type Manager
  • func (m *Manager) Start(persister operator.Persister) error
  • func (m *Manager) Stop() error
• type PositionalScanner
  • func NewPositionalScanner(r io.Reader, maxLogSize int, startOffset int64, splitFunc bufio.SplitFunc) *PositionalScanner
  • func (ps *PositionalScanner) Pos() int64
• type Reader
  • func (r *Reader) Close()
  • func (r *Reader) Read(dst []byte) (int, error)
  • func (r *Reader) ReadToEnd(ctx context.Context)

Constants

const DefaultFingerprintSize = 1000 // bytes

const MinFingerprintSize = 16 // bytes

Variables

var AllowHeaderMetadataParsing = featuregate.GlobalRegistry().MustRegister(
	"filelog.allowHeaderMetadataParsing",
	featuregate.StageAlpha,
	featuregate.WithRegisterDescription("When enabled, allows usage of the `header` setting."),
	featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/18198"),
)

Types

type Config

type Config struct {
	Finder                  `mapstructure:",squash"`
	IncludeFileName         bool                  `mapstructure:"include_file_name,omitempty"`
	IncludeFilePath         bool                  `mapstructure:"include_file_path,omitempty"`
	IncludeFileNameResolved bool                  `mapstructure:"include_file_name_resolved,omitempty"`
	IncludeFilePathResolved bool                  `mapstructure:"include_file_path_resolved,omitempty"`
	PollInterval            time.Duration         `mapstructure:"poll_interval,omitempty"`
	StartAt                 string                `mapstructure:"start_at,omitempty"`
	FingerprintSize         helper.ByteSize       `mapstructure:"fingerprint_size,omitempty"`
	MaxLogSize              helper.ByteSize       `mapstructure:"max_log_size,omitempty"`
	MaxConcurrentFiles      int                   `mapstructure:"max_concurrent_files,omitempty"`
	MaxBatches              int                   `mapstructure:"max_batches,omitempty"`
	DeleteAfterRead         bool                  `mapstructure:"delete_after_read,omitempty"`
	Splitter                helper.SplitterConfig `mapstructure:",squash,omitempty"`
	Header                  *HeaderConfig         `mapstructure:"header,omitempty"`
}

Config is the configuration of a file input operator

func NewConfig

func NewConfig() *Config

NewConfig creates a new input config with default values

func (Config) Build

func (c Config) Build(logger *zap.SugaredLogger, emit EmitFunc) (*Manager, error)

Build will build a file input operator from the supplied configuration

func (Config) BuildWithSplitFunc

func (c Config) BuildWithSplitFunc(
	logger *zap.SugaredLogger, emit EmitFunc, splitFunc bufio.SplitFunc) (*Manager, error)

BuildWithSplitFunc will build a file input operator with customized splitFunc function

type EmitFunc

type EmitFunc func(ctx context.Context, attrs *FileAttributes, token []byte)

type FileAttributes

type FileAttributes struct {
	Name             string `json:"-"`
	Path             string `json:"-"`
	NameResolved     string `json:"-"`
	PathResolved     string `json:"-"`
	HeaderAttributes map[string]any
}

func (*FileAttributes) HeaderAttributesCopy

func (f *FileAttributes) HeaderAttributesCopy() map[string]any

HeaderAttributesCopy gives a copy of the HeaderAttributes, in order to restrict mutation of the HeaderAttributes.

type Finder

type Finder struct {
	Include []string `mapstructure:"include,omitempty"`
	Exclude []string `mapstructure:"exclude,omitempty"`
}

func (Finder) FindFiles

func (f Finder) FindFiles() []string

FindFiles gets a list of paths given an array of glob patterns to include and exclude

type Fingerprint

type Fingerprint struct {
	FirstBytes []byte
}

Fingerprint is used to identify a file A file's fingerprint is the first N bytes of the file

func NewFingerprint

func NewFingerprint(file *os.File, size int) (*Fingerprint, error)

NewFingerprint creates a new fingerprint from an open file

func (Fingerprint) Copy

func (f Fingerprint) Copy() *Fingerprint

Copy creates a new copy of the fingerprint

func (Fingerprint) StartsWith

func (f Fingerprint) StartsWith(old *Fingerprint) bool

StartsWith returns true if the fingerprints are the same or if the new fingerprint starts with the old one This is important functionality for tracking new files, since their initial size is typically less than that of a fingerprint. As the file grows, its fingerprint is updated until it reaches a maximum size, as configured on the operator

type HeaderConfig

type HeaderConfig struct {
	Pattern           string            `mapstructure:"pattern"`
	MetadataOperators []operator.Config `mapstructure:"metadata_operators"`
}

type Manager

type Manager struct {
	*zap.SugaredLogger
	// contains filtered or unexported fields
}

func (*Manager) Start

func (m *Manager) Start(persister operator.Persister) error

func (*Manager) Stop

func (m *Manager) Stop() error

Stop will stop the file monitoring process

type PositionalScanner

type PositionalScanner struct {
	*bufio.Scanner
	// contains filtered or unexported fields
}

PositionalScanner is a scanner that maintains position

func NewPositionalScanner

func NewPositionalScanner(r io.Reader, maxLogSize int, startOffset int64, splitFunc bufio.SplitFunc) *PositionalScanner

NewPositionalScanner creates a new positional scanner

func (*PositionalScanner) Pos

func (ps *PositionalScanner) Pos() int64

Pos returns the current position of the scanner

type Reader

type Reader struct {
	*zap.SugaredLogger `json:"-"` // json tag excludes embedded fields from storage

	Fingerprint *Fingerprint
	Offset      int64

	FileAttributes *FileAttributes

	HeaderFinalized bool
	// contains filtered or unexported fields
}

Reader manages a single file

func (*Reader) Close

func (r *Reader) Close()

Close will close the file

func (*Reader) Read

func (r *Reader) Read(dst []byte) (int, error)

Read from the file and update the fingerprint if necessary

func (*Reader) ReadToEnd

func (r *Reader) ReadToEnd(ctx context.Context)

ReadToEnd will read until the end of the file