buffer

Documentation

Index

• type Buffer
• type BufferBuilder
• type Config
  • func NewConfig() Config
  • func (bc *Config) UnmarshalJSON(data []byte) error
  • func (bc *Config) UnmarshalYAML(f func(interface{}) error) error
• type DiskBuffer
  • func NewDiskBuffer(maxDiskSize int) *DiskBuffer
  • func (d *DiskBuffer) Add(ctx context.Context, newEntry *entry.Entry) error
  • func (d *DiskBuffer) Close() error
  • func (d *DiskBuffer) Compact() error
  • func (d *DiskBuffer) Open(path string, sync bool) error
  • func (d *DiskBuffer) Read(dst []*entry.Entry) (f FlushFunc, i int, err error)
  • func (d *DiskBuffer) ReadWait(ctx context.Context, dst []*entry.Entry) (FlushFunc, int, error)
• type DiskBufferConfig
  • func NewDiskBufferConfig() *DiskBufferConfig
  • func (c DiskBufferConfig) Build(context operator.BuildContext, _ string) (Buffer, error)
• type FlushFunc
• type MemoryBuffer
  • func (m *MemoryBuffer) Add(ctx context.Context, e *entry.Entry) error
  • func (m *MemoryBuffer) Close() error
  • func (m *MemoryBuffer) Read(dst []*entry.Entry) (FlushFunc, int, error)
  • func (m *MemoryBuffer) ReadWait(ctx context.Context, dst []*entry.Entry) (FlushFunc, int, error)
• type MemoryBufferConfig
  • func NewMemoryBufferConfig() *MemoryBufferConfig
  • func (c MemoryBufferConfig) Build(context operator.BuildContext, pluginID string) (Buffer, error)
• type Metadata
  • func OpenMetadata(path string, sync bool) (*Metadata, error)
  • func (m *Metadata) Close() error
  • func (m *Metadata) MarshalBinary(wr io.Writer) (err error)
  • func (m *Metadata) Sync() error
  • func (m *Metadata) UnmarshalBinary(r io.Reader) error

Types

type Buffer

type Buffer interface {
	Add(context.Context, *entry.Entry) error
	Read([]*entry.Entry) (FlushFunc, int, error)
	ReadWait(context.Context, []*entry.Entry) (FlushFunc, int, error)
	Close() error
}

type BufferBuilder

type BufferBuilder interface {
	Build(context operator.BuildContext, pluginID string) (Buffer, error)
}

type Config

type Config struct {
	BufferBuilder
}

func NewConfig

func NewConfig() Config

func (*Config) UnmarshalJSON

func (bc *Config) UnmarshalJSON(data []byte) error

func (*Config) UnmarshalYAML

func (bc *Config) UnmarshalYAML(f func(interface{}) error) error

type DiskBuffer

type DiskBuffer struct {
	sync.Mutex
	// contains filtered or unexported fields
}

DiskBuffer is a buffer for storing entries on disk until they are flushed to their final destination.

func NewDiskBuffer

func NewDiskBuffer(maxDiskSize int) *DiskBuffer

NewDiskBuffer creates a new DiskBuffer

func (*DiskBuffer) Add

func (d *DiskBuffer) Add(ctx context.Context, newEntry *entry.Entry) error

Add adds an entry to the buffer, blocking until it is either added or the context is cancelled.

func (*DiskBuffer) Close

func (d *DiskBuffer) Close() error

Close flushes the current metadata to disk, then closes the underlying files

func (*DiskBuffer) Compact

func (d *DiskBuffer) Compact() error

Compact removes all flushed entries from disk

func (*DiskBuffer) Open

func (d *DiskBuffer) Open(path string, sync bool) error

Open opens the disk buffer files from a database directory

func (*DiskBuffer) Read

func (d *DiskBuffer) Read(dst []*entry.Entry) (f FlushFunc, i int, err error)

Read copies entries from the disk into the destination buffer. It returns a function that, when called, marks the entries as flushed, the number of entries read, and an error.

func (*DiskBuffer) ReadWait

