firehose

package
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 19, 2026 License: MIT Imports: 31 Imported by: 0

README

Kinesis Data Firehose

Parity grade: A · SDK aws-sdk-go-v2/service/firehose@v1.42.11 · last audited 2026-07-11 (2b2086c9)

Coverage
Operations audited 12 (12 ok)
Feature families 2 (2 ok)
Known gaps 3
Deferred items 3
Resource leaks clean
Known gaps

  • Redshift destination does not model AWS's actual two-hop delivery (Firehose stages records to the S3Configuration bucket, then issues a COPY command referencing CopyCommand against that staged data). This backend instead executes a synthesized INSERT statement directly via the Redshift Data API. Wire shape for CreateDeliveryStream/ UpdateDestination/DescribeDeliveryStream is now correct (S3Configuration and CopyCommand round-trip accurately), but the actual data-movement mechanics diverge behaviorally. Deferred — larger rework than a wire-shape fix, no bd id filed yet.

  • PutRecord/PutRecordBatch responses omit the optional Encrypted boolean field that real AWS returns. Non-breaking (SDK treats it as optional/pointer), noted for completeness only.

  • CreateDeliveryStream does not validate that at most one destination configuration is supplied per call (UpdateDestination does enforce "exactly one" via applyDestinationUpdate, but Create has no equivalent check). Real AWS rejects a CreateDeliveryStream request naming more than one destination type. Low traffic path; deferred.

Deferred
  • Redshift real S3-staging + COPY delivery mechanics (see gaps)
  • CreateDeliveryStream multi-destination input validation (see gaps)
  • MSK source ingestion path (present via SourceDescription wire shape; poller behavior not re-verified this pass beyond existing kinesis_source_test.go coverage)

More

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotFound is returned when a delivery stream is not found.
	ErrNotFound = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	// ErrAlreadyExists is returned when a delivery stream already exists.
	ErrAlreadyExists = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	// ErrTransformPayload is a sentinel error indicating the Lambda transform
	// payload could not be built. Use [errors.Is] to check for this condition.
	ErrTransformPayload = errors.New("failed to build Lambda transform payload")
	// ErrRecordTooLarge is returned when a record exceeds the 1,000 KB per-record limit.
	ErrRecordTooLarge = awserr.New("InvalidArgumentException", awserr.ErrInvalidParameter)
	// ErrBatchTooLarge is returned when a PutRecordBatch request exceeds the 500-record limit.
	ErrBatchTooLarge = awserr.New("InvalidArgumentException", awserr.ErrInvalidParameter)
	// ErrValidation is returned for invalid input parameters.
	ErrValidation = awserr.New("InvalidArgumentException", awserr.ErrInvalidParameter)
)
View Source 
var ErrNilAppContext = errors.New("nil AppContext passed to Firehose Provider.Init")

ErrNilAppContext is returned by Init when a nil AppContext is passed.

Functions

This section is empty.

Types

type BufferingHints 

type BufferingHints struct {
	SizeInMBs         int `json:"SizeInMBs"`
	IntervalInSeconds int `json:"IntervalInSeconds"`
}

BufferingHints controls when buffered records are delivered to S3.

type CloudWatchLoggingOptions 

type CloudWatchLoggingOptions struct {
	LogGroupName  string `json:"LogGroupName,omitempty"`
	LogStreamName string `json:"LogStreamName,omitempty"`
	Enabled       bool   `json:"Enabled"`
}

CloudWatchLoggingOptions configures CloudWatch logging for a destination.

type CreateDeliveryStreamInput 

type CreateDeliveryStreamInput struct {
	S3Destination           *S3DestinationDescription
	HTTPEndpointDestination *HTTPEndpointDestinationDescription
	RedshiftDestination     *RedshiftDestinationDescription
	OpenSearchDestination   *OpenSearchDestinationDescription
	SplunkDestination       *SplunkDestinationDescription
	Source                  *SourceDescription
	Name                    string
	DeliveryStreamType      string
}

CreateDeliveryStreamInput holds the input for creating a delivery stream.

type DataFormatConversionConfig 

