firehose

package
v1.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 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-23 (2b2086c9)

Coverage

Metric Value
Operations audited 12 (12 ok)
Feature families 2 (2 ok)
Known gaps 4
Deferred items 5
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 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.

  • Iceberg and Snowflake destinations (new this pass) land processed records into their required S3Configuration staging bucket rather than driving a real Apache Iceberg/Glue Data Catalog commit or a real Snowflake Snowpipe Streaming ingest — this backend has no Iceberg-table or Snowflake-account backend to connect to. Wire shape for CreateDeliveryStream/UpdateDestination/DescribeDeliveryStream is fully field-diffed and correct (including CatalogConfiguration, DestinationTableConfigurationList, SchemaEvolutionConfiguration, TableCreationConfiguration for Iceberg, and SecretsManagerConfiguration/SnowflakeRoleConfiguration/SnowflakeVpcConfiguration for Snowflake); only the data-movement mechanics diverge, same documented-simplification pattern as the existing Redshift gap above. Deferred — no bd id filed yet.

  • Legacy Elasticsearch (ElasticsearchDestinationConfiguration, new this pass) and the pre-existing Amazonopensearchservice family both omit VpcConfiguration/ VpcConfigurationDescription (private-VPC ENI delivery) and DocumentIdOptions (Firehose-generated vs. OpenSearch-generated document IDs) — both are real, optional SDK fields on those destination types that are not modeled. Newly identified this pass; not a regression (OpenSearch was previously marked ok without this having been field-diffed). Deferred — low-traffic advanced configuration, no bd id filed yet.

  • AmazonOpenSearchServerlessDestinationConfiguration (a real, distinct 11th destination type in the SDK, separate from Amazonopensearchservice) is not implemented. Not in this pass's explicit destination scope; newly identified, deferred.

Deferred
  • Redshift real S3-staging + COPY delivery mechanics (see gaps)
  • Iceberg/Snowflake real catalog-commit / Snowpipe-Streaming ingest mechanics (see gaps)
  • Elasticsearch/OpenSearch VpcConfiguration and DocumentIdOptions fields (see gaps)
  • AmazonOpenSearchServerlessDestinationConfiguration destination family (see gaps)
  • MSK source ingestion path (present via SourceDescription wire shape, CreateDeliveryStream/ DescribeDeliveryStream round-trip correctly). Real polling/ingestion is genuinely unimplemented: unlike KinesisStreamAsSource (wired via the KinesisReader interface, set by cli.go's service-wiring step), there is no MSK/Kafka backend wiring — adding one would require a new KafkaReader-style interface plus cli.go changes to wire services/kafka's backend in, and this pass's instructions explicitly forbid editing cli.go. Left exactly as found; not reclassified to ok.

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 CatalogConfiguration added in v1.2.0 

type CatalogConfiguration struct {
	CatalogARN        string `json:"CatalogARN,omitempty"`
	WarehouseLocation string `json:"WarehouseLocation,omitempty"`
}

CatalogConfiguration describes where destination Apache Iceberg tables are persisted.

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
	ElasticsearchDestination *ElasticsearchDestinationDescription
	SplunkDestination        *SplunkDestinationDescription
	IcebergDestination       *IcebergDestinationDescription
	SnowflakeDestination     *SnowflakeDestinationDescription
	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"`
	ElasticsearchDestination *ElasticsearchDestinationDescription `json:"elasticsearchDestination,omitempty"`
	SplunkDestination        *SplunkDestinationDescription        `json:"splunkDestination,omitempty"`
	SnowflakeDestination     *SnowflakeDestinationDescription     `json:"snowflakeDestination,omitempty"`
	IcebergDestination       *IcebergDestinationDescription       `json:"icebergDestination,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 DestinationTableConfiguration added in v1.2.0 

type DestinationTableConfiguration struct {
	PartitionSpec           *PartitionSpec `json:"PartitionSpec,omitempty"`
	DestinationDatabaseName string         `json:"DestinationDatabaseName"`
	DestinationTableName    string         `json:"DestinationTableName"`
	S3ErrorOutputPrefix     string         `json:"S3ErrorOutputPrefix,omitempty"`
	UniqueKeys              []string       `json:"UniqueKeys,omitempty"`
}

DestinationTableConfiguration configures delivery to a single Apache Iceberg table.

type DynamicPartitioningConfiguration 

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

DynamicPartitioningConfiguration controls dynamic partitioning.

type ElasticsearchDestinationDescription added in v1.2.0 

type ElasticsearchDestinationDescription 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"`
}

