pipes

package
v1.5.0 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: MIT Imports: 24 Imported by: 0

README

EventBridge Pipes

Parity grade: A · SDK aws-sdk-go-v2/service/pipes@v1.26.4 · last audited 2026-08-23 (7f68d2d24)

Coverage

Metric Value
PARITY entries audited 10 (10 ok)
Feature families 1 (1 ok)
Known gaps 1
Deferred items 0
Resource leaks clean
Known gaps
  • MSK, self-managed Kafka, RabbitMQ, and ActiveMQ pipe sources are modeled in full in CreatePipe/UpdatePipe/DescribePipe wire shapes (sources.go) but are never polled by the runner, and this is a genuine impossibility rather than a deferred implementation: gopherstack has no in-process Kafka-wire-protocol broker or AMQP/OpenWire broker anywhere in the repo to read messages from. Verified by inspecting both candidate backends before writing this line: services/kafka (Amazon MSK) implements only the AWS control-plane HTTP API (CreateCluster/DescribeCluster/GetBootstrapBrokers/topic metadata CRUD) -- confirmed via grep -rl 'func.*Produce\\|func.*Consume\\|func.*SendMessage\\|func.*ReceiveMessage' returning nothing message-plane-shaped; services/mq (Amazon MQ, backs both RabbitMQ and ActiveMQ engine types) is the same shape (broker/user/configuration lifecycle CRUD only, zero produce/consume methods anywhere in the package). Neither package speaks the real wire protocol (Kafka's binary TCP protocol; AMQP 0-9-1 for RabbitMQ; OpenWire/STOMP for ActiveMQ), so even a cluster/broker created via those services' control planes has no data-plane to poll. runner.go's pollPipe routes only SQS/Kinesis/DynamoDB-Streams ARNs and leaves these four source types unrouted (with a doc comment explaining why) rather than faking delivery.

More

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNotFound      = awserr.New("NotFoundException", awserr.ErrNotFound)
	ErrAlreadyExists = awserr.New("ConflictException", awserr.ErrConflict)
	ErrValidation    = awserr.New("ValidationException", awserr.ErrInvalidParameter)
	ErrConflict      = awserr.New("ConflictException", awserr.ErrConflict)
	ErrQuota         = awserr.New("ServiceQuotaExceededException", awserr.ErrConflict)
)
View Source 
var (
	// ErrUnsupportedPipeTarget is returned when a pipe target ARN service is not
	// handled by the runner.
	ErrUnsupportedPipeTarget = errors.New("pipes: unsupported target ARN")
	// ErrUnsupportedPipeEnrichment is returned when a pipe enrichment ARN service
	// is not handled by the runner.
	ErrUnsupportedPipeEnrichment = errors.New("pipes: unsupported enrichment ARN")
	// ErrEnrichmentInvokerUnwired is returned when an enrichment ARN is recognized
	// but the corresponding invoker (Lambda/StepFunctions) has not been wired into
	// the runner. This surfaces a real misconfiguration instead of silently
	// dropping the event.
	ErrEnrichmentInvokerUnwired = errors.New("pipes: enrichment invoker not configured")
	// ErrTargetInvokerUnwired is returned when a target ARN service is recognized but
	// the corresponding invoker has not been wired into the runner.
	ErrTargetInvokerUnwired = errors.New("pipes: target invoker not configured")
)

Functions

This section is empty.

Types

type ActiveMQBrokerSourceParameters 

