Support aws ecs readonly_root_filesystem

docs/content/dagster-plus/deployment/agents/amazon-ecs/configuration-reference.mdx

@@ -73,6 +73,7 @@ locations:
             value: MyEcsTagValue
         run_ecs_tags:
           - key: MyEcsTagKeyWithoutValue
+        readonly_root_filesystem: true
         repository_credentials: MyRepositoryCredentialsSecretArn
 ```
 
@@ -171,6 +172,7 @@ user_code_launcher:
         value: MyEcsTagValue
     run_ecs_tags:
       - key: MyEcsTagKeyWithoutValue
+    readonly_root_filesystem: true
     repository_credentials: MyRepositoryCredentialsSecretArn
 
 isolated_agents:
@@ -308,6 +310,11 @@ agent_queues:
   Additional ECS tags to include in the task for each run. If set, must be a list of dictionaries, each with a <code>key</code> key and optional <code>value</code> key.
   </ReferenceTableItem>
   <ReferenceTableItem
+  propertyName="config.readonly_root_filesystem">
+  Optional. When this parameter is true, the container is given read-only access to its root file system. This parameter maps to ReadonlyRootfs in the docker container create command and the --read-only option to docker run.
+  <a href="https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-ecs-taskdefinition-containerdefinition.html#cfn-ecs-taskdefinition-containerdefinition-readonlyrootfilesystem">See AWS ECS docs.</a>.
+  </ReferenceTableItem>
+  <ReferenceTableItem
   propertyName="config.repository_credentials">
   Optional arn of the secret to authenticate into your private container registry. This does not apply if you are leveraging ECR for your images, see <a href="https://docs.aws.amazon.com/AmazonECS/latest/userguide/private-auth.html">the AWS private auth guide</a>.
   </ReferenceTableItem>

docs/docs-beta/docs/dagster-plus/deployment/deployment-types/hybrid/amazon-ecs/configuration-reference.md

@@ -72,6 +72,7 @@ locations:
             value: MyEcsTagValue
         run_ecs_tags:
           - key: MyEcsTagKeyWithoutValue
+        readonly_root_filesystem: true
         repository_credentials: MyRepositoryCredentialsSecretArn
 ```
 
@@ -195,6 +196,7 @@ user_code_launcher:
         value: MyEcsTagValue
     run_ecs_tags:
       - key: MyEcsTagKeyWithoutValue
+    readonly_root_filesystem: true
     repository_credentials: MyRepositoryCredentialsSecretArn
 
 isolated_agents:
@@ -239,6 +241,7 @@ agent_queues:
 | config.server_ecs_tags | Additional ECS tags to include in the service for each code location. If set, must be a list of dictionaries, each with a `key` key and optional `value` key. |
 | config.run_ecs_tags | AAdditional ECS tags to include in the task for each run. If set, must be a list of dictionaries, each with a `key` key and optional `value` key. |
 | config.repository_credentials | Optional arn of the secret to authenticate into your private container registry. This does not apply if you are leveraging ECR for your images, see the [AWS private auth guide.](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/private-auth.html) |
+| config.readonly_root_filesystem | Optional. When this parameter is true, the container is given read-only access to its root file system. This parameter maps to ReadonlyRootfs in the docker container create command and the --read-only option to docker run. [See AWS ECS docs.](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-ecs-taskdefinition-containerdefinition.html#cfn-ecs-taskdefinition-containerdefinition-readonlyrootfilesystem) |
 | config.enable_ecs_exec | Boolean that determines whether tasks created by the agent should be configured with the needed linuxParameters and permissions to use [ECS Exec](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/ecs-exec.html) to shell into the task. Also grants the `SYS_PTRACE` linux capability to enable running tools like py-spy to debug slow or hanging tasks. Defaults to false. **Note**: For ECS Exec to work, the task IAM role must be granted [certain permissions](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-iam-roles.html#ecs-exec-required-iam-permissions). |
 
 

python_modules/libraries/dagster-aws/dagster_aws/ecs/container_context.py

@@ -130,6 +130,14 @@
             " for more information."
         ),
     ),
