apmproxy

Documentation

Index

• Variables
• func GetUncompressedBytes(rawBytes []byte, encodingType string) ([]byte, error)
• func ProcessMetadata(data APMData) ([]byte, error)
• type APMData
• type BatchData
  • func NewBatch(maxSize int, maxAge time.Duration, metadata []byte) *BatchData
  • func (b *BatchData) Add(d []byte) error
  • func (b *BatchData) Count() int
  • func (b *BatchData) Reset()
  • func (b *BatchData) ShouldShip() bool
  • func (b *BatchData) ToAPMData() APMData
• type Client
  • func NewClient(opts ...Option) (*Client, error)
  • func (c *Client) ComputeGracePeriod() time.Duration
  • func (c *Client) FlushAPMData(ctx context.Context)
  • func (c *Client) ForwardApmData(ctx context.Context) error
  • func (c *Client) IsUnhealthy() bool
  • func (c *Client) PostToApmServer(ctx context.Context, apmData APMData) error
  • func (c *Client) ResetFlush()
  • func (c *Client) ShouldFlush() bool
  • func (c *Client) Shutdown() error
  • func (c *Client) StartReceiver() error
  • func (c *Client) UpdateStatus(ctx context.Context, status Status)
  • func (c *Client) WaitForFlush() <-chan struct{}
• type Option
  • func WithAPIKey(key string) Option
  • func WithAgentDataBufferSize(size int) Option
  • func WithDataForwarderTimeout(timeout time.Duration) Option
  • func WithLogger(logger *zap.SugaredLogger) Option
  • func WithMaxBatchAge(age time.Duration) Option
  • func WithMaxBatchSize(size int) Option
  • func WithReceiverAddress(addr string) Option
  • func WithReceiverTimeout(timeout time.Duration) Option
  • func WithSecretToken(secret string) Option
  • func WithSendStrategy(strategy SendStrategy) Option
  • func WithURL(url string) Option
• type SendStrategy
• type Status

Variables

var (
	// ErrBatchFull signfies that the batch has reached full capacity
	// and cannot accept more entries.
	ErrBatchFull = errors.New("batch is full")
	// ErrInvalidEncoding is returned for any APMData that is encoded
	// with any encoding format
	ErrInvalidEncoding = errors.New("encoded data not supported")
)

Functions

func GetUncompressedBytes

func GetUncompressedBytes(rawBytes []byte, encodingType string) ([]byte, error)

func ProcessMetadata

func ProcessMetadata(data APMData) ([]byte, error)

ProcessMetadata return a byte array containing the Metadata marshaled in JSON In case we want to update the Metadata values, usage of https://github.com/tidwall/sjson is advised

Types

type APMData

type APMData struct {
	Data            []byte
	ContentEncoding string
}

APMData represents data to be sent to APMServer. `Agent` type data will have `metadata` as ndjson whereas `lambda` type data will be without metadata.

type BatchData

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

BatchData represents a batch of data without metadata that will be sent to APMServer. BatchData is not safe concurrent access.

func NewBatch

func NewBatch(maxSize int, maxAge time.Duration, metadata []byte) *BatchData

NewBatch creates a new BatchData which can accept a maximum number of entries as specified by the argument

func (*BatchData) Add

func (b *BatchData) Add(d []byte) error

Add adds a new entry to the batch. Returns ErrBatchFull if batch has reached its maximum size.

func (*BatchData) Count

func (b *BatchData) Count() int

Count return the number of APMData entries in batch.

func (*BatchData) Reset

func (b *BatchData) Reset()

Reset resets the batch to prepare for new set of data

func (*BatchData) ShouldShip

func (b *BatchData) ShouldShip() bool

ShouldShip indicates when a batch is ready for sending. A batch is marked as ready for flush when one of the below conditions is reached: 1. max size is greater than threshold (90% of maxSize) 2. batch is older than maturity age

func (*BatchData) ToAPMData

func (b *BatchData) ToAPMData() APMData

ToAPMData returns APMData with metadata and the accumulated batch

type Client

type Client struct {
	AgentDataChannel  chan APMData
	LambdaDataChannel chan []byte

	Status            Status
	ReconnectionCount int
	ServerAPIKey      string
	ServerSecretToken string
	// contains filtered or unexported fields
}

Client is the client used to communicate with the apm server.

func NewClient

func NewClient(opts ...Option) (*Client, error)

func (*Client) ComputeGracePeriod

func (c *Client) ComputeGracePeriod() time.Duration

ComputeGracePeriod https://github.com/elastic/apm/blob/main/specs/agents/transport.md#transport-errors