type DataFormatConversionConfig struct {
	SchemaConfiguration       *SchemaConfiguration       `json:"SchemaConfiguration,omitempty"`
	InputFormatConfiguration  *InputFormatConfiguration  `json:"InputFormatConfiguration,omitempty"`
	OutputFormatConfiguration *OutputFormatConfiguration `json:"OutputFormatConfiguration,omitempty"`
	Enabled                   bool                       `json:"Enabled"`
}

DataFormatConversionConfig controls record-format conversion (JSON → Parquet/ORC). It mirrors the AWS DataFormatConversionConfiguration API shape.

type DeliveryMetrics 

type DeliveryMetrics struct {
	TotalRecords  int64 `json:"TotalRecords"`
	FailedRecords int64 `json:"FailedRecords"`
	TotalBytes    int64 `json:"TotalBytes"`
}

DeliveryMetrics tracks delivery statistics for a stream.

type DeliveryStream 

type DeliveryStream struct {
	CreateTimestamp         time.Time                           `json:"createTimestamp"`
	LastUpdateTimestamp     time.Time                           `json:"lastUpdateTimestamp"`
	Tags                    *tags.Tags                          `json:"tags,omitempty"`
	S3Destination           *S3DestinationDescription           `json:"s3Destination,omitempty"`
	HTTPEndpointDestination *HTTPEndpointDestinationDescription `json:"httpEndpointDestination,omitempty"`
	RedshiftDestination     *RedshiftDestinationDescription     `json:"redshiftDestination,omitempty"`
	OpenSearchDestination   *OpenSearchDestinationDescription   `json:"openSearchDestination,omitempty"`
	SplunkDestination       *SplunkDestinationDescription       `json:"splunkDestination,omitempty"`
	Encryption              *EncryptionConfig                   `json:"encryption,omitempty"`
	Source                  *SourceDescription                  `json:"source,omitempty"`
	DeliveryStreamType      string                              `json:"deliveryStreamType,omitempty"`
	Name                    string                              `json:"name"`
	ARN                     string                              `json:"arn"`
	VersionID               string                              `json:"versionID,omitempty"`
	Status                  string                              `json:"status"`
	AccountID               string                              `json:"accountID"`
	Region                  string                              `json:"region"`
	Records                 [][]byte                            `json:"records,omitempty"`
	BackupRecords           [][]byte                            `json:"backupRecords,omitempty"`
	Metrics                 DeliveryMetrics                     `json:"metrics"`
	// contains filtered or unexported fields
}

DeliveryStream represents a Kinesis Firehose delivery stream.

type Deserializer 

type Deserializer struct {
	OpenXJSONSerDe *OpenXJSONSerDe `json:"OpenXJsonSerDe,omitempty"`
	HiveJSONSerDe  *HiveJSONSerDe  `json:"HiveJsonSerDe,omitempty"`
}

Deserializer selects one of the supported JSON deserializers.

type DynamicPartitioningConfiguration 

type DynamicPartitioningConfiguration struct {
	RetryOptions *RetryOptions `json:"RetryOptions,omitempty"`
	Enabled      bool          `json:"Enabled"`
}

DynamicPartitioningConfiguration controls dynamic partitioning.

type EncryptionConfig 

type EncryptionConfig struct {
	FailureDescription *FailureDescription `json:"FailureDescription,omitempty"`
	KeyARN             string              `json:"KeyARN,omitempty"`
	KeyType            string              `json:"KeyType"`
	Status             string              `json:"Status"`
}

EncryptionConfig holds the effective SSE configuration for a delivery stream.

type EncryptionConfigInput 

type EncryptionConfigInput struct {
	KeyARN  string `json:"KeyARN,omitempty"`
	KeyType string `json:"KeyType"`
}

EncryptionConfigInput holds the optional SSE configuration for a delivery stream.

type FailureDescription 

type FailureDescription struct {
	Details string `json:"Details,omitempty"`
	Type    string `json:"Type,omitempty"`
}

FailureDescription holds error context for SSE failures.

type HTTPEndpointCommonAttribute 