+    "readonly_root_filesystem": Field(
+        BoolSource,
+        is_required=False,
+        description=(
+            "When true, the root filesystem of the container is read-only."
+            "See https://docs.aws.amazon.com/AmazonECS/latest/APIReference/API_ContainerDefinition.html"
+        ),
+    ),
     "repository_credentials": Field(
         StringSource,
         is_required=False,
@@ -246,6 +254,7 @@ class EcsContainerContext(
             ("run_sidecar_containers", Sequence[Mapping[str, Any]]),
             ("server_ecs_tags", Sequence[Mapping[str, Optional[str]]]),
             ("run_ecs_tags", Sequence[Mapping[str, Optional[str]]]),
+            ("readonly_root_filesystem", Optional[bool]),
             ("repository_credentials", Optional[str]),
             ("server_health_check", Optional[Mapping[str, Any]]),
         ],
@@ -271,6 +280,7 @@ def __new__(
         run_sidecar_containers: Optional[Sequence[Mapping[str, Any]]] = None,
         server_ecs_tags: Optional[Sequence[Mapping[str, Optional[str]]]] = None,
         run_ecs_tags: Optional[Sequence[Mapping[str, Optional[str]]]] = None,
+        readonly_root_filesystem: Optional[bool] = None,
         repository_credentials: Optional[str] = None,
         server_health_check: Optional[Mapping[str, Any]] = None,
     ):
@@ -298,6 +308,7 @@ def __new__(
             ),
             server_ecs_tags=check.opt_sequence_param(server_ecs_tags, "server_ecs_tags"),
             run_ecs_tags=check.opt_sequence_param(run_ecs_tags, "run_tags"),
+            readonly_root_filesystem=check.opt_bool_param(readonly_root_filesystem, "readonly_root_filesystem"),
             repository_credentials=check.opt_str_param(
                 repository_credentials, "repository_credentials"
             ),
@@ -325,6 +336,7 @@ def merge(self, other: "EcsContainerContext") -> "EcsContainerContext":
             run_sidecar_containers=[*other.run_sidecar_containers, *self.run_sidecar_containers],
             server_ecs_tags=[*other.server_ecs_tags, *self.server_ecs_tags],
             run_ecs_tags=[*other.run_ecs_tags, *self.run_ecs_tags],
+            readonly_root_filesystem=other.readonly_root_filesystem if other.readonly_root_filesystem is not None else self.readonly_root_filesystem,
             repository_credentials=other.repository_credentials or self.repository_credentials,
             server_health_check=other.server_health_check or self.server_health_check,
         )
@@ -357,6 +369,7 @@ def create_for_run(dagster_run: DagsterRun, run_launcher: Optional["EcsRunLaunch
                     volumes=run_launcher.volumes,
                     run_sidecar_containers=run_launcher.run_sidecar_containers,
                     run_ecs_tags=run_launcher.run_ecs_tags,
+                    readonly_root_filesystem=run_launcher.readonly_root_filesystem,
                     repository_credentials=run_launcher.repository_credentials,
                 )
             )
@@ -419,6 +432,7 @@ def create_from_config(run_container_context) -> "EcsContainerContext":
                 run_sidecar_containers=processed_context_value.get("run_sidecar_containers"),
                 server_ecs_tags=processed_context_value.get("server_ecs_tags"),
                 run_ecs_tags=processed_context_value.get("run_ecs_tags"),
+                readonly_root_filesystem=processed_context_value.get("readonly_root_filesystem"),
                 repository_credentials=processed_context_value.get("repository_credentials"),
                 server_health_check=processed_context_value.get("server_health_check"),
             )

python_modules/libraries/dagster-aws/dagster_aws/ecs/launcher.py

@@ -252,6 +252,12 @@ def volumes(self) -> Optional[Sequence[Mapping[str, Any]]]:
             return None
         return self.task_definition_dict.get("volumes")
 
+    @property
+    def readonly_root_filesystem(self) -> Optional[bool]:
+        if not self.task_definition_dict:
+            return None
+        return self.task_definition_dict.get("readonly_root_filesystem")
+
     @property
     def repository_credentials(self) -> Optional[str]:
         if not self.task_definition_dict:
@@ -712,6 +718,7 @@ def _run_task_kwargs(
                     runtime_platform=runtime_platform,
                     volumes=container_context.volumes,
                     mount_points=container_context.mount_points,
+                    readonly_root_filesystem=container_context.readonly_root_filesystem,
                     repository_credentials=container_context.repository_credentials,
                     linux_parameters=self.linux_parameters,
                 )
@@ -734,6 +741,7 @@ def _run_task_kwargs(
                     ephemeral_storage=container_context.run_resources.get("ephemeral_storage"),
                     volumes=container_context.volumes,
                     mount_points=container_context.mount_points,
+                    readonly_root_filesystem=container_context.readonly_root_filesystem,
                     additional_sidecars=container_context.run_sidecar_containers,
                     repository_credentials=container_context.repository_credentials,
                 )

python_modules/libraries/dagster-aws/dagster_aws/ecs/tasks.py