type ActiveMQBrokerSourceParameters struct {
	Credentials                    *MQBrokerCredentials `json:"Credentials,omitempty"`
	QueueName                      string               `json:"QueueName,omitempty"`
	BatchSize                      int                  `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int                  `json:"MaximumBatchingWindowInSeconds,omitempty"`
}

ActiveMQBrokerSourceParameters holds ActiveMQ broker source configuration.

type AwsVpcConfiguration 

type AwsVpcConfiguration struct {
	AssignPublicIP string   `json:"AssignPublicIp,omitempty"`
	Subnets        []string `json:"Subnets,omitempty"`
	SecurityGroups []string `json:"SecurityGroups,omitempty"`
}

AwsVpcConfiguration is the VPC network configuration for ECS tasks.

type BatchArrayProperties 

type BatchArrayProperties struct {
	Size int `json:"Size,omitempty"`
}

BatchArrayProperties holds Batch array job properties.

type BatchContainerOverrides 

type BatchContainerOverrides struct {
	Environment  []BatchEnvironmentVariable `json:"Environment,omitempty"`
	InstanceType string                     `json:"InstanceType,omitempty"`
	Command      []string                   `json:"Command,omitempty"`
}

BatchContainerOverrides holds container override values for a Batch job.

type BatchEnvironmentVariable added in v1.5.0 

type BatchEnvironmentVariable struct {
	Name  string `json:"Name,omitempty"`
	Value string `json:"Value,omitempty"`
}

BatchEnvironmentVariable is a single name/value pair in BatchContainerOverrides.Environment. The real Batch shape (types.go) is a list of these, not a bare map -- both serializers.go (request) and deserializers.go (response) reuse the same BatchContainerOverrides type, so a map here breaks CreatePipe's request decode for any real client setting environment variables, and would equally break DescribePipe's response decode once fixed on the input side alone.

type BatchJobDependency 

type BatchJobDependency struct {
	JobID string `json:"JobId,omitempty"`
	Type  string `json:"Type,omitempty"`
}

BatchJobDependency represents a dependency between Batch jobs.

type BatchJobTargetParameters 

type BatchJobTargetParameters struct {
	ArrayProperties    *BatchArrayProperties    `json:"ArrayProperties,omitempty"`
	RetryStrategy      *BatchRetryStrategy      `json:"RetryStrategy,omitempty"`
	ContainerOverrides *BatchContainerOverrides `json:"ContainerOverrides,omitempty"`
	Parameters         map[string]string        `json:"Parameters,omitempty"`
	JobDefinition      string                   `json:"JobDefinition,omitempty"`
	JobName            string                   `json:"JobName,omitempty"`
	DependsOn          []BatchJobDependency     `json:"DependsOn,omitempty"`
}

BatchJobTargetParameters holds Batch job target configuration.

type BatchRetryStrategy 

type BatchRetryStrategy struct {
	Attempts int `json:"Attempts,omitempty"`
}

BatchRetryStrategy holds Batch retry configuration.

type CapacityProviderStrategyItem 

type CapacityProviderStrategyItem struct {
	CapacityProvider string `json:"CapacityProvider,omitempty"`
	Weight           int    `json:"Weight,omitempty"`
	Base             int    `json:"Base,omitempty"`
}

CapacityProviderStrategyItem is a single entry in an ECS capacity provider strategy.

type CloudWatchLogsTargetParameters 

type CloudWatchLogsTargetParameters struct {
	LogStreamName string `json:"LogStreamName,omitempty"`
	Timestamp     string `json:"Timestamp,omitempty"`
}

CloudWatchLogsTargetParameters holds CloudWatch Logs target configuration.

type CloudWatchMetricsDestination 

type CloudWatchMetricsDestination struct {
	Namespace string `json:"Namespace,omitempty"`
}

CloudWatchMetricsDestination configures a CloudWatch metrics destination.

type CloudwatchLogsLogDestination 

type CloudwatchLogsLogDestination struct {
	LogGroupArn string `json:"LogGroupArn,omitempty"`
}

CloudwatchLogsLogDestination is a CloudWatch Logs target.

type CreatePipeInput 

type CreatePipeInput struct {
	Tags                    map[string]string
	SourceParameters        *SourceParameters
	TargetParameters        *TargetParameters
	LogConfiguration        *LogConfiguration
	EnrichmentParameters    *EnrichmentParameters
	RuntimeMetricsStreaming *RuntimeMetricsStreaming
	Name                    string
	RoleARN                 string
	Source                  string
	Target                  string
	Description             string
	Enrichment              string
	KmsKeyIdentifier        string
	DesiredState            string
}

CreatePipeInput holds the full set of fields for pipe creation.

type DeadLetterConfig 

type DeadLetterConfig struct {
	Arn string `json:"Arn,omitempty"`
}

DeadLetterConfig identifies the DLQ for failed pipe events.

type DynamoDBStreamRecord added in v1.2.0 

type DynamoDBStreamRecord struct {
	NewImage                    map[string]any
	OldImage                    map[string]any
	Keys                        map[string]any
	EventID                     string
	EventName                   string // INSERT, MODIFY, or REMOVE
	SequenceNumber              string
	StreamViewType              string
	ApproximateCreationDateTime float64 // Unix epoch seconds
	SizeBytes                   int64
}

DynamoDBStreamRecord is a single record read from a DynamoDB stream for a pipe source.

type DynamoDBStreamSourceParameters 

type DynamoDBStreamSourceParameters struct {
	DeadLetterConfig               *DeadLetterConfig `json:"DeadLetterConfig,omitempty"`
	StartingPosition               string            `json:"StartingPosition,omitempty"`
	OnPartialBatchItemFailure      string            `json:"OnPartialBatchItemFailure,omitempty"`
	BatchSize                      int               `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int               `json:"MaximumBatchingWindowInSeconds,omitempty"`
	MaximumRecordAgeInSeconds      int               `json:"MaximumRecordAgeInSeconds,omitempty"`
	MaximumRetryAttempts           int               `json:"MaximumRetryAttempts,omitempty"`
	ParallelizationFactor          int               `json:"ParallelizationFactor,omitempty"`
}

DynamoDBStreamSourceParameters holds DynamoDB stream source configuration.

type EBEventBusTargetParameters 

type EBEventBusTargetParameters struct {
	DetailType string   `json:"DetailType,omitempty"`
	EndpointID string   `json:"EndpointId,omitempty"`
	Source     string   `json:"Source,omitempty"`
	Time       string   `json:"Time,omitempty"`
	Resources  []string `json:"Resources,omitempty"`
}

EBEventBusTargetParameters holds EventBridge event bus target configuration.

type ECSTaskTargetParameters 

