fluent

package
v0.8.260826 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Overview

Package fluent sends structured events using Fluent Forward Protocol v1.

A Client owns its connection, retry loop, bounded buffer, and batching worker. Send waits for delivery. Submit and TrySubmit transfer ownership of encoded entries to the client and return a Receipt for the eventual result.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrClosed             = errors.New("fluent client closed")
	ErrQueueFull          = errors.New("fluent buffer full")
	ErrTooLarge           = errors.New("fluent request exceeds buffer limits")
	ErrInvalidTag         = errors.New("invalid fluent tag")
	ErrInvalidRecord      = errors.New("invalid fluent record")
	ErrDeliveryUnknown    = errors.New("fluent delivery status unknown")
	ErrACKMismatch        = errors.New("fluent ACK mismatch")
	ErrAuthRejected       = errors.New("fluent authentication rejected")
	ErrServerVerification = errors.New("fluent server verification failed")
	ErrProtocol           = errors.New("invalid fluent protocol message")
)

Functions

This section is empty.

Types

type Auth 

type Auth struct {
	SharedKey string `json:"shared-key" desc:"Shared key for Forward authentication"`
	Username  string `json:"username" desc:"Username for Forward authentication"`
	Password  string `json:"password" desc:"Password for Forward authentication"`
	Hostname  string `json:"hostname" desc:"Client hostname used during Forward authentication"`
}

Auth configures the Forward Protocol shared-key handshake.

type BufferConfig added in v0.3.260716 

type BufferConfig struct {
	MaxEvents      int           `json:"max-events" desc:"Maximum number of unfinished events"`
	MaxBytes       int           `json:"max-bytes" desc:"Maximum encoded bytes held by unfinished events"`
	BatchMaxEvents int           `json:"batch-max-events" desc:"Maximum number of events in one wire batch"`
	BatchMaxBytes  int           `json:"batch-max-bytes" desc:"Maximum encoded bytes in one wire batch"`
	BatchWait      time.Duration `json:"batch-wait" desc:"Maximum wait for collecting an automatic batch"`
}

BufferConfig bounds accepted but unfinished work and controls automatic batching. MaxBytes counts encoded record data and protocol overhead.

type Client 

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

Client is a concurrency-safe Fluent Forward client.

func New 

func New(config Config, options ...Option) (*Client, error)

New creates a lazy client. Runtime options are applied in order before final validation. The first batch establishes the connection.

func (*Client) Abort added in v0.3.260716 

func (c *Client) Abort()

Abort immediately cancels in-flight work and rejects pending work. It does not wait for a custom Connector that violates context cancellation.

func (*Client) Flush 

func (c *Client) Flush(ctx context.Context) error

Flush waits until all requests accepted before the barrier have completed.

func (*Client) Send 

func (c *Client) Send(ctx context.Context, tag string, entries ...Entry) error

Send submits entries and waits for their delivery. ctx controls admission, delivery, and waiting. Synchronous requests are not combined with requests from other callers.

func (*Client) Shutdown 

func (c *Client) Shutdown(ctx context.Context) error

Shutdown stops admission, drains accepted work, and closes the connection. If ctx expires, remaining work is aborted before Shutdown returns.

func (*Client) Stats added in v0.3.260716 

func (c *Client) Stats() Stats

Stats returns a concurrency-safe snapshot.

func (*Client) Submit 

func (c *Client) Submit(ctx context.Context, tag string, entries ...Entry) (*Receipt, error)

Submit waits for buffer capacity and transfers ownership of the encoded entries to the client. ctx only controls admission.

func (*Client) TrySubmit 

func (c *Client) TrySubmit(tag string, entries ...Entry) (*Receipt, error)

TrySubmit transfers entries to the client without waiting for capacity.

type Config 

type Config struct {
	// Enabled is an application-level switch provided for config and flag
	// binding. This package intentionally does not inspect it; callers decide
	// whether to construct and use a Client.
	Enabled              bool            `json:"enabled" desc:"Enable Fluent client in the calling application"`
	Endpoint             string          `json:"endpoint" desc:"Fluent endpoint using tcp, tls, unix, ws, or wss"`
	Auth                 *Auth           `json:"auth" desc:"Forward authentication settings"`
	TagPrefix            string          `json:"tag-prefix" desc:"Prefix added to every event tag"`
	ACK                  bool            `json:"ack" desc:"Require Forward acknowledgements"`
	Buffer               BufferConfig    `json:"buffer" desc:"Buffer and automatic batching settings"`
	Retry                RetryConfig     `json:"retry" desc:"Delivery retry settings"`
	Timeout              TimeoutConfig   `json:"timeout" desc:"Connection and I/O timeout settings"`
	CompressionThreshold int             `json:"compression-threshold" desc:"Minimum packed bytes required for gzip compression"`
	TLS                  TLSConfig       `json:"tls" desc:"TLS and mutual TLS settings"`
	WebSocket            WebSocketConfig `json:"websocket" desc:"WebSocket upgrade settings"`
}

