firetap

README

firetap

firetap is an AWS Lambda extension to transport logs to Kinesis Data Firehose or Kinesis Data Streams.

This is an alpha version and not recommended for production use.

Architecture

Run firetap as an AWS Lambda extension.

• Runs HTTP server that receive telemetries using Lambda Telemetry API.
• Sends function logs to Kinesis Data Firehose / Streams.

Limitations

The execution of the Lambda function is stopped after it returns the response to the invoker. The telemetry API is called asynchronously, so all logs may not be sent to the Kinesis Data Firehose / Streams if the function is stopped before they are sent.

firetap waits for the SHUTDOWN event to be called before terminating the Lambda function. When the SHUTDOWN event is received, if the remaining logs exist, they are sent to the Kinesis Data Firehose / Streams.

Normally, the SHUTDOWN event is sent after about 5 minutes of the function being idle. So, the logs delayed by up to 5 minutes may be sent. (However, the AWS Lambda specification does not guarantee the delay time.)

Usage

Lambda Extension

firetap works as a Lambda extension.

To deploy firetap as a Lambda extension, you need to create a Lambda layer that contains firetap binary in extensions/ directory.

$ mkdir extensions
$ cp /path/to/firetap extensions/firetap
$ zip -r layer.zip extensions
$ aws lambda publish-layer-version \
		--layer-name firetap \
		--zip-file fileb://layer.zip \
		--compatible-runtimes provided.al2023 provided.al2

Configurations

You can configure firetap by setting environment variables.

• FIRETAP_STREAM_NAME: The name of the Kinesis Data Firehose or Kinesis Data Streams stream.
• FIRETAP_USE_DATA_STREAM: Set true if you want to use Kinesis Data Streams. Default is false (use Firehose).

LICENSE

MIT

Author

Fujiwara Shunichiro

Documentation

Index

• func Fatal(err error)
• func LogType(name string)
• func Run(ctx context.Context, opt *Option) error
• type ExtensionClient
  • func NewExtensionClient() *ExtensionClient
  • func (c *ExtensionClient) Register(ctx context.Context) error
  • func (c *ExtensionClient) Run(ctx context.Context, cancel func()) error
  • func (c *ExtensionClient) SubscribeTelemetry(ctx context.Context, endpoint string) error
• type LogSender
  • func NewSender(ctx context.Context, streamName string, dataStream bool) (*LogSender, error)
  • func (s *LogSender) Flush(ctx context.Context) error
  • func (s *LogSender) Send(ctx context.Context, msg []byte) error
• type Option
  • func NewOption() (*Option, error)
• type Receiver
  • func (r *Receiver) Run(ctx context.Context, sender *LogSender) error
• type TelemetryBuffering
• type TelemetryDestination
• type TelemetryEvent
• type TelemetrySubscription
  • func NewTelemetrySubscription(endpoint string) *TelemetrySubscription

Functions

func Fatal

func Fatal(err error)

func LogType

func LogType(name string)

func Run

func Run(ctx context.Context, opt *Option) error

Types

type ExtensionClient

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

func NewExtensionClient

func NewExtensionClient() *ExtensionClient

func (*ExtensionClient) Register

func (c *ExtensionClient) Register(ctx context.Context) error

func (*ExtensionClient) Run

func (c *ExtensionClient) Run(ctx context.Context, cancel func()) error

func (*ExtensionClient) SubscribeTelemetry

func (c *ExtensionClient) SubscribeTelemetry(ctx context.Context, endpoint string) error

type LogSender

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

func NewSender

func NewSender(ctx context.Context, streamName string, dataStream bool) (*LogSender, error)

func (*LogSender) Flush

func (s *LogSender) Flush(ctx context.Context) error

func (*LogSender) Send

func (s *LogSender) Send(ctx context.Context, msg []byte) error

type Option

type Option struct {
	StreamName string // StreamName is the name of the Kinesis Firehose/DataStream stream name.
	DataStream bool   // DataStream is the flag to use DataStream instead of Firehose.
}

func NewOption

func NewOption() (*Option, error)

type Receiver

type Receiver struct {
	Endpoint string
	// contains filtered or unexported fields
}

func (*Receiver) Run

func (r *Receiver) Run(ctx context.Context, sender *LogSender) error

type TelemetryBuffering

type TelemetryBuffering struct {
	MaxItems  int `json:"maxItems"`
	MaxBytes  int `json:"maxBytes"`
	TimeoutMs int `json:"timeoutMs"`
}

type TelemetryDestination

type TelemetryDestination struct {
	Protocol string `json:"protocol"`
	URI      string `json:"URI"`
}

type TelemetryEvent

type TelemetryEvent struct {
	Time   string           `json:"time"`
	Type   string           `json:"type"`
	Record *json.RawMessage `json:"record"`
}

type TelemetrySubscription

type TelemetrySubscription struct {
	SchemaVersion string               `json:"schemaVersion"`
	Types         []string             `json:"types"`
	Buffering     TelemetryBuffering   `json:"buffering"`
	Destination   TelemetryDestination `json:"destination"`
}

func NewTelemetrySubscription

func NewTelemetrySubscription(endpoint string) *TelemetrySubscription