type ECSTaskTargetParameters struct {
	NetworkConfiguration     *NetworkConfiguration          `json:"NetworkConfiguration,omitempty"`
	Overrides                *EcsTaskOverride               `json:"Overrides,omitempty"`
	TaskDefinitionArn        string                         `json:"TaskDefinitionArn,omitempty"`
	LaunchType               string                         `json:"LaunchType,omitempty"`
	Group                    string                         `json:"Group,omitempty"`
	PlatformVersion          string                         `json:"PlatformVersion,omitempty"`
	CapacityProviderStrategy []CapacityProviderStrategyItem `json:"CapacityProviderStrategy,omitempty"`
	PlacementConstraints     []PlacementConstraint          `json:"PlacementConstraints,omitempty"`
	PlacementStrategy        []PlacementStrategy            `json:"PlacementStrategy,omitempty"`
	TaskCount                int                            `json:"TaskCount,omitempty"`
	EnableECSManagedTags     bool                           `json:"EnableECSManagedTags,omitempty"`
	EnableExecuteCommand     bool                           `json:"EnableExecuteCommand,omitempty"`
}

ECSTaskTargetParameters holds ECS task target configuration.

type EcsTaskOverride 

type EcsTaskOverride struct {
	TaskRoleArn      string `json:"TaskRoleArn,omitempty"`
	ExecutionRoleArn string `json:"ExecutionRoleArn,omitempty"`
	CPU              string `json:"Cpu,omitempty"`
	Memory           string `json:"Memory,omitempty"`
}

EcsTaskOverride holds override values for an ECS task execution.

type EnrichmentHTTPParameters 

type EnrichmentHTTPParameters struct {
	HeaderParameters      map[string]string `json:"HeaderParameters,omitempty"`
	QueryStringParameters map[string]string `json:"QueryStringParameters,omitempty"`
	PathParameterValues   []string          `json:"PathParameterValues,omitempty"`
}

EnrichmentHTTPParameters holds HTTP parameters for enrichment calls.

type EnrichmentParameters 

type EnrichmentParameters struct {
	HTTPParameters *EnrichmentHTTPParameters `json:"HttpParameters,omitempty"`
	InputTemplate  string                    `json:"InputTemplate,omitempty"`
}

EnrichmentParameters holds enrichment-specific configuration.

type Filter 

type Filter struct {
	Pattern string `json:"Pattern,omitempty"`
}

Filter is a single JSON-pattern filter.

type FilterCriteria 

type FilterCriteria struct {
	Filters []Filter `json:"Filters,omitempty"`
}

FilterCriteria holds event filter patterns applied before forwarding to the target.

type FirehoseLogDestination 

type FirehoseLogDestination struct {
	DeliveryStreamArn string `json:"DeliveryStreamArn,omitempty"`
}

FirehoseLogDestination is a Firehose delivery stream log target.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

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

Handler is the HTTP handler for the EventBridge Pipes REST API.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler creates a new Pipes handler.

func (*Handler) ChaosOperations 

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

func (*Handler) ChaosRegions 

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

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

func (*Handler) ExtractOperation 

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

ExtractOperation determines the operation name from the HTTP request.

func (*Handler) ExtractResource 

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

func (*Handler) GetRunner 

func (h *Handler) GetRunner() *Runner

GetRunner returns the handler's Runner so callers can configure target invokers.

func (*Handler) GetSupportedOperations 

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

GetSupportedOperations returns the list of supported Pipes operations.

func (*Handler) Handler 

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

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all backend state.

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

func (*Handler) Shutdown 

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

Shutdown implements service.Shutdowner. It stops the background runner and cancels any in-flight delayed state-transition goroutines in the backend so they cannot mutate pipe state after the process begins shutting down.

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 implements service.BackgroundWorker.

type InMemoryBackend 

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

InMemoryBackend is the in-memory store for pipes.

All resource collections are nested by region (outer key = region) so that same-named pipes are isolated across regions. The per-region *store.Table / *store.Index values are created lazily via the pipesTable / pipesByARNIndex / enrichmentCountsTable accessors in store_setup.go — mirroring the lazy map creation the hand-rolled code did before Phase 3.3's pkgs/store conversion. Callers must hold b.mu while accessing them.

func NewInMemoryBackend 

func NewInMemoryBackend(accountID, region string) *InMemoryBackend

NewInMemoryBackend creates a new InMemoryBackend with a background lifecycle context. Prefer NewInMemoryBackendWithContext when a service context is available so delayed state transitions are cancelled on shutdown.

func NewInMemoryBackendWithContext 

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

NewInMemoryBackendWithContext creates a new InMemoryBackend whose delayed state-transition goroutines are tied to svcCtx, so they are cancelled when the service shuts down. If svcCtx is nil, context.Background is used.

func (*InMemoryBackend) CreatePipe 

func (b *InMemoryBackend) CreatePipe(ctx context.Context, in CreatePipeInput) (*Pipe, error)

func (*InMemoryBackend) DeletePipe 

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

func (*InMemoryBackend) GetEnrichmentCallCount 

func (b *InMemoryBackend) GetEnrichmentCallCount(ctx context.Context, pipeName string) int64

GetEnrichmentCallCount returns the number of enrichment calls for a pipe.

