forwardprotocol

Documentation

Overview

Package forwardprotocol provides definitions and a reference implementation for Fluentd "Forward" protocol

Functions here are not considered performance critical

Index

• Constants
• func DoClientHandshake(conn net.Conn, sharedKey string, timeout time.Duration) (bool, string, error)
• func DoServerHandshake(conn net.Conn, sharedKey string, timeout time.Duration, auth AuthCallback) (bool, error)
• type Ack
• type AuthCallback
• type EventEntry
  • func (e *EventEntry) ResolvePath(path ...string) (interface{}, error)
• type EventTime
  • func (tm EventTime) MarshalJSON() ([]byte, error)
  • func (tm EventTime) MarshalMsgpack() ([]byte, error)
  • func (tm *EventTime) UnmarshalMsgpack(b []byte) error
• type Helo
• type HeloOptions
• type Message
  • func (msg *Message) DecodeMsgpack(decoder *msgpack.Decoder) error
• type MessageMode
• type Ping
• type Pong
• type TransportOption

Constants

const CompressionFormat = "gzip"

CompressionFormat defines the compression format, only "gzip" is supported

Functions

func DoClientHandshake

func DoClientHandshake(conn net.Conn, sharedKey string, timeout time.Duration) (bool, string, error)

DoClientHandshake performs client-side handshake on the given forward protocol connection.

Returns (success?, failure reason, network error)

func DoServerHandshake

func DoServerHandshake(conn net.Conn, sharedKey string, timeout time.Duration, auth AuthCallback) (bool, error)

DoServerHandshake performs server-side handshake on the given forward protocol connection.

Returns (success?, network error)

Types

type Ack

type Ack struct {
	Ack string `msgpack:"ack"` // equals to ForwardTransportOption.Chunk
}

Ack is the acknowledgement or response from server to client for receiving a chunk

type AuthCallback

type AuthCallback func(hostname, username, password string) (bool, string)

AuthCallback is the definition of callback for (test) server to authenticate a client

Returns (success?, reason)

type EventEntry

type EventEntry struct {
	Time   EventTime              `msgpack:"time"`
	Record map[string]interface{} `msgpack:"record"`
	// contains filtered or unexported fields
}

EventEntry represents a single log record in forward messages

The struct is not used directly for encoding but serves as a reference

func (*EventEntry) ResolvePath

func (e *EventEntry) ResolvePath(path ...string) (interface{}, error)

ResolvePath attempts to fetch field by the given path, represented as one or more map keys

Supports nested maps - the type should be map[string]interface{}

type EventTime

type EventTime struct {
	time.Time
}

EventTime represents the custom timestamp type used by Fluentd

func (EventTime) MarshalJSON

func (tm EventTime) MarshalJSON() ([]byte, error)

MarshalJSON defines custom JSON marshaling for log record to match its msgpack format (the simplest Forward mode)

func (EventTime) MarshalMsgpack

func (tm EventTime) MarshalMsgpack() ([]byte, error)

MarshalMsgpack encodes EventTime in msgpack format

func (*EventTime) UnmarshalMsgpack

func (tm *EventTime) UnmarshalMsgpack(b []byte) error

UnmarshalMsgpack decodes EventTime from msgpack bytes

type Helo

type Helo struct {
	Type    string      `msgpack:"type"`
	Options HeloOptions `msgpack:"options"`
	// contains filtered or unexported fields
}

Helo is the HELO message from server to client during forward protocol handshake step 1

type HeloOptions

type HeloOptions struct {
	Nonce     string `msgpack:"nonce"`
	Auth      string `msgpack:"auth"`
	KeepAlive bool   `msgpack:"keepalive"`
}

HeloOptions is a map of options returned from fluent server

type Message

type Message struct {
	Tag     string          `msgpack:"tag"`
	Entries []EventEntry    `msgpack:"entries"` // Depending on MessageMode, the entries may be serialized as-is or in other formats
	Option  TransportOption `msgpack:"option"`
	// contains filtered or unexported fields
}

Message represents a request to forward a batch of log events to Fluentd

The struct is not used directly for encoding but serves as a reference

func (*Message) DecodeMsgpack

func (msg *Message) DecodeMsgpack(decoder *msgpack.Decoder) error

DecodeMsgpack is the custom msgpack decoding implementation for Message, in order to decode Entries properly

See MessageMode for different types of Entries encoding

type MessageMode

type MessageMode string

MessageMode determines the format in which Message.Entries are serialized The mode is to be detected by upstream, not itself specified during communication

const (
	// ModeForward serializes logs as a msgpack array, the original and fluent-bit compatible format
	ModeForward MessageMode = "Forward"

	// ModePackedForward packs serialized logs as a msgpack binary (double msgpack)
	ModePackedForward MessageMode = "PackedForward"

	// ModeCompressedPackedForward packs gzipped and serialized logs as a msgpack binary (double msgpack)
	// In production this should always be used because the saving of space and network bandwidth is 20-50x
	ModeCompressedPackedForward MessageMode = "CompressedPackedForward"
)

type Ping

type Ping struct {
	Type               string `msgpack:"type"`
	ClientHostname     string `msgpack:"client_hostname"`
	SharedKeySalt      string `msgpack:"shared_key_salt"`
	SharedKeyHexdigest string `msgpack:"shared_key_hexdigest"`
	Username           string `msgpack:"username"`
	Password           string `msgpack:"password"`
	// contains filtered or unexported fields
}

Ping is the PING message from client to server during forward protocol handshake step 2

type Pong

type Pong struct {
	Type               string `msgpack:"type"`
	AuthResult         bool   `msgpack:"auth_result"`
	Reason             string `msgpack:"reason"`
	ServerHostname     string `msgpack:"server_hostname"`
	SharedKeyHexdigest string `msgpack:"shared_key_hexdigest"`
	// contains filtered or unexported fields
}

Pong is the PONG message from server to client during forward protocol handshake step 3

type TransportOption

type TransportOption struct {
	Size       int    `msgpack:"size" json:"size"`             // The numbers of log records in this msg
	Chunk      string `msgpack:"chunk" json:"chunk"`           // Chunk ID, omitted if a response from server as ACK is not needed
	Compressed string `msgpack:"compressed" json:"compressed"` // set to ForwardCompressionFormat for "CompressedPackedForward" mode
	// contains filtered or unexported fields
}

TransportOption is the option of each transport request (last value of array)