ElasticsearchDestinationDescription holds a legacy (pre-OpenSearch-rename) Elasticsearch destination config. AWS still exposes this as a distinct API shape (ElasticsearchDestinationConfiguration/-Update/-Description) alongside the newer AmazonopensearchserviceDestinationConfiguration family; the two are wire-distinct even though the field sets are nearly identical.

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 IcebergDestinationDescription added in v1.2.0 

type IcebergDestinationDescription struct {
	SchemaEvolutionConfiguration      *SchemaEvolutionConfiguration   `json:"SchemaEvolutionConfiguration,omitempty"`
	CatalogConfiguration              *CatalogConfiguration           `json:"CatalogConfiguration,omitempty"`
	CloudWatchLoggingOptions          *CloudWatchLoggingOptions       `json:"CloudWatchLoggingOptions,omitempty"`
	ProcessingConfiguration           *ProcessingConfiguration        `json:"ProcessingConfiguration,omitempty"`
	RetryOptions                      *RetryOptions                   `json:"RetryOptions,omitempty"`
	S3Destination                     *S3DestinationDescription       `json:"S3DestinationDescription,omitempty"`
	BufferingHints                    *BufferingHints                 `json:"BufferingHints,omitempty"`
	TableCreationConfiguration        *TableCreationConfiguration     `json:"TableCreationConfiguration,omitempty"`
	RoleARN                           string                          `json:"RoleARN,omitempty"`
	S3BackupMode                      string                          `json:"S3BackupMode,omitempty"`
	DestinationID                     string                          `json:"DestinationId,omitempty"`
	DestinationTableConfigurationList []DestinationTableConfiguration `json:"DestinationTableConfigurationList,omitempty"`
	AppendOnly                        bool                            `json:"AppendOnly,omitempty"`
}

IcebergDestinationDescription holds an Apache Iceberg Tables destination config.

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) IsStreamEncrypted added in v1.2.0 

func (b *InMemoryBackend) IsStreamEncrypted(ctx context.Context, streamName string) bool

IsStreamEncrypted reports whether server-side encryption is currently enabled on the named delivery stream. Used to populate the optional Encrypted field on PutRecord/ PutRecordBatch responses without changing those methods' established signatures (an external adapter in cli.go forwards to InMemoryBackend.PutRecordBatch directly and depends on its existing (int, error) return shape).

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) TaggedResources added in v1.2.0 

func (b *InMemoryBackend) TaggedResources() []TaggedEntry

TaggedResources returns every delivery stream ARN, across all regions, that currently has at least one tag. b.streams is a single store.Table keyed by "region|name" (see regionKey), so All() already spans every region without a per-region loop.

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 PartitionField added in v1.2.0 

type PartitionField struct {
	SourceName string `json:"SourceName"`
}

PartitionField is a single identity-transform partition column for an Iceberg table.

type PartitionSpec added in v1.2.0 

type PartitionSpec struct {
	Identity []PartitionField `json:"Identity,omitempty"`
}

PartitionSpec holds the partition-spec configuration used by automatic table creation.

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 SchemaEvolutionConfiguration added in v1.2.0 

type SchemaEvolutionConfiguration struct {
	Enabled bool `json:"Enabled"`
}

SchemaEvolutionConfiguration toggles automatic schema evolution for Iceberg delivery.

type SecretsManagerConfiguration added in v1.2.0 

type SecretsManagerConfiguration struct {
	RoleARN   string `json:"RoleARN,omitempty"`
	SecretARN string `json:"SecretARN,omitempty"`
	Enabled   bool   `json:"Enabled"`
}

SecretsManagerConfiguration describes how Firehose accesses secrets for a destination.

type Serializer 

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

Serializer selects one of the supported columnar serializers.

type SnowflakeBufferingHints added in v1.2.0 

type SnowflakeBufferingHints struct {
	IntervalInSeconds int `json:"IntervalInSeconds,omitempty"`
	SizeInMBs         int `json:"SizeInMBs,omitempty"`
}