func (*InMemoryBackend) GetPipe 

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

func (*InMemoryBackend) ListPipes 

func (b *InMemoryBackend) ListPipes(ctx context.Context, f ListPipesFilter) (ListPipesResult, error)

func (*InMemoryBackend) ListTagsForResource 

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

func (*InMemoryBackend) MarkPipeFailed 

func (b *InMemoryBackend) MarkPipeFailed(name, state, reason string)

MarkPipeFailed updates a pipe to a failed state with a reason message. It searches all regions for the named pipe.

func (*InMemoryBackend) RecordEnrichmentCall 

func (b *InMemoryBackend) RecordEnrichmentCall(ctx context.Context, pipeName string)

RecordEnrichmentCall increments the enrichment invocation counter for a pipe.

func (*InMemoryBackend) Region 

func (b *InMemoryBackend) Region() string

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

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

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

Shutdown cancels in-flight delayed state transitions and waits for their goroutines to exit, bounded by ctx. It implements the service shutdown contract used by the handler.

func (*InMemoryBackend) Snapshot 

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

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

func (*InMemoryBackend) StartPipe 

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

func (*InMemoryBackend) StopPipe 

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

func (*InMemoryBackend) TagResource 

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

func (*InMemoryBackend) TaggedResources added in v1.3.1 

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

TaggedResources returns every pipe ARN, across every region, that currently has at least one tag applied via TagResource.

func (*InMemoryBackend) UntagResource 

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

func (*InMemoryBackend) UpdatePipe 

func (b *InMemoryBackend) UpdatePipe(ctx context.Context, name string, in UpdatePipeInput) (*Pipe, error)

type KinesisRecord added in v1.2.0 

type KinesisRecord struct {
	ArrivalTime    time.Time
	PartitionKey   string
	SequenceNumber string
	Data           []byte
}

KinesisRecord is a single record read from a Kinesis stream for a pipe source.

type KinesisStreamSourceParameters 

type KinesisStreamSourceParameters struct {
	StartingPositionTimestamp      *time.Time        `json:"StartingPositionTimestamp,omitempty"`
	DeadLetterConfig               *DeadLetterConfig `json:"DeadLetterConfig,omitempty"`
	StartingPosition               string            `json:"StartingPosition,omitempty"`
	OnPartialBatchItemFailure      string            `json:"OnPartialBatchItemFailure,omitempty"`
	BatchSize                      int               `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int               `json:"MaximumBatchingWindowInSeconds,omitempty"`
	MaximumRecordAgeInSeconds      int               `json:"MaximumRecordAgeInSeconds,omitempty"`
	MaximumRetryAttempts           int               `json:"MaximumRetryAttempts,omitempty"`
	ParallelizationFactor          int               `json:"ParallelizationFactor,omitempty"`
}

KinesisStreamSourceParameters holds Kinesis-specific source configuration.

func (KinesisStreamSourceParameters) MarshalJSON added in v1.5.0 

func (k KinesisStreamSourceParameters) MarshalJSON() ([]byte, error)

MarshalJSON encodes StartingPositionTimestamp as the epoch-seconds JSON number real clients expect, not encoding/json's default RFC3339 string.

func (*KinesisStreamSourceParameters) UnmarshalJSON added in v1.5.0 

func (k *KinesisStreamSourceParameters) UnmarshalJSON(data []byte) error

UnmarshalJSON is the inverse of MarshalJSON. Without it, a real client's StartingPositionTimestamp (a JSON number on the wire) fails time.Time.UnmarshalJSON ("input is not a JSON string"), which rejects the entire CreatePipe/UpdatePipe request body, not just this field.

type KinesisStreamTargetParameters 

type KinesisStreamTargetParameters struct {
	PartitionKey string `json:"PartitionKey,omitempty"`
}

KinesisStreamTargetParameters holds Kinesis stream target configuration.

type LambdaFunctionParameters 

type LambdaFunctionParameters struct {
	InvocationType string `json:"InvocationType,omitempty"`
}

LambdaFunctionParameters holds Lambda-specific target configuration.

type ListPipesFilter 

type ListPipesFilter struct {
	NamePrefix   string
	DesiredState string
	CurrentState string
	SourcePrefix string
	TargetPrefix string
	NextToken    string
	Limit        int
}

ListPipesFilter holds optional query parameters for ListPipes.

type ListPipesResult 

type ListPipesResult struct {
	NextToken string
	Pipes     []*Pipe
}

ListPipesResult holds the paginated result of a ListPipes call.

type LogConfiguration 

type LogConfiguration struct {
	CloudwatchLogsLogDestination *CloudwatchLogsLogDestination `json:"CloudwatchLogsLogDestination,omitempty"`
	FirehoseLogDestination       *FirehoseLogDestination       `json:"FirehoseLogDestination,omitempty"`
	S3LogDestination             *S3LogDestination             `json:"S3LogDestination,omitempty"`
	Level                        string                        `json:"Level,omitempty"`
	IncludeExecutionData         []string                      `json:"IncludeExecutionData,omitempty"`
}

LogConfiguration controls pipe execution logging. Matches the real API's flat shape (aws-sdk-go-v2/service/pipes/types.PipeLogConfiguration): the three destinations are direct top-level fields, not wrapped in a "Destinations" list.

type MQBrokerCredentials 

type MQBrokerCredentials struct {
	BasicAuth string `json:"BasicAuth,omitempty"`
}

MQBrokerCredentials holds credentials for ActiveMQ or RabbitMQ broker sources.

type MSKAccessCredentials 

type MSKAccessCredentials struct {
	ClientCertificateTLSAuth string `json:"ClientCertificateTlsAuth,omitempty"`
	SaslScram512Auth         string `json:"SaslScram512Auth,omitempty"`
}

MSKAccessCredentials holds authentication credentials for MSK sources. Exactly one field is populated (models an AWS union type).

type MSKSourceParameters 

type MSKSourceParameters struct {
	Credentials                    *MSKAccessCredentials `json:"Credentials,omitempty"`
	TopicName                      string                `json:"TopicName,omitempty"`
	StartingPosition               string                `json:"StartingPosition,omitempty"`
	ConsumerGroupID                string                `json:"ConsumerGroupId,omitempty"`
	BatchSize                      int                   `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int                   `json:"MaximumBatchingWindowInSeconds,omitempty"`
}