type HTTPEndpointCommonAttribute struct {
	AttributeName  string `json:"AttributeName"`
	AttributeValue string `json:"AttributeValue"`
}

HTTPEndpointCommonAttribute is a key-value attribute sent with HTTP requests.

type HTTPEndpointConfiguration 

type HTTPEndpointConfiguration struct {
	URL       string `json:"Url,omitempty"`
	Name      string `json:"Name,omitempty"`
	AccessKey string `json:"AccessKey,omitempty"`
}

HTTPEndpointConfiguration holds the HTTP endpoint URL and name.

type HTTPEndpointDestinationDescription 

type HTTPEndpointDestinationDescription struct {
	ProcessingConfiguration  *ProcessingConfiguration          `json:"ProcessingConfiguration,omitempty"`
	EndpointConfiguration    *HTTPEndpointConfiguration        `json:"EndpointConfiguration,omitempty"`
	RequestConfiguration     *HTTPEndpointRequestConfiguration `json:"RequestConfiguration,omitempty"`
	BufferingHints           *BufferingHints                   `json:"BufferingHints,omitempty"`
	RetryOptions             *RetryOptions                     `json:"RetryOptions,omitempty"`
	CloudWatchLoggingOptions *CloudWatchLoggingOptions         `json:"CloudWatchLoggingOptions,omitempty"`
	S3BackupMode             string                            `json:"S3BackupMode,omitempty"`
	S3BackupDescription      *S3BackupDescription              `json:"S3BackupDescription,omitempty"`
	DestinationID            string                            `json:"DestinationId,omitempty"`
}

HTTPEndpointDestinationDescription holds the HTTP endpoint destination config.

type HTTPEndpointRequestConfiguration 

type HTTPEndpointRequestConfiguration struct {
	ContentEncoding  string                        `json:"ContentEncoding,omitempty"`
	CommonAttributes []HTTPEndpointCommonAttribute `json:"CommonAttributes,omitempty"`
}

HTTPEndpointRequestConfiguration holds the content-encoding and attributes for HTTP requests.

type Handler 

type Handler struct {
	Backend StorageBackend
	// contains filtered or unexported fields
}

Handler is the Echo HTTP handler for Kinesis Firehose operations.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new Firehose handler.

func (*Handler) ChaosOperations 

func (h *Handler) ChaosOperations() []string

ChaosOperations returns all operations that can be fault-injected.

func (*Handler) ChaosRegions 

func (h *Handler) ChaosRegions() []string

ChaosRegions returns all regions this Firehose instance handles.

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

ChaosServiceName returns the lowercase AWS service name for fault rule matching.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation extracts the Firehose action from the X-Amz-Target header.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource extracts the delivery stream name from the request body.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations returns the list of supported Firehose operations.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler function.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all state in the backend.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher returns a function that matches Firehose requests.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown implements service.Shutdowner. It flushes any buffered records to their destinations before the process exits so that records received since the last interval flush are not lost. If ctx expires before FlushAll returns, Shutdown returns immediately.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker starts the background interval flusher. It implements service.BackgroundWorker.

type HiveJSONSerDe 

type HiveJSONSerDe struct {
	TimestampFormats []string `json:"TimestampFormats,omitempty"`
}

HiveJSONSerDe configures the Hive JSON deserializer.

type InMemoryBackend 

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

InMemoryBackend is the in-memory store for Firehose resources.

func NewInMemoryBackend 

func NewInMemoryBackend(accountID, region string) *InMemoryBackend

NewInMemoryBackend creates a new InMemoryBackend.

func NewInMemoryBackendWithContext 

func NewInMemoryBackendWithContext(svcCtx context.Context, accountID, region string) *InMemoryBackend

NewInMemoryBackendWithContext creates a new InMemoryBackend whose delivery operations are bounded by the provided parent context. Use this in production to ensure in-flight deliveries are cancelled on server shutdown. If svcCtx is nil, context.Background is used.

func (*InMemoryBackend) AddStreamInternal 

func (b *InMemoryBackend) AddStreamInternal(s *DeliveryStream)

AddStreamInternal deep-copies s into the backend, used for seeding test data.

