kinesis

package
v1.7.0 Latest Latest
Warning

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

Go to latest
Published: Oct 6, 2026 License: MIT Imports: 47 Imported by: 0

README

Kinesis

Parity grade: A · SDK aws-sdk-go-v2/service/kinesis@v1.53.0 · last audited 2026-09-19 (49cff86c4)

Coverage

Metric Value
PARITY entries audited 39 (39 ok)
Feature families 9 (9 ok)
Known gaps 7
Deferred items 0
Resource leaks clean
Known gaps
  • AWS/Kinesis stream-level metrics are emitted; enhanced shard-level metrics (ShardId dimension), SubscribeToShard.* and the Read/WriteProvisionedThroughputExceeded counts outside the FIS throughput fault are not (no per-shard throughput model). (gopherstack-4m1qr)
  • Channel S3Tables (Iceberg) delivery is unmodeled (no services/s3tables data-file write path): such a channel accepts PutRecord but never buffers or flushes. Plain S3 delivery is real. (gopherstack-s781r)
  • Channel S3-delivery details are inferences, not verified against AWS: object-key suffix placement, delivered byte layout, dead-letter JSON schema/prefix, channel ARN format; OutputKeyTemplate's documented validation rules (length cap, no traversal) are unenforced at Create/UpdateChannel (rules live only in AWS docs, not the SDK). (gopherstack-s781r)
  • Buffered-but-unflushed channel records are not persisted (channelBuffers is in-memory by design); Shutdown/DeleteChannel/DeleteStream flush first, only a crash loses them. (gopherstack-s781r)
  • Channel control-plane 5 TPS throttle (LimitExceededException) and S3Tables PartitionSpec verification are not modeled (no Iceberg partitioning backend).
  • KMSAccessDeniedException (Start/StopStreamEncryption) and AccessDeniedException (UpdateMaxRecordSize/UpdateStreamWarmThroughput) are wire-mapped but have no trigger: no IAM policy engine exists. UpdateMaxRecordSize applies synchronously (SDK docs state no UPDATING transition for it). (gopherstack-ud2, gopherstack-nbg8)
  • DryRun (PutRecord, PutRecords, GetRecords, GetShardIterator, SubscribeToShard) is accepted but not honoured: the SDK doc says only 'Checks if your request will succeed' and the dry-run response shape is undocumented, so a DryRun PutRecord still writes. StreamId is 'Not Implemented. Reserved for future use.' in the SDK.

More

Documentation

Index

Constants

View Source
const (

	// StreamModeProvisioned is the PROVISIONED stream mode.
	StreamModeProvisioned = "PROVISIONED"
	// StreamModeOnDemand is the ON_DEMAND stream mode.
	StreamModeOnDemand = "ON_DEMAND"
)

Variables

View Source
var (
	ErrStreamNotFound      = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	ErrStreamAlreadyExists = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	// ErrStreamHasConsumers is returned by DeleteStream when the stream has
	// registered enhanced fan-out consumers and EnforceConsumerDeletion is
	// unset or false (real DeleteStreamInput.EnforceConsumerDeletion doc
	// comment: "the call to DeleteStream fails with a ResourceInUseException").
	ErrStreamHasConsumers = awserr.New("ResourceInUseException", awserr.ErrConflict)
	// ErrStreamNotActive is returned by control-plane mutations real AWS only
	// accepts against an ACTIVE stream (DeleteStream, MergeShards, SplitShard,
	// UpdateShardCount, StartStreamEncryption, StopStreamEncryption,
	// UpdateStreamMode, UpdateMaxRecordSize, UpdateStreamWarmThroughput) when
	// called while the stream is still CREATING/UPDATING/DELETING.
	ErrStreamNotActive        = awserr.New("ResourceInUseException", awserr.ErrConflict)
	ErrInvalidArgument        = awserr.New("InvalidArgumentException", awserr.ErrInvalidParameter)
	ErrUnknownAction          = errors.New("UnknownOperationException")
	ErrShardIteratorExpired   = errors.New("ExpiredIteratorException")
	ErrConsumerNotFound       = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	ErrConsumerAlreadyExists  = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	ErrResourcePolicyNotFound = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	// ErrChannelNotFound is returned by DescribeChannel/UpdateChannel/
	// DeleteChannel when ChannelARN does not match any channel, and by
	// ListTagsForResource/TagResource/UntagResource for an unrecognized
	// channel ARN.
	ErrChannelNotFound = awserr.New("ResourceNotFoundException", awserr.ErrNotFound)
	// ErrChannelAlreadyExists is returned by CreateChannel when ChannelName
	// collides with an existing channel: "The name is unique within your
	// Amazon Web Services account and Amazon Web Services Region"
	// (api_op_CreateChannel.go doc comment on ChannelName). ResourceInUseException
	// is one of CreateChannel's declared exceptions (deserializers.go
	// deserializeOpErrorCreateChannel).
	ErrChannelAlreadyExists          = awserr.New("ResourceInUseException", awserr.ErrAlreadyExists)
	ErrProvisionedThroughputExceeded = awserr.New(
		"ProvisionedThroughputExceededException",
		errRateExceeded,
	)
	ErrTagLimitExceeded = awserr.New("LimitExceededException", awserr.ErrInvalidParameter)
	ErrLimitExceeded    = awserr.New("LimitExceededException", awserr.ErrInvalidParameter)

	// ErrKMSNotFound indicates the KMS key referenced by StartStreamEncryption's
	// KeyId does not exist. Only reachable when a KMSKeyValidator is wired via
	// WithKMSValidator (see stream_encryption.go); with no validator wired,
	// KeyId is format-checked only, matching pre-existing behavior for
	// deployments that don't wire cross-service KMS validation.
	ErrKMSNotFound = errors.New("KMSNotFoundException")
	// ErrKMSDisabled indicates the KMS key exists but is disabled or pending
	// deletion/import, matching the real KMSDisabledException/
	// KMSInvalidStateException split (see stream_encryption.go for which
	// applies to which key state).
	ErrKMSDisabled = errors.New("KMSDisabledException")
	// ErrKMSInvalidState indicates the KMS key is in a state (e.g. pending
	// deletion, pending import) that doesn't allow use for encryption.
	ErrKMSInvalidState = errors.New("KMSInvalidStateException")
	// ErrKMSAccessDenied is modeled per the real aws-sdk-go-v2/service/kinesis
	// error set for StartStreamEncryption/StopStreamEncryption/
	// UpdateMaxRecordSize (types.KMSAccessDeniedException), but gopherstack has
	// no IAM/key-policy evaluation engine to ever produce it -- defined for
	// wire-shape completeness and left with no trigger path (see PARITY.md gaps).
	ErrKMSAccessDenied = errors.New("KMSAccessDeniedException")
)

Sentinel errors for Kinesis operations.

View Source
var ErrNilAppContext = errors.New("kinesis: AppContext must not be nil")

ErrNilAppContext is returned by Init when the AppContext is nil.

View Source
var ErrShardCountScaling = errors.New(
	"UpdateShardCount cannot scale by more than double or less than half " +
		"of the current shard count within a single call",
)

ErrShardCountScaling indicates that an UpdateShardCount target fell outside the AWS per-call scaling window: within a single call the target shard count may not be more than double or less than half of the current open shard count. AWS surfaces this as ValidationException. The sentinel's message is the error text returned to the client.

View Source
var ErrValidation = errors.New("kinesis: validation error")

ErrValidation is the sentinel error for Kinesis input validation failures.

Functions

func ContextAndNameFromStreamARN

func ContextAndNameFromStreamARN(ctx context.Context, streamARN string) (context.Context, string)

ContextAndNameFromStreamARN parses a Kinesis stream ARN (arn:aws:kinesis:{region}:{account}:stream/{name}) and returns ctx with the ARN's region attached alongside the plain stream name, for callers outside this package that only hold the ARN -- cli.go's wireKinesisLambda (gopherstack-qowd), so Kinesis-to-Lambda event source mappings resolve the stream's actual region instead of always the account default.

Types

type Channel

type Channel struct {
	ChannelCreationTimestamp         time.Time                   `json:"channelCreationTimestamp"`
	EncryptionConfiguration          *ChannelEncryptionConfig    `json:"encryptionConfiguration,omitempty"`
	S3DestinationConfiguration       *ChannelS3Destination       `json:"s3DestinationConfiguration,omitempty"`
	S3TablesDestinationConfiguration *ChannelS3TablesDestination `json:"s3TablesDestinationConfiguration,omitempty"`
	Tags                             map[string]string           `json:"tags,omitempty"`
	ChannelID                        string                      `json:"channelID"`
	ChannelARN                       string                      `json:"channelARN"`
	ChannelName                      string                      `json:"channelName"`
	ChannelStatus                    string                      `json:"channelStatus"`
	ServiceExecutionRoleARN          string                      `json:"serviceExecutionRoleARN"`
	Region                           string                      `json:"region,omitempty"`
	StreamConfigurationList          []ChannelStreamConfig       `json:"streamConfigurationList"`
	LoggingConfiguration             ChannelCloudWatchLogsConfig `json:"loggingConfiguration"`
}