MSKSourceParameters holds MSK source configuration.

type MetricsDestination 

type MetricsDestination struct {
	CloudwatchMetrics *CloudWatchMetricsDestination `json:"CloudwatchMetrics,omitempty"`
}

MetricsDestination wraps the destination for pipe runtime metrics.

type NetworkConfiguration 

type NetworkConfiguration struct {
	AwsvpcConfiguration *AwsVpcConfiguration `json:"AwsvpcConfiguration,omitempty"`
}

NetworkConfiguration wraps VPC configuration for ECS task targets.

type Pipe 

type Pipe struct {
	SourceParameters        *SourceParameters        `json:"sourceParameters,omitempty"`
	TargetParameters        *TargetParameters        `json:"targetParameters,omitempty"`
	LogConfiguration        *LogConfiguration        `json:"logConfiguration,omitempty"`
	EnrichmentParameters    *EnrichmentParameters    `json:"enrichmentParameters,omitempty"`
	RuntimeMetricsStreaming *RuntimeMetricsStreaming `json:"runtimeMetricsStreaming,omitempty"`
	LastModifiedTime        time.Time                `json:"lastModifiedTime"`
	CreationTime            time.Time                `json:"creationTime"`
	Tags                    map[string]string        `json:"tags,omitempty"`
	Description             string                   `json:"description,omitempty"`
	Enrichment              string                   `json:"enrichment,omitempty"`
	KmsKeyIdentifier        string                   `json:"kmsKeyIdentifier,omitempty"`
	Source                  string                   `json:"source"`
	Target                  string                   `json:"target"`
	RoleARN                 string                   `json:"roleArn"`
	StateReason             string                   `json:"stateReason,omitempty"`
	DesiredState            string                   `json:"desiredState"`
	CurrentState            string                   `json:"currentState"`
	AccountID               string                   `json:"accountID"`
	Region                  string                   `json:"region"`
	ARN                     string                   `json:"arn"`
	Name                    string                   `json:"name"`
}

Pipe represents an EventBridge Pipe.

type PipeCloudWatchLogsPutter 

type PipeCloudWatchLogsPutter interface {
	PutLogEvents(
		ctx context.Context,
		logGroupARN, logStreamName string,
		messages []string,
	) error
}

PipeCloudWatchLogsPutter puts log events to a CloudWatch Logs log group.

type PipeDynamoDBStreamsReader added in v1.2.0 

type PipeDynamoDBStreamsReader interface {
	// DescribeStreamShards returns the ordered list of shard IDs for the stream.
	// streamARN is the full DynamoDB stream ARN.
	DescribeStreamShards(streamARN string) ([]string, error)
	// GetStreamShardIterator returns an iterator for a specific shard of the stream.
	GetStreamShardIterator(streamARN, shardID, iteratorType string) (string, error)
	// GetStreamRecords reads up to limit records from the given iterator.
	GetStreamRecords(iteratorToken string, limit int) ([]DynamoDBStreamRecord, string, error)
}

PipeDynamoDBStreamsReader reads records from a DynamoDB stream for a pipe source.

type PipeEventBridgePutter 

type PipeEventBridgePutter interface {
	PutEvents(ctx context.Context, eventBusARN string, events []map[string]any) error
}

PipeEventBridgePutter puts events to an EventBridge event bus.

type PipeFirehosePutter 

type PipeFirehosePutter interface {
	PutRecord(ctx context.Context, deliveryStreamARN string, data []byte) error
}

PipeFirehosePutter puts a record into a Kinesis Data Firehose delivery stream.

type PipeKinesisPutter 

type PipeKinesisPutter interface {
	PutRecord(ctx context.Context, streamARN, partitionKey string, data []byte) error
}

PipeKinesisPutter puts a record into a Kinesis stream.

type PipeKinesisReader added in v1.2.0 

