mwaa

package
v1.3.2 Latest Latest
Warning

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

 Go to latest Published: Aug 16, 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.43.4 · last audited 2026-07-23 (e15f163e+uncommitted)

Coverage

Metric Value
Operations audited 12 (10 ok, 2 partial)
Feature families 3 (3 ok)
Known gaps 3
Deferred items 1
Resource leaks clean
Known gaps
  • CreateWebLoginToken does not populate AirflowIdentity/IamIdentity (real AWS returns the calling IAM identity's username/ARN). Re-investigated end-to-end this pass rather than assuming the blocker (the codedeploy/mgn siblingServices/GetXHandler() pattern exists and *CLI already exposes GetSTSHandler()/GetIAMHandler() at cli.go:1043-1046). The actual missing piece is upstream of any mwaa-specific wiring: gopherstack has NO per-request caller-identity plumbing anywhere in the codebase. Confirmed by grepping every ctxval.NewKey call site (repo-wide): only two context keys exist at all, pkgs/awsmeta (Account/Region/Partition/RequestID -- no principal/ARN field) and pkgs/logger. pkgs/httputils/sigv4.go's SigV4Validator parses the Authorization header's Credential (which contains the access-key-id) purely to verify the signature and explicitly discards it afterward -- its own doc comment states "the access-key-id in the request is informational only -- gopherstack is a single-tenant simulator". So even though services/sts.GetCallerIdentity(accessKeyID, sessionToken) exists and could resolve an access-key-id to an ARN, no access-key-id is ever threaded from the request into a handler's context anywhere in gopherstack today. Making IamIdentity real requires NEW cross-cutting plumbing (a context key carrying the parsed Credential access-key-id, populated by an Echo middleware, consumed via a services/mwaa/cross_service.go siblingServices accessor into STS) -- not a mwaa-local fix. AirflowIdentity is a second, independent gap on top of that: it requires mapping the resolved IAM principal to an Airflow RBAC username, which AWS derives from environment-specific IdP/role-mapping configuration that has no field anywhere in mwaa's Environment model (verified: models.go has no such member). Populating either field with a fabricated value would violate the no-fabricated-data rule, so both are left absent.
  • InvokeRestApi always synthesizes a 200 success with an empty RestApiResponse for any AVAILABLE environment, regardless of the caller-supplied Path/Method. Re-investigated this pass against botocore's mwaa/2020-07-01/service-2.json (not just the Go SDK): the operation's HTTP binding is "responseCode": 200 for the success shape, so the AWS-transport-level 200 gopherstack already returns is not itself wrong -- real MWAA's InvokeRestApiOutput/RestApiClientException/RestApiServerException shapes ALL carry the same RestApiStatusCode/RestApiResponse pair (types/errors.go:94-150), meaning the actual downstream Airflow HTTP status (e.g. 404 for an unknown path, 405 for a wrong method) is meant to be surfaced as data inside that pair, not necessarily as a distinct SDK-visible exception -- and the SDK model does not document which of {success w/ non-2xx RestApiStatusCode, RestApiClientException, RestApiServerException} a given downstream failure maps to. Enumerating the real Apache Airflow REST API's actual path/method surface (which varies by AirflowVersion: /api/v1 for Airflow 2.x, /api/v2 for Airflow 3.x per the AWS user guide) to decide per-request which of those three shapes applies would mean inventing a route table gopherstack cannot verify -- exactly the fabrication class today's campaign already reverted once (an invented xray formula). Declined to implement path/method-based rejection for this reason; RestAPIStatusCode remains a fixed, documented mock simplification (see Notes) rather than a per-path guess.
  • 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) TaggedResources added in v1.3.1 

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

TaggedResources returns every MWAA environment ARN that currently has at least one tag applied via TagResource.

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 TaggedEntry added in v1.3.1 

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

TaggedEntry pairs a resource ARN with its tags.

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.