Channel represents an in-memory Kinesis Data Streams channel -- a CreateChannel-provisioned delivery pipe from a stream to either a general purpose S3 bucket or a streaming table (Apache Iceberg / Amazon S3 Tables) destination (types.ChannelDescription, kinesis@v1.53.0 types/types.go:9-79). Records put to the source stream are NOT delivered to either destination by this backend -- see PARITY.md.

type ChannelCloudWatchLogsConfig

type ChannelCloudWatchLogsConfig struct {
	LogGroupName  string `json:"logGroupName,omitempty"`
	LogStreamName string `json:"logStreamName,omitempty"`
	Enabled       bool   `json:"enabled"`
}

ChannelCloudWatchLogsConfig mirrors types.CloudWatchLogs / types.CloudWatchLogsUpdateInput (kinesis@v1.53.0 types/types.go:256-289).

type ChannelDeadLetterQueueS3Config

type ChannelDeadLetterQueueS3Config struct {
	BucketARN           string `json:"bucketARN"`
	ExpectedBucketOwner string `json:"expectedBucketOwner,omitempty"`
	ErrorOutputPrefix   string `json:"errorOutputPrefix,omitempty"`
}

ChannelDeadLetterQueueS3Config mirrors types.DeadLetterQueueS3Configuration (kinesis@v1.53.0 types/types.go:364-381).

type ChannelEncryptionConfig

type ChannelEncryptionConfig struct {
	EncryptionType string `json:"encryptionType"`
	KeyID          string `json:"keyID"`
}

ChannelEncryptionConfig mirrors types.ChannelEncryptionConfiguration (kinesis@v1.53.0 types/types.go:83-97).

type ChannelPartitionField

type ChannelPartitionField struct {
	SourceName string `json:"sourceName"`
	Transform  string `json:"transform"`
}

ChannelPartitionField mirrors types.PartitionField (kinesis@v1.53.0 types/types.go:479-493).

type ChannelPartitionSpec

type ChannelPartitionSpec struct {
	PartitionFields []ChannelPartitionField `json:"partitionFields"`
}

ChannelPartitionSpec mirrors types.PartitionSpec (kinesis@v1.53.0 types/types.go:495-504).

type ChannelRecordConfig

type ChannelRecordConfig struct {
	RecordFormatType string `json:"recordFormatType"`
	GSRSchemaARN     string `json:"gsrSchemaARN,omitempty"`
}

ChannelRecordConfig mirrors types.RecordConfiguration (kinesis@v1.53.0 types/types.go:600-624).

type ChannelS3Destination

type ChannelS3Destination struct {
	DeadLetterQueueS3Configuration *ChannelDeadLetterQueueS3Config `json:"deadLetterQueueS3Configuration,omitempty"`
	StorageConfiguration           ChannelS3StorageConfig          `json:"storageConfiguration"`
	DataFreshnessInSeconds         int                             `json:"dataFreshnessInSeconds"`
}

ChannelS3Destination mirrors the merged shape of types.S3DestinationConfiguration (CreateChannelInput) and types.S3DestinationDescription (ChannelDescription) -- both carry the same members, only required-ness differs between the two (kinesis@v1.53.0 types/types.go:626-663).

type ChannelS3StorageConfig

type ChannelS3StorageConfig struct {
	BucketARN           string `json:"bucketARN"`
	CompressionType     string `json:"compressionType"`
	ExpectedBucketOwner string `json:"expectedBucketOwner,omitempty"`
	OutputKeyTemplate   string `json:"outputKeyTemplate,omitempty"`
	StorageClass        string `json:"storageClass,omitempty"`
}

ChannelS3StorageConfig mirrors types.S3StorageConfiguration (kinesis@v1.53.0 types/types.go:678-714).

type ChannelS3TablesConfig

type ChannelS3TablesConfig struct {
	PartitionSpec   *ChannelPartitionSpec `json:"partitionSpec,omitempty"`
	TableBucketARN  string                `json:"tableBucketARN"`
	Namespace       string                `json:"namespace"`
	TableName       string                `json:"tableName"`
	CompressionType string                `json:"compressionType"`
}

ChannelS3TablesConfig mirrors types.S3TablesConfiguration (kinesis@v1.53.0 types/types.go:718-746).

type ChannelS3TablesDestination

type ChannelS3TablesDestination struct {
	DeadLetterQueueS3Configuration *ChannelDeadLetterQueueS3Config `json:"deadLetterQueueS3Configuration,omitempty"`
	S3TablesConfigurationList      []ChannelS3TablesConfig         `json:"s3TablesConfigurationList"`
	DataFreshnessInSeconds         int                             `json:"dataFreshnessInSeconds"`
}

ChannelS3TablesDestination mirrors types.S3TablesDestinationConfiguration / types.S3TablesDestinationDescription (kinesis@v1.53.0 types/types.go:762-802).

type ChannelS3Writer

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

ChannelS3Writer is the subset of S3 operations that channel record delivery needs to write an object to a general purpose S3 destination. Mirrors firehose.S3Storer (services/firehose/interfaces.go).

type ChannelStreamConfig

type ChannelStreamConfig struct {
	StreamCreationTimestamp time.Time           `json:"streamCreationTimestamp"`
	StreamARN               string              `json:"streamARN"`
	RecordConfiguration     ChannelRecordConfig `json:"recordConfiguration"`
}