func (*Client) FlushAPMData

func (c *Client) FlushAPMData(ctx context.Context)

FlushAPMData reads all the apm data in the apm data channel and sends it to the APM server.

func (*Client) ForwardApmData

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

ForwardApmData receives apm data as it comes in and posts it to the APM server. Stop checking for, and sending apm data when the function invocation has completed, signaled via a channel.

func (*Client) IsUnhealthy

func (c *Client) IsUnhealthy() bool

IsUnhealthy returns true if the apmproxy is not healthy.

func (*Client) PostToApmServer

func (c *Client) PostToApmServer(ctx context.Context, apmData APMData) error

PostToApmServer takes a chunk of APM agent data and posts it to the APM server.

The function compresses the APM agent data, if it's not already compressed. It sets the APM transport status to failing upon errors, as part of the backoff strategy.

func (*Client) ResetFlush

func (c *Client) ResetFlush()

ResetFlush resets the client's "agent flushed" state, such that subsequent calls to WaitForFlush will block until another request is received from the agent indicating it has flushed.

func (*Client) ShouldFlush

func (c *Client) ShouldFlush() bool

ShouldFlush returns true if the client should flush APM data after processing the event.

func (*Client) Shutdown

func (c *Client) Shutdown() error

Shutdown shutdowns the apm receiver gracefully.

func (*Client) StartReceiver

func (c *Client) StartReceiver() error

StartReceiver starts the server listening for APM agent data.

func (*Client) UpdateStatus

func (c *Client) UpdateStatus(ctx context.Context, status Status)

UpdateStatus takes a state of the APM server transport and updates the current state of the transport. For a change to a failing state, the grace period is calculated and a go routine is started that waits for that period to complete before changing the status to "pending". This would allow a subsequent send attempt to the APM server.

This function is public for use in tests.

func (*Client) WaitForFlush

func (c *Client) WaitForFlush() <-chan struct{}

WaitForFlush returns a channel that is closed when the agent has signalled that the Lambda invocation has completed, and there is no more APM data coming.

type Option

type Option func(*Client)

func WithAPIKey

func WithAPIKey(key string) Option

func WithAgentDataBufferSize

func WithAgentDataBufferSize(size int) Option

WithAgentDataBufferSize sets the agent data buffer size.

func WithDataForwarderTimeout

func WithDataForwarderTimeout(timeout time.Duration) Option

func WithLogger

func WithLogger(logger *zap.SugaredLogger) Option

WithLogger configures a custom zap logger to be used by the client.

func WithMaxBatchAge

func WithMaxBatchAge(age time.Duration) Option

WithMaxBatchAge configures the maximum age of the batch before it is sent to APMServer. Age is measured from the time the first entry is added in the batch.

It is possible for batch age to be greater than the configured max batch age when sending since a send is triggered by a new log event and log events can be delayed due to various factors.

func WithMaxBatchSize

func WithMaxBatchSize(size int) Option

WithMaxBatchSize configures the maximum batch size for the payload sent to the APMServer

func WithReceiverAddress

func WithReceiverAddress(addr string) Option

WithReceiverAddress sets the receiver address.

func WithReceiverTimeout

func WithReceiverTimeout(timeout time.Duration) Option

WithReceiverTimeout sets the timeout receiver.

func WithSecretToken

func WithSecretToken(secret string) Option

func WithSendStrategy

func WithSendStrategy(strategy SendStrategy) Option

WithSendStrategy sets the sendstrategy.

func WithURL

func WithURL(url string) Option

type SendStrategy

type SendStrategy string

SendStrategy represents the type of sending strategy the extension uses

const (
	// Background send strategy allows the extension to send remaining buffered
	// agent data on the next function invocation
	Background SendStrategy = "background"

	// SyncFlush send strategy indicates that the extension will synchronously
	// flush remaining buffered agent data when it receives a signal that the
	// function is complete
	SyncFlush SendStrategy = "syncflush"
)

type Status

type Status string

Constants for the state of the transport used in the backoff implementation.

const (
	// The apmproxy started but no information can be
	// inferred on the status of the transport.
	// Either because the apmproxy just started and no
	// request was forwarded or because it recovered
	// from a failure.
	Started Status = "Started"

	// Last request completed successfully.
	Healthy Status = "Healthy"

	// Last request failed.
	Failing Status = "Failing"

	// The APM Server returned status 429 and the extension
	// was ratelimited.
	RateLimited Status = "RateLimited"

	// A failure on the client was observed. This does not
	// trigger any backoff mechanism.
	ClientFailing Status = "ClientFailing"
)