"""Serverless mode implementation for Parsl Ephemeral AWS Provider.
This mode uses AWS Lambda and ECS/Fargate for executing jobs without EC2 instances,
providing cost-effective serverless execution for suitable workloads.
SPDX-License-Identifier: Apache-2.0
SPDX-FileCopyrightText: 2025-2026 Scott Friedman and Project Contributors
"""
import logging
import time
from typing import Any, Dict, List, Optional
import boto3
from botocore.exceptions import ClientError
from parsl_ephemeral_provider.constants import (
DEFAULT_INSTANCE_TYPE,
DEFAULT_REGION,
LAUNCH_TEMPLATE_NAME_PREFIX,
RESOURCE_TYPE_LAMBDA_FUNCTION,
RESOURCE_TYPE_ECS_TASK,
RESOURCE_TYPE_SPOT_FLEET,
WORKER_TYPE_LAMBDA,
WORKER_TYPE_ECS,
WORKER_TYPE_AUTO,
STATUS_PENDING,
STATUS_RUNNING,
STATUS_SUCCEEDED,
STATUS_FAILED,
STATUS_CANCELLED,
STATUS_COMPLETED,
STATUS_INTERRUPTED,
STATUS_UNKNOWN,
DEFAULT_LAMBDA_TIMEOUT,
DEFAULT_LAMBDA_MEMORY,
DEFAULT_LAMBDA_RUNTIME,
DEFAULT_LAMBDA_HANDLER,
DEFAULT_ECS_CPU,
DEFAULT_ECS_MEMORY,
DEFAULT_ECS_CONTAINER_IMAGE,
)
from parsl_ephemeral_provider.exceptions import (
ConfigurationError,
JobSubmissionError,
OperatingModeError,
ResourceCreationError,
)
from parsl_ephemeral_provider.modes.base import OperatingMode
from parsl_ephemeral_provider.state.base import STATE_KEY_MODE
from parsl_ephemeral_provider.compute.lambda_func import LambdaManager
from parsl_ephemeral_provider.compute.ecs import ECSManager
from parsl_ephemeral_provider.compute.spot_interruption import SpotInterruptionMonitor
from parsl_ephemeral_provider.utils.aws import (
architecture_for_instance_type,
build_fleet_launch_template_configs,
build_launch_template_data,
create_ec2_fleet,
create_launch_template,
delete_ec2_fleet,
delete_launch_template,
describe_ec2_fleet,
get_cf_template,
get_default_ami,
get_ec2_fleet_instance_ids,
)
logger = logging.getLogger(__name__)
[docs]
class ServerlessMode(OperatingMode):
"""Serverless operating mode implementation.
In serverless mode, AWS Lambda and/or ECS/Fargate are used to execute tasks
without any EC2 instances. This mode is suitable for event-driven or sporadic
workloads with short-running tasks. It also supports EC2 SpotFleet for improved
reliability and cost savings when more substantial compute resources are needed.
Attributes
----------
worker_type : str
Type of worker to use (lambda, ecs, or auto)
lambda_timeout : int
Timeout for Lambda functions in seconds
lambda_memory : int
Memory for Lambda functions in MB
ecs_task_cpu : int
CPU units for ECS tasks
ecs_task_memory : int
Memory for ECS tasks in MB
ecs_container_image : str
Container image for ECS tasks
use_spot : bool
Whether to use spot instances for ECS tasks (Fargate Spot)
use_spot_fleet : bool
Whether to use Spot Fleet for EC2 instance deployment
instance_types : List[str]
List of instance types to use with Spot Fleet
nodes_per_block : int
Number of nodes per block for Spot Fleet
spot_max_price_percentage : Optional[float]
Maximum spot price as percentage of on-demand price
lambda_code_bucket : Optional[str]
Caller-supplied S3 bucket for staging Lambda deployment packages
lambda_manager : LambdaManager
Manager for Lambda functions
ecs_manager : ECSManager
Manager for ECS tasks
"""
[docs]
def __init__(
self,
provider_id: str,
session: boto3.Session,
state_store: Any,
worker_type: str = WORKER_TYPE_AUTO,
lambda_timeout: int = DEFAULT_LAMBDA_TIMEOUT,
lambda_memory: int = DEFAULT_LAMBDA_MEMORY,
lambda_runtime: str = DEFAULT_LAMBDA_RUNTIME,
ecs_task_cpu: int = DEFAULT_ECS_CPU,
ecs_task_memory: int = DEFAULT_ECS_MEMORY,
ecs_container_image: str = DEFAULT_ECS_CONTAINER_IMAGE,
vpc_id: Optional[str] = None,
subnet_id: Optional[str] = None,
security_group_id: Optional[str] = None,
use_public_ips: bool = True,
use_spot: bool = False,
use_spot_fleet: bool = False,
instance_types: Optional[List[str]] = None,
nodes_per_block: int = 1,
spot_max_price_percentage: Optional[float] = None,
additional_tags: Optional[Dict[str, str]] = None,
debug: bool = False,
compute_type: Optional[str] = None,
memory_size: Optional[int] = None,
timeout: Optional[int] = None,
lambda_code_bucket: Optional[str] = None,
**kwargs: Any,
) -> None:
"""Initialize the serverless mode.
Parameters
----------
provider_id : str
Unique identifier for the provider instance
session : boto3.Session
AWS session for API calls
state_store : Any
Store for persisting state
worker_type : str, optional
Type of worker to use (lambda, ecs, or auto), by default WORKER_TYPE_AUTO
lambda_timeout : int, optional
Timeout for Lambda functions in seconds, by default DEFAULT_LAMBDA_TIMEOUT
lambda_memory : int, optional
Memory for Lambda functions in MB, by default DEFAULT_LAMBDA_MEMORY
lambda_runtime : str, optional
Runtime for Lambda functions, by default DEFAULT_LAMBDA_RUNTIME
ecs_task_cpu : int, optional
CPU units for ECS tasks, by default DEFAULT_ECS_CPU
ecs_task_memory : int, optional
Memory for ECS tasks in MB, by default DEFAULT_ECS_MEMORY
ecs_container_image : str, optional
Container image for ECS tasks, by default DEFAULT_ECS_CONTAINER_IMAGE
vpc_id : Optional[str], optional
Existing VPC ID to use, by default None
subnet_id : Optional[str], optional
Existing subnet ID to use, by default None
security_group_id : Optional[str], optional
Existing security group ID to use, by default None
use_public_ips : bool, optional
Whether to assign public IPs to ECS tasks, by default True
use_spot : bool, optional
Whether to use spot instances for ECS tasks (Fargate Spot), by default False
use_spot_fleet : bool, optional
Whether to use Spot Fleet for EC2 instance deployment, by default False.
If True, this overrides the use_spot parameter and uses EC2 Spot Fleet
instead of Fargate Spot.
instance_types : Optional[List[str]], optional
List of instance types to use with Spot Fleet, by default None.
If not provided but use_spot_fleet is True, a default set of instance
types will be used.
nodes_per_block : int, optional
Number of nodes per block for Spot Fleet, by default 1
spot_max_price_percentage : Optional[float], optional
Maximum spot price as percentage of on-demand price, by default None.
If None, AWS will use the current spot market price up to the on-demand price.
additional_tags : Optional[Dict[str, str]], optional
Tags to apply to created resources, by default None
debug : bool, optional
Whether to enable debug logging, by default False
compute_type : Optional[str], optional
Compute type forwarded by ``EphemeralProvider`` ("lambda" or "ecs").
When supplied it takes precedence over ``worker_type``; "ec2" is
treated as unset since it has no meaning for serverless mode.
memory_size : Optional[int], optional
Lambda memory in MB, forwarded by ``EphemeralProvider``. Overrides
``lambda_memory`` when supplied.
timeout : Optional[int], optional
Lambda timeout in seconds, forwarded by ``EphemeralProvider``.
Overrides ``lambda_timeout`` when supplied.
lambda_code_bucket : Optional[str], optional
Existing S3 bucket to stage Lambda deployment packages in. A
caller-supplied bucket is reused as-is and never deleted; when
omitted, a provider-scoped bucket is created on first use and removed
by ``cleanup_infrastructure()``.
This is the surviving half of the old ``checkpoint_bucket``
parameter. Its checkpointing purpose was removed in #137 along with
the unimplementable task-recovery API, but the staging override was
real and is kept under a name that says what it does.
"""
# compute_type is the provider-facing name for worker_type. Map it before
# validating so an invalid value is reported against the real input.
# It arrives as a ComputeType enum member, whose str() is the qualified
# name rather than the value — read .value when present.
if compute_type is not None:
compute_type = str(getattr(compute_type, "value", compute_type)).lower()
if compute_type != "ec2":
worker_type = compute_type
# Validate worker type before calling super(), so a bad value is reported
# even when the network guard below would also have failed.
if worker_type not in [WORKER_TYPE_LAMBDA, WORKER_TYPE_ECS, WORKER_TYPE_AUTO]:
raise ConfigurationError(
f"Serverless mode requires worker_type to be '{WORKER_TYPE_LAMBDA}', "
f"'{WORKER_TYPE_ECS}', or '{WORKER_TYPE_AUTO}'"
)
# Lambda runs in the Lambda-managed VPC and needs none of the three
# network IDs; ECS/Fargate requires a subnet and security group for its
# mandatory awsvpcConfiguration. Only enforce the base-class requirement
# when ECS is reachable.
requires_network = worker_type in [WORKER_TYPE_ECS, WORKER_TYPE_AUTO]
super().__init__(
provider_id=provider_id,
session=session,
state_store=state_store,
vpc_id=vpc_id,
subnet_id=subnet_id,
security_group_id=security_group_id,
use_public_ips=use_public_ips,
additional_tags=additional_tags,
debug=debug,
require_network_resources=requires_network,
**kwargs,
)
# Set serverless mode specific attributes
self.worker_type = worker_type
self.lambda_timeout = timeout if timeout is not None else lambda_timeout
self.lambda_memory = memory_size if memory_size is not None else lambda_memory
self.lambda_runtime = lambda_runtime
self.ecs_task_cpu = ecs_task_cpu
self.ecs_task_memory = ecs_task_memory
self.ecs_container_image = ecs_container_image
# Spot and Spot Fleet configuration
self.use_spot = use_spot
self.use_spot_fleet = use_spot_fleet
self.instance_types = instance_types or [
"t3.small",
"t3a.small",
"t3.medium",
"t3a.medium",
"m5.large",
"m5a.large",
"c5.large",
"c5a.large",
]
self.nodes_per_block = nodes_per_block
self.spot_max_price_percentage = spot_max_price_percentage
# LambdaManager and ECSManager were written against EphemeralProvider
# and are handed this mode as their `provider`. Define the attributes they
# read so the contract is satisfied without duplicating the mode.
#
# Credentials are deliberately left as None: self.session is already
# fully authenticated by the provider via create_session(), and the
# managers fall back to the ambient credential chain when these are unset.
self.workflow_id = self.provider_id
self.aws_access_key_id: Optional[str] = None
self.aws_secret_access_key: Optional[str] = None
self.aws_session_token: Optional[str] = None
self.aws_profile: Optional[str] = None
self.security_config: Optional[Any] = None
# ECSManager reads this to decide whether to request Fargate Spot.
self.use_spot_instances = use_spot
# ECSManager prefers a subnet_ids list and falls back to subnet_id.
self.subnet_ids = [self.subnet_id] if self.subnet_id else None
# Initialize compute managers
self.lambda_manager: Optional[LambdaManager] = None
self.ecs_manager: Optional[ECSManager] = None
self.cf_client = self.session.client("cloudformation")
# Bucket that stages Lambda deployment packages. A caller-supplied one is
# adopted here; otherwise it is created on first use.
# _owns_lambda_code_bucket records that we created it, so cleanup only
# ever deletes a bucket of our own making.
self._lambda_code_bucket: Optional[str] = lambda_code_bucket
self._owns_lambda_code_bucket = False
# Initialize spot interruption handling if enabled. Detection needs no
# S3 bucket -- requiring one meant a caller who asked for interruption
# handling and gave no bucket silently got none at all (#137).
self.spot_interruption_monitor = None
if (self.use_spot or self.use_spot_fleet) and self.spot_interruption_handling:
logger.debug("Initializing SpotInterruptionMonitor for ServerlessMode")
self.spot_interruption_monitor = SpotInterruptionMonitor(
self.session, provider_id=self.provider_id
)
self.spot_interruption_monitor.start_monitoring()
[docs]
def initialize(self) -> None:
"""Initialize serverless mode.
Verifies the caller-supplied network resources (when the worker type
needs them) and initializes the Lambda and/or ECS managers. Network
resources are never created by this mode.
Raises
------
ResourceNotFoundError
If a configured VPC, subnet, or security group is missing.
ResourceCreationError
If initialization fails
"""
# Idempotent: if already initialized, do nothing.
if self.initialized:
return
# Confirm the caller-supplied network resources exist before we rely on
# them; Lambda-only deployments have nothing to verify. This runs ahead
# of the try block, as in the other two modes, so ResourceNotFoundError
# reaches the caller naming the unusable resource instead of being
# re-wrapped as a generic ResourceCreationError. Verifying inside the try
# also triggered cleanup_infrastructure() before anything had been
# created.
self._verify_resources()
# Try to load state first
if self.load_state():
logger.debug("Loaded state, resources verified")
# Initialize compute managers
self._initialize_compute_managers()
self.initialized = True
return
logger.debug("Initializing serverless mode")
try:
# Initialize compute managers
self._initialize_compute_managers()
self.initialized = True
# Save state
self.save_state()
logger.info(
f"Initialized serverless mode: "
f"worker_type={self.worker_type}, "
f"vpc_id={self.vpc_id}, subnet_id={self.subnet_id}, "
f"security_group_id={self.security_group_id}"
)
except Exception as e:
logger.error(f"Failed to initialize serverless mode: {e}")
# Release anything we did stand up. Network resources belong to the
# caller and are left untouched.
self.cleanup_infrastructure()
raise ResourceCreationError(
f"Failed to initialize serverless mode: {e}"
) from e
def _initialize_compute_managers(self) -> None:
"""Initialize compute managers based on worker type."""
if self.worker_type in [WORKER_TYPE_LAMBDA, WORKER_TYPE_AUTO]:
logger.debug("Initializing Lambda manager")
self.lambda_manager = LambdaManager(self)
if self.worker_type in [WORKER_TYPE_ECS, WORKER_TYPE_AUTO]:
logger.debug("Initializing ECS manager")
self.ecs_manager = ECSManager(self)
def _select_worker_type(self, command: str, tasks_per_node: int) -> str:
"""Select the appropriate worker type for a job.
Parameters
----------
command : str
Command to execute
tasks_per_node : int
Number of tasks per node
Returns
-------
str
Worker type to use (lambda or ecs)
"""
# If worker type is not auto, use the configured type
if self.worker_type != WORKER_TYPE_AUTO:
return self.worker_type
# For auto mode, select based on job characteristics
# Use Lambda for short, simple jobs
if len(command) < 5000 and tasks_per_node <= 1:
return WORKER_TYPE_LAMBDA
# Otherwise use ECS
return WORKER_TYPE_ECS
[docs]
def submit_job(
self,
job_id: str,
command: str,
tasks_per_node: int,
job_name: Optional[str] = None,
) -> str:
"""Submit a job for execution.
Parameters
----------
job_id : str
Unique identifier for the job
command : str
Command to execute
tasks_per_node : int
Number of tasks to run per node
job_name : Optional[str], optional
Human-readable name for the job, by default None
Returns
-------
str
Resource ID for tracking the job
Raises
------
OperatingModeError
If job submission fails
"""
# Ensure the mode is initialized
self.ensure_initialized()
logger.info(
f"Submitting job {job_id} ({job_name if job_name else 'unnamed'}) in serverless mode"
)
# Select worker type
worker_type = self._select_worker_type(command, tasks_per_node)
# Resource ID will be the CloudFormation stack for the job
resource_id = f"serverless-{worker_type}-{job_id}"
# Track the resource *before* dispatching. Both submit helpers finish by
# calling self.resources[resource_id].update(...) to record the stack
# name, so creating the record afterwards made every submit -- Lambda and
# ECS alike -- raise KeyError from inside the helper's blanket
# `except Exception`, surfacing as an opaque
# "Failed to submit ECS job: 'serverless-ecs-<job>'". The stack was
# created before that point and left untracked, so nothing could ever
# clean it up (#115).
self.resources[resource_id] = {
"id": resource_id,
"job_id": job_id,
"job_name": job_name or "unnamed",
"worker_type": worker_type,
"command": command,
"tasks_per_node": tasks_per_node,
"status": STATUS_PENDING,
"created_at": time.time(),
}
try:
# Submit job to the appropriate service using CloudFormation
if worker_type == WORKER_TYPE_LAMBDA:
if not self.lambda_manager:
raise JobSubmissionError("Lambda manager not initialized")
self._submit_lambda_job(job_id, command, job_name, resource_id)
elif worker_type == WORKER_TYPE_ECS:
if not self.ecs_manager:
raise JobSubmissionError("ECS manager not initialized")
# Make sure we have the required network resources for ECS
if not self.vpc_id or not self.subnet_id or not self.security_group_id:
raise JobSubmissionError(
"Missing required network resources for ECS tasks. "
"VPC, subnet, and security group are required."
)
self._submit_ecs_job(
job_id, command, tasks_per_node, job_name, resource_id
)
# Save state
self.save_state()
logger.info(f"Submitted job {job_id} with resource ID {resource_id}")
return resource_id
except Exception as e:
logger.error(f"Failed to submit job {job_id}: {e}")
# A partially created stack must not outlive the failed submit.
# cleanup_resources() deletes the stack when one was recorded and
# drops the tracking entry either way.
try:
self.cleanup_resources([resource_id])
except Exception as cleanup_error: # pragma: no cover - best effort
logger.error(
f"Failed to clean up after failed submit of {job_id}: "
f"{cleanup_error}"
)
raise OperatingModeError(f"Failed to submit job {job_id}: {e}") from e
def _submit_lambda_job(
self, job_id: str, command: str, job_name: Optional[str], resource_id: str
) -> None:
"""Submit a job to AWS Lambda.
Parameters
----------
job_id : str
Unique identifier for the job
command : str
Command to execute
job_name : Optional[str]
Human-readable name for the job
resource_id : str
Resource ID for tracking
Raises
------
JobSubmissionError
If job submission fails
"""
logger.debug(f"Submitting job {job_id} to Lambda")
try:
# Generate Lambda function code
code_zip = self.lambda_manager._generate_lambda_code(command)
# Stage the zip in S3 and reference it by key. The archive cannot
# travel through a CloudFormation string parameter: it was previously
# latin1-decoded into `CodeZipContent`, which is neither the base64
# the template documents nor legal XML -- 117 of its codepoints are
# control characters that XML 1.0 forbids in character data, so
# CloudFormation's own DescribeStacks echo of the parameter came back
# unparseable and every Lambda job reported UNKNOWN forever. And
# `AWS::Lambda::Function`'s ZipFile takes inline source text (capped
# at 4096 bytes), not archive bytes, so even correct base64 would
# deploy a broken function (#116).
code_bucket = self._ensure_lambda_code_bucket()
code_key = f"lambda-code/{self.provider_id}/{job_id}.zip"
self.session.client("s3").put_object(
Bucket=code_bucket, Key=code_key, Body=code_zip
)
# Deploy Lambda function using CloudFormation
stack_name = f"parsl-lambda-{job_id[:8]}"
template_body = get_cf_template("lambda_worker.yml")
# Create CloudFormation stack
self.cf_client.create_stack(
StackName=stack_name,
TemplateBody=template_body,
Parameters=[
{
"ParameterKey": "FunctionName",
"ParameterValue": f"parsl-lambda-{job_id}",
},
{
"ParameterKey": "Runtime",
"ParameterValue": self.lambda_runtime,
},
{
"ParameterKey": "Handler",
"ParameterValue": DEFAULT_LAMBDA_HANDLER,
},
{
"ParameterKey": "MemorySize",
"ParameterValue": str(self.lambda_memory),
},
{
"ParameterKey": "Timeout",
"ParameterValue": str(self.lambda_timeout),
},
{"ParameterKey": "CodeS3Bucket", "ParameterValue": code_bucket},
{"ParameterKey": "CodeS3Key", "ParameterValue": code_key},
{
"ParameterKey": "WorkflowId",
"ParameterValue": self.provider_id,
},
{"ParameterKey": "JobId", "ParameterValue": job_id},
],
Tags=[
{"Key": "CreatedBy", "Value": "ParslEphemeralProvider"},
{"Key": "ProviderId", "Value": self.provider_id},
{"Key": "JobId", "Value": job_id},
],
Capabilities=["CAPABILITY_IAM"],
)
# Store reference to stack in resource data. The code location is
# recorded so cleanup_resources() can delete the object.
self.resources[resource_id].update(
{
"stack_name": stack_name,
"resource_type": RESOURCE_TYPE_LAMBDA_FUNCTION,
"code_bucket": code_bucket,
"code_key": code_key,
}
)
logger.debug(
f"Created CloudFormation stack {stack_name} for Lambda job {job_id}"
)
except Exception as e:
logger.error(f"Failed to submit Lambda job {job_id}: {e}")
raise JobSubmissionError(f"Failed to submit Lambda job: {e}") from e
def _ensure_lambda_code_bucket(self) -> str:
"""Return the S3 bucket used to stage Lambda deployment packages.
A provider-scoped bucket is created on first use and removed by
``cleanup_infrastructure()``. A ``lambda_code_bucket`` supplied by the
caller is returned unchanged and left alone at cleanup --
``_owns_lambda_code_bucket`` stays False, which is what protects it.
Returns
-------
str
Name of the bucket to stage deployment packages in.
"""
if self._lambda_code_bucket:
return self._lambda_code_bucket
s3 = self.session.client("s3")
region = self.session.region_name
bucket = f"parsl-lambda-code-{self.provider_id[:8]}"
try:
# us-east-1 is the one region CreateBucket rejects a
# LocationConstraint for.
if region and region != "us-east-1":
s3.create_bucket(
Bucket=bucket,
CreateBucketConfiguration={"LocationConstraint": region},
)
else:
s3.create_bucket(Bucket=bucket)
self._owns_lambda_code_bucket = True
logger.debug(f"Created Lambda code bucket {bucket}")
except ClientError as e:
code = e.response.get("Error", {}).get("Code")
# Already ours (a restart, or a second job in the same run).
if code not in ("BucketAlreadyOwnedByYou", "BucketAlreadyExists"):
raise
logger.debug(f"Reusing existing Lambda code bucket {bucket}")
self._lambda_code_bucket = bucket
return bucket
def _submit_ecs_job(
self,
job_id: str,
command: str,
tasks_per_node: int,
job_name: Optional[str],
resource_id: str,
) -> None:
"""Submit a job to ECS/Fargate or EC2 using SpotFleet.
This method supports two deployment modes:
1. ECS/Fargate: The default mode, which uses serverless containers
2. EC2 SpotFleet: When use_spot_fleet=True, deploys EC2 instances using SpotFleet
for improved reliability and cost savings
Parameters
----------
job_id : str
Unique identifier for the job
command : str
Command to execute
tasks_per_node : int
Number of tasks per node
job_name : Optional[str]
Human-readable name for the job
resource_id : str
Resource ID for tracking
Raises
------
JobSubmissionError
If job submission fails
"""
logger.debug(f"Submitting job {job_id} to ECS")
# An EC2 Fleet is created directly rather than through CloudFormation.
# CloudFormation cannot express this resource: the override list is
# variable-length, and CFN has no way to build one -- Fn::ForEach expands
# to a map, and padding a fixed set of !Select slots is rejected by EC2
# ("duplicate instance pools"). CreateFleet takes the list natively.
# Deploying the stack also created an ECS cluster, task definition, two
# IAM roles, and a log group that an EC2 fleet never uses.
if self.use_spot_fleet:
self._create_job_fleet(job_id, command, resource_id)
return
try:
# Deploy ECS task using CloudFormation
stack_name = f"parsl-ecs-{job_id[:8]}"
template_body = get_cf_template("ecs_worker.yml")
# Create CloudFormation stack
self.cf_client.create_stack(
StackName=stack_name,
TemplateBody=template_body,
Parameters=[
{
"ParameterKey": "ClusterName",
"ParameterValue": f"parsl-ecs-cluster-{self.provider_id[:8]}",
},
{
"ParameterKey": "TaskFamily",
"ParameterValue": f"parsl-ecs-task-{job_id[:8]}",
},
{
"ParameterKey": "ContainerImage",
"ParameterValue": self.ecs_container_image,
},
{
"ParameterKey": "TaskCpu",
"ParameterValue": str(self.ecs_task_cpu),
},
{
"ParameterKey": "TaskMemory",
"ParameterValue": str(self.ecs_task_memory),
},
# Passed through unaltered. The template exec's it with
# /bin/sh -c since #226, so this is an ordinary shell string
# like every other mode's, and a multi-line command works.
# Newlines used to be collapsed to ';' here, which served the
# old argv-by-comma-split encoding and silently broke any
# command with a comment line -- everything after a '#'
# became part of the comment. CloudFormation preserves
# newlines in a String parameter verbatim; verified against
# the live service by reading a parameter echo back out of a
# stack output.
{"ParameterKey": "Command", "ParameterValue": command},
{"ParameterKey": "VpcId", "ParameterValue": self.vpc_id},
{"ParameterKey": "SubnetIds", "ParameterValue": self.subnet_id},
{
"ParameterKey": "SecurityGroupIds",
"ParameterValue": self.security_group_id,
},
{
"ParameterKey": "AssignPublicIp",
"ParameterValue": "ENABLED"
if self.use_public_ips
else "DISABLED",
},
{"ParameterKey": "WorkflowId", "ParameterValue": self.provider_id},
{"ParameterKey": "JobId", "ParameterValue": job_id},
{
"ParameterKey": "TaskCount",
"ParameterValue": str(max(1, tasks_per_node)),
},
{
"ParameterKey": "UseSpot",
"ParameterValue": "true" if self.use_spot else "false",
},
],
Tags=[
{"Key": "CreatedBy", "Value": "ParslEphemeralProvider"},
{"Key": "ProviderId", "Value": self.provider_id},
{"Key": "JobId", "Value": job_id},
],
Capabilities=["CAPABILITY_IAM"],
)
# Store reference to stack in resource data
self.resources[resource_id].update(
{
"stack_name": stack_name,
"resource_type": RESOURCE_TYPE_ECS_TASK,
"use_spot_fleet": False,
}
)
logger.debug(
f"Created CloudFormation stack {stack_name} for ECS job {job_id}"
)
except Exception as e:
logger.error(f"Failed to submit ECS job {job_id}: {e}")
raise JobSubmissionError(f"Failed to submit ECS job: {e}") from e
[docs]
def get_job_status(self, resource_ids: List[str]) -> Dict[str, str]:
"""Get the status of jobs.
Parameters
----------
resource_ids : List[str]
List of resource IDs to check
Returns
-------
Dict[str, str]
Dictionary mapping resource IDs to status strings
"""
if not resource_ids:
return {}
status_map = {}
for resource_id in resource_ids:
resource = self.resources.get(resource_id)
if not resource:
status_map[resource_id] = STATUS_UNKNOWN
continue
# An interruption is sticky. The fleet and stack lookups below both
# re-derive status, and neither can see a reclaim -- a fleet whose
# instances AWS is taking back still reports itself active -- so
# without this the marker set by handle_fleet_interruption is
# overwritten on the very next poll (#137).
if resource.get("status") == STATUS_INTERRUPTED:
status_map[resource_id] = STATUS_INTERRUPTED
continue
# A fleet-backed job has no stack: the fleet is created directly, so
# its status comes from the fleet rather than from CloudFormation.
fleet_request_id = resource.get("fleet_request_id")
if resource.get("use_spot_fleet") and fleet_request_id:
status = self._get_spot_fleet_status(fleet_request_id)
self.resources[resource_id]["status"] = status
status_map[resource_id] = status
continue
# Get stack status
stack_name = resource.get("stack_name")
if not stack_name:
status_map[resource_id] = resource.get("status", STATUS_UNKNOWN)
continue
try:
response = self.cf_client.describe_stacks(StackName=stack_name)
stack_status = response["Stacks"][0]["StackStatus"]
# Map CloudFormation status to our status
if stack_status.startswith("CREATE_IN_PROGRESS"):
status = STATUS_PENDING
elif stack_status == "CREATE_COMPLETE":
# When stack is complete, we need to check the actual resource status
worker_type = resource.get("worker_type")
if worker_type == WORKER_TYPE_LAMBDA:
if self.lambda_manager:
# For Lambda, we get the function name from stack outputs
outputs = response["Stacks"][0].get("Outputs", [])
function_name = None
for output in outputs:
if output["OutputKey"] == "LambdaFunctionName":
function_name = output["OutputValue"]
break
if function_name:
# Check Lambda invocation status
# Note: This is simplified as true Lambda status tracking
# would require additional mechanisms
status = self._get_lambda_status(
function_name, resource_id
)
else:
status = STATUS_RUNNING
elif worker_type == WORKER_TYPE_ECS:
# No fleet branch here: a fleet-backed job never reaches
# this point, having been answered from the fleet above.
if self.ecs_manager:
# For standard ECS, we get the cluster and service name from stack outputs
outputs = response["Stacks"][0].get("Outputs", [])
cluster_name = None
service_name = None
for output in outputs:
if output["OutputKey"] == "ClusterName":
cluster_name = output["OutputValue"]
elif output["OutputKey"] == "ServiceName":
service_name = output["OutputValue"]
if cluster_name and service_name:
# Check ECS service status
status = self._get_ecs_status(
cluster_name, service_name
)
else:
status = STATUS_RUNNING
else:
status = STATUS_RUNNING
# Any rollback means the stack failed. Checked before the
# affix tests below because ROLLBACK_COMPLETE and
# UPDATE_ROLLBACK_COMPLETE match neither of them and so used to
# fall through to RUNNING (#106) -- and ROLLBACK_COMPLETE is the
# *usual* CloudFormation failure state, since automatic rollback
# on CREATE_FAILED is the default. Reporting RUNNING there left
# the job outside _TERMINAL_STATES and polled forever, so Parsl
# never learned the task had failed and never retried it, while
# the stack sat in a state that can only be deleted.
elif "ROLLBACK" in stack_status:
status = STATUS_FAILED
elif stack_status.endswith("FAILED"):
status = STATUS_FAILED
elif stack_status.startswith("DELETE"):
status = STATUS_CANCELLED
else:
status = STATUS_RUNNING
# Update resource status
self.resources[resource_id]["status"] = status
status_map[resource_id] = status
except ClientError as e:
logger.error(f"Failed to get stack status for {stack_name}: {e}")
# Handle case where stack doesn't exist anymore
if "does not exist" in str(e):
# Assume job completed
status = STATUS_SUCCEEDED
self.resources[resource_id]["status"] = status
status_map[resource_id] = status
else:
status_map[resource_id] = STATUS_UNKNOWN
except Exception as e:
logger.error(f"Unexpected error getting status for {resource_id}: {e}")
status_map[resource_id] = STATUS_UNKNOWN
# Save state with updated status
self.save_state()
return status_map
def _get_lambda_status(self, function_name: str, resource_id: str) -> str:
"""Get the status of a Lambda job.
Parameters
----------
function_name : str
Lambda function name
resource_id : str
Resource ID for tracking
Returns
-------
str
Job status
"""
# In a real implementation, we would use CloudWatch Logs or a state store
# to track Lambda execution. For now, we'll simulate based on time.
resource = self.resources.get(resource_id, {})
elapsed = time.time() - resource.get("created_at", 0)
if elapsed < 5:
return STATUS_PENDING
elif elapsed < self.lambda_timeout:
return STATUS_RUNNING
else:
# After timeout, assume success (in a real impl, we'd check CloudWatch)
return STATUS_SUCCEEDED
def _create_job_fleet(self, job_id: str, command: str, resource_id: str) -> None:
"""Launch this job's workers as an ``instant`` EC2 Fleet.
Bypasses CloudFormation, which the fleet used to go through. Two reasons,
both established against real AWS:
* The ``Overrides`` list is variable-length -- one entry per instance type
-- and CloudFormation cannot build one. ``Fn::ForEach`` expands to a map
rather than a list, and a fixed set of ``!Select`` slots cannot be left
partly unfilled because an out-of-range ``!Select`` fails validation even
in an untaken ``!If`` branch. Padding the slots by repeating a type is
what the previous revision did, and EC2 rejects it outright:
``InvalidFleetConfig: The fleet configuration contains duplicate
instance pools``.
* The stack's other resources -- an ECS cluster, a task definition, two
IAM roles, a log group -- are Fargate machinery that an EC2 fleet never
touches, so deploying it to get one fleet created five resources to
leak and made the fleet ID reachable only by polling stack outputs.
The launch template is mandatory: ``CreateFleet`` has no
``LaunchSpecifications`` member, and the per-job user data has nowhere
else to live -- ``Overrides`` cannot carry ``UserData``.
Parameters
----------
job_id : str
Job this fleet serves; names the launch template.
command : str
Command the instances run, via user data.
resource_id : str
Resource record to annotate with the fleet and template IDs.
Raises
------
JobSubmissionError
If the template or the fleet cannot be created, or if EC2 filled none
of the requested capacity.
"""
ec2_client = self.session.client("ec2")
instance_types = self._fleet_instance_types()
tags = self._build_fleet_tags(job_id)
# Restated here rather than relied upon from the ECS guard in
# submit_job. Every fleet override names the subnet, so an unset one
# reaches CreateFleet as a null and is refused there -- after the launch
# template has already been created, leaving it to be cleaned up.
if not self.subnet_id:
raise JobSubmissionError(
"subnet_id is required to launch an EC2 Fleet; every launch "
"template override names the subnet to launch into."
)
template_name = f"{LAUNCH_TEMPLATE_NAME_PREFIX}-ecs-{job_id[:8]}"
template_id = None
try:
template_data = build_launch_template_data(
image_id=self._resolve_fleet_image_id(),
instance_type=instance_types[0],
subnet_id=self.subnet_id,
security_group_id=self.security_group_id,
associate_public_ip=self.use_public_ips,
# Fleet instances are reclaimed by deleting the fleet, which
# always terminates them, so one that shuts itself down should
# terminate too rather than linger as a billed volume.
shutdown_behavior="terminate",
user_data=self._build_fleet_user_data(command),
)
template_data["TagSpecifications"] = [
{"ResourceType": "instance", "Tags": list(tags)}
]
template_id, version = create_launch_template(
ec2_client, template_name, template_data, list(tags)
)
max_total_price = self._resolve_max_total_price()
fleet_id, instance_ids = create_ec2_fleet(
ec2_client,
build_fleet_launch_template_configs(
template_id, version, instance_types, self.subnet_id
),
target_capacity=max(1, self.nodes_per_block),
allocation_strategy=self.spot_allocation_strategy,
tags=tags,
max_total_price=max_total_price or None,
)
except Exception as e:
# The template outlives a failed CreateFleet, so drop it here rather
# than leaving a per-job resource behind for the orphan sweep.
if template_id:
try:
delete_launch_template(ec2_client, template_id)
except Exception as cleanup_error:
logger.warning(
f"Failed to delete launch template {template_id} after a "
f"failed fleet submit: {cleanup_error}"
)
logger.error(f"Failed to submit fleet job {job_id}: {e}")
raise JobSubmissionError(f"Failed to submit fleet job: {e}") from e
self.resources[resource_id].update(
{
"resource_type": RESOURCE_TYPE_SPOT_FLEET,
"use_spot_fleet": True,
"fleet_request_id": fleet_id,
"launch_template_id": template_id,
"instance_ids": instance_ids,
# No stack_name: there is no stack. get_job_status() keys off
# this, and cleanup_resources() must not try to delete one.
}
)
if not instance_ids:
# An instant fleet reports a pool it could not fill inline instead of
# failing, so an empty fleet is a successful API call. The fleet is
# still deleted, since it holds nothing and will never grow -- an
# instant fleet makes no further attempts.
try:
delete_ec2_fleet(ec2_client, fleet_id)
except Exception as e:
logger.warning(f"Failed to delete unfilled fleet {fleet_id}: {e}")
raise JobSubmissionError(
f"EC2 Fleet {fleet_id} launched no instances for job {job_id}; "
"no spot capacity was available in any of the requested pools "
f"({', '.join(instance_types)})."
)
# Registration is immediate now: the fleet ID comes back from CreateFleet
# itself. It used to require polling stack outputs for up to 3 minutes,
# during which an interruption would have gone unhandled.
if self.spot_interruption_handling and self.spot_interruption_monitor:
self.spot_interruption_monitor.register_fleet(
fleet_id,
self.handle_fleet_interruption,
)
logger.info(f"Registered EC2 Fleet {fleet_id} for interruption handling")
logger.info(
f"Created EC2 Fleet {fleet_id} with {len(instance_ids)} instance(s) "
f"for job {job_id}"
)
def _reclaim_fleet(self, ec2_client: Any, resource: Dict[str, Any]) -> None:
"""Delete a job's fleet and its launch template.
Deleting the fleet terminates its instances; that is not optional for an
``instant`` fleet, which AWS refuses to leave running without its fleet
(#86). The template is deleted afterwards -- doing so does not affect
instances already launched from it, but leaving it behind would strand a
per-job resource that only the orphan sweep could find.
Both steps log rather than raise: this runs from cancellation and cleanup
paths, where one unreclaimable resource must not prevent the rest from
being reclaimed.
Parameters
----------
ec2_client : Any
A boto3 EC2 client.
resource : Dict[str, Any]
Resource record carrying ``fleet_request_id`` and, optionally,
``launch_template_id``.
"""
fleet_request_id = resource.get("fleet_request_id")
if fleet_request_id:
try:
delete_ec2_fleet(ec2_client, fleet_request_id)
logger.info(
f"Deleted EC2 Fleet {fleet_request_id} for job "
f"{resource.get('job_id')}"
)
except Exception as e:
logger.warning(f"Error deleting EC2 Fleet {fleet_request_id}: {e}")
template_id = resource.get("launch_template_id")
if template_id:
try:
delete_launch_template(ec2_client, template_id)
except Exception as e:
logger.warning(f"Error deleting launch template {template_id}: {e}")
def _build_fleet_tags(self, job_id: str) -> List[Dict[str, str]]:
"""Build the tags applied to a job's fleet, instances, and template.
The keys match what the orphan sweep looks for, so anything this leaves
behind is findable: ``ParslWorkflowId`` is one of
``spot_fleet_cleanup.WORKFLOW_ID_TAG_KEYS``.
"""
return [
{"Key": "ParslResource", "Value": "true"},
{"Key": "ParslWorkflowId", "Value": self.provider_id},
{"Key": "ParslJobId", "Value": job_id},
{
"Key": "Name",
"Value": f"parsl-fleet-{self.provider_id[:8]}-{job_id[:8]}",
},
]
def _build_fleet_user_data(self, command: str) -> str:
"""Build the user data a fleet instance runs.
Mirrors what the template's ``UserData`` did, with the command written to
a file and executed. ``shutdown -h now`` is deliberate and pairs with the
template's ``InstanceInitiatedShutdownBehavior=terminate``: the instance
exists to run one command, so it should terminate when that is done
instead of idling at full price.
``worker_init`` runs first, as it does on every other EC2 path
(``StandardMode._build_user_data`` and
``SpotFleetManager._generate_user_data``). It was previously dropped
here, so a fleet instance in this mode booted a bare Amazon Linux image
with no Parsl installed and ran ``command`` against it -- the one
opportunity to install anything, silently discarded (#198).
"""
script = '#!/bin/bash\necho "Starting Parsl worker..."\n'
if self.worker_init:
script += f"\n# User-provided worker initialization\n{self.worker_init}\n"
return script + (
"mkdir -p /tmp/parsl\n"
f"cat > /tmp/parsl/command.sh <<'PARSL_EOF'\n{command}\nPARSL_EOF\n"
"chmod +x /tmp/parsl/command.sh\n"
"/tmp/parsl/command.sh\n"
'echo "Parsl worker completed."\n'
"shutdown -h now\n"
)
def _resolve_fleet_image_id(self) -> str:
"""Return the AMI the EC2 Fleet's instances should boot.
Prefers an explicitly configured ``image_id``, otherwise resolves the
current Amazon Linux 2023 image for the session's region from AWS's
public SSM alias (#83).
This has to be resolved here rather than in the template: the template's
old ``RegionMap`` was never wired to a ``FindInMap``, so its fleet
launched with no ``ImageId`` at all and EC2 rejected every request with
"Parameter 'amiIdList' cannot be empty" -- confirmed against both fleet
APIs. The fleet path in this mode has therefore never worked.
Returns
-------
str
AMI ID.
"""
if self.image_id:
return self.image_id
region = self.session.region_name or DEFAULT_REGION
architecture = architecture_for_instance_type(self.instance_types[0])
return get_default_ami(region, architecture, session=self.session)
def _fleet_instance_types(self) -> List[str]:
"""Return the instance types the fleet may draw from, deduplicated.
Order is preserved, since the allocation strategy treats the list as a
preference order. Duplicates are dropped because ``CreateFleet`` rejects
the request outright when two overrides name the same capacity pool:
InvalidFleetConfig: The fleet configuration contains duplicate
instance pools.
A ``DryRun`` does *not* catch this -- verified against real EC2, which
accepted the identical duplicate-bearing request with ``DryRun=True`` and
rejected it without. This is why the fleet is no longer built by
CloudFormation at all; see :meth:`_create_job_fleet`.
Returns
-------
List[str]
Instance types in preference order, each appearing once.
"""
seen: Dict[str, None] = {}
for instance_type in self.instance_types:
seen.setdefault(instance_type, None)
return list(seen) or [DEFAULT_INSTANCE_TYPE]
def _resolve_max_total_price(self) -> str:
"""Translate ``spot_max_price_percentage`` into a fleet ``MaxTotalPrice``.
Returns an empty string when no cap is configured, which is the
recommended setting -- AWS: "We do not recommend using this parameter
because it can lead to increased interruptions." An uncapped fleet pays
the prevailing spot price, already far below on-demand.
The percentage is of on-demand, which is what the setting has always
documented. There is no cheap API for an on-demand price, so the same
3x-current-spot proxy that ``SpotFleetManager`` uses is applied here.
Unlike the legacy per-instance-hour ``SpotPrice``, ``MaxTotalPrice``
covers the whole fleet, so it is multiplied by the node count.
Returns
-------
str
Fleet-wide hourly maximum in USD, or "" for no cap.
"""
if not self.spot_max_price_percentage:
return ""
try:
history = (
self.session.client("ec2")
.describe_spot_price_history(
InstanceTypes=[self.instance_types[0]],
ProductDescriptions=["Linux/UNIX"],
MaxResults=1,
)
.get("SpotPriceHistory", [])
)
current_spot = float(history[0]["SpotPrice"]) if history else 1.0
on_demand_price = current_spot * 3
except Exception as e:
logger.warning(f"Could not read spot price history, assuming $1.00/hr: {e}")
on_demand_price = 1.0
per_instance = on_demand_price * (self.spot_max_price_percentage / 100.0)
return str(per_instance * max(1, self.nodes_per_block))
def _get_spot_fleet_status(self, fleet_request_id: str) -> str:
"""Get the status of an EC2 Fleet.
Parameters
----------
fleet_request_id : str
ID of the EC2 Fleet
Returns
-------
str
Job status
"""
ec2_client = self.session.client("ec2")
try:
fleet = describe_ec2_fleet(ec2_client, fleet_request_id)
if fleet is None:
# EC2 has forgotten the fleet, so its instances are long gone.
# Terminal, so Parsl can free the block.
return STATUS_COMPLETED
fleet_status = fleet["FleetState"]
if fleet_status in ("submitted", "modifying"):
return STATUS_PENDING
elif fleet_status == "active":
# An instant fleet does not maintain capacity, so FleetState
# stays "active" for the fleet's whole life regardless of what
# became of its instances. Capacity counters cannot decide this
# either: they reflect the original launch. Only the instances
# can, so ask them.
#
# The capacity comparison this replaced was also the site of
# #114: FulfilledCapacity was read from the wrong nesting level
# and always came back 0, so a fully provisioned fleet reported
# PENDING forever and Parsl never freed the block.
if get_ec2_fleet_instance_ids(ec2_client, fleet_request_id):
return STATUS_RUNNING
return STATUS_COMPLETED
elif fleet_status == "deleted_running":
# Fleet is being deleted but instances are still running
return STATUS_RUNNING
elif fleet_status in ("deleted", "deleted_terminating"):
return STATUS_CANCELLED
elif fleet_status == "failed":
return STATUS_FAILED
else:
return STATUS_UNKNOWN
except Exception as e:
logger.error(f"Error getting EC2 Fleet status for {fleet_request_id}: {e}")
return STATUS_UNKNOWN
def _get_ecs_status(self, cluster_name: str, service_name: str) -> str:
"""Get the status of an ECS service.
Parameters
----------
cluster_name : str
ECS cluster name
service_name : str
ECS service name
Returns
-------
str
Job status
"""
ecs_client = self.session.client("ecs")
try:
# Get service details
response = ecs_client.describe_services(
cluster=cluster_name, services=[service_name]
)
if not response["services"]:
return STATUS_UNKNOWN
service = response["services"][0]
# Check if service has tasks
task_response = ecs_client.list_tasks(
cluster=cluster_name, serviceName=service_name
)
# If no tasks, check service events to determine status
if not task_response.get("taskArns"):
# Check deployment status
deployments = service.get("deployments", [])
if not deployments:
return STATUS_COMPLETED
# Look at recent events for status info
events = service.get("events", [])
if events:
# Look for completion or failure events
for event in events[:5]: # Check recent events
if "has reached a steady state" in event.get("message", ""):
return STATUS_SUCCEEDED
if "was unable to place a task" in event.get("message", ""):
return STATUS_FAILED
# If desired count is 0, job is considered complete
if service.get("desiredCount", 0) == 0:
return STATUS_COMPLETED
# Otherwise still pending
return STATUS_PENDING
# Get task details
task_arns = task_response["taskArns"]
if task_arns:
task_details = ecs_client.describe_tasks(
cluster=cluster_name,
tasks=[task_arns[0]], # Check first task
)
if task_details["tasks"]:
task = task_details["tasks"][0]
last_status = task["lastStatus"]
if last_status == "PROVISIONING" or last_status == "PENDING":
return STATUS_PENDING
elif last_status == "RUNNING":
return STATUS_RUNNING
elif last_status == "STOPPED":
# Check if task stopped with error
if task.get(
"stoppedReason"
) and "Essential container" in task.get("stoppedReason"):
# Look at container exit codes
for container in task.get("containers", []):
exit_code = container.get("exitCode")
if exit_code is not None and exit_code != 0:
return STATUS_FAILED
return STATUS_SUCCEEDED
else:
return STATUS_RUNNING
# Default to running if service exists but status is unclear
return STATUS_RUNNING
except Exception as e:
logger.error(f"Error getting ECS service status: {e}")
return STATUS_UNKNOWN
[docs]
def cancel_jobs(self, resource_ids: List[str]) -> Dict[str, str]:
"""Cancel jobs.
Parameters
----------
resource_ids : List[str]
List of resource IDs to cancel
Returns
-------
Dict[str, str]
Dictionary mapping resource IDs to status strings
"""
if not resource_ids:
return {}
cancel_map = {}
ec2_client = self.session.client("ec2")
for resource_id in resource_ids:
resource = self.resources.get(resource_id)
if not resource:
cancel_map[resource_id] = STATUS_UNKNOWN
continue
# A fleet-backed job has no stack; deleting the fleet *is* the
# cancellation. Handled before the stack_name check below, which
# would otherwise report UNKNOWN and leave the instances running.
if resource.get("resource_type") == RESOURCE_TYPE_SPOT_FLEET:
self._reclaim_fleet(ec2_client, resource)
self.resources[resource_id]["status"] = STATUS_CANCELLED
cancel_map[resource_id] = STATUS_CANCELLED
continue
# Get stack name
stack_name = resource.get("stack_name")
if not stack_name:
cancel_map[resource_id] = STATUS_UNKNOWN
continue
try:
# Delete the CloudFormation stack to cancel the job
self.cf_client.delete_stack(StackName=stack_name)
# Mark as cancelled
self.resources[resource_id]["status"] = STATUS_CANCELLED
cancel_map[resource_id] = STATUS_CANCELLED
logger.info(
f"Cancelled job {resource.get('job_id')} (stack: {stack_name})"
)
except ClientError as e:
logger.error(f"Failed to cancel job (stack: {stack_name}): {e}")
# Handle case where stack doesn't exist anymore
if "does not exist" in str(e):
# Assume job completed
cancel_map[resource_id] = STATUS_COMPLETED
self.resources[resource_id]["status"] = STATUS_COMPLETED
else:
cancel_map[resource_id] = STATUS_FAILED
except Exception as e:
logger.error(
f"Unexpected error cancelling job (stack: {stack_name}): {e}"
)
cancel_map[resource_id] = STATUS_FAILED
# Save state with updated status
self.save_state()
return cancel_map
[docs]
def cleanup_resources(self, resource_ids: List[str]) -> None:
"""Clean up resources.
Parameters
----------
resource_ids : List[str]
List of resource IDs to clean up
"""
if not resource_ids:
return
for resource_id in resource_ids:
resource = self.resources.get(resource_id)
if not resource:
continue
# Drop the staged Lambda deployment package, if any. Done before the
# stack delete so the object goes even when the stack is already gone.
self._delete_staged_lambda_code(resource)
# A fleet-backed job has no stack, so it has to be reclaimed here.
# Falling through to the stack_name check below would delete the
# tracking record and leave the fleet's instances running with
# nothing left that knows their IDs.
if resource.get("resource_type") == RESOURCE_TYPE_SPOT_FLEET:
self._reclaim_fleet(self.session.client("ec2"), resource)
del self.resources[resource_id]
continue
# Get stack name
stack_name = resource.get("stack_name")
if not stack_name:
# Remove resource from tracking
if resource_id in self.resources:
del self.resources[resource_id]
continue
try:
# Delete the CloudFormation stack
self.cf_client.delete_stack(StackName=stack_name)
logger.info(
f"Deleted stack {stack_name} for job {resource.get('job_id')}"
)
# Remove resource from tracking
if resource_id in self.resources:
del self.resources[resource_id]
except ClientError as e:
# If the stack is already deleted or doesn't exist, that's fine
if "does not exist" not in str(e):
logger.error(f"Failed to delete stack {stack_name}: {e}")
# Still remove resource from tracking
if resource_id in self.resources:
del self.resources[resource_id]
except Exception as e:
logger.error(f"Unexpected error deleting stack {stack_name}: {e}")
# Still remove resource from tracking
if resource_id in self.resources:
del self.resources[resource_id]
# Save state with updated resources
self.save_state()
def _delete_staged_lambda_code(self, resource: Dict[str, Any]) -> None:
"""Delete the S3 object staging a job's Lambda deployment package.
Parameters
----------
resource : Dict[str, Any]
Resource tracking record; ignored unless it carries both
``code_bucket`` and ``code_key``.
"""
bucket = resource.get("code_bucket")
key = resource.get("code_key")
if not bucket or not key:
return
try:
self.session.client("s3").delete_object(Bucket=bucket, Key=key)
logger.debug(f"Deleted staged Lambda code s3://{bucket}/{key}")
except Exception as e:
logger.warning(f"Failed to delete staged Lambda code {key}: {e}")
def _delete_lambda_code_bucket(self) -> None:
"""Delete the Lambda code bucket, but only if this mode created it.
``_owns_lambda_code_bucket`` is the guard: a bucket this mode merely
found and reused is left alone.
"""
if not self._owns_lambda_code_bucket or not self._lambda_code_bucket:
return
bucket = self._lambda_code_bucket
s3 = self.session.client("s3")
try:
# A bucket must be empty before it can be deleted; any objects left
# here are packages whose jobs never reached cleanup_resources().
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket=bucket):
objects = [{"Key": obj["Key"]} for obj in page.get("Contents", [])]
if objects:
s3.delete_objects(Bucket=bucket, Delete={"Objects": objects})
s3.delete_bucket(Bucket=bucket)
logger.debug(f"Deleted Lambda code bucket {bucket}")
except Exception as e:
logger.warning(f"Failed to delete Lambda code bucket {bucket}: {e}")
finally:
self._lambda_code_bucket = None
self._owns_lambda_code_bucket = False
[docs]
def cleanup_infrastructure(self) -> None:
"""Clean up infrastructure created by this mode.
The VPC, subnet, and security group are supplied by the caller and are
never created — or deleted — by this mode. Only the Lambda functions,
ECS tasks, and Spot Fleet resources created here are removed.
"""
logger.info("Cleaning up serverless mode infrastructure")
# Delete all resources first
if self.resources:
self.cleanup_all()
# Stop spot interruption monitoring if enabled
if self.spot_interruption_monitor:
try:
self.spot_interruption_monitor.stop_monitoring()
logger.info("Stopped spot interruption monitoring")
except Exception as e:
logger.error(f"Failed to stop spot interruption monitoring: {e}")
self.spot_interruption_monitor = None
# Clean up compute managers
if self.lambda_manager:
try:
self.lambda_manager.cleanup_all_resources()
except Exception as e:
logger.error(f"Error cleaning up Lambda manager resources: {e}")
# Drop the deployment-package bucket if this mode created it.
self._delete_lambda_code_bucket()
if self.ecs_manager:
try:
self.ecs_manager.cleanup_all_resources()
except Exception as e:
logger.error(f"Error cleaning up ECS manager resources: {e}")
# Sweep any fleet resources the per-job stack deletes missed.
if self.use_spot_fleet:
try:
from parsl_ephemeral_provider.compute.spot_fleet_cleanup import (
cleanup_all_spot_fleet_resources,
)
cleanup_result = cleanup_all_spot_fleet_resources(
session=self.session,
workflow_id=self.provider_id,
cancel_active_requests=True,
cleanup_iam_roles=True,
)
# Log cleanup results
if cleanup_result:
if cleanup_result.get("deleted_fleets"):
logger.info(
f"Deleted {len(cleanup_result['deleted_fleets'])} EC2 Fleets"
)
# Only non-empty for a workflow that predates #86.
if cleanup_result.get("cancelled_requests"):
logger.info(
f"Cancelled {len(cleanup_result['cancelled_requests'])} "
"legacy Spot Fleet requests"
)
if cleanup_result.get("cleaned_roles"):
logger.info(
f"Cleaned up {len(cleanup_result['cleaned_roles'])} IAM roles"
)
# Log errors
if cleanup_result.get("errors"):
for error in cleanup_result["errors"]:
logger.warning(f"Fleet cleanup error: {error}")
except Exception as e:
logger.error(f"Error cleaning up fleet resources: {e}")
# Clear initialization flag
self.initialized = False
# Save state
self.save_state()
logger.info("Serverless mode infrastructure cleanup complete")
[docs]
def list_resources(self) -> Dict[str, List[Dict[str, Any]]]:
"""List all resources created by this mode.
Returns
-------
Dict[str, List[Dict[str, Any]]]
Dictionary of resource types and their details
"""
result: Dict[str, List[Dict[str, Any]]] = {
"lambda_functions": [],
"ecs_tasks": [],
"spot_fleet_requests": [],
"vpc": [],
"subnet": [],
"security_group": [],
}
# Add jobs by resource type
for resource_id, resource in self.resources.items():
worker_type = resource.get("worker_type")
if worker_type == WORKER_TYPE_LAMBDA:
result["lambda_functions"].append(
{
"id": resource_id,
"job_id": resource.get("job_id"),
"job_name": resource.get("job_name"),
"status": resource.get("status"),
"created_at": resource.get("created_at"),
"stack_name": resource.get("stack_name"),
}
)
elif worker_type == WORKER_TYPE_ECS:
# Check if this is a SpotFleet resource
if resource.get("use_spot_fleet") and resource.get("fleet_request_id"):
result["spot_fleet_requests"].append(
{
"id": resource_id,
"job_id": resource.get("job_id"),
"job_name": resource.get("job_name"),
"status": resource.get("status"),
"created_at": resource.get("created_at"),
# No stack_name: a fleet is created directly (#86).
"fleet_request_id": resource.get("fleet_request_id"),
"launch_template_id": resource.get("launch_template_id"),
"instance_ids": resource.get("instance_ids", []),
}
)
else:
result["ecs_tasks"].append(
{
"id": resource_id,
"job_id": resource.get("job_id"),
"job_name": resource.get("job_name"),
"status": resource.get("status"),
"created_at": resource.get("created_at"),
"stack_name": resource.get("stack_name"),
}
)
# Add VPC if available
if self.vpc_id:
result["vpc"].append(
{
"id": self.vpc_id,
}
)
# Add subnet if available
if self.subnet_id:
result["subnet"].append(
{
"id": self.subnet_id,
"vpc_id": self.vpc_id,
}
)
# Add security group if available
if self.security_group_id:
result["security_group"].append(
{
"id": self.security_group_id,
"vpc_id": self.vpc_id,
}
)
return result
[docs]
def cleanup_all(self) -> None:
"""Clean up all resources created by this mode."""
logger.info("Cleaning up all serverless mode resources")
# Get all resource IDs
resource_ids = list(self.resources.keys())
if resource_ids:
self.cleanup_resources(resource_ids)
logger.info(f"Cleaned up {len(resource_ids)} resources")
else:
logger.debug("No resources to clean up")
[docs]
def save_state(self) -> None:
"""Save the current state to the state store."""
state = {
"resources": self.resources,
"provider_id": self.provider_id,
"mode": self.__class__.__name__,
"vpc_id": self.vpc_id,
"subnet_id": self.subnet_id,
"security_group_id": self.security_group_id,
"initialized": self.initialized,
# The routing decision belongs in the document: it is what determines
# whether the three network IDs above were required at all (Lambda
# needs none of them, ECS needs subnet + SG), so a document without it
# cannot be interpreted. Not restored by load_state() — the
# constructor value is authoritative for every configuration field,
# the same rule _restore_network_ids() follows (#118).
"worker_type": self.worker_type,
"use_spot": self.use_spot,
"use_spot_fleet": self.use_spot_fleet,
"spot_interruption_handling": self.spot_interruption_handling,
# Persisted so a restarted provider can still delete a bucket it
# created, rather than leaking it.
"lambda_code_bucket": self._lambda_code_bucket,
"owns_lambda_code_bucket": self._owns_lambda_code_bucket,
}
try:
self.state_store.save_state(STATE_KEY_MODE, state)
except Exception as e:
logger.error(f"Failed to save state: {e}")
[docs]
def load_state(self) -> bool:
"""Load state from the state store.
Returns
-------
bool
True if state was loaded successfully, False otherwise
"""
try:
state = self.state_store.load_state(STATE_KEY_MODE)
if state and state.get("provider_id") == self.provider_id:
self.resources = state.get("resources", {})
self._restore_network_ids(state)
self.initialized = state.get("initialized", False)
self._lambda_code_bucket = state.get("lambda_code_bucket")
self._owns_lambda_code_bucket = state.get(
"owns_lambda_code_bucket", False
)
# Check if spot interruption handling was previously enabled
previous_spot_handling = state.get("spot_interruption_handling", False)
if previous_spot_handling != self.spot_interruption_handling:
logger.info(
f"Spot interruption handling changed from {previous_spot_handling} to {self.spot_interruption_handling}"
)
# Initialize or clean up spot interruption handling based on new setting
if self.spot_interruption_handling and (
self.use_spot or self.use_spot_fleet
):
if not self.spot_interruption_monitor:
logger.debug(
"Initializing SpotInterruptionMonitor after state load"
)
self.spot_interruption_monitor = SpotInterruptionMonitor(
self.session,
provider_id=self.provider_id,
)
self.spot_interruption_monitor.start_monitoring()
elif (
not self.spot_interruption_handling
and self.spot_interruption_monitor
):
logger.debug(
"Stopping SpotInterruptionMonitor after state load"
)
self.spot_interruption_monitor.stop_monitoring()
self.spot_interruption_monitor = None
# Re-register existing spot fleet resources with interruption monitor if needed
if self.spot_interruption_handling and self.spot_interruption_monitor:
for resource_id, resource in self.resources.items():
from parsl_ephemeral_provider.constants import (
RESOURCE_TYPE_SPOT_FLEET,
)
if resource.get(
"resource_type"
) == RESOURCE_TYPE_SPOT_FLEET and resource.get(
"fleet_request_id"
):
fleet_request_id = resource.get("fleet_request_id")
self.spot_interruption_monitor.register_fleet(
fleet_request_id,
self.handle_fleet_interruption,
)
logger.info(
f"Re-registered spot fleet {fleet_request_id} for interruption handling"
)
logger.debug(f"Loaded state with {len(self.resources)} resources")
return True
except Exception as e:
logger.error(f"Failed to load state: {e}")
return False