func (*InMemoryBackend) CreateDeliveryStream 

func (b *InMemoryBackend) CreateDeliveryStream(
	ctx context.Context, input CreateDeliveryStreamInput,
) (*DeliveryStream, error)

CreateDeliveryStream creates a new delivery stream.

func (*InMemoryBackend) DeleteDeliveryStream 

func (b *InMemoryBackend) DeleteDeliveryStream(ctx context.Context, name string) error

DeleteDeliveryStream deletes a delivery stream.

func (*InMemoryBackend) DescribeDeliveryStream 

func (b *InMemoryBackend) DescribeDeliveryStream(ctx context.Context, name string) (*DeliveryStream, error)

DescribeDeliveryStream returns a delivery stream by name.

func (*InMemoryBackend) FlushAll 

func (b *InMemoryBackend) FlushAll(ctx context.Context)

FlushAll forces delivery of all buffered records across all streams in all regions. Used by tests and for graceful shutdown.

func (*InMemoryBackend) ListDeliveryStreams 

func (b *InMemoryBackend) ListDeliveryStreams(ctx context.Context) []string

ListDeliveryStreams returns all delivery stream names in the request's region in alphabetical order.

func (*InMemoryBackend) ListDeliveryStreamsByType 

func (b *InMemoryBackend) ListDeliveryStreamsByType(ctx context.Context, streamType string) []string

ListDeliveryStreamsByType returns delivery stream names in the request's region in alphabetical order, optionally filtered to a single DeliveryStreamType (DirectPut or KinesisStreamAsSource). An empty streamType returns all streams. The full sorted-name list is cached per region and reused across calls until a create/delete invalidates it, so repeated listing does not re-sort the whole namespace every time.

func (*InMemoryBackend) ListTagsForDeliveryStream 

func (b *InMemoryBackend) ListTagsForDeliveryStream(ctx context.Context, name string) (map[string]string, error)

ListTagsForDeliveryStream returns tags for a delivery stream.

func (*InMemoryBackend) PutRecord 

func (b *InMemoryBackend) PutRecord(ctx context.Context, streamName string, data []byte) error

PutRecord appends a record to the delivery stream and flushes if buffer threshold is met.

func (*InMemoryBackend) PutRecordBatch 

func (b *InMemoryBackend) PutRecordBatch(ctx context.Context, streamName string, records [][]byte) (int, error)

PutRecordBatch appends multiple records to the delivery stream and flushes if buffer threshold is met.

func (*InMemoryBackend) Region 

func (b *InMemoryBackend) Region() string

Region returns the AWS region this backend is configured for.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all delivery streams, closing their tag registries to prevent leaks.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a JSON snapshot. It implements persistence.Persistable.

func (*InMemoryBackend) RunFlusher 

func (b *InMemoryBackend) RunFlusher(ctx context.Context)

RunFlusher starts the background interval flusher goroutine.

func (*InMemoryBackend) SetKinesisBackend 

func (b *InMemoryBackend) SetKinesisBackend(k KinesisReader)

SetKinesisBackend wires the Kinesis backend for polling KinesisStreamAsSource streams.

func (*InMemoryBackend) SetLambdaBackend 

func (b *InMemoryBackend) SetLambdaBackend(lambda LambdaInvoker)

SetLambdaBackend wires the Lambda backend for record transformation.

func (*InMemoryBackend) SetS3Backend 

func (b *InMemoryBackend) SetS3Backend(s3 S3Storer)

SetS3Backend wires the S3 backend for actual record delivery.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state to JSON. It implements persistence.Persistable.

func (*InMemoryBackend) StartDeliveryStreamEncryption 

func (b *InMemoryBackend) StartDeliveryStreamEncryption(
	ctx context.Context, name string, input *EncryptionConfigInput,
) error

StartDeliveryStreamEncryption enables server-side encryption for a delivery stream. In this in-memory implementation the status transitions directly to ENABLED.

func (*InMemoryBackend) StopDeliveryStreamEncryption 

func (b *InMemoryBackend) StopDeliveryStreamEncryption(ctx context.Context, name string) error

