stream

Documentation

Overview

Package stream implements streamers that publish AWS events periodically. A streamer fetches AWS events periodically and notifies subscribed channels of them.

Index

• func Stream(ctx context.Context, streamer Streamer) error
• type ECSDeployment
• type ECSDeploymentStreamer
  • func NewECSDeploymentStreamer(ecs ECSServiceDescriber, cluster, service string, ...) *ECSDeploymentStreamer
  • func (s *ECSDeploymentStreamer) Close()
  • func (s *ECSDeploymentStreamer) Fetch() (next time.Time, done bool, err error)
  • func (s *ECSDeploymentStreamer) Notify()
  • func (s *ECSDeploymentStreamer) Subscribe() <-chan ECSService
• type ECSService
• type ECSServiceDescriber
• type StackEvent
• type StackEventsDescriber
• type StackSetDescriber
• type StackSetOpEvent
• type StackSetStreamer
  • func NewStackSetStreamer(cfn StackSetDescriber, ssName, opID string, opStartTime time.Time) *StackSetStreamer
  • func (s *StackSetStreamer) Close()
  • func (s *StackSetStreamer) Fetch() (next time.Time, done bool, err error)
  • func (s *StackSetStreamer) InstanceStreamers(cfnClientFor func(region string) StackEventsDescriber) ([]*StackStreamer, error)
  • func (s *StackSetStreamer) Name() string
  • func (s *StackSetStreamer) Notify()
  • func (s *StackSetStreamer) Subscribe() <-chan StackSetOpEvent
• type StackStreamer
  • func NewStackStreamer(cfn StackEventsDescriber, stackID string, csCreationTime time.Time) *StackStreamer
  • func (s *StackStreamer) Close()
  • func (s *StackStreamer) Fetch() (next time.Time, done bool, err error)
  • func (s *StackStreamer) Name() string
  • func (s *StackStreamer) Notify()
  • func (s *StackStreamer) Region() (string, bool)
  • func (s *StackStreamer) Subscribe() <-chan StackEvent
• type Streamer

Functions

func Stream

func Stream(ctx context.Context, streamer Streamer) error

Stream streams event updates by calling Fetch followed with Notify until there are no more events left. If the context is canceled or Fetch errors, then Stream short-circuits and returns the error.

Types

type ECSDeployment

type ECSDeployment struct {
	Status          string
	TaskDefRevision string
	DesiredCount    int
	RunningCount    int
	FailedCount     int
	PendingCount    int
	RolloutState    string
	CreatedAt       time.Time
	UpdatedAt       time.Time
}

ECSDeployment represent an ECS rolling update deployment.

type ECSDeploymentStreamer

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

ECSDeploymentStreamer is a Streamer for ECSService descriptions until the deployment is completed.

func NewECSDeploymentStreamer

func NewECSDeploymentStreamer(ecs ECSServiceDescriber, cluster, service string, deploymentCreationTime time.Time) *ECSDeploymentStreamer

NewECSDeploymentStreamer creates a new ECSDeploymentStreamer that streams service descriptions since the deployment creation time and until the primary deployment is completed.

func (*ECSDeploymentStreamer) Close

func (s *ECSDeploymentStreamer) Close()

Close closes all subscribed channels notifying them that no more events will be sent.

func (*ECSDeploymentStreamer) Fetch

func (s *ECSDeploymentStreamer) Fetch() (next time.Time, done bool, err error)

Fetch retrieves and stores ECSService descriptions since the deployment's creation time until the primary deployment's running count is equal to its desired count. If an error occurs from describe service, returns a wrapped err. Otherwise, returns the time the next Fetch should be attempted.

func (*ECSDeploymentStreamer) Notify

func (s *ECSDeploymentStreamer) Notify()

Notify flushes all new events to the streamer's subscribers.

func (*ECSDeploymentStreamer) Subscribe

func (s *ECSDeploymentStreamer) Subscribe() <-chan ECSService

Subscribe returns a read-only channel that will receive service descriptions from the ECSDeploymentStreamer.

type ECSService

type ECSService struct {
	Deployments         []ECSDeployment
	LatestFailureEvents []string
}

ECSService is a description of an ECS service.

type ECSServiceDescriber

type ECSServiceDescriber interface {
	Service(clusterName, serviceName string) (*ecs.Service, error)
}

ECSServiceDescriber is the interface to describe an ECS service.

type StackEvent

type StackEvent struct {
	LogicalResourceID    string
	PhysicalResourceID   string
	ResourceType         string
	ResourceStatus       string
	ResourceStatusReason string
	Timestamp            time.Time
}

StackEvent is a CloudFormation stack event.