SnowflakeBufferingHints controls when buffered records are delivered to Snowflake.

type SnowflakeDestinationDescription added in v1.2.0 

type SnowflakeDestinationDescription struct {
	BufferingHints           *SnowflakeBufferingHints  `json:"BufferingHints,omitempty"`
	CloudWatchLoggingOptions *CloudWatchLoggingOptions `json:"CloudWatchLoggingOptions,omitempty"`
	ProcessingConfiguration  *ProcessingConfiguration  `json:"ProcessingConfiguration,omitempty"`
	RetryOptions             *SnowflakeRetryOptions    `json:"RetryOptions,omitempty"`
	// S3Destination is the required S3 location Snowflake delivery stages through
	// (wire field "S3DestinationDescription").
	S3Destination               *S3DestinationDescription    `json:"S3DestinationDescription,omitempty"`
	SecretsManagerConfiguration *SecretsManagerConfiguration `json:"SecretsManagerConfiguration,omitempty"`
	SnowflakeRoleConfiguration  *SnowflakeRoleConfiguration  `json:"SnowflakeRoleConfiguration,omitempty"`
	SnowflakeVpcConfiguration   *SnowflakeVpcConfiguration   `json:"SnowflakeVpcConfiguration,omitempty"`
	AccountURL                  string                       `json:"AccountUrl,omitempty"`
	ContentColumnName           string                       `json:"ContentColumnName,omitempty"`
	DataLoadingOption           string                       `json:"DataLoadingOption,omitempty"`
	Database                    string                       `json:"Database,omitempty"`
	MetaDataColumnName          string                       `json:"MetaDataColumnName,omitempty"`
	RoleARN                     string                       `json:"RoleARN,omitempty"`
	S3BackupMode                string                       `json:"S3BackupMode,omitempty"`
	Schema                      string                       `json:"Schema,omitempty"`
	Table                       string                       `json:"Table,omitempty"`
	User                        string                       `json:"User,omitempty"`
	DestinationID               string                       `json:"DestinationId,omitempty"`
}

SnowflakeDestinationDescription holds a Snowflake destination config.

type SnowflakeRetryOptions added in v1.2.0 

type SnowflakeRetryOptions struct {
	DurationInSeconds int `json:"DurationInSeconds,omitempty"`
}

SnowflakeRetryOptions holds a retry duration for Snowflake delivery.

type SnowflakeRoleConfiguration added in v1.2.0 

type SnowflakeRoleConfiguration struct {
	SnowflakeRole string `json:"SnowflakeRole,omitempty"`
	Enabled       bool   `json:"Enabled"`
}

SnowflakeRoleConfiguration optionally configures a Snowflake role.

type SnowflakeVpcConfiguration added in v1.2.0 

type SnowflakeVpcConfiguration struct {
	PrivateLinkVpceID string `json:"PrivateLinkVpceId"`
}

SnowflakeVpcConfiguration holds the PrivateLink VPCE ID for private Snowflake connectivity.

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)
	// IsStreamEncrypted reports whether server-side encryption is currently enabled on the
	// named stream, used to populate PutRecord/PutRecordBatch's optional Encrypted field.
	IsStreamEncrypted(ctx context.Context, streamName string) bool
	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 TableCreationConfiguration added in v1.2.0 

type TableCreationConfiguration struct {
	Enabled bool `json:"Enabled"`
}

TableCreationConfiguration toggles automatic table creation for Iceberg delivery.

type TaggedEntry added in v1.2.0 

type TaggedEntry struct {
	Tags map[string]string
	ARN  string
}

TaggedEntry pairs a resource ARN with its tag map, for cross-service tag enumeration by the Resource Groups Tagging API (see cli.go's wireTaggingFirehose).

type UpdateDestinationInput 

type UpdateDestinationInput struct {
	S3Destination            *S3DestinationDescription
	HTTPEndpointDestination  *HTTPEndpointDestinationDescription
	RedshiftDestination      *RedshiftDestinationDescription
	OpenSearchDestination    *OpenSearchDestinationDescription
	ElasticsearchDestination *ElasticsearchDestinationDescription
	SplunkDestination        *SplunkDestinationDescription
	IcebergDestination       *IcebergDestinationDescription
	SnowflakeDestination     *SnowflakeDestinationDescription
}

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.