Config contains all file-friendly client settings. Endpoint selects a built-in transport using tcp, tls, unix, ws, or wss. Runtime Go objects are supplied to New through Option values.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns a complete client configuration targeting Fluent's conventional local TCP endpoint. ParseConfig overlays JSON on this value so omitted fields retain their defaults.

func ParseConfig added in v0.7.260825 

func ParseConfig(data []byte) (Config, error)

ParseConfig decodes a strict JSON configuration over DefaultConfig and validates the result. Durations use time.ParseDuration syntax, such as "250ms" or "3s".

func (Config) Validate added in v0.3.260716 

func (c Config) Validate() error

Validate checks transport and cross-field constraints without reading TLS files or establishing a connection.

type Connector added in v0.3.260716 

type Connector interface {
	Dial(context.Context) (net.Conn, error)
}

Connector establishes a complete transport connection. Implementations must honor context cancellation for the entire connection setup.

type Entry 

type Entry struct {
	Time   time.Time
	Record Record
}

Entry is one structured event. A zero Time is replaced once, during submission, so retries preserve the original timestamp.

type Error added in v0.3.260716 

type Error struct {
	Operation Operation
	Endpoint  string
	Attempt   int
	Retryable bool
	Err       error
}

Error adds stable operation metadata while preserving the underlying error.

func (*Error) Error added in v0.3.260716 

func (e *Error) Error() string

func (*Error) Temporary added in v0.3.260716 

func (e *Error) Temporary() bool

Temporary reports whether retrying the operation on a new connection may succeed. It supports existing net.Error-style classifiers.

func (*Error) Unwrap added in v0.3.260716 

func (e *Error) Unwrap() error

type Failure added in v0.3.260716 

type Failure struct {
	Tag       string
	Entries   int
	Delivered int
	Err       error
}

Failure describes the final failure of an asynchronously submitted request.

type FailureHandler added in v0.3.260716 

type FailureHandler func(Failure)

FailureHandler observes final asynchronous delivery failures. Calls are serialized on a separate goroutine and must return promptly.

type HeaderProvider added in v0.3.260716 

type HeaderProvider func(context.Context) (http.Header, error)

HeaderProvider returns headers for each WebSocket connection attempt. The returned headers override static headers with the same names.

func BearerToken added in v0.3.260716 

func BearerToken(provider func(context.Context) (string, error)) HeaderProvider

BearerToken adapts a rotating token provider into a HeaderProvider.

type Operation added in v0.3.260716 

type Operation string

Operation identifies the failed client operation. 

const (
	OperationConnect Operation = "connect"
	OperationSend    Operation = "send"
)

type Option added in v0.3.260716 

type Option interface {
	// contains filtered or unexported methods
}

Option configures code-only runtime dependencies. File-friendly settings belong in Config. Options are applied in order before final validation.

func WithConnector added in v0.4.260716 

func WithConnector(connector Connector) Option

WithConnector replaces the built-in Endpoint transport. Endpoint remains the error-context label. Built-in TLS, WebSocket, and NetDialer settings cannot be combined with a custom Connector.

func WithFailureHandler added in v0.3.260716 

func WithFailureHandler(handler FailureHandler) Option

WithFailureHandler observes final asynchronous delivery failures.

func WithHeaderProvider added in v0.4.260716 

func WithHeaderProvider(provider HeaderProvider) Option

WithHeaderProvider supplies dynamic headers for ws:// or wss:// endpoints.

func WithNetDialer added in v0.4.260716 

func WithNetDialer(dialer *net.Dialer) Option

WithNetDialer supplies a custom net.Dialer for a built-in transport.

func WithTLSConfig added in v0.4.260716 

func WithTLSConfig(config *tls.Config) Option

WithTLSConfig supplies a native TLS configuration for tls:// or wss:// endpoints. It cannot be combined with Config.TLS file settings.

type PartialDeliveryError added in v0.3.260716 

type PartialDeliveryError struct {
	Delivered int
	Total     int
	Err       error
}

