pipes

package
v1.1.4 Latest Latest
Warning

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

 Go to latest Published: Jul 20, 2026 License: MIT Imports: 23 Imported by: 0

README

EventBridge Pipes

Parity grade: B · SDK aws-sdk-go-v2/service/pipes@v1.23.18 · last audited 2026-07-13 (5d5b2188)

Coverage

Metric Value
Operations audited 10 (10 ok)
Feature families 1 (1 ok)
Known gaps 3
Deferred items 0
Resource leaks clean
Known gaps
  • CreatePipe/UpdatePipe do not validate RoleArn is non-empty, though the real SDK marks it required (validateOpCreatePipeInput/validateOpUpdatePipeInput). Deliberately NOT fixed this pass: a real aws-sdk-go-v2 client client-side-validates RoleArn before ever sending the request (smithy validators.go), so this only matters for non-SDK/raw-HTTP callers, and enforcing it breaks ~340 existing subtests across audit_batch1-4/handler_test/isolation_test/etc. that build CreatePipe/UpdatePipe bodies without RoleArn. Low value / high test-churn tradeoff -- revisit only if a raw-HTTP parity test specifically needs it.
  • Runner (runner.go) only polls SQS sources (isSQSARN gate in pollPipe) -- Kinesis, DynamoDB Streams, MSK, self-managed Kafka, RabbitMQ, and ActiveMQ sources are modeled in CreatePipe/UpdatePipe/DescribePipe wire shapes but a RUNNING pipe with one of those sources never actually polls or forwards events. No control-plane bug (Describe/List still report CurrentState=RUNNING correctly), but a real EXECUTION gap. Cross-service follow-up, not in services/pipes/'s edit scope to fix (would need new source-reader adapters in cli.go plus backend hooks in kinesis/dynamodb/kafka-shaped services).
  • cli.go's wirePipesRunner (cli.go:6592) only wires SQS as source and Lambda+StepFunctions as target/enrichment invokers. SNS, SQS, Kinesis, EventBridge, CloudWatchLogs, and Firehose TARGET invokers (Runner.Set{SNSPublisher,SQSSender,KinesisPutter,EventBridgePutter,CloudWatchLogsPutter,FirehosePutter}) and both DLQ senders (sqsSender/sns for handlePipeFailure) are never set, so a RUNNING SQS-sourced pipe targeting any of those services will poll+enrich correctly but every invokeXTarget call returns ErrTargetInvokerUnwired and the source message is left unconsumed (and DLQ delivery silently fails the same way). Cross-service wiring gap in cli.go, out of services/pipes/'s edit scope; needs adapter structs analogous to pipesSQSReaderAdapter/pipesSFNStarterAdapter for each target service's backend.

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  map[string]string `json:"Environment,omitempty"`
	InstanceType string            `json:"InstanceType,omitempty"`
	Command      []string          `json:"Command,omitempty"`
}

BatchContainerOverrides holds container override values for a Batch job.

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
	DeadLetterConfig        *DeadLetterConfig
	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 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) 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 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.

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 {
	Level                string           `json:"Level,omitempty"`
	Destinations         []LogDestination `json:"Destinations,omitempty"`
	IncludeExecutionData []string         `json:"IncludeExecutionData,omitempty"`
}

LogConfiguration controls pipe execution logging.

type LogDestination 

type LogDestination struct {
	CloudwatchLogsLogDestination *CloudwatchLogsLogDestination `json:"CloudwatchLogsLogDestination,omitempty"`
	FirehoseLogDestination       *FirehoseLogDestination       `json:"FirehoseLogDestination,omitempty"`
	S3LogDestination             *S3LogDestination             `json:"S3LogDestination,omitempty"`
}

LogDestination wraps possible log destination types.

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"`
	DeadLetterConfig        *DeadLetterConfig        `json:"deadLetterConfig,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 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 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) 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) 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 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
	DeadLetterConfig        *DeadLetterConfig
	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.