StopDeliveryStreamEncryption disables server-side encryption for a delivery stream. In this in-memory implementation the status transitions directly to DISABLED.

func (*InMemoryBackend) TagDeliveryStream 

func (b *InMemoryBackend) TagDeliveryStream(ctx context.Context, name string, kv map[string]string) error

TagDeliveryStream adds or updates tags on a delivery stream.

func (*InMemoryBackend) UntagDeliveryStream 

func (b *InMemoryBackend) UntagDeliveryStream(ctx context.Context, name string, keys []string) error

UntagDeliveryStream removes tag keys from a delivery stream.

func (*InMemoryBackend) UpdateDestination 

func (b *InMemoryBackend) UpdateDestination(
	ctx context.Context,
	streamName, currentVersionID string,
	input UpdateDestinationInput,
) error

UpdateDestination updates the destination configuration of an existing stream. AWS allows updating exactly one destination type per call.

type InputFormatConfiguration 

type InputFormatConfiguration struct {
	Deserializer *Deserializer `json:"Deserializer,omitempty"`
}

InputFormatConfiguration selects the deserializer applied to incoming records.

type KMSEncryptionConfig 

type KMSEncryptionConfig struct {
	AWSKMSKeyARN string `json:"AWSKMSKeyARN"`
}

KMSEncryptionConfig holds a KMS key ARN for S3 encryption.

type KinesisReader 

type KinesisReader interface {
	// ListShards returns all open shard IDs for the named stream.
	ListShards(streamName string) ([]string, error)
	// GetShardIterator returns a TRIM_HORIZON iterator token for the given stream/shard.
	GetShardIterator(streamName, shardID string) (string, error)
	// GetRecords reads up to limit records. Returns raw data slices, next iterator token, and error.
	GetRecords(shardIterator string, limit int) (records [][]byte, nextIterator string, err error)
}

KinesisReader is the subset of Kinesis operations that Firehose needs to poll source streams.

type KinesisStreamSourceDescription 

type KinesisStreamSourceDescription struct {
	DeliveryStartTimestamp string `json:"DeliveryStartTimestamp,omitempty"`
	KinesisStreamARN       string `json:"KinesisStreamARN,omitempty"`
	RoleARN                string `json:"RoleARN,omitempty"`
}

KinesisStreamSourceDescription describes a Kinesis stream source.

type LambdaInvoker 

type LambdaInvoker interface {
	InvokeFunction(ctx context.Context, name string, invocationType string, payload []byte) ([]byte, int, error)
}

LambdaInvoker is the subset of Lambda operations that Firehose needs for transformation.

type MSKAuthenticationConfiguration 

type MSKAuthenticationConfiguration struct {
	Connectivity string `json:"Connectivity,omitempty"`
	RoleARN      string `json:"RoleARN,omitempty"`
}

MSKAuthenticationConfiguration holds MSK connectivity and role config.

type MSKSourceDescription 

type MSKSourceDescription struct {
	AuthenticationConfiguration *MSKAuthenticationConfiguration `json:"AuthenticationConfiguration,omitempty"`
	MSKClusterARN               string                          `json:"MSKClusterARN,omitempty"`
	TopicName                   string                          `json:"TopicName,omitempty"`
	ReadFromTimestamp           string                          `json:"ReadFromTimestamp,omitempty"`
}

MSKSourceDescription describes an MSK cluster source.

type OpenSearchDestinationDescription 

type OpenSearchDestinationDescription struct {
	ProcessingConfiguration  *ProcessingConfiguration  `json:"ProcessingConfiguration,omitempty"`
	BufferingHints           *BufferingHints           `json:"BufferingHints,omitempty"`
	RetryOptions             *RetryOptions             `json:"RetryOptions,omitempty"`
	CloudWatchLoggingOptions *CloudWatchLoggingOptions `json:"CloudWatchLoggingOptions,omitempty"`
	S3BackupDescription      *S3BackupDescription      `json:"S3BackupDescription,omitempty"`
	DomainARN                string                    `json:"DomainARN,omitempty"`
	ClusterEndpoint          string                    `json:"ClusterEndpoint,omitempty"`
	IndexName                string                    `json:"IndexName,omitempty"`
	TypeName                 string                    `json:"TypeName,omitempty"`
	IndexRotationPeriod      string                    `json:"IndexRotationPeriod,omitempty"`
	S3BackupMode             string                    `json:"S3BackupMode,omitempty"`
	RoleARN                  string                    `json:"RoleARN,omitempty"`
	DestinationID            string                    `json:"DestinationId,omitempty"`
}