type PipeKinesisReader interface {
	// GetShardIDs returns the shard IDs for the given stream.
	GetShardIDs(streamName string) ([]string, error)
	// GetShardIterator returns an iterator token for a shard.
	GetShardIterator(streamName, shardID, iteratorType, startingSeqNum string) (string, error)
	// GetRecords reads up to limit records from the given iterator, returning
	// records and the next iterator token.
	GetRecords(iteratorToken string, limit int) ([]KinesisRecord, string, error)
}

PipeKinesisReader reads records from a Kinesis stream for a pipe source.

type PipeLambdaInvoker 

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

PipeLambdaInvoker invokes a Lambda function with a payload.

type PipeStepFunctionsStarter 

type PipeStepFunctionsStarter interface {
	StartExecution(stateMachineARN, name, input string) error
}

PipeStepFunctionsStarter starts a StepFunctions state machine execution.

type PlacementConstraint 

type PlacementConstraint struct {
	Expression string `json:"Expression,omitempty"`
	Type       string `json:"Type,omitempty"`
}

PlacementConstraint is a constraint for ECS task placement.

type PlacementStrategy 

type PlacementStrategy struct {
	Field string `json:"Field,omitempty"`
	Type  string `json:"Type,omitempty"`
}

PlacementStrategy is a placement strategy rule for ECS tasks.

type Provider 

type Provider struct{}

Provider implements service.Provider for EventBridge Pipes.

func (*Provider) Init 

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

Init initializes the Pipes service backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the provider name.

type RabbitMQBrokerSourceParameters 

type RabbitMQBrokerSourceParameters struct {
	Credentials                    *MQBrokerCredentials `json:"Credentials,omitempty"`
	QueueName                      string               `json:"QueueName,omitempty"`
	VirtualHost                    string               `json:"VirtualHost,omitempty"`
	BatchSize                      int                  `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int                  `json:"MaximumBatchingWindowInSeconds,omitempty"`
}

RabbitMQBrokerSourceParameters holds RabbitMQ broker source configuration.

type RedshiftDataTargetParameters 

type RedshiftDataTargetParameters struct {
	Database         string   `json:"Database,omitempty"`
	DBUser           string   `json:"DbUser,omitempty"`
	SecretManagerArn string   `json:"SecretManagerArn,omitempty"`
	StatementName    string   `json:"StatementName,omitempty"`
	Sqls             []string `json:"Sqls,omitempty"`
	WithEvent        bool     `json:"WithEvent,omitempty"`
}

RedshiftDataTargetParameters holds Redshift Data API target configuration.

type Runner 

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

Runner polls pipe sources and forwards records to pipe targets for RUNNING pipes.

func NewRunner 

func NewRunner(backend *InMemoryBackend) *Runner

func (*Runner) SetCloudWatchLogsPutter 

func (r *Runner) SetCloudWatchLogsPutter(c PipeCloudWatchLogsPutter)

func (*Runner) SetDynamoDBStreamsReader added in v1.2.0 

func (r *Runner) SetDynamoDBStreamsReader(d PipeDynamoDBStreamsReader)

func (*Runner) SetEventBridgePutter 

func (r *Runner) SetEventBridgePutter(e PipeEventBridgePutter)

func (*Runner) SetFirehosePutter 

func (r *Runner) SetFirehosePutter(f PipeFirehosePutter)

func (*Runner) SetKinesisPutter 

func (r *Runner) SetKinesisPutter(k PipeKinesisPutter)

func (*Runner) SetKinesisReader added in v1.2.0 

func (r *Runner) SetKinesisReader(k PipeKinesisReader)

func (*Runner) SetLambdaInvoker 

func (r *Runner) SetLambdaInvoker(l PipeLambdaInvoker)

func (*Runner) SetSNSPublisher 

func (r *Runner) SetSNSPublisher(s SNSPublisher)

func (*Runner) SetSQSReader 

func (r *Runner) SetSQSReader(s SQSReader)

func (*Runner) SetSQSSender 

func (r *Runner) SetSQSSender(s SQSSender)

func (*Runner) SetStepFunctionsStarter 

func (r *Runner) SetStepFunctionsStarter(s PipeStepFunctionsStarter)

func (*Runner) Start 

func (r *Runner) Start(ctx context.Context)

func (*Runner) Wait 

func (r *Runner) Wait(ctx context.Context)

Wait blocks until all runner goroutines have exited, or ctx expires.

type RuntimeMetricsStreaming 

type RuntimeMetricsStreaming struct {
	MetricsDestination *MetricsDestination `json:"MetricsDestination,omitempty"`
	Level              string              `json:"Level,omitempty"`
}

RuntimeMetricsStreaming configures runtime metrics streaming for a pipe.

type S3LogDestination 

type S3LogDestination struct {
	BucketName   string `json:"BucketName,omitempty"`
	BucketOwner  string `json:"BucketOwner,omitempty"`
	Prefix       string `json:"Prefix,omitempty"`
	OutputFormat string `json:"OutputFormat,omitempty"`
}

S3LogDestination is an S3 bucket log target.

type SNSPublisher 

type SNSPublisher interface {
	PublishMessage(ctx context.Context, topicARN, message string) error
}

SNSPublisher publishes a message to an SNS topic.

type SQSMessage 

type SQSMessage struct {
	Attributes    map[string]string
	MessageID     string
	ReceiptHandle string
	Body          string
	MD5OfBody     string
}

SQSMessage is a single SQS message pulled by the pipe runner.

type SQSReader 

type SQSReader interface {
	ReceivePipeMessages(queueARN string, maxMessages int) ([]*SQSMessage, error)
	DeletePipeMessages(queueARN string, receiptHandles []string) error
}

SQSReader reads and deletes SQS messages for a pipe source.

type SQSSender 

type SQSSender interface {
	SendMessage(
		ctx context.Context,
		queueARN, body, groupID, dedupID string,
	) error
}

SQSSender sends a message to an SQS queue.

type SQSSourceParameters 

type SQSSourceParameters struct {
	BatchSize                      int `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int `json:"MaximumBatchingWindowInSeconds,omitempty"`
}