PartialDeliveryError reports that earlier entries from one request completed before a later wire batch failed. The completed prefix must not be retried blindly in unconfirmed mode.

func (*PartialDeliveryError) Error added in v0.3.260716 

func (e *PartialDeliveryError) Error() string

func (*PartialDeliveryError) Unwrap added in v0.3.260716 

func (e *PartialDeliveryError) Unwrap() error

type Receipt 

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

Receipt represents the eventual result of an accepted asynchronous request.

func (*Receipt) Done added in v0.3.260716 

func (r *Receipt) Done() <-chan struct{}

Done is closed when delivery finishes.

func (*Receipt) Wait 

func (r *Receipt) Wait(ctx context.Context) error

Wait waits for delivery. Canceling ctx only stops this wait; it never cancels an asynchronously submitted request.

type Record added in v0.3.260716 

type Record map[string]any

Record is a Forward Protocol record. Its top level is always a map with string keys; nested values must be supported by MessagePack.

type RetryConfig 

type RetryConfig struct {
	MaxAttempts int           `json:"max-attempts" desc:"Maximum number of delivery attempts"`
	MinBackoff  time.Duration `json:"min-backoff" desc:"Minimum delay between delivery attempts"`
	MaxBackoff  time.Duration `json:"max-backoff" desc:"Maximum delay between delivery attempts"`
}

RetryConfig configures the total number of attempts and exponential backoff.

type Stats added in v0.3.260716 

type Stats struct {
	PendingEvents               int
	PendingBytes                int
	SubmittedEntries            uint64
	DeliveredEntries            uint64
	FailedRequests              uint64
	Retries                     uint64
	Connections                 uint64
	RejectedRequests            uint64
	FailureNotificationsDropped uint64
}

Stats is a point-in-time snapshot of client activity.

type TLSConfig added in v0.4.260716 

type TLSConfig struct {
	CAFile             string `json:"ca-file" desc:"PEM file containing trusted server certificate authorities"`
	CertificateFile    string `json:"certificate-file" desc:"PEM client certificate file for mutual TLS"`
	KeyFile            string `json:"key-file" desc:"PEM client private key file for mutual TLS"`
	ServerName         string `json:"server-name" desc:"TLS server name used for certificate verification"`
	MinVersion         string `json:"min-version" desc:"Minimum TLS version: 1.2 or 1.3"`
	InsecureSkipVerify bool   `json:"insecure-skip-verify" desc:"Skip TLS server certificate verification"`
}

TLSConfig contains file-friendly TLS and mTLS settings.

type TimeoutConfig 

type TimeoutConfig struct {
	Connect   time.Duration `json:"connect" desc:"Timeout for complete transport connection setup"`
	Handshake time.Duration `json:"handshake" desc:"Timeout for the Forward authentication handshake"`
	Write     time.Duration `json:"write" desc:"Timeout for writing one wire batch"`
	ACK       time.Duration `json:"ack" desc:"Timeout for receiving one Forward acknowledgement"`
}

TimeoutConfig configures each connection, Forward handshake, write, and ACK operation. A zero duration disables that individual timeout.

type WebSocketConfig 

type WebSocketConfig struct {
	Header []WebSocketHeader `json:"header" desc:"Static HTTP headers sent during the WebSocket upgrade"`
}

WebSocketConfig contains WebSocket-specific connection settings.

type WebSocketHandshakeError added in v0.3.260716 

type WebSocketHandshakeError struct {
	StatusCode int
	Body       string
	Err        error
}

WebSocketHandshakeError reports a failed HTTP upgrade. Body is bounded by the WebSocket implementation and omitted from Error to avoid accidental logging of sensitive response content.

func (*WebSocketHandshakeError) Error added in v0.3.260716 

func (e *WebSocketHandshakeError) Error() string

func (*WebSocketHandshakeError) Temporary added in v0.3.260716 

func (e *WebSocketHandshakeError) Temporary() bool

func (*WebSocketHandshakeError) Unwrap added in v0.3.260716 

func (e *WebSocketHandshakeError) Unwrap() error

type WebSocketHeader added in v0.6.260716 

type WebSocketHeader struct {
	Name   string   `json:"name" desc:"HTTP header name"`
	Values []string `json:"values" desc:"HTTP header values"`
}

WebSocketHeader describes one static HTTP header sent during the WebSocket upgrade. Values preserves repeated header values without exposing a map to file/CLI configuration decoders.

Source Files

View all Source files

Directories

Path Synopsis
internal
protocol
transport

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.