OpenSearchDestinationDescription holds an OpenSearch (Elasticsearch) destination config.

type OpenXJSONSerDe 

type OpenXJSONSerDe struct {
	ColumnToJSONKeyMappings            map[string]string `json:"ColumnToJsonKeyMappings,omitempty"`
	ConvertDotsInJSONKeysToUnderscores bool              `json:"ConvertDotsInJsonKeysToUnderscores"`
	CaseInsensitive                    bool              `json:"CaseInsensitive"`
}

OpenXJSONSerDe configures the OpenX JSON deserializer.

type OrcSerDe 

type OrcSerDe struct {
	Compression                         string   `json:"Compression,omitempty"`
	FormatVersion                       string   `json:"FormatVersion,omitempty"`
	BloomFilterColumns                  []string `json:"BloomFilterColumns,omitempty"`
	StripeSizeBytes                     int      `json:"StripeSizeBytes,omitempty"`
	BlockSizeBytes                      int      `json:"BlockSizeBytes,omitempty"`
	RowIndexStride                      int      `json:"RowIndexStride,omitempty"`
	BloomFilterFalsePositiveProbability float64  `json:"BloomFilterFalsePositiveProbability,omitempty"`
	DictionaryKeyThreshold              float64  `json:"DictionaryKeyThreshold,omitempty"`
	EnablePadding                       bool     `json:"EnablePadding"`
}

OrcSerDe configures the ORC serializer.

type OutputFormatConfiguration 

type OutputFormatConfiguration struct {
	Serializer *Serializer `json:"Serializer,omitempty"`
}

OutputFormatConfiguration selects the serializer used to write converted records.

type ParquetSerDe 

type ParquetSerDe struct {
	Compression                 string `json:"Compression,omitempty"`
	WriterVersion               string `json:"WriterVersion,omitempty"`
	BlockSizeBytes              int    `json:"BlockSizeBytes,omitempty"`
	PageSizeBytes               int    `json:"PageSizeBytes,omitempty"`
	MaxPaddingBytes             int    `json:"MaxPaddingBytes,omitempty"`
	EnableDictionaryCompression bool   `json:"EnableDictionaryCompression"`
}

ParquetSerDe configures the Parquet serializer.

type ProcessingConfiguration 

type ProcessingConfiguration struct {
	Processors []Processor `json:"Processors,omitempty"`
	Enabled    bool        `json:"Enabled"`
}

ProcessingConfiguration describes Lambda-based transformation.

type Processor 

type Processor struct {
	Type       string               `json:"Type"`
	Parameters []ProcessorParameter `json:"Parameters,omitempty"`
}

Processor describes a single transformation step.

type ProcessorParameter 

type ProcessorParameter struct {
	ParameterName  string `json:"ParameterName"`
	ParameterValue string `json:"ParameterValue"`
}

ProcessorParameter is a key-value parameter for a processor.

type Provider 

type Provider struct{}

Provider implements service.Provider for Kinesis Firehose.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init initializes the Firehose service backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the provider name.

type RedshiftCopyCommand 

type RedshiftCopyCommand struct {
	DataTableName    string `json:"DataTableName"`
	DataTableColumns string `json:"DataTableColumns,omitempty"`
	CopyOptions      string `json:"CopyOptions,omitempty"`
}

RedshiftCopyCommand holds the Redshift COPY command configuration. On the wire this nests under RedshiftDestinationDescription.CopyCommand (and, on the request side, RedshiftDestinationConfiguration.CopyCommand) rather than as flat fields.

type RedshiftDestinationDescription 