func (d *DiskBuffer) ReadWait(ctx context.Context, dst []*entry.Entry) (FlushFunc, int, error)

ReadWait reads entries from the buffer, waiting until either there are enough entries in the buffer to fill dst or the context is cancelled. This amortizes the cost of reading from the disk. It returns a function that, when called, marks the read entries as flushed, the number of entries read, and an error.

type DiskBufferConfig

type DiskBufferConfig struct {
	Type string `json:"type" yaml:"type"`

	// MaxSize is the maximum size in bytes of the data file on disk
	MaxSize int `json:"max_size" yaml:"max_size"`

	// Path is a path to a directory which contains the data and metadata files
	Path string `json:"path" yaml:"path"`

	// Sync indicates whether to open the files with O_SYNC. If this is set to false,
	// in cases like power failures or unclean shutdowns, logs may be lost or the
	// database may become corrupted.
	Sync bool `json:"sync" yaml:"sync"`
}

DiskBufferConfig is a configuration struct for a DiskBuffer

func NewDiskBufferConfig

func NewDiskBufferConfig() *DiskBufferConfig

NewDiskBufferConfig creates a new default disk buffer config

func (DiskBufferConfig) Build

func (c DiskBufferConfig) Build(context operator.BuildContext, _ string) (Buffer, error)

Build creates a new Buffer from a DiskBufferConfig

type FlushFunc

type FlushFunc func() error

type MemoryBuffer

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

MemoryBuffer is a buffer that holds all entries in memory until Close() is called, at which point it saves the entries into a database. It provides no guarantees about lost entries if shut down uncleanly.

func (*MemoryBuffer) Add

func (m *MemoryBuffer) Add(ctx context.Context, e *entry.Entry) error

Add inserts an entry into the memory database, blocking until there is space

func (*MemoryBuffer) Close

func (m *MemoryBuffer) Close() error

Close closes the memory buffer, saving all entries currently in the memory buffer to the agent's database.

func (*MemoryBuffer) Read

func (m *MemoryBuffer) Read(dst []*entry.Entry) (FlushFunc, int, error)

Read reads entries until either there are no entries left in the buffer or the destination slice is full. The returned function must be called once the entries are flushed to remove them from the memory buffer.

func (*MemoryBuffer) ReadWait

func (m *MemoryBuffer) ReadWait(ctx context.Context, dst []*entry.Entry) (FlushFunc, int, error)

ReadWait reads entries until either the destination slice is full, or the context passed to it is cancelled. The returned function must be called once the entries are flushed to remove them from the memory buffer

type MemoryBufferConfig

type MemoryBufferConfig struct {
	Type       string `json:"type" yaml:"type"`
	MaxEntries int    `json:"max_entries" yaml:"max_entries"`
}

MemoryBufferConfig holds the configuration for a memory buffer

func NewMemoryBufferConfig

func NewMemoryBufferConfig() *MemoryBufferConfig

NewMemoryBufferConfig creates a new default MemoryBufferConfig

func (MemoryBufferConfig) Build

func (c MemoryBufferConfig) Build(context operator.BuildContext, pluginID string) (Buffer, error)

Build builds a MemoryBufferConfig into a Buffer, loading any entries that were previously unflushed back into memory

type Metadata

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

Metadata is a representation of the on-disk metadata file. It contains information about the layout, location, and flushed status of entries stored in the data file

func OpenMetadata

func OpenMetadata(path string, sync bool) (*Metadata, error)

OpenMetadata opens and parses the metadata

func (*Metadata) Close

func (m *Metadata) Close() error

Close syncs metadata to disk and closes the underlying file descriptor

func (*Metadata) MarshalBinary

func (m *Metadata) MarshalBinary(wr io.Writer) (err error)

MarshalBinary marshals a metadata struct to a binary stream

func (*Metadata) Sync

func (m *Metadata) Sync() error

Sync persists the metadata to disk

func (*Metadata) UnmarshalBinary

func (m *Metadata) UnmarshalBinary(r io.Reader) error

UnmarshalBinary unmarshals metadata from a binary stream (usually a file)