SQSSourceParameters holds SQS-specific source configuration.

type SQSTargetParameters 

type SQSTargetParameters struct {
	MessageGroupID         string `json:"MessageGroupId,omitempty"`
	MessageDeduplicationID string `json:"MessageDeduplicationId,omitempty"`
}

SQSTargetParameters holds SQS-specific target configuration.

type SageMakerPipelineParameter 

type SageMakerPipelineParameter struct {
	Name  string `json:"Name,omitempty"`
	Value string `json:"Value,omitempty"`
}

SageMakerPipelineParameter is a name/value pair for a SageMaker pipeline.

type SageMakerPipelineTargetParameters 

type SageMakerPipelineTargetParameters struct {
	PipelineParameterList []SageMakerPipelineParameter `json:"PipelineParameterList,omitempty"`
}

SageMakerPipelineTargetParameters holds SageMaker pipeline target configuration.

type SelfManagedKafkaAccessCredentials 

type SelfManagedKafkaAccessCredentials struct {
	BasicAuth                string `json:"BasicAuth,omitempty"`
	ClientCertificateTLSAuth string `json:"ClientCertificateTlsAuth,omitempty"`
	SaslScram256Auth         string `json:"SaslScram256Auth,omitempty"`
	SaslScram512Auth         string `json:"SaslScram512Auth,omitempty"`
}

SelfManagedKafkaAccessCredentials holds authentication credentials for self-managed Kafka. Exactly one field is populated (models an AWS union type).

type SelfManagedKafkaSourceParameters 

type SelfManagedKafkaSourceParameters struct {
	Credentials                    *SelfManagedKafkaAccessCredentials `json:"Credentials,omitempty"`
	Vpc                            *SelfManagedKafkaVpc               `json:"Vpc,omitempty"`
	TopicName                      string                             `json:"TopicName,omitempty"`
	StartingPosition               string                             `json:"StartingPosition,omitempty"`
	ConsumerGroupID                string                             `json:"ConsumerGroupId,omitempty"`
	ServerRootCaCertificate        string                             `json:"ServerRootCaCertificate,omitempty"`
	AdditionalBootstrapServers     []string                           `json:"AdditionalBootstrapServers,omitempty"`
	BatchSize                      int                                `json:"BatchSize,omitempty"`
	MaximumBatchingWindowInSeconds int                                `json:"MaximumBatchingWindowInSeconds,omitempty"`
}

SelfManagedKafkaSourceParameters holds self-managed Kafka source configuration.

type SelfManagedKafkaVpc 

type SelfManagedKafkaVpc struct {
	SecurityGroup []string `json:"SecurityGroup,omitempty"`
	Subnets       []string `json:"Subnets,omitempty"`
}

SelfManagedKafkaVpc holds VPC configuration for self-managed Kafka connectivity.

type SourceParameters 

type SourceParameters struct {
	FilterCriteria                  *FilterCriteria                   `json:"FilterCriteria,omitempty"`
	SqsQueueParameters              *SQSSourceParameters              `json:"SqsQueueParameters,omitempty"`
	KinesisStreamParameters         *KinesisStreamSourceParameters    `json:"KinesisStreamParameters,omitempty"`
	DynamoDBStreamParameters        *DynamoDBStreamSourceParameters   `json:"DynamoDBStreamParameters,omitempty"`
	ManagedStreamingKafkaParameters *MSKSourceParameters              `json:"ManagedStreamingKafkaParameters,omitempty"`
	SelfManagedKafkaParameters      *SelfManagedKafkaSourceParameters `json:"SelfManagedKafkaParameters,omitempty"`
	RabbitMQBrokerParameters        *RabbitMQBrokerSourceParameters   `json:"RabbitMQBrokerParameters,omitempty"`
	ActiveMQBrokerParameters        *ActiveMQBrokerSourceParameters   `json:"ActiveMQBrokerParameters,omitempty"`
}

SourceParameters holds source-specific configuration.

type StepFunctionTargetParameters 

type StepFunctionTargetParameters struct {
	InvocationType string `json:"InvocationType,omitempty"`
}

StepFunctionTargetParameters holds Step Functions target configuration.