type RedshiftDestinationDescription struct {
	ProcessingConfiguration *ProcessingConfiguration `json:"ProcessingConfiguration,omitempty"`
	RetryOptions            *RetryOptions            `json:"RetryOptions,omitempty"`
	S3BackupDescription     *S3BackupDescription     `json:"S3BackupDescription,omitempty"`
	// S3Destination is the required intermediate S3 staging location that Amazon
	// Redshift's COPY command reads from (wire field "S3DestinationDescription").
	S3Destination  *S3DestinationDescription `json:"S3DestinationDescription,omitempty"`
	CopyCommand    *RedshiftCopyCommand      `json:"CopyCommand,omitempty"`
	ClusterJDBCURL string                    `json:"ClusterJDBCURL,omitempty"`
	Username       string                    `json:"Username,omitempty"`
	RoleARN        string                    `json:"RoleARN,omitempty"`
	S3BackupMode   string                    `json:"S3BackupMode,omitempty"`
	DestinationID  string                    `json:"DestinationId,omitempty"`
}

RedshiftDestinationDescription holds a Redshift destination config.

type RetryOptions 

type RetryOptions struct {
	DurationInSeconds int `json:"DurationInSeconds"`
}

RetryOptions holds a retry duration.

type S3BackupDescription 

type S3BackupDescription struct {
	BufferingHints    *BufferingHints `json:"BufferingHints,omitempty"`
	BucketARN         string          `json:"BucketARN,omitempty"`
	RoleARN           string          `json:"RoleARN,omitempty"`
	Prefix            string          `json:"Prefix,omitempty"`
	CompressionFormat string          `json:"CompressionFormat,omitempty"`
}

S3BackupDescription holds the S3 backup destination configuration.

type S3DestinationDescription 

type S3DestinationDescription struct {
	BufferingHints                   *BufferingHints                   `json:"BufferingHints,omitempty"`
	ProcessingConfiguration          *ProcessingConfiguration          `json:"ProcessingConfiguration,omitempty"`
	S3BackupDescription              *S3BackupDescription              `json:"S3BackupDescription,omitempty"`
	EncryptionConfiguration          *S3EncryptionConfiguration        `json:"EncryptionConfiguration,omitempty"`
	CloudWatchLoggingOptions         *CloudWatchLoggingOptions         `json:"CloudWatchLoggingOptions,omitempty"`
	DynamicPartitioningConfiguration *DynamicPartitioningConfiguration `json:"DynamicPartitioningConfiguration,omitempty"`
	DataFormatConversion             *DataFormatConversionConfig       `json:"DataFormatConversionConfiguration,omitempty"`
	BucketARN                        string                            `json:"BucketARN,omitempty"`
	RoleARN                          string                            `json:"RoleARN,omitempty"`
	Prefix                           string                            `json:"Prefix,omitempty"`
	ErrorOutputPrefix                string                            `json:"ErrorOutputPrefix,omitempty"`
	CompressionFormat                string                            `json:"CompressionFormat,omitempty"`
	FileExtension                    string                            `json:"FileExtension,omitempty"`
	CustomTimeZone                   string                            `json:"CustomTimeZone,omitempty"`
	DestinationID                    string                            `json:"DestinationId,omitempty"`
	S3BackupMode                     string                            `json:"S3BackupMode,omitempty"`
}

S3DestinationDescription holds the effective S3 destination config stored on the stream.

type S3EncryptionConfiguration 

type S3EncryptionConfiguration struct {
	KMSEncryptionConfig *KMSEncryptionConfig `json:"KMSEncryptionConfig,omitempty"`
	NoEncryptionConfig  string               `json:"NoEncryptionConfig,omitempty"`
}

S3EncryptionConfiguration holds the S3 object encryption config.

type S3Storer 

type S3Storer interface {
	PutObject(ctx context.Context, input *sdk_s3.PutObjectInput) (*sdk_s3.PutObjectOutput, error)
}

S3Storer is the subset of S3 operations that Firehose needs to deliver objects.

type SchemaConfiguration 