ChannelStreamConfig mirrors the source-stream binding shared by types.ChannelStreamConfiguration (CreateChannelInput's StreamConfigurationList), types.ChannelStreamDescription (ChannelDescription's StreamConfigurationList), and types.ChannelStreamIdentifier (ChannelSummary's Streams) -- kinesis@v1.53.0 types/types.go:122-172.

type ChannelStreamFilter

type ChannelStreamFilter struct {
	StreamARN string
}

ChannelStreamFilter mirrors types.StreamFilter (kinesis@v1.53.0 types/types.go:1141-1153), used by ListChannels to filter by source stream.

type ChildShard added in v1.3.1

type ChildShard struct {
	ShardID           string
	HashKeyRangeStart string
	HashKeyRangeEnd   string
	ParentShards      []string
}

ChildShard describes a shard that resulted from splitting or merging the shard a GetRecords call just finished reading (aws-sdk-go-v2 types.ChildShard). Real AWS only returns this "when the end of the current shard is reached" -- i.e. exactly when NextShardIterator is empty because the shard is Closed and fully consumed.

type ConfigProvider

type ConfigProvider interface {
	GetKinesisSettings() Settings
}

ConfigProvider is a private interface to extract Kinesis configuration from the abstract AppContext Config.

type Consumer

type Consumer struct {
	ConsumerCreationTimestamp time.Time         `json:"consumerCreationTimestamp"`
	Tags                      map[string]string `json:"tags,omitempty"`
	ConsumerName              string            `json:"consumerName"`
	ConsumerARN               string            `json:"consumerARN"`
	ConsumerStatus            string            `json:"consumerStatus"`
	StreamARN                 string            `json:"streamARN"`
}

Consumer represents a registered Kinesis enhanced fan-out consumer.

type CreateChannelInput

type CreateChannelInput struct {
	EncryptionConfiguration          *ChannelEncryptionConfig
	LoggingConfiguration             *ChannelCloudWatchLogsConfig
	S3DestinationConfiguration       *ChannelS3Destination
	S3TablesDestinationConfiguration *ChannelS3TablesDestination
	Tags                             map[string]string
	ChannelName                      string
	ServiceExecutionRoleARN          string
	StreamConfigurationList          []ChannelStreamConfig
}

CreateChannelInput is the input for CreateChannel.

type CreateChannelOutput

type CreateChannelOutput struct {
	ChannelDescription Channel
}

CreateChannelOutput is the output for CreateChannel.

type CreateStreamInput

type CreateStreamInput struct {
	StreamName string
	Region     string
	AccountID  string
	StreamMode string
	ShardCount int
	// MaxRecordSizeInKiB mirrors CreateStreamInput's own field of the same
	// name (kinesis@v1.46.4 api_op_CreateStream.go:101-103), not just
	// UpdateMaxRecordSize's. Zero means "not specified" -- CreateStream keeps
	// defaultMaxRecordSizeBytes, same as an omitted request member.
	MaxRecordSizeInKiB int
	// WarmThroughputMiBps mirrors CreateStreamInput's own field of the same
	// name (kinesis@v1.46.4 api_op_CreateStream.go:119-121). Zero means "not
	// specified".
	WarmThroughputMiBps int
}

CreateStreamInput is the input for CreateStream.

type DecreaseStreamRetentionPeriodInput

type DecreaseStreamRetentionPeriodInput struct {
	StreamName           string
	RetentionPeriodHours int
}

DecreaseStreamRetentionPeriodInput is the input for DecreaseStreamRetentionPeriod.

type DeleteChannelInput

type DeleteChannelInput struct {
	ChannelARN string
}

DeleteChannelInput is the input for DeleteChannel.

type DeleteResourcePolicyInput

type DeleteResourcePolicyInput struct {
	ResourceARN string
}

DeleteResourcePolicyInput is the input for DeleteResourcePolicy.

type DeleteStreamInput

type DeleteStreamInput struct {
	StreamName string
	// EnforceConsumerDeletion mirrors the real DeleteStreamInput field: unset
	// or false with registered consumers fails the call with
	// ResourceInUseException instead of deleting the stream.
	EnforceConsumerDeletion bool
}

DeleteStreamInput is the input for DeleteStream.

type DeregisterStreamConsumerInput

type DeregisterStreamConsumerInput struct {
	StreamARN    string
	ConsumerARN  string
	ConsumerName string
}

DeregisterStreamConsumerInput is the input for DeregisterStreamConsumer.

type DescribeAccountSettingsOutput

type DescribeAccountSettingsOutput struct {
	MinimumThroughputBillingCommitment MinimumThroughputBillingCommitmentOutput
}

DescribeAccountSettingsOutput is the output for DescribeAccountSettings.

type DescribeChannelInput

type DescribeChannelInput struct {
	ChannelARN string
}

DescribeChannelInput is the input for DescribeChannel.

type DescribeChannelOutput

type DescribeChannelOutput struct {
	ChannelDescription Channel
}

DescribeChannelOutput is the output for DescribeChannel.

type DescribeStreamConsumerInput

type DescribeStreamConsumerInput struct {
	StreamARN    string
	ConsumerARN  string
	ConsumerName string
}

DescribeStreamConsumerInput is the input for DescribeStreamConsumer.

type DescribeStreamConsumerOutput

type DescribeStreamConsumerOutput struct {
	ConsumerDescription Consumer
}

DescribeStreamConsumerOutput is the output for DescribeStreamConsumer.

type DescribeStreamInput

type DescribeStreamInput struct {
	StreamName string
	// ExclusiveStartShardID resumes shard pagination after the given shard ID.
	ExclusiveStartShardID string
	// Limit caps the number of ShardDescription entries returned (AWS default
	// 100, max 10000). Zero means "use the AWS default".
	Limit int
}

DescribeStreamInput is the input for DescribeStream.

type DescribeStreamOutput

type DescribeStreamOutput struct {
	StreamCreationTimestamp time.Time
	StreamName              string
	StreamARN               string
	StreamStatus            string
	EncryptionType          string
	StreamMode              string
	KeyID                   string
	Shards                  []ShardDescription
	EnhancedMonitoring      []string
	RetentionPeriodHours    int
	// HasMoreShards indicates the shard list was truncated by Limit and more
	// shards can be fetched with a follow-up call using ExclusiveStartShardID.
	HasMoreShards bool
	// MaxRecordSizeBytes and WarmThroughputMiBps mirror the same-named Stream
	// fields (see UpdateMaxRecordSize/UpdateStreamWarmThroughput). Real
	// StreamDescriptionSummary carries both (MaxRecordSizeInKiB/WarmThroughput);
	// StreamDescription (DescribeStream) does not, so only
	// handleDescribeStreamSummary reads these.
	MaxRecordSizeBytes  int
	WarmThroughputMiBps int
	// CurrentWarmThroughputMiBps lags WarmThroughputMiBps while UPDATING.
	CurrentWarmThroughputMiBps int
}

DescribeStreamOutput is the output for DescribeStream.

type DisableEnhancedMonitoringInput

type DisableEnhancedMonitoringInput struct {
	StreamName        string
	ShardLevelMetrics []string
}

DisableEnhancedMonitoringInput is the input for DisableEnhancedMonitoring.

type DisableEnhancedMonitoringOutput

type DisableEnhancedMonitoringOutput struct {
	StreamName               string
	StreamARN                string
	CurrentShardLevelMetrics []string
	DesiredShardLevelMetrics []string
}

DisableEnhancedMonitoringOutput is the output for DisableEnhancedMonitoring.

type EnableEnhancedMonitoringInput

type EnableEnhancedMonitoringInput struct {
	StreamName        string
	ShardLevelMetrics []string
}

EnableEnhancedMonitoringInput is the input for EnableEnhancedMonitoring.

type EnableEnhancedMonitoringOutput

type EnableEnhancedMonitoringOutput struct {
	StreamName               string
	StreamARN                string
	CurrentShardLevelMetrics []string
	DesiredShardLevelMetrics []string
}

EnableEnhancedMonitoringOutput is the output for EnableEnhancedMonitoring.

type GetRecordResult

type GetRecordResult struct {
	ApproximateArrivalTimestamp time.Time
	PartitionKey                string
	SequenceNumber              string
	EncryptionType              string
	Data                        []byte
}

GetRecordResult is a single record returned by GetRecords.

type GetRecordsInput

type GetRecordsInput struct {
	ShardIterator string
	Limit         int
}

GetRecordsInput is the input for GetRecords.

type GetRecordsOutput

type GetRecordsOutput struct {
	NextShardIterator  string
	Records            []GetRecordResult
	ChildShards        []ChildShard
	MillisBehindLatest int64
}

GetRecordsOutput is the output for GetRecords.

type GetResourcePolicyInput

type GetResourcePolicyInput struct {
	ResourceARN string
}

GetResourcePolicyInput is the input for GetResourcePolicy.

type GetResourcePolicyOutput

type GetResourcePolicyOutput struct {
	Policy string
}

GetResourcePolicyOutput is the output for GetResourcePolicy.

type GetShardIteratorInput

type GetShardIteratorInput struct {
	Timestamp              *time.Time
	StreamName             string
	ShardID                string
	ShardIteratorType      string
	StartingSequenceNumber string
}

GetShardIteratorInput is the input for GetShardIterator. Timestamp is a pointer so a genuinely omitted value (nil) can be distinguished from an explicit epoch-zero timestamp; required (non-nil) when ShardIteratorType is AT_TIMESTAMP.

type GetShardIteratorOutput

type GetShardIteratorOutput struct {
	ShardIterator string
}

GetShardIteratorOutput is the output for GetShardIterator.

type Handler

type Handler struct {
	Backend StorageBackend

	DefaultRegion string
	AccountID     string
	// contains filtered or unexported fields
}

Handler is the Echo HTTP handler for Kinesis operations.

func NewHandler

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new Kinesis 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 Kinesis instance handles.

func (*Handler) ChaosServiceName

func (h *Handler) ChaosServiceName() string

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

func (*Handler) ExecuteFISAction

func (h *Handler) ExecuteFISAction(ctx context.Context, action service.FISActionExecution) error

ExecuteFISAction executes a FIS action against resolved Kinesis targets.

func (*Handler) ExtractOperation

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

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

func (*Handler) ExtractResource

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

ExtractResource extracts the stream name from the JSON request body.

func (*Handler) FISActions

func (h *Handler) FISActions() []service.FISActionDefinition

FISActions returns the FIS action definitions that the Kinesis service supports.

func (*Handler) GetSupportedOperations

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

GetSupportedOperations returns the list of supported Kinesis operations.

func (*Handler) Handler

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

Handler returns the Echo handler function for Kinesis operations.

func (*Handler) MatchPriority

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority for the Kinesis handler.

func (*Handler) Name

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Purge

func (h *Handler) Purge(ctx context.Context, cutoff time.Time)

Purge implements service.Purgeable by removing all Kinesis streams older than cutoff.

func (*Handler) Reset

func (h *Handler) Reset()

Reset clears all in-memory state from the backend. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

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 incoming Kinesis requests.

func (*Handler) Shutdown

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

Shutdown implements service.Shutdowner: it flushes any channel-buffered records to S3 before the process exits, so records accepted since the last interval flush are not lost (this backend does not persist buffered records across a snapshot/restore cycle -- see PARITY.md). If ctx expires before the flush finishes, 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 janitor (if configured) and the channel delivery interval flusher (see channel_delivery.go's runChannelFlusher).

func (*Handler) StopWorker

func (h *Handler) StopWorker()

StopWorker stops the background worker if one is configured.

func (*Handler) WithJanitor

func (h *Handler) WithJanitor(interval time.Duration, taskTimeout ...time.Duration) *Handler

WithJanitor attaches a background janitor to the handler. If the backend is not an *InMemoryBackend, this is a no-op.

func (*Handler) WithSubscribeToShardTiming

func (h *Handler) WithSubscribeToShardTiming(streamDuration, pollInterval, heartbeatInterval time.Duration) *Handler

WithSubscribeToShardTiming overrides SubscribeToShard's stream duration, poll interval, and heartbeat interval (see handler_consumers.go). A zero argument keeps that setting's current value, mirroring services/polly's WithStreamLimits zero-means-keep-default pattern.

type InMemoryBackend

type InMemoryBackend struct {
	OnStreamPurged func(string)
	// contains filtered or unexported fields
}

InMemoryBackend implements StorageBackend using in-memory maps.

Streams are keyed by a composite "region/name" key (see streamKey) inside a single flat store.Table, with a secondary store.Index grouping them by region — so same-named streams in different regions remain fully isolated, including their shards, records, and consumers, all of which stay inline fields on Stream (see pkgs/store's package doc: shards/records are the per-stream hot path and are not decomposed into their own tables). fisThroughputFaults and resourcePolicies are NOT store.Table candidates: their value types (a pointer with no self-describing identity, and a bare string) carry no key of their own to hand a Table keyFn, so they remain plain nested maps guarded the same way as before.

func NewInMemoryBackend

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new empty InMemoryBackend with default account/region.

func NewInMemoryBackendWithConfig

func NewInMemoryBackendWithConfig(accountID, region string) *InMemoryBackend

NewInMemoryBackendWithConfig creates a new InMemoryBackend with the given account ID and region.

func (*InMemoryBackend) AddStreamInternal

func (b *InMemoryBackend) AddStreamInternal(stream *Stream)

AddStreamInternal seeds a stream directly into the backend for testing. Caller must provide a non-nil stream with at least Name and ARN set. The stream is placed in the region encoded in its ARN, falling back to the backend's default region when the ARN carries none.

func (*InMemoryBackend) CountOnDemandStreams added in v1.3.1

func (b *InMemoryBackend) CountOnDemandStreams(ctx context.Context) int

CountOnDemandStreams returns the number of ON_DEMAND streams in the region carried on ctx, for DescribeLimits' required OnDemandStreamCount member (kinesis@v1.46.4 api_op_DescribeLimits.go:34-45). DescribeLimits is region-scoped in AWS, matching CountOpenShards' convention.

func (*InMemoryBackend) CountOpenShards

func (b *InMemoryBackend) CountOpenShards(ctx context.Context) int

CountOpenShards returns the total number of open (non-closed) shards across every stream in the region carried on ctx. DescribeLimits is region-scoped in AWS, so this counts within a single region.

func (*InMemoryBackend) CreateChannel

func (b *InMemoryBackend) CreateChannel(ctx context.Context, input *CreateChannelInput) (*CreateChannelOutput, error)

CreateChannel creates a channel that delivers records from a Kinesis data stream to an S3 or S3-Tables destination (api_op_CreateChannel.go). Records are not actually delivered -- see PARITY.md. CreateChannel is documented as asynchronous (CREATING then ACTIVE); this backend applies it synchronously, matching the precedent already set for UpdateStreamWarmThroughput/UpdateStreamMode.

func (*InMemoryBackend) CreateStream

func (b *InMemoryBackend) CreateStream(ctx context.Context, input *CreateStreamInput) error

CreateStream creates a new Kinesis stream.

func (*InMemoryBackend) DecreaseStreamRetentionPeriod

func (b *InMemoryBackend) DecreaseStreamRetentionPeriod(
	ctx context.Context,
	input *DecreaseStreamRetentionPeriodInput,
) error

DecreaseStreamRetentionPeriod decreases the retention period for a stream. Mirroring IncreaseStreamRetentionPeriod, a target equal to the current retention period is an idempotent no-op returning success (HTTP 200), matching real AWS behaviour. A target above the current period or below minRetentionHours (24h) is rejected with InvalidArgumentException.

func (*InMemoryBackend) DeleteChannel

func (b *InMemoryBackend) DeleteChannel(ctx context.Context, input *DeleteChannelInput) error

DeleteChannel deletes the specified channel (api_op_DeleteChannel.go). Deletion is synchronous -- DeleteChannel's own doc comment, unlike CreateChannel/UpdateChannel's, describes no CREATING/UPDATING-style asynchronous transition, so there is no documented DELETING state to model. Any buffered records are flushed to S3 (best-effort) after the channel row is removed, so records already accepted are not silently dropped.

func (*InMemoryBackend) DeleteResourcePolicy

func (b *InMemoryBackend) DeleteResourcePolicy(ctx context.Context, input *DeleteResourcePolicyInput) error

DeleteResourcePolicy removes the resource-based policy for the given stream or consumer ARN.

func (*InMemoryBackend) DeleteStream

func (b *InMemoryBackend) DeleteStream(ctx context.Context, input *DeleteStreamInput) error

DeleteStream removes a stream, then flushes (best-effort) the buffer of any channel sourced from it, so records already accepted by PutRecord are not silently dropped by the removal (see channel_delivery.go).

func (*InMemoryBackend) DeregisterStreamConsumer

func (b *InMemoryBackend) DeregisterStreamConsumer(ctx context.Context, input *DeregisterStreamConsumerInput) error

DeregisterStreamConsumer removes a registered consumer from a stream.

func (*InMemoryBackend) DescribeAccountSettings

func (b *InMemoryBackend) DescribeAccountSettings(_ context.Context) (*DescribeAccountSettingsOutput, error)

DescribeAccountSettings returns the account's minimum throughput billing commitment configuration (kinesis@v1.46.4 api_op_DescribeAccountSettings.go:34-45).

func (*InMemoryBackend) DescribeChannel

func (b *InMemoryBackend) DescribeChannel(
	_ context.Context,
	input *DescribeChannelInput,
) (*DescribeChannelOutput, error)

DescribeChannel describes the specified channel (api_op_DescribeChannel.go).

func (*InMemoryBackend) DescribeStream

func (b *InMemoryBackend) DescribeStream(
	ctx context.Context,
	input *DescribeStreamInput,
) (*DescribeStreamOutput, error)

DescribeStream returns full stream details including shards.

func (*InMemoryBackend) DescribeStreamConsumer

func (b *InMemoryBackend) DescribeStreamConsumer(
	ctx context.Context,
	input *DescribeStreamConsumerInput,
) (*DescribeStreamConsumerOutput, error)

DescribeStreamConsumer returns details about a registered consumer. Lookup is by ConsumerARN, or by StreamARN + ConsumerName.

func (*InMemoryBackend) DisableEnhancedMonitoring

DisableEnhancedMonitoring removes shard-level metrics from a stream.

func (*InMemoryBackend) EnableEnhancedMonitoring

func (b *InMemoryBackend) EnableEnhancedMonitoring(
	ctx context.Context,
	input *EnableEnhancedMonitoringInput,
) (*EnableEnhancedMonitoringOutput, error)

EnableEnhancedMonitoring adds shard-level metrics to a stream.

func (*InMemoryBackend) FlushAllChannels

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

FlushAllChannels forces immediate delivery of every channel's buffered records. Used by Handler.Shutdown and tests.

func (*InMemoryBackend) FlushChannel

func (b *InMemoryBackend) FlushChannel(ctx context.Context, channelARN string)

FlushChannel forces immediate delivery of channelARN's buffered records, regardless of its DataFreshnessInSeconds window. Used by DeleteChannel, graceful shutdown, and tests (this backend has no injectable clock for interval-based flush -- see PARITY.md).

func (*InMemoryBackend) GetRecords

func (b *InMemoryBackend) GetRecords(ctx context.Context, input *GetRecordsInput) (*GetRecordsOutput, error)

GetRecords retrieves records starting at the given shard iterator position.

The region is taken from the iterator token (encoded by GetShardIterator), not from ctx, so an iterator issued for one region always reads that region's records even if the GetRecords call carries a different ctx region.

func (*InMemoryBackend) GetResourcePolicy

func (b *InMemoryBackend) GetResourcePolicy(
	ctx context.Context,
	input *GetResourcePolicyInput,
) (*GetResourcePolicyOutput, error)

GetResourcePolicy retrieves the resource-based policy for the given stream or consumer ARN.

func (*InMemoryBackend) GetShardIterator

func (b *InMemoryBackend) GetShardIterator(
	ctx context.Context,
	input *GetShardIteratorInput,
) (*GetShardIteratorOutput, error)

GetShardIterator returns an iterator for reading records from a shard.

func (*InMemoryBackend) IncreaseStreamRetentionPeriod

func (b *InMemoryBackend) IncreaseStreamRetentionPeriod(
	ctx context.Context,
	input *IncreaseStreamRetentionPeriodInput,
) error

IncreaseStreamRetentionPeriod increases the retention period for a stream. A target equal to the current retention period is treated as an idempotent no-op returning success (HTTP 200), matching real AWS behaviour: the Terraform AWS provider calls IncreaseStreamRetentionPeriod on stream create for ANY configured retention_period > 0 (see the provider's resourceStreamCreate, guard `v.(int) > 0`), so a stream created with the default retention_period of 24h receives IncreaseStreamRetentionPeriod(24) against a stream already at 24h. Rejecting that equal value with InvalidArgumentException (as a strict reading of the SDK doc "Must be more than the current retention period" would suggest) breaks every default-retention Terraform stream, so real AWS accepts it. A target below the current period, below minRetentionHours (24h), or above maxRetentionHours (8760h) is rejected with InvalidArgumentException.

func (*InMemoryBackend) ListAll

func (b *InMemoryBackend) ListAll(_ context.Context) []StreamInfo

ListAll returns a snapshot of all streams as StreamInfo values across every region. It is used by the dashboard, which presents a global inventory.

func (*InMemoryBackend) ListChannels

func (b *InMemoryBackend) ListChannels(ctx context.Context, input *ListChannelsInput) (*ListChannelsOutput, error)

ListChannels lists the channels in the caller's account/region, optionally filtered by source stream (api_op_ListChannels.go).

func (*InMemoryBackend) ListShards

func (b *InMemoryBackend) ListShards(ctx context.Context, input *ListShardsInput) (*ListShardsOutput, error)

ListShards returns the shards for a stream.

func (*InMemoryBackend) ListStreamConsumers

func (b *InMemoryBackend) ListStreamConsumers(
	ctx context.Context,
	input *ListStreamConsumersInput,
) (*ListStreamConsumersOutput, error)

ListStreamConsumers lists all registered consumers for a stream.

func (*InMemoryBackend) ListStreams

func (b *InMemoryBackend) ListStreams(ctx context.Context, input *ListStreamsInput) (*ListStreamsOutput, error)

ListStreams returns stream names with optional pagination.

AWS contract: results are returned in alphabetical order. When `Limit` is set the response contains at most that many names. Pagination is keyed on either `ExclusiveStartStreamName` or the opaque `NextToken` (which we treat as the previously returned last stream name) so that callers can iterate over arbitrarily large account inventories.

func (*InMemoryBackend) ListTagsForResource

func (b *InMemoryBackend) ListTagsForResource(
	ctx context.Context,
	input *ListTagsForResourceInput,
) (*ListTagsForResourceOutput, error)

ListTagsForResource returns the tags associated with a Kinesis resource identified by its ARN -- a stream (tags stored on the stream's internal Tags store, set via TagResource) or, per the real op's doc comment ("the specified Kinesis resource"), an enhanced fan-out consumer (tags stored on Consumer.Tags, set via RegisterStreamConsumer's Tags parameter or TagResource against the consumer's own ARN).

func (*InMemoryBackend) MergeShards

func (b *InMemoryBackend) MergeShards(ctx context.Context, input *MergeShardsInput) error

MergeShards merges two adjacent shards into one. The merged shard spans the combined hash key range of both parent shards.

func (*InMemoryBackend) OnDemandStreamCountLimit added in v1.3.1

func (b *InMemoryBackend) OnDemandStreamCountLimit(_ context.Context) int

OnDemandStreamCountLimit returns the account's current cap on ON_DEMAND streams, for DescribeLimits' required OnDemandStreamCountLimit member. Real AWS manages this as a Service Quota, not adjustable via UpdateAccountSettings; see SetOnDemandStreamCountLimit for how this backend exposes changing it (a Go-level config knob, not a wire operation).

func (*InMemoryBackend) Purge

func (b *InMemoryBackend) Purge(ctx context.Context, cutoff time.Time)

Purge removes all Kinesis streams and consumers created before the cutoff time.

func (*InMemoryBackend) PutRecord

func (b *InMemoryBackend) PutRecord(ctx context.Context, input *PutRecordInput) (*PutRecordOutput, error)

PutRecord writes a single record to a stream shard, then delivers it to any ACTIVE channel sourced from the stream (see deliverPutToChannels).

func (*InMemoryBackend) PutRecords

func (b *InMemoryBackend) PutRecords(ctx context.Context, input *PutRecordsInput) (*PutRecordsOutput, error)

PutRecords writes multiple records to a stream.

Request-level validation errors (empty/oversized batch, unknown stream) fail the whole call with a single top-level exception, matching the AWS contract: only per-record issues (throughput, per-record validation) surface as per-entry ErrorCode/ErrorMessage inside a 200 response.

func (*InMemoryBackend) PutResourcePolicy

func (b *InMemoryBackend) PutResourcePolicy(ctx context.Context, input *PutResourcePolicyInput) error

PutResourcePolicy stores a resource-based policy for the given stream or consumer ARN.

func (*InMemoryBackend) Region

func (b *InMemoryBackend) Region() string

Region returns the AWS region this backend is configured to use as its default.

func (*InMemoryBackend) RegisterStreamConsumer

func (b *InMemoryBackend) RegisterStreamConsumer(
	ctx context.Context,
	input *RegisterStreamConsumerInput,
) (*RegisterStreamConsumerOutput, error)

RegisterStreamConsumer registers a new enhanced fan-out consumer on a stream.

func (*InMemoryBackend) Reset

func (b *InMemoryBackend) Reset()

Reset clears all in-memory state from the backend. It is used by the POST /_gopherstack/reset endpoint for CI pipelines and rapid local development.

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) SetMetricEmitter

func (b *InMemoryBackend) SetMetricEmitter(e cwmetric.Emitter)

SetMetricEmitter sets the emitter that publishes AWS/Kinesis metrics to CloudWatch.

func (*InMemoryBackend) SetOnDemandStreamCountLimit added in v1.3.1

func (b *InMemoryBackend) SetOnDemandStreamCountLimit(n int)

SetOnDemandStreamCountLimit configures the account-level cap CreateStream enforces for ON_DEMAND streams (default defaultOnDemandStreamCountLimit). No real Kinesis wire operation can change this account setting -- it was previously (and incorrectly) exposed as a fabricated field on UpdateAccountSettingsInput; real AWS manages it as a Service Quota. This method is the Go-level replacement, mirroring how WithKMSValidator wires cross-service config outside the wire protocol.

func (*InMemoryBackend) SetS3Writer

func (b *InMemoryBackend) SetS3Writer(w ChannelS3Writer)

SetS3Writer wires the S3 backend used to deliver channel-buffered records to their configured S3DestinationConfiguration bucket. See cli.go's wireKinesisS3Delivery.

func (*InMemoryBackend) Snapshot

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

Snapshot serialises the backend state to JSON. It implements persistence.Persistable. Note: shard sequence number counters are now serialised via the NextSeq field.

func (*InMemoryBackend) SplitShard

func (b *InMemoryBackend) SplitShard(ctx context.Context, input *SplitShardInput) error

SplitShard splits a shard into two at the given new starting hash key.

func (*InMemoryBackend) StartStreamEncryption

func (b *InMemoryBackend) StartStreamEncryption(ctx context.Context, input *StartStreamEncryptionInput) error

StartStreamEncryption enables server-side encryption on a stream.

func (*InMemoryBackend) StopStreamEncryption

func (b *InMemoryBackend) StopStreamEncryption(ctx context.Context, input *StopStreamEncryptionInput) error

StopStreamEncryption disables server-side encryption on a stream.

func (*InMemoryBackend) SubscribeToShard

func (b *InMemoryBackend) SubscribeToShard(
	ctx context.Context,
	input *SubscribeToShardInput,
) (*SubscribeToShardOutput, error)

SubscribeToShard delivers records from a shard to an enhanced fan-out consumer. For mock purposes this is a single-shot delivery of all available records.

func (*InMemoryBackend) TagResource

func (b *InMemoryBackend) TagResource(ctx context.Context, input *TagResourceInput) error

TagResource adds or updates tags on a Kinesis resource identified by its ARN -- a stream (the ARN-based counterpart to AddTagsToStream) or an enhanced fan-out consumer.

func (*InMemoryBackend) TaggedStreams added in v1.2.0

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

TaggedStreams returns every Kinesis stream that currently has at least one tag, across every region this backend holds streams for.

func (*InMemoryBackend) UntagResource

func (b *InMemoryBackend) UntagResource(ctx context.Context, input *UntagResourceInput) error

UntagResource removes tags from a Kinesis resource identified by its ARN -- a stream (the ARN-based counterpart to RemoveTagsFromStream) or an enhanced fan-out consumer.

func (*InMemoryBackend) UpdateAccountSettings

UpdateAccountSettings sets the account's minimum throughput billing commitment status (kinesis@v1.46.4 api_op_UpdateAccountSettings.go:42-51). This backend has no billing engine: no billing behaviour follows from enabling the commitment. Status/StartedAt/EndedAt only track the requested transition; see MinimumThroughputBillingCommitmentOutput.

func (*InMemoryBackend) UpdateChannel

UpdateChannel updates a channel's data-freshness interval or CloudWatch Logs configuration (api_op_UpdateChannel.go). UpdateChannel is documented as asynchronous (UPDATING then ACTIVE); this backend applies it synchronously, matching CreateChannel's disclosed simplification.

func (*InMemoryBackend) UpdateMaxRecordSize

func (b *InMemoryBackend) UpdateMaxRecordSize(ctx context.Context, input *UpdateMaxRecordSizeInput) error

UpdateMaxRecordSize changes the per-record data payload size limit for a stream. The wire input is MaxRecordSizeInKiB (kinesis@v1.46.4 api_op_UpdateMaxRecordSize.go:30-47); this backend stores the limit in bytes on Stream.MaxRecordSizeBytes, so the requested KiB value is converted via bytesPerKiB. The valid range is [defaultMaxRecordSizeBytes, absoluteMaxRecordSizeBytes] (1 MiB - 10 MiB). The real Input has no StreamName member, only StreamARN.

func (*InMemoryBackend) UpdateShardCount

func (b *InMemoryBackend) UpdateShardCount(
	ctx context.Context,
	input *UpdateShardCountInput,
) (*UpdateShardCountOutput, error)

UpdateShardCount resizes a stream to the given number of shards. Existing records in the stream are not migrated; new shards start empty.

func (*InMemoryBackend) UpdateStreamMode

func (b *InMemoryBackend) UpdateStreamMode(ctx context.Context, input *UpdateStreamModeInput) error

UpdateStreamMode changes the mode of a stream identified by its ARN.

func (*InMemoryBackend) UpdateStreamWarmThroughput

UpdateStreamWarmThroughput configures pre-warmed throughput for a stream (kinesis@v1.53.0 api_op_UpdateStreamWarmThroughput.go:24-27: the stream goes UPDATING, then back to ACTIVE; Current stays at the old value until then).

func (*InMemoryBackend) WithClock

func (b *InMemoryBackend) WithClock(now func() time.Time) *InMemoryBackend

WithClock overrides the backend's time source, used by tests to drive AT_TRIM_HORIZON/AT_TIMESTAMP retention math and ON_DEMAND write-throughput auto-scaling deterministically -- no time.Sleep, no real wall-clock waits. Mirrors services/polly/throttle.go's WithClock.

func (*InMemoryBackend) WithKMSValidator added in v1.2.0

func (b *InMemoryBackend) WithKMSValidator(v KMSKeyValidator) *InMemoryBackend

WithKMSValidator attaches a KMSKeyValidator so StartStreamEncryption can verify a KeyId resolves to a real, usable KMS key. Returns b for chaining.

type IncreaseStreamRetentionPeriodInput

type IncreaseStreamRetentionPeriodInput struct {
	StreamName           string
	RetentionPeriodHours int
}

IncreaseStreamRetentionPeriodInput is the input for IncreaseStreamRetentionPeriod.

type Janitor

type Janitor struct {
	Backend *InMemoryBackend

	Interval time.Duration
	// TaskTimeout bounds each individual janitor task. When non-zero, each task
	// runs with a child context that expires after this duration, preventing a
	// stalled operation from blocking the janitor loop indefinitely.
	TaskTimeout time.Duration
	// contains filtered or unexported fields
}

Janitor is the Kinesis background worker that enforces per-stream retention periods by evicting records older than stream.RetentionPeriod hours.

func NewJanitor

func NewJanitor(backend *InMemoryBackend, interval time.Duration) *Janitor

NewJanitor creates a new Kinesis Janitor for the given backend. A zero interval falls back to defaultJanitorInterval.

func (*Janitor) Run

func (j *Janitor) Run(ctx context.Context)

Run runs the janitor loop until ctx is cancelled.

func (*Janitor) Stop

func (j *Janitor) Stop()

Stop explicitly shuts down the janitor.

func (*Janitor) SweepOnce

func (j *Janitor) SweepOnce(ctx context.Context)

SweepOnce executes a single retention sweep. Exposed for testing.

type KMSKeyValidator added in v1.2.0

type KMSKeyValidator interface {
	// ValidateKMSKey resolves keyID against the KMS backend and returns nil if
	// the key exists and is usable, or one of ErrKMSNotFound/ErrKMSDisabled/
	// ErrKMSInvalidState (kinesis sentinels) describing why it is not.
	ValidateKMSKey(ctx context.Context, keyID string) error
}

KMSKeyValidator optionally validates a KMS KeyId against a real KMS backend. Implemented by an adapter wrapping the KMS service backend and attached via WithKMSValidator (mirrors services/ssm's KMSEncryptor injection pattern -- see cli.go's wireKinesisKMS). When no validator is wired, StartStreamEncryption still enforces the KeyId *shape* AWS documents (UUID / key ARN / alias ARN / "alias/..." name) but cannot know whether the key actually exists, is disabled, or is pending deletion -- those KMS-specific exceptions require real cross-service key state.

type ListChannelsInput

type ListChannelsInput struct {
	NextToken    string
	StreamFilter []ChannelStreamFilter
	MaxResults   int
}

ListChannelsInput is the input for ListChannels.

type ListChannelsOutput

type ListChannelsOutput struct {
	NextToken        string
	ChannelSummaries []Channel
}

ListChannelsOutput is the output for ListChannels.

type ListShardsInput

type ListShardsInput struct {
	ShardFilterTimestamp  *time.Time
	StreamCreationTime    *time.Time
	StreamName            string
	NextToken             string
	ExclusiveStartShardID string
	ShardFilter           string
	ShardFilterType       string
	ShardFilterShardID    string
	MaxResults            int
}

ListShardsInput is the input for ListShards.

type ListShardsOutput

type ListShardsOutput struct {
	NextToken string
	Shards    []ShardDescription
}

ListShardsOutput is the output for ListShards.

type ListStreamConsumersInput

type ListStreamConsumersInput struct {
	StreamARN  string
	NextToken  string
	MaxResults int
}

ListStreamConsumersInput is the input for ListStreamConsumers.

type ListStreamConsumersOutput

type ListStreamConsumersOutput struct {
	NextToken string
	Consumers []Consumer
}

ListStreamConsumersOutput is the output for ListStreamConsumers.

type ListStreamsInput

type ListStreamsInput struct {
	NextToken                string
	ExclusiveStartStreamName string
	Limit                    int
}

ListStreamsInput is the input for ListStreams.

type ListStreamsOutput

type ListStreamsOutput struct {
	NextToken       string
	StreamNames     []string
	StreamSummaries []StreamSummary
	HasMoreStreams  bool
}

ListStreamsOutput is the output for ListStreams.

type ListTagsForResourceInput

type ListTagsForResourceInput struct {
	ResourceARN string
}

ListTagsForResourceInput is the input for ListTagsForResource.

type ListTagsForResourceOutput

type ListTagsForResourceOutput struct {
	Tags map[string]string
}

ListTagsForResourceOutput is the output for ListTagsForResource.

type MergeShardsInput

type MergeShardsInput struct {
	StreamName           string
	StreamARN            string
	ShardToMerge         string
	AdjacentShardToMerge string
}

MergeShardsInput is the input for MergeShards.

type MinimumThroughputBillingCommitmentInput added in v1.3.1

type MinimumThroughputBillingCommitmentInput struct {
	// Status is required: minimumThroughputBillingCommitmentEnabled or
	// minimumThroughputBillingCommitmentDisabled.
	Status string
}

MinimumThroughputBillingCommitmentInput is the input shape for the commitment status requested via UpdateAccountSettings (types.MinimumThroughputBillingCommitmentInput; Status is its only, required, member -- kinesis@v1.46.4 types/types.go:168-176).

type MinimumThroughputBillingCommitmentOutput added in v1.3.1

type MinimumThroughputBillingCommitmentOutput struct {
	EarliestAllowedEndAt time.Time `json:"earliestAllowedEndAt"`
	EndedAt              time.Time `json:"endedAt"`
	StartedAt            time.Time `json:"startedAt"`
	Status               string    `json:"status"`
}

MinimumThroughputBillingCommitmentOutput is the account's current minimum throughput billing commitment (types.MinimumThroughputBillingCommitmentOutput, kinesis@v1.46.4 types/types.go:178-197). This backend has no billing engine: Status/StartedAt/EndedAt only track the state transitions UpdateAccountSettings requests; EarliestAllowedEndAt is never populated since computing it needs a commitment-window model this backend doesn't have (see PARITY.md gaps), and Status never reports minimumThroughputBillingCommitmentEnabledUntilEnd for the same reason.

type Provider

type Provider struct{}

Provider implements service.Provider for the Kinesis service.

func (*Provider) Init

Init initializes the Kinesis service backend and handler.

func (*Provider) Name

func (p *Provider) Name() string

Name returns the service provider name.

type PutRecordInput

type PutRecordInput struct {
	StreamName      string
	PartitionKey    string
	ExplicitHashKey string
	Data            []byte
}

PutRecordInput is the input for PutRecord.

type PutRecordOutput

type PutRecordOutput struct {
	ShardID        string
	SequenceNumber string
	EncryptionType string
}

PutRecordOutput is the output for PutRecord.

type PutRecordsEntry

type PutRecordsEntry struct {
	PartitionKey    string
	ExplicitHashKey string
	Data            []byte
}

PutRecordsEntry is a single entry in a PutRecords request.

type PutRecordsInput

type PutRecordsInput struct {
	StreamName string
	Records    []PutRecordsEntry
}

PutRecordsInput is the input for PutRecords.

type PutRecordsOutput

type PutRecordsOutput struct {
	Records           []PutRecordsResultEntry
	FailedRecordCount int
}

PutRecordsOutput is the output for PutRecords.

type PutRecordsResultEntry

type PutRecordsResultEntry struct {
	ShardID        string
	SequenceNumber string
	ErrorCode      string
	ErrorMessage   string
}

PutRecordsResultEntry is a single result entry in a PutRecords response.

type PutResourcePolicyInput

type PutResourcePolicyInput struct {
	ResourceARN string
	Policy      string
}

PutResourcePolicyInput is the input for PutResourcePolicy.

type Record

type Record struct {
	ApproximateArrivalTimestamp time.Time `json:"approximateArrivalTimestamp"`
	PartitionKey                string    `json:"partitionKey"`
	SequenceNumber              string    `json:"sequenceNumber"`
	Data                        []byte    `json:"data"`
}

Record represents a single Kinesis data record.

type RegisterStreamConsumerInput

type RegisterStreamConsumerInput struct {
	Tags         map[string]string
	StreamARN    string
	ConsumerName string
}

RegisterStreamConsumerInput is the input for RegisterStreamConsumer.

type RegisterStreamConsumerOutput

type RegisterStreamConsumerOutput struct {
	Consumer Consumer
}

RegisterStreamConsumerOutput is the output for RegisterStreamConsumer.

type Settings

type Settings struct {
	JanitorInterval time.Duration `json:"janitor_interval" env:"KINESIS_JANITOR_INTERVAL" default:"1m" help:"Janitor tick interval."` //nolint:lll // Kong struct tag makes this line long
}

Settings holds service-level configuration for the Kinesis backend. Fields are picked up by the Kong CLI parser when this struct is embedded in the root CLI command.

type Shard

type Shard struct {
	// StartedAt is when this shard became open (stream creation for the initial
	// shard set, or reshard time for shards born from SplitShard/MergeShards/
	// UpdateShardCount/UpdateStreamMode). Used by ListShards' AT_TIMESTAMP/
	// FROM_TIMESTAMP/AT_TRIM_HORIZON ShardFilter to bound shard lineage by time.
	StartedAt time.Time `json:"startedAt"`
	// ClosedAt is when this shard was closed (zero if still open). Populated
	// alongside Closed by closeShard. omitempty has no effect on a struct
	// field like time.Time, so it is intentionally omitted here.
	ClosedAt time.Time `json:"closedAt"`

	ID                    string       `json:"id"`
	HashKeyRangeStart     string       `json:"hashKeyRangeStart"`
	HashKeyRangeEnd       string       `json:"hashKeyRangeEnd"`
	ParentShardID         string       `json:"parentShardId,omitempty"`
	AdjacentParentShardID string       `json:"adjacentParentShardId,omitempty"`
	Records               shardRecords `json:"records"`
	NextSeq               uint64       `json:"nextSeq"`
	Closed                bool         `json:"closed,omitempty"`
	// contains filtered or unexported fields
}

Shard represents a single Kinesis shard within a stream.

type ShardDescription

type ShardDescription struct {
	ShardID                  string
	HashKeyRangeStart        string
	HashKeyRangeEnd          string
	SequenceNumberRangeStart string
	SequenceNumberRangeEnd   string
	ParentShardID            string
	AdjacentParentShardID    string
	Closed                   bool
}

ShardDescription describes a shard in a DescribeStream response.

type ShardIterator

type ShardIterator struct {
	CreatedAt      time.Time `json:"CreatedAt"`
	StreamName     string    `json:"StreamName"`
	ShardID        string    `json:"ShardID"`
	SequenceNumber string    `json:"SequenceNumber"`
	Region         string    `json:"Region"`
	Position       int       `json:"Position"`
}

ShardIterator holds the position within a shard for GetRecords. Region is encoded into the iterator token so that GetRecords resolves the record store of the same region the iterator was issued in, keeping same-named streams in different regions isolated on the record hot path.

type SplitShardInput

type SplitShardInput struct {
	StreamName         string
	StreamARN          string
	ShardToSplit       string
	NewStartingHashKey string
}

SplitShardInput is the input for SplitShard.

type StartStreamEncryptionInput

type StartStreamEncryptionInput struct {
	StreamName     string
	StreamARN      string
	EncryptionType string
	KeyID          string
}

StartStreamEncryptionInput is the input for StartStreamEncryption.

type StartingPosition

type StartingPosition struct {
	Timestamp      *time.Time `json:"Timestamp,omitempty"`
	Type           string     `json:"Type"`
	SequenceNumber string     `json:"SequenceNumber,omitempty"`
}

StartingPosition describes where to start reading in SubscribeToShard.

type StopStreamEncryptionInput

type StopStreamEncryptionInput struct {
	StreamName     string
	StreamARN      string
	EncryptionType string
	KeyID          string
}

StopStreamEncryptionInput is the input for StopStreamEncryption.

type StorageBackend

type StorageBackend interface {
	CreateStream(ctx context.Context, input *CreateStreamInput) error
	DeleteStream(ctx context.Context, input *DeleteStreamInput) error
	DescribeStream(ctx context.Context, input *DescribeStreamInput) (*DescribeStreamOutput, error)
	ListStreams(ctx context.Context, input *ListStreamsInput) (*ListStreamsOutput, error)
	PutRecord(ctx context.Context, input *PutRecordInput) (*PutRecordOutput, error)
	PutRecords(ctx context.Context, input *PutRecordsInput) (*PutRecordsOutput, error)
	GetShardIterator(ctx context.Context, input *GetShardIteratorInput) (*GetShardIteratorOutput, error)
	GetRecords(ctx context.Context, input *GetRecordsInput) (*GetRecordsOutput, error)
	ListShards(ctx context.Context, input *ListShardsInput) (*ListShardsOutput, error)
	RegisterStreamConsumer(
		ctx context.Context,
		input *RegisterStreamConsumerInput,
	) (*RegisterStreamConsumerOutput, error)
	DescribeStreamConsumer(
		ctx context.Context,
		input *DescribeStreamConsumerInput,
	) (*DescribeStreamConsumerOutput, error)
	ListStreamConsumers(ctx context.Context, input *ListStreamConsumersInput) (*ListStreamConsumersOutput, error)
	DeregisterStreamConsumer(ctx context.Context, input *DeregisterStreamConsumerInput) error
	SubscribeToShard(ctx context.Context, input *SubscribeToShardInput) (*SubscribeToShardOutput, error)
	UpdateShardCount(ctx context.Context, input *UpdateShardCountInput) (*UpdateShardCountOutput, error)
	EnableEnhancedMonitoring(
		ctx context.Context,
		input *EnableEnhancedMonitoringInput,
	) (*EnableEnhancedMonitoringOutput, error)
	DisableEnhancedMonitoring(
		ctx context.Context,
		input *DisableEnhancedMonitoringInput,
	) (*DisableEnhancedMonitoringOutput, error)
	IncreaseStreamRetentionPeriod(ctx context.Context, input *IncreaseStreamRetentionPeriodInput) error
	DecreaseStreamRetentionPeriod(ctx context.Context, input *DecreaseStreamRetentionPeriodInput) error
	MergeShards(ctx context.Context, input *MergeShardsInput) error
	SplitShard(ctx context.Context, input *SplitShardInput) error
	StartStreamEncryption(ctx context.Context, input *StartStreamEncryptionInput) error
	StopStreamEncryption(ctx context.Context, input *StopStreamEncryptionInput) error
	DeleteResourcePolicy(ctx context.Context, input *DeleteResourcePolicyInput) error
	GetResourcePolicy(ctx context.Context, input *GetResourcePolicyInput) (*GetResourcePolicyOutput, error)
	PutResourcePolicy(ctx context.Context, input *PutResourcePolicyInput) error
	ListTagsForResource(ctx context.Context, input *ListTagsForResourceInput) (*ListTagsForResourceOutput, error)
	TagResource(ctx context.Context, input *TagResourceInput) error
	UntagResource(ctx context.Context, input *UntagResourceInput) error
	UpdateStreamMode(ctx context.Context, input *UpdateStreamModeInput) error
	UpdateAccountSettings(ctx context.Context, input *UpdateAccountSettingsInput) (*UpdateAccountSettingsOutput, error)
	UpdateMaxRecordSize(ctx context.Context, input *UpdateMaxRecordSizeInput) error
	UpdateStreamWarmThroughput(
		ctx context.Context,
		input *UpdateStreamWarmThroughputInput,
	) (*UpdateStreamWarmThroughputOutput, error)
	DescribeAccountSettings(ctx context.Context) (*DescribeAccountSettingsOutput, error)
	CreateChannel(ctx context.Context, input *CreateChannelInput) (*CreateChannelOutput, error)
	DeleteChannel(ctx context.Context, input *DeleteChannelInput) error
	DescribeChannel(ctx context.Context, input *DescribeChannelInput) (*DescribeChannelOutput, error)
	ListChannels(ctx context.Context, input *ListChannelsInput) (*ListChannelsOutput, error)
	UpdateChannel(ctx context.Context, input *UpdateChannelInput) (*UpdateChannelOutput, error)
	CountOpenShards(ctx context.Context) int
	CountOnDemandStreams(ctx context.Context) int
	OnDemandStreamCountLimit(ctx context.Context) int
	ListAll(ctx context.Context) []StreamInfo
}

StorageBackend defines the interface for a Kinesis backend.

Every method takes a context.Context so the per-request AWS region can be threaded through and resources kept isolated per region. The region is read from the context via getRegion, falling back to the backend's default region when the context carries no region.

type Stream

type Stream struct {
	CreatedAt time.Time `json:"createdAt"`
	// ReadyAt is the deadline at which a CREATING/UPDATING/DELETING stream
	// lazily advances to its terminal state (ACTIVE, or removed for
	// DELETING) -- see resolveStreamTransitionLocked. Zero when Status is
	// not mid-transition. Additive persisted field.
	ReadyAt time.Time `json:"readyAt"`

	Tags      *tags.Tags           `json:"tags,omitempty"`
	Consumers map[string]*Consumer `json:"consumers,omitempty"`
	Name      string               `json:"name"`
	ARN       string               `json:"arn"`
	// Region is the AWS region this stream lives in. It is the second half of
	// the composite key (see streamKey in backend.go) that keeps same-named
	// streams in different regions isolated inside the single flat
	// store.Table[Stream] — the region-nested map it replaced used the
	// region as an outer map key instead of a field on Stream itself.
	Region             string   `json:"region,omitempty"`
	Status             string   `json:"status"`
	EncryptionType     string   `json:"encryptionType,omitempty"`
	KeyID              string   `json:"keyId,omitempty"`
	StreamMode         string   `json:"streamMode,omitempty"`
	Shards             []*Shard `json:"shards"`
	EnhancedMonitoring []string `json:"enhancedMonitoring,omitempty"`
	RetentionPeriod    int      `json:"retentionPeriod"`
	// MaxRecordSizeBytes is the per-record data payload size limit for this stream.
	// Defaults to defaultMaxRecordSizeBytes (1 MiB); updatable via UpdateMaxRecordSize
	// (wire unit is MaxRecordSizeInKiB; converted to bytes on write via bytesPerKiB).
	MaxRecordSizeBytes int `json:"maxRecordSizeBytes,omitempty"`
	// WarmThroughputMiBps is the target; PrevWarmThroughputMiBps is the
	// Current value reported while the stream is UPDATING.
	WarmThroughputMiBps     int `json:"warmThroughputMiBps,omitempty"`
	PrevWarmThroughputMiBps int `json:"prevWarmThroughputMiBps,omitempty"`
	// contains filtered or unexported fields
}

Stream represents an in-memory Kinesis stream.

func (*Stream) MarshalJSON

func (stream *Stream) MarshalJSON() ([]byte, error)

MarshalJSON serialises the stream under stream.mu.RLock, since Snapshot must not race consumer/setting updates. Those mutate stream fields, including Consumers, under stream.mu without ever taking b.mu (gopherstack-fwd0g).

type StreamInfo

type StreamInfo struct {
	Name       string
	ARN        string
	Status     string
	ShardCount int
}

StreamInfo holds summary information about a stream, safe to return without lock.

type StreamModeDetails

type StreamModeDetails struct {
	StreamMode string
}

StreamModeDetails describes the mode of a Kinesis stream.

type StreamSummary

type StreamSummary struct {
	StreamCreationTimestamp time.Time
	StreamName              string
	StreamARN               string
	StreamStatus            string
	StreamMode              string
}

StreamSummary is the optional, richer per-stream shape ListStreamsOutput carries alongside the required StreamNames (types.StreamSummary).

type SubscribeToShardEvent

type SubscribeToShardEvent struct {
	ContinuationSequenceNumber string
	Records                    []GetRecordResult
	MillisBehindLatest         int64
}

SubscribeToShardEvent is a single event in the SubscribeToShard response.

type SubscribeToShardInput

type SubscribeToShardInput struct {
	ConsumerARN      string
	ShardID          string
	StartingPosition StartingPosition
}

SubscribeToShardInput is the input for SubscribeToShard.

type SubscribeToShardOutput

type SubscribeToShardOutput struct {
	Event SubscribeToShardEvent
}

SubscribeToShardOutput is the output for SubscribeToShard.

type TagResourceInput

type TagResourceInput struct {
	Tags        map[string]string
	ResourceARN string
}

TagResourceInput is the input for TagResource (ARN-based tagging).

type TaggedEntry added in v1.2.0

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

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

type UntagResourceInput

type UntagResourceInput struct {
	ResourceARN string
	TagKeys     []string
}

UntagResourceInput is the input for UntagResource (ARN-based tag removal).

type UpdateAccountSettingsInput

type UpdateAccountSettingsInput struct {
	// MinimumThroughputBillingCommitment is required.
	MinimumThroughputBillingCommitment *MinimumThroughputBillingCommitmentInput
}

UpdateAccountSettingsInput is the input for UpdateAccountSettings.

type UpdateAccountSettingsOutput added in v1.3.1

type UpdateAccountSettingsOutput struct {
	MinimumThroughputBillingCommitment MinimumThroughputBillingCommitmentOutput
}

UpdateAccountSettingsOutput is the output for UpdateAccountSettings.

type UpdateChannelInput

type UpdateChannelInput struct {
	LoggingConfiguration             *ChannelCloudWatchLogsConfig
	S3DestinationConfiguration       *ChannelS3Destination
	S3TablesDestinationConfiguration *ChannelS3TablesDestination
	ChannelARN                       string
}

UpdateChannelInput is the input for UpdateChannel. Per the real op's doc comment, only LoggingConfiguration and the active destination's DataFreshnessInSeconds can be changed: "You cannot change the destination, source stream, record format, schema, encryption configuration, or service execution role of an existing channel".

type UpdateChannelOutput

type UpdateChannelOutput struct {
	ChannelDescription Channel
}

UpdateChannelOutput is the output for UpdateChannel.

type UpdateMaxRecordSizeInput

type UpdateMaxRecordSizeInput struct {
	StreamARN string
	// MaxRecordSizeInKiB is required; wire unit is KiB, not bytes.
	MaxRecordSizeInKiB int
}

UpdateMaxRecordSizeInput is the input for UpdateMaxRecordSize. Unlike most stream-identifying inputs in this file, the real shape has no StreamName member -- only StreamARN (and StreamId, reserved for future use) -- kinesis@v1.46.4 api_op_UpdateMaxRecordSize.go:30-47.

type UpdateShardCountInput

type UpdateShardCountInput struct {
	StreamName       string
	ScalingType      string
	TargetShardCount int
}

UpdateShardCountInput is the input for UpdateShardCount.

type UpdateShardCountOutput

type UpdateShardCountOutput struct {
	StreamName        string
	StreamARN         string
	CurrentShardCount int
	TargetShardCount  int
}

UpdateShardCountOutput is the output for UpdateShardCount.

type UpdateStreamModeInput

type UpdateStreamModeInput struct {
	// WarmThroughputMiBps mirrors UpdateStreamModeInput's own field
	// (kinesis@v1.53.0 api_op_UpdateStreamMode.go, "only valid when the
	// stream mode is being updated to on-demand"). Optional on the real
	// SDK (*int32, no "This member is required." doc) -- nil means
	// "not specified" (preserve the stored value); a non-nil value,
	// including 0, is applied.
	WarmThroughputMiBps *int
	StreamARN           string
	StreamModeDetails   StreamModeDetails
}

UpdateStreamModeInput is the input for UpdateStreamMode.

type UpdateStreamWarmThroughputInput

type UpdateStreamWarmThroughputInput struct {
	StreamName string
	StreamARN  string
	// WarmThroughputMiBps is required (api_op_UpdateStreamWarmThroughput.go:63-70).
	WarmThroughputMiBps int
}

UpdateStreamWarmThroughputInput is the input for UpdateStreamWarmThroughput.

type UpdateStreamWarmThroughputOutput added in v1.3.1

type UpdateStreamWarmThroughputOutput struct {
	StreamARN      string
	StreamName     string
	WarmThroughput WarmThroughputObject
}

UpdateStreamWarmThroughputOutput is the output for UpdateStreamWarmThroughput.

type WarmThroughputObject added in v1.3.1

type WarmThroughputObject struct {
	CurrentMiBps int
	TargetMiBps  int
}

WarmThroughputObject mirrors types.WarmThroughputObject (kinesis@v1.46.4 types/types.go:729-740).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL