mwaa

package
v1.3.0 Latest Latest
Warning

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

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

README

Managed Workflows for Apache Airflow

Parity grade: A · SDK aws-sdk-go-v2/service/mwaa@v1.40.1 · last audited 2026-07-23 (e15f163e+uncommitted)

Coverage

Metric Value
Operations audited 12 (10 ok, 2 partial)
Feature families 3 (3 ok)
Known gaps 4
Deferred items 1
Resource leaks clean
Known gaps
  • CreateWebLoginToken does not populate AirflowIdentity/IamIdentity (real AWS returns the calling IAM identity's username/ARN). No caller-identity extraction helper exists in pkgs/ or is derivable within services/mwaa alone (STS's assumed-role/session tracking lives in services/sts and is out of this audit's edit scope); populating these with fabricated values would violate the no-fabricated-data rule, so they are left absent rather than invented. Re-confirmed this pass: grepped pkgs/ for CallerIdentity/AssumedRoleUser/GetCallerIdentity/ExtractCallerIdentity helpers -- none exist. Candidate follow-up: a shared pkgs/ helper that derives caller identity from the SigV4 Authorization header, usable by any service that needs it.
  • InvokeRestApi always synthesizes a 200 success with an empty RestApiResponse for any AVAILABLE environment, regardless of the caller-supplied Path/Method; it never returns RestApiClientException/RestApiServerException (which real AWS returns when the underlying Airflow REST call itself 4xx/5xx's). Faithfully emulating arbitrary Airflow REST API behavior per-path is out of scope for this pass. This pass DID fix the adjacent, previously-missing AVAILABLE-status precondition (see ops.InvokeRestApi note) -- the always-200-for-available-envs limitation is the only remaining piece.
  • EnvironmentClass mw1.micro is now accepted (fixed this pass), but its AWS-documented special-case default/bounds for MaxWebservers/MinWebservers ("Defaults to 2 for all environment sizes except mw1.micro, which defaults to 1") are NOT modeled -- gopherstack applies the same default (2) and bounds (1-5) to mw1.micro as every other class. Confirmed via aws-sdk-go-v2/service/mwaa@v1.40.1/types/types.go's MaxWebservers/MinWebservers doc comments. Narrow, newly-discovered gap; not fixed this pass to keep the NetworkConfiguration/WorkerReplacementStrategy/status fixes (which have real client-facing impact) as the priority.
  • MethodNotAllowedException (405) is used for HTTP-verb mismatches on matched MWAA path prefixes (e.g. GET /clitoken/{name}). This exception name is not part of the real MWAA API model, but the code path is unreachable by any conformant aws-sdk-go-v2 client (which always sends the correct verb per operation) -- and the same pattern is used consistently across 15+ other gopherstack services (apigatewayv2, pinpoint, lambda, opensearch, etc.), so it was left as-is rather than special-cased here.
Deferred
  • Chaos/fault-injection interaction with this pass's status-constant and NetworkConfiguration-validation changes (not re-audited; ChaosOperations() surface is GetSupportedOperations() minus nothing new -- it shrank by one entry this pass since GetMetrics was removed, see Notes).

More

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrEnvironmentNotFound is returned when an environment does not exist.
	ErrEnvironmentNotFound = awserr.New("ResourceNotFoundException: environment not found", awserr.ErrNotFound)
	// ErrEnvironmentAlreadyExists is returned when an environment already exists.
	// The message deliberately does not say "AlreadyExistsException" -- MWAA's
	// API model has no such exception (see handler.go's writeEnvironmentResult),
	// so this sentinel is mapped to a real ValidationException/400 on the wire;
	// embedding the fabricated exception name in the message text would leak it
	// right back into the response body's "message" field.
	ErrEnvironmentAlreadyExists = awserr.New(
		"ValidationException: environment already exists",
		awserr.ErrAlreadyExists,
	)
	// ErrInvalidParameter is returned when an invalid or missing parameter is provided.
	ErrInvalidParameter = awserr.New("ValidationException: invalid parameter", awserr.ErrInvalidParameter)
)

Errors used by the backend.

View Source 
var ErrNilAppContext = errors.New("nil AppContext passed to MWAA Provider.Init")

ErrNilAppContext is returned by Init when a nil AppContext is passed.

Functions

This section is empty.

Types

type Dimension 

type Dimension struct {
	Name  string `json:"Name"`
	Value string `json:"Value"`
}

Dimension represents an internal MWAA metric dimension.

type Environment 

type Environment struct {
	Tags                         map[string]string     `json:"Tags,omitempty"`
	NetworkConfiguration         *NetworkConfig        `json:"NetworkConfiguration,omitempty"`
	AirflowConfigurationOptions  map[string]string     `json:"AirflowConfigurationOptions,omitempty"`
	LoggingConfiguration         *LoggingConfiguration `json:"LoggingConfiguration,omitempty"`
	LastUpdate                   *LastUpdate           `json:"LastUpdate,omitempty"`
	Name                         string                `json:"Name"`
	AirflowVersion               string                `json:"AirflowVersion"`
	ExecutionRoleArn             string                `json:"ExecutionRoleArn"`
	SourceBucketArn              string                `json:"SourceBucketArn"`
	EnvironmentClass             string                `json:"EnvironmentClass"`
	WebserverURL                 string                `json:"WebserverUrl"`
	WebserverAccessMode          string                `json:"WebserverAccessMode"`
	DagS3Path                    string                `json:"DagS3Path"`
	Status                       string                `json:"Status"`
	ARN                          string                `json:"Arn"`
	KmsKey                       string                `json:"KmsKey,omitempty"`
	PluginsS3Path                string                `json:"PluginsS3Path,omitempty"`
	PluginsS3ObjectVersion       string                `json:"PluginsS3ObjectVersion,omitempty"`
	RequirementsS3Path           string                `json:"RequirementsS3Path,omitempty"`
	RequirementsS3ObjectVersion  string                `json:"RequirementsS3ObjectVersion,omitempty"`
	StartupScriptS3Path          string                `json:"StartupScriptS3Path,omitempty"`
	StartupScriptS3ObjectVersion string                `json:"StartupScriptS3ObjectVersion,omitempty"`
	EndpointManagement           string                `json:"EndpointManagement,omitempty"`
	ServiceRoleArn               string                `json:"ServiceRoleArn,omitempty"`
	CeleryExecutorQueue          string                `json:"CeleryExecutorQueue,omitempty"`
	DatabaseVpcEndpointService   string                `json:"DatabaseVpcEndpointService,omitempty"`
	WebserverVpcEndpointService  string                `json:"WebserverVpcEndpointService,omitempty"`
	WeeklyMaintenanceWindowStart string                `json:"WeeklyMaintenanceWindowStart,omitempty"`
	CreatedAt                    float64               `json:"CreatedAt"`
	MaxWorkers                   int32                 `json:"MaxWorkers"`
	MinWorkers                   int32                 `json:"MinWorkers"`
	MaxWebservers                int32                 `json:"MaxWebservers,omitempty"`
	MinWebservers                int32                 `json:"MinWebservers,omitempty"`
	Schedulers                   int32                 `json:"Schedulers,omitempty"`
	// contains filtered or unexported fields
}

Environment represents an MWAA environment.

type ExportedCreateEnvironmentRequest 

type ExportedCreateEnvironmentRequest = createEnvironmentRequest

ExportedCreateEnvironmentRequest aliases createEnvironmentRequest for testing.

type ExportedEnvironment 

type ExportedEnvironment = Environment

ExportedEnvironment is a compatibility alias used by the dashboard package.

type ExportedInvokeRestAPIRequest 

type ExportedInvokeRestAPIRequest = invokeRestAPIRequest

ExportedInvokeRestAPIRequest aliases invokeRestAPIRequest for testing.

type ExportedMetricDatum 

type ExportedMetricDatum = MetricDatum

ExportedMetricDatum aliases MetricDatum for testing.

type ExportedPublishMetricsRequest 

type ExportedPublishMetricsRequest = publishMetricsRequest

ExportedPublishMetricsRequest aliases publishMetricsRequest for testing.

type ExportedUpdateEnvironmentRequest 

type ExportedUpdateEnvironmentRequest = updateEnvironmentRequest

ExportedUpdateEnvironmentRequest aliases updateEnvironmentRequest for testing.

type Handler 

type Handler struct {
	Backend       StorageBackend
	AccountID     string
	DefaultRegion string
}

Handler is the HTTP handler for the AWS MWAA REST API.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new MWAA 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 handler handles.

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

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

func (*Handler) ExtractOperation 

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

ExtractOperation extracts the operation name from the request path and method.

func (*Handler) ExtractResource 

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

ExtractResource extracts the environment name or ARN from the request path.

func (*Handler) GetSupportedOperations 

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

GetSupportedOperations returns the list of supported MWAA operations.

func (*Handler) Handler 

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

Handler returns the echo.HandlerFunc for this service.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset resets the handler's 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

RouteMatcher returns a function that matches MWAA API requests. All path-based matches are gated on the SigV4 service name to prevent routing conflicts with other services that share similar REST paths.

func (*Handler) ServeHTTP 

func (h *Handler) ServeHTTP(c *echo.Context) error

ServeHTTP dispatches MWAA API requests.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

h.Backend is the StorageBackend interface, which already declares Snapshot(ctx context.Context) []byte (see interfaces.go) with a shape matching persistence.Persistable exactly, so this can call it directly -- no local type assertion needed. But interface membership alone does not help: h.Backend is a named field, not an embedded one, so InMemoryBackend's methods are never promoted onto *Handler. Without this delegation, cli.go's setupPersistence type-asserts the registered service.Registerable (this *Handler) against persistence.Persistable, fails silently, and never registers mwaa for snapshot/restore despite the backend being fully capable. Mirrors services/securityhub's Handler-level delegation.

type InMemoryBackend 

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

InMemoryBackend is the in-memory implementation of StorageBackend.

func NewInMemoryBackend 

func NewInMemoryBackend(region, accountID string) *InMemoryBackend

NewInMemoryBackend creates a new MWAA in-memory backend.

func (*InMemoryBackend) AccountID 

func (b *InMemoryBackend) AccountID() string

AccountID returns the configured account ID.

func (*InMemoryBackend) AddEnvironmentInternal 

func (b *InMemoryBackend) AddEnvironmentInternal(name string) *Environment

AddEnvironmentInternal creates an environment with minimal defaults, bypassing validation, intended for use in tests only. It uses the backend's default region.

func (*InMemoryBackend) AddEnvironmentInternalRegion 

func (b *InMemoryBackend) AddEnvironmentInternalRegion(region, name string) *Environment

AddEnvironmentInternalRegion creates an environment with minimal defaults in the given region, bypassing validation, intended for use in tests only.

func (*InMemoryBackend) CreateCliToken 

func (b *InMemoryBackend) CreateCliToken(ctx context.Context, envName string) (string, string, error)

CreateCliToken validates that the environment exists and is AVAILABLE, then returns a JWT-shaped CLI token and the environment's webserver hostname. AWS returns ResourceNotFoundException when the environment is in any non-AVAILABLE state.

func (*InMemoryBackend) CreateEnvironment 

func (b *InMemoryBackend) CreateEnvironment(
	ctx context.Context,
	name string,
	req *createEnvironmentRequest,
) (*Environment, error)

CreateEnvironment creates a new MWAA environment in the region resolved from ctx.

func (*InMemoryBackend) CreateWebLoginToken 

func (b *InMemoryBackend) CreateWebLoginToken(ctx context.Context, envName string) (string, string, error)

CreateWebLoginToken validates that the environment exists and is AVAILABLE, then returns a JWT-shaped web login token and the environment's webserver hostname. AWS returns ResourceNotFoundException when the environment is in any non-AVAILABLE state.

func (*InMemoryBackend) DeleteEnvironment 

func (b *InMemoryBackend) DeleteEnvironment(ctx context.Context, name string) (*Environment, error)

DeleteEnvironment deletes an MWAA environment by name and cascades to metrics.

func (*InMemoryBackend) GetEnvironment 

func (b *InMemoryBackend) GetEnvironment(ctx context.Context, name string) (*Environment, error)

GetEnvironment retrieves a deep copy of an MWAA environment by name.

func (*InMemoryBackend) GetMetrics 

func (b *InMemoryBackend) GetMetrics(ctx context.Context, envName string) ([]MetricDatum, error)

GetMetrics returns the stored metrics for the specified environment.

func (*InMemoryBackend) InvokeRestAPI 

func (b *InMemoryBackend) InvokeRestAPI(
	ctx context.Context,
	envName string,
	req *invokeRestAPIRequest,
) (*InvokeRestAPIResponse, error)

InvokeRestAPI simulates calling the Apache Airflow REST API on the specified environment's webserver. Like CreateCliToken/CreateWebLoginToken (the other two operations that reach the environment's Airflow webserver), the environment must be AVAILABLE: the webserver process doesn't exist yet while an environment is CREATING/UPDATING/etc, so AWS returns ResourceNotFoundException for any non-AVAILABLE state, not just a missing name.

func (*InMemoryBackend) ListEnvironments 

func (b *InMemoryBackend) ListEnvironments(ctx context.Context) ([]string, error)

ListEnvironments returns a sorted list of environment names.

func (*InMemoryBackend) ListEnvironmentsPage 

func (b *InMemoryBackend) ListEnvironmentsPage(
	ctx context.Context,
	nextToken string,
	pageSize int,
) ([]string, string, error)

ListEnvironmentsPage returns a paginated, sorted list of environment names. pageSize is clamped to [1, listEnvMaxPageSize]; 0 falls back to listEnvDefaultPageSize. nextToken is the name of the first environment to include in this page (exclusive start cursor of the previous page); empty starts at the beginning.

func (*InMemoryBackend) ListTagsForResource 

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

ListTagsForResource returns all tags for a resource identified by its ARN.

func (*InMemoryBackend) PublishMetrics 

func (b *InMemoryBackend) PublishMetrics(ctx context.Context, envName string, req *publishMetricsRequest) error

PublishMetrics stores internal environment metrics for the specified environment. The total number of metrics per environment is capped at maxMetricsPerEnv.

func (*InMemoryBackend) Region 

func (b *InMemoryBackend) Region() string

Region returns the configured region.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset closes the current mutex and reinitialises all state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a JSON snapshot.

func (*InMemoryBackend) Snapshot 

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

Snapshot serialises the backend state to JSON.

func (*InMemoryBackend) TagResource 

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

TagResource adds or updates tags on a resource identified by its ARN.

func (*InMemoryBackend) UntagResource 

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

UntagResource removes tags from a resource identified by its ARN.

func (*InMemoryBackend) UpdateEnvironment 

func (b *InMemoryBackend) UpdateEnvironment(
	ctx context.Context,
	name string,
	req *updateEnvironmentRequest,
) (*Environment, error)

UpdateEnvironment updates an existing MWAA environment.

type InvokeRestAPIResponse 

type InvokeRestAPIResponse struct {
	RestAPIResponse   any   `json:"RestApiResponse"`
	RestAPIStatusCode int32 `json:"RestApiStatusCode"`
}

InvokeRestAPIResponse is the response body for InvokeRestAPI.

type LastUpdate 

type LastUpdate struct {
	Error                     *UpdateError `json:"Error,omitempty"`
	Status                    string       `json:"Status,omitempty"`
	Source                    string       `json:"Source,omitempty"`
	WorkerReplacementStrategy string       `json:"WorkerReplacementStrategy,omitempty"`
	CreatedAt                 float64      `json:"CreatedAt,omitempty"`
}

LastUpdate captures the result of the most recent environment update.

type LoggingConfiguration 

type LoggingConfiguration struct {
	DagProcessingLogs *ModuleLoggingConfiguration `json:"DagProcessingLogs,omitempty"`
	SchedulerLogs     *ModuleLoggingConfiguration `json:"SchedulerLogs,omitempty"`
	TaskLogs          *ModuleLoggingConfiguration `json:"TaskLogs,omitempty"`
	WebserverLogs     *ModuleLoggingConfiguration `json:"WebserverLogs,omitempty"`
	WorkerLogs        *ModuleLoggingConfiguration `json:"WorkerLogs,omitempty"`
}

LoggingConfiguration aggregates the five Airflow module logging configs.

type MetricDatum 

type MetricDatum struct {
	StatisticValues *StatisticSet `json:"StatisticValues,omitempty"`
	Timestamp       *float64      `json:"Timestamp,omitempty"`
	Value           *float64      `json:"Value,omitempty"`
	MetricName      string        `json:"MetricName"`
	Unit            string        `json:"Unit,omitempty"`
	Dimensions      []Dimension   `json:"Dimensions,omitempty"`
}

MetricDatum represents a single metric data point for PublishMetrics.

type ModuleLoggingConfiguration 

type ModuleLoggingConfiguration struct {
	Enabled               *bool  `json:"Enabled,omitempty"`
	LogLevel              string `json:"LogLevel,omitempty"`
	CloudWatchLogGroupArn string `json:"CloudWatchLogGroupArn,omitempty"`
}