type StackEventsDescriber

type StackEventsDescriber interface {
	DescribeStackEvents(*cloudformation.DescribeStackEventsInput) (*cloudformation.DescribeStackEventsOutput, error)
}

StackEventsDescriber is the CloudFormation interface needed to describe stack events.

type StackSetDescriber

type StackSetDescriber interface {
	InstanceSummaries(name string, opts ...stackset.InstanceSummariesOption) ([]stackset.InstanceSummary, error)
	DescribeOperation(name, opID string) (stackset.Operation, error)
}

StackSetDescriber is the CloudFormation interface needed to describe the health of a stack set operation.

type StackSetOpEvent

type StackSetOpEvent struct {
	Name      string // The name of the stack set.
	Operation stackset.Operation
}

StackSetOpEvent represents a stack set operation status update message.

type StackSetStreamer

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

StackSetStreamer is a Streamer emitting StackSetOpEvent messages for instances under modification.

func NewStackSetStreamer

func NewStackSetStreamer(cfn StackSetDescriber, ssName, opID string, opStartTime time.Time) *StackSetStreamer

NewStackSetStreamer creates a StackSetStreamer for the given stack set name and operation.

func (*StackSetStreamer) Close

func (s *StackSetStreamer) Close()

Close closes all subscribed channels notifying them that no more events will be sent and causes the streamer to no longer accept any new subscribers.

func (*StackSetStreamer) Fetch

func (s *StackSetStreamer) Fetch() (next time.Time, done bool, err error)

Fetch retrieves and stores the latest CloudFormation stack set operation. If an error occurs from describing stack set operation, returns a wrapped error. Otherwise, returns the time the next Fetch should be attempted and whether or not there are more operations to fetch.

func (*StackSetStreamer) InstanceStreamers

func (s *StackSetStreamer) InstanceStreamers(cfnClientFor func(region string) StackEventsDescriber) ([]*StackStreamer, error)

InstanceStreamers initializes Streamers for each stack instance that's in progress part of the stack set. As long as the operation is in progress, [InstanceStreamers] will keep looking for at least one stack instance that's outdated and return only then.

func (*StackSetStreamer) Name

func (s *StackSetStreamer) Name() string

Name returns the CloudFormation stack set's name.

func (*StackSetStreamer) Notify

func (s *StackSetStreamer) Notify()

Notify publishes the stack set's operation description to subscribers only if the content changed from the last time Notify was called.

func (*StackSetStreamer) Subscribe

func (s *StackSetStreamer) Subscribe() <-chan StackSetOpEvent

Subscribe returns a read-only channel to receive stack set operation events.

type StackStreamer

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

StackStreamer is a Streamer for StackEvent events started by a change set.

func NewStackStreamer

func NewStackStreamer(cfn StackEventsDescriber, stackID string, csCreationTime time.Time) *StackStreamer

NewStackStreamer creates a StackStreamer from a cloudformation client, stack name, and the change set creation timestamp.

func (*StackStreamer) Close

func (s *StackStreamer) Close()

Close closes all subscribed channels notifying them that no more events will be sent and causes the streamer to no longer accept any new subscribers.

func (*StackStreamer) Fetch

func (s *StackStreamer) Fetch() (next time.Time, done bool, err error)

Fetch retrieves and stores any new CloudFormation stack events since the ChangeSetCreationTime in chronological order. If an error occurs from describe stack events, returns a wrapped error. Otherwise, returns the time the next Fetch should be attempted and whether or not there are more events to fetch.

func (*StackStreamer) Name

func (s *StackStreamer) Name() string

Name returns the CloudFormation stack's name.

func (*StackStreamer) Notify

func (s *StackStreamer) Notify()

Notify flushes all new events to the streamer's subscribers.

func (*StackStreamer) Region

func (s *StackStreamer) Region() (string, bool)

Region returns the region of the CloudFormation stack. If the region cannot be parsed from the input stack ID, then return "", false.

func (*StackStreamer) Subscribe

func (s *StackStreamer) Subscribe() <-chan StackEvent

Subscribe returns a read-only channel that will receive stack events from the StackStreamer.

type Streamer

type Streamer interface {
	// Fetch fetches events, updates the internal state of the Streamer with new events and returns the next time
	// the Fetch call should be attempted. On failure, Fetch returns an error.
	Fetch() (next time.Time, done bool, err error)

	// Notify publishes all new event updates to subscribers.
	Notify()

	// Close notifies all subscribers that no more events will be sent.
	Close()
}

Streamer is the interface that groups methods to periodically retrieve events, publish them to subscribers, and stop publishing once there are no more events left.