@@ -43,6 +43,7 @@ class DagsterEcsTaskDefinitionConfig(
             ("runtime_platform", Mapping[str, Any]),
             ("mount_points", Sequence[Mapping[str, Any]]),
             ("volumes", Sequence[Mapping[str, Any]]),
+            ("readonly_root_filesystem", Optional[bool]),
             ("repository_credentials", Optional[str]),
             ("linux_parameters", Optional[Mapping[str, Any]]),
             ("health_check", Optional[Mapping[str, Any]]),
@@ -72,6 +73,7 @@ def __new__(
         runtime_platform: Optional[Mapping[str, Any]] = None,
         mount_points: Optional[Sequence[Mapping[str, Any]]] = None,
         volumes: Optional[Sequence[Mapping[str, Any]]] = None,
+        readonly_root_filesystem: Optional[bool] = None,
         repository_credentials: Optional[str] = None,
         linux_parameters: Optional[Mapping[str, Any]] = None,
         health_check: Optional[Mapping[str, Any]] = None,
@@ -95,6 +97,7 @@ def __new__(
             check.opt_mapping_param(runtime_platform, "runtime_platform"),
             check.opt_sequence_param(mount_points, "mount_points"),
             check.opt_sequence_param(volumes, "volumes"),
+            check.opt_bool_param(readonly_root_filesystem, "readonly_root_filesystem"),
             check.opt_str_param(repository_credentials, "repository_credentials"),
             check.opt_mapping_param(linux_parameters, "linux_parameters"),
             check.opt_mapping_param(health_check, "health_check"),
@@ -130,6 +133,11 @@ def task_definition_dict(self):
                         else {}
                     ),
                     ({"linuxParameters": self.linux_parameters} if self.linux_parameters else {}),
+                    (
+                        {"readonlyRootFilesystem": self.readonly_root_filesystem}
+                        if self.readonly_root_filesystem is not None
+                        else {}
+                    ),
                     ({"healthCheck": self.health_check} if self.health_check else {}),
                 ),
                 *self.sidecars,
@@ -232,6 +240,7 @@ def from_task_definition_dict(task_definition_dict, container_name):
             runtime_platform=task_definition_dict.get("runtimePlatform"),
             mount_points=container_definition.get("mountPoints"),
             volumes=task_definition_dict.get("volumes"),
+            readonly_root_filesystem=container_definition.get("readonlyRootFilesystem"),
             repository_credentials=container_definition.get("repositoryCredentials", {}).get(
                 "credentialsParameter"
             ),
@@ -274,6 +283,7 @@ def get_task_definition_dict_from_current_task(
     mount_points=None,
     volumes=None,
     additional_sidecars=None,
+    readonly_root_filesystem=None,
     repository_credentials=None,
 ):
     current_container_name = current_ecs_container_name()
@@ -331,6 +341,11 @@ def get_task_definition_dict_from_current_task(
         ),
         **({"secrets": secrets} if secrets else {}),
         **({} if include_sidecars else {"dependsOn": []}),
+        **(
+            {"readonlyRootFilesystem": readonly_root_filesystem}
+            if readonly_root_filesystem is not None
+            else {}
+        ),
     }
     if environment:
         new_container_definition["environment"] = [

python_modules/libraries/dagster-aws/dagster_aws_tests/ecs_tests/launcher_tests/test_container_context.py

@@ -115,6 +115,8 @@ def test_merge(
         },
     ]
 
+    assert merged.readonly_root_filesystem == None
+
     assert merged.repository_credentials == "fake-secret-arn"
 
     assert merged.volumes == [

python_modules/libraries/dagster-aws/dagster_aws_tests/ecs_tests/launcher_tests/test_launching.py

@@ -924,6 +924,51 @@ def test_launching_custom_task_definition(ecs, instance_cm, run, workspace, job,
         assert run.run_id in str(override["command"])
 
 
+@pytest.mark.parametrize("readonly_root_filesystem", [True, False])
+def test_readonly_root_filesystem(ecs, instance_cm, run, workspace, job, remote_job, readonly_root_filesystem):
+    with instance_cm(
+        {
+            "task_definition": {
+                "task_role_arn": "fake-task-role",
+                "execution_role_arn": "fake-execution-role",
+                "readonly_root_filesystem": readonly_root_filesystem,
+            },
+        }
+    ) as instance:
+        run = instance.create_run_for_job(
+            job,
+            remote_job_origin=remote_job.get_remote_origin(),
+            job_code_origin=remote_job.get_python_origin(),
+        )
+
+        initial_task_definitions = ecs.list_task_definitions()["taskDefinitionArns"]
+        initial_tasks = ecs.list_tasks()["taskArns"]
+
+        # Launch the run
+        instance.launch_run(run.run_id, workspace)
+
+        # A new task definition is created
+        task_definitions = ecs.list_task_definitions()["taskDefinitionArns"]
+        assert len(task_definitions) == len(initial_task_definitions) + 1
+
+        # A new task is launched
+        tasks = ecs.list_tasks()["taskArns"]
+        assert len(tasks) == len(initial_tasks) + 1
+
+        task_arn = next(iter(set(tasks).difference(initial_tasks)))
+        task = ecs.describe_tasks(tasks=[task_arn])["tasks"][0]
+        task_definition_arn = task["taskDefinitionArn"]
+
+        # Get the task definition and container definition
+        task_definition = ecs.describe_task_definition(taskDefinition=task_definition_arn)[
+            "taskDefinition"
+        ]
+        container_definition = task_definition["containerDefinitions"][0]
+
+        # Assert that readonlyRootFilesystem is set to correctly
+        assert container_definition.get("readonlyRootFilesystem") is readonly_root_filesystem
+
+
 def test_eventual_consistency(ecs, instance, workspace, run, monkeypatch):
     initial_tasks = ecs.list_tasks()["taskArns"]