ModuleLoggingConfiguration is a single Airflow module logging configuration.

type NetworkConfig 

type NetworkConfig struct {
	SecurityGroupIDs []string `json:"SecurityGroupIds"`
	SubnetIDs        []string `json:"SubnetIds"`
}

NetworkConfig holds the VPC networking configuration.

type Provider 

type Provider struct{}

Provider implements service.Provider for the MWAA service.

func (*Provider) Init 

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

Init initializes the MWAA service backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the provider name.

type StatisticSet 

type StatisticSet struct {
	Maximum     *float64 `json:"Maximum,omitempty"`
	Minimum     *float64 `json:"Minimum,omitempty"`
	SampleCount *int32   `json:"SampleCount,omitempty"`
	Sum         *float64 `json:"Sum,omitempty"`
}

StatisticSet represents the statistical values for a metric.

type StorageBackend 

type StorageBackend interface {
	// Environment CRUD
	CreateEnvironment(ctx context.Context, name string, req *createEnvironmentRequest) (*Environment, error)
	GetEnvironment(ctx context.Context, name string) (*Environment, error)
	DeleteEnvironment(ctx context.Context, name string) (*Environment, error)
	UpdateEnvironment(ctx context.Context, name string, req *updateEnvironmentRequest) (*Environment, error)
	ListEnvironments(ctx context.Context) ([]string, error)
	ListEnvironmentsPage(ctx context.Context, nextToken string, pageSize int) ([]string, string, error)

	// Tag operations
	TagResource(ctx context.Context, resourceARN string, tags map[string]string) error
	UntagResource(ctx context.Context, resourceARN string, tagKeys []string) error
	ListTagsForResource(ctx context.Context, resourceARN string) (map[string]string, error)

	// REST API / metrics
	InvokeRestAPI(ctx context.Context, envName string, req *invokeRestAPIRequest) (*InvokeRestAPIResponse, error)
	PublishMetrics(ctx context.Context, envName string, req *publishMetricsRequest) error
	GetMetrics(ctx context.Context, envName string) ([]MetricDatum, error)

	// Token operations — return (token, webserverHostname, error).
	CreateCliToken(ctx context.Context, envName string) (string, string, error)
	CreateWebLoginToken(ctx context.Context, envName string) (string, string, error)

	// Lifecycle
	Reset()
	Region() string
	AccountID() string
	Snapshot(ctx context.Context) []byte
	Restore(ctx context.Context, data []byte) error
}

StorageBackend is the interface for the MWAA in-memory backend. All per-resource operations take a context.Context carrying the request's AWS region so resources are isolated per region.

type UpdateError 

type UpdateError struct {
	ErrorCode    string `json:"ErrorCode,omitempty"`
	ErrorMessage string `json:"ErrorMessage,omitempty"`
}

UpdateError describes a failed update attempt.

type UpdateNetworkConfig 

type UpdateNetworkConfig struct {
	SecurityGroupIDs []string `json:"SecurityGroupIds"`
}

UpdateNetworkConfig is the network configuration shape accepted by UpdateEnvironment. Unlike NetworkConfig (used by CreateEnvironment and returned by GetEnvironment), AWS's UpdateNetworkConfigurationInput shape has NO SubnetIds member -- subnets cannot be changed after an environment is created, only SecurityGroupIds can.

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.