type TaggedEntry added in v1.3.1 

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

TaggedEntry pairs a resource ARN with its tags.

type TargetHTTPParameters 

type TargetHTTPParameters struct {
	HeaderParameters      map[string]string `json:"HeaderParameters,omitempty"`
	QueryStringParameters map[string]string `json:"QueryStringParameters,omitempty"`
	PathParameterValues   []string          `json:"PathParameterValues,omitempty"`
}

TargetHTTPParameters holds HTTP-specific parameters for API Gateway and API destination targets.

type TargetParameters 

type TargetParameters struct {
	LambdaFunctionParameters      *LambdaFunctionParameters          `json:"LambdaFunctionParameters,omitempty"`
	SFNStateMachineParameters     *StepFunctionTargetParameters      `json:"StepFunctionStateMachineParameters,omitempty"`
	SqsQueueParameters            *SQSTargetParameters               `json:"SqsQueueParameters,omitempty"`
	KinesisStreamParameters       *KinesisStreamTargetParameters     `json:"KinesisStreamParameters,omitempty"`
	CloudWatchLogsParameters      *CloudWatchLogsTargetParameters    `json:"CloudWatchLogsParameters,omitempty"`
	EventBridgeEventBusParameters *EBEventBusTargetParameters        `json:"EventBridgeEventBusParameters,omitempty"`
	RedshiftDataParameters        *RedshiftDataTargetParameters      `json:"RedshiftDataParameters,omitempty"`
	SageMakerPipelineParameters   *SageMakerPipelineTargetParameters `json:"SageMakerPipelineParameters,omitempty"`
	BatchJobParameters            *BatchJobTargetParameters          `json:"BatchJobParameters,omitempty"`
	EcsTaskParameters             *ECSTaskTargetParameters           `json:"EcsTaskParameters,omitempty"`
	TimestreamParameters          *TimestreamParameters              `json:"TimestreamParameters,omitempty"`
	HTTPParameters                *TargetHTTPParameters              `json:"HttpParameters,omitempty"`
	InputTemplate                 string                             `json:"InputTemplate,omitempty"`
}

TargetParameters holds target-specific configuration.

type TimestreamDimensionMapping 

type TimestreamDimensionMapping struct {
	DimensionName      string `json:"DimensionName,omitempty"`
	DimensionValue     string `json:"DimensionValue,omitempty"`
	DimensionValueType string `json:"DimensionValueType,omitempty"`
}

TimestreamDimensionMapping maps an event field to a Timestream dimension.

type TimestreamMultiMeasureAttributeMapping 

type TimestreamMultiMeasureAttributeMapping struct {
	MeasureValue              string `json:"MeasureValue,omitempty"`
	MeasureValueType          string `json:"MeasureValueType,omitempty"`
	MultiMeasureAttributeName string `json:"MultiMeasureAttributeName,omitempty"`
}

TimestreamMultiMeasureAttributeMapping maps an event field to a multi-measure attribute.

type TimestreamMultiMeasureMapping 

type TimestreamMultiMeasureMapping struct {
	MultiMeasureName              string                                   `json:"MultiMeasureName,omitempty"`
	MultiMeasureAttributeMappings []TimestreamMultiMeasureAttributeMapping `json:"MultiMeasureAttributeMappings,omitempty"`
}

TimestreamMultiMeasureMapping maps event fields to a Timestream multi-measure record.

type TimestreamParameters 

type TimestreamParameters struct {
	TimeValue             string                           `json:"TimeValue,omitempty"`
	TimeFieldType         string                           `json:"TimeFieldType,omitempty"`
	TimestampFormat       string                           `json:"TimestampFormat,omitempty"`
	EpochTimeUnit         string                           `json:"EpochTimeUnit,omitempty"`
	VersionValue          string                           `json:"VersionValue,omitempty"`
	DimensionMappings     []TimestreamDimensionMapping     `json:"DimensionMappings,omitempty"`
	SingleMeasureMappings []TimestreamSingleMeasureMapping `json:"SingleMeasureMappings,omitempty"`
	MultiMeasureMappings  []TimestreamMultiMeasureMapping  `json:"MultiMeasureMappings,omitempty"`
}

TimestreamParameters holds Timestream target configuration.

type TimestreamSingleMeasureMapping 

type TimestreamSingleMeasureMapping struct {
	MeasureName      string `json:"MeasureName,omitempty"`
	MeasureValue     string `json:"MeasureValue,omitempty"`
	MeasureValueType string `json:"MeasureValueType,omitempty"`
}

TimestreamSingleMeasureMapping maps an event field to a single Timestream measure.

type UpdatePipeInput 

type UpdatePipeInput struct {
	SourceParameters        *SourceParameters
	TargetParameters        *TargetParameters
	LogConfiguration        *LogConfiguration
	EnrichmentParameters    *EnrichmentParameters
	RuntimeMetricsStreaming *RuntimeMetricsStreaming
	Description             *string
	RoleARN                 string
	Target                  string
	Enrichment              string
	KmsKeyIdentifier        string
	DesiredState            string
}

UpdatePipeInput holds the fields that can be updated on an existing pipe.

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.