outctx

Documentation

Index

• Constants
• type EventManager
  • func (m *EventManager) ToS3(config S3Config, uploader *manager.Uploader) error
• type S3Config
  • func NewS3Config(plugin unsafe.Pointer) (*S3Config, error)
• type S3Context
  • func NewS3Context(plugin unsafe.Pointer) (*S3Context, error)
  • func (ctx *S3Context) GetBufferFilePaths(tag string) (string, string)
  • func (ctx *S3Context) GetBufferPaths() (string, string)
  • func (ctx *S3Context) GetEventManager(tag string, size int) (*EventManager, error)
  • func (ctx *S3Context) RecoverEventManager(tag string, size int) (*EventManager, error)

Constants

const (
	IrDir   = "ir"
	ZstdDir = "zstd"
)

Names of disk buffering directories.

Types

type EventManager

type EventManager struct {
	Tag    string
	Index  int
	Writer irzstd.Writer
}

Resources and metadata to process Fluent Bit events with the same tag.

func (*EventManager) ToS3

func (m *EventManager) ToS3(config S3Config, uploader *manager.Uploader) error

Sends Zstd buffer to s3 and reset writer and buffers for future uploads. Prior to upload, IR buffer is flushed and IR/Zstd streams are terminated. The [EventManager.Index] is incremented on successful upload.

Parameters:

• config: Plugin configuration
• uploader: S3 uploader manager

Returns:

• err: Error creating closing streams, error uploading to s3, error resetting writer

type S3Config

type S3Config struct {
	S3Region        string `conf:"s3_region"         validate:"required"`
	S3Bucket        string `conf:"s3_bucket"         validate:"required"`
	S3BucketPrefix  string `conf:"s3_bucket_prefix"  validate:"dirpath"`
	RoleArn         string `conf:"role_arn"          validate:"omitempty,startswith=arn:aws:iam"`
	Id              string `conf:"id"                validate:"required"`
	UseSingleKey    bool   `conf:"use_single_key"    validate:"-"`
	AllowMissingKey bool   `conf:"allow_missing_key" validate:"-"`
	SingleKey       string `conf:"single_key"        validate:"required_if=use_single_key true"`
	UseDiskBuffer   bool   `conf:"use_disk_buffer"   validate:"-"`
	DiskBufferPath  string `conf:"disk_buffer_path"  validate:"omitempty,dirpath"`
	UploadSizeMb    int    `conf:"upload_size_mb"    validate:"omitempty,gte=2,lt=1000"`
	TimeZone        string `conf:"time_zone"         validate:"timezone"`
}

Holds settings for S3 CLP plugin from user-defined Fluent Bit configuration file. The "conf" struct tags are the plugin options described to user in README, and allow user to see snake case "use_single_key" vs. camel case "SingleKey" in validation error messages. The "validate" struct tags are rules to be consumed by validator. The functionality of each rule can be found in docs for validator.

func NewS3Config

func NewS3Config(plugin unsafe.Pointer) (*S3Config, error)

Generates configuration struct containing user-defined settings. In addition, sets default values and validates user input.

Parameters:

• plugin: Fluent Bit plugin reference

Returns:

• S3Config: Configuration based on fluent-bit.conf
• err: All validation errors in config wrapped, parse bool error

type S3Context

type S3Context struct {
	Config        S3Config
	Uploader      *manager.Uploader
	EventManagers map[string]*EventManager
}

Holds objects accessible to plugin during flush. Fluent Bit uses a single thread for Go output plugin instance so no need to consider synchronization issues. C plugins use "coroutines" which could cause synchronization issues for C plugins according to [docs] but "coroutines" are not used in Go plugins. [docs]: https://github.com/fluent/fluent-bit/blob/master/DEVELOPER_GUIDE.md#concurrency

func NewS3Context

func NewS3Context(plugin unsafe.Pointer) (*S3Context, error)

Creates a new context. Loads configuration from user. Loads and tests aws credentials.

Parameters:

• plugin: Fluent Bit plugin reference

Returns:

• S3Context: Plugin context
• err: User configuration load failed, aws errors

func (*S3Context) GetBufferFilePaths

func (ctx *S3Context) GetBufferFilePaths(
	tag string,
) (string, string)

Retrieves paths for IR and Zstd disk buffer files.

Parameters:

• tag: Fluent Bit tag

Returns:

• irPath: Path to IR disk buffer file
• zstdPath: Path to Zstd disk buffer file

func (*S3Context) GetBufferPaths

func (ctx *S3Context) GetBufferPaths() (string, string)

Retrieves paths for IR and Zstd disk buffer directories.

Returns:

• irBufferPath: Path of IR disk buffer directory
• zstdBufferPath: Path of Zstd disk buffer directory

func (*S3Context) GetEventManager

func (ctx *S3Context) GetEventManager(tag string, size int) (*EventManager, error)

If the event manager for the tag has been initialized, get the corresponding event manager. If not, create new one.

Parameters:

• tag: Fluent Bit tag
• size: Byte length

Returns:

• err: Could not create buffers or tag

func (*S3Context) RecoverEventManager

func (ctx *S3Context) RecoverEventManager(
	tag string,
	size int,
) (*EventManager, error)

Recovers EventManager from previous execution using existing disk buffers.

Parameters:

• tag: Fluent Bit tag
• size: Byte length

Returns:

• eventManager: Manager for Fluent Bit events with the same tag
• err: Error creating new writer