type SchemaConfiguration struct {
	CatalogID    string `json:"CatalogId,omitempty"`
	DatabaseName string `json:"DatabaseName,omitempty"`
	TableName    string `json:"TableName,omitempty"`
	Region       string `json:"Region,omitempty"`
	RoleARN      string `json:"RoleARN,omitempty"`
	VersionID    string `json:"VersionId,omitempty"`
}

SchemaConfiguration references the Glue table describing the record schema.

type Serializer 

type Serializer struct {
	ParquetSerDe *ParquetSerDe `json:"ParquetSerDe,omitempty"`
	OrcSerDe     *OrcSerDe     `json:"OrcSerDe,omitempty"`
}

Serializer selects one of the supported columnar serializers.

type SourceDescription 

type SourceDescription struct {
	KinesisStreamSourceDescription *KinesisStreamSourceDescription `json:"KinesisStreamSourceDescription,omitempty"`
	MSKSourceDescription           *MSKSourceDescription           `json:"MSKSourceDescription,omitempty"`
}

SourceDescription holds source details for non-DirectPut streams.

type SplunkDestinationDescription 

type SplunkDestinationDescription struct {
	ProcessingConfiguration           *ProcessingConfiguration  `json:"ProcessingConfiguration,omitempty"`
	RetryOptions                      *RetryOptions             `json:"RetryOptions,omitempty"`
	CloudWatchLoggingOptions          *CloudWatchLoggingOptions `json:"CloudWatchLoggingOptions,omitempty"`
	S3BackupDescription               *S3BackupDescription      `json:"S3BackupDescription,omitempty"`
	HECEndpoint                       string                    `json:"HECEndpoint,omitempty"`
	HECEndpointType                   string                    `json:"HECEndpointType,omitempty"`
	HECToken                          string                    `json:"HECToken,omitempty"`
	S3BackupMode                      string                    `json:"S3BackupMode,omitempty"`
	DestinationID                     string                    `json:"DestinationId,omitempty"`
	HECAcknowledgmentTimeoutInSeconds int                       `json:"HECAcknowledgmentTimeoutInSeconds,omitempty"`
}

SplunkDestinationDescription holds a Splunk HEC destination config.

type StorageBackend 

type StorageBackend interface {
	CreateDeliveryStream(ctx context.Context, input CreateDeliveryStreamInput) (*DeliveryStream, error)
	DeleteDeliveryStream(ctx context.Context, name string) error
	DescribeDeliveryStream(ctx context.Context, name string) (*DeliveryStream, error)
	ListDeliveryStreams(ctx context.Context) []string
	ListDeliveryStreamsByType(ctx context.Context, streamType string) []string
	PutRecord(ctx context.Context, streamName string, data []byte) error
	PutRecordBatch(ctx context.Context, streamName string, records [][]byte) (int, error)
	UpdateDestination(ctx context.Context, streamName, currentVersionID string, input UpdateDestinationInput) error
	ListTagsForDeliveryStream(ctx context.Context, name string) (map[string]string, error)
	TagDeliveryStream(ctx context.Context, name string, kv map[string]string) error
	UntagDeliveryStream(ctx context.Context, name string, keys []string) error
	StartDeliveryStreamEncryption(ctx context.Context, name string, input *EncryptionConfigInput) error
	StopDeliveryStreamEncryption(ctx context.Context, name string) error
	Reset()
	Region() string
	RunFlusher(ctx context.Context)
	FlushAll(ctx context.Context)
	Snapshot(ctx context.Context) []byte
	Restore(ctx context.Context, data []byte) error
	AddStreamInternal(s *DeliveryStream)
}

StorageBackend defines the interface for Firehose backend implementations. All mutating methods must be safe for concurrent use.

type UpdateDestinationInput 

type UpdateDestinationInput struct {
	S3Destination           *S3DestinationDescription
	HTTPEndpointDestination *HTTPEndpointDestinationDescription
	RedshiftDestination     *RedshiftDestinationDescription
	OpenSearchDestination   *OpenSearchDestinationDescription
	SplunkDestination       *SplunkDestinationDescription
}

UpdateDestinationInput holds the destination update fields for UpdateDestination. Exactly one destination field should be non-nil.

Source Files

View all Source files

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.