Enable removing networkConfiguration from AWS ECS Run Task API call (#…

…26893)

## Summary & Motivation
When ECS Run Launcher is used it calculates ECS Task
`networkConfiguration`
[here](https://github.com/dagster-io/dagster/blob/0f2de9b6a006e45cd70f886cad0c8eebd7512036/python_modules/libraries/dagster-aws/dagster_aws/ecs/tasks.py#L375).
This works well for AWS Tasks with `awsvpc` networkMode, but for other
networkModes the configuration should not be passed to the run task API
call
([documentation](https://docs.aws.amazon.com/cli/latest/reference/ecs/run-task.html)).

The problem is that if you set `networkConfiguration` to null, example:
```
run_launcher:
  module: "dagster_aws.ecs"
  class: "EcsRunLauncher"
  config:
    task_definition: arn:aws:ecs:...
    container_name: ...
    run_task_kwargs:
      networkConfiguration: null
```
then the null value will be passed "as is" to the API call, causing a
following error:
```
botocore.exceptions.ParamValidationError: Parameter validation failed:
Invalid type for parameter networkConfiguration, value: None, type: <class 'NoneType'>, valid types: <class 'dict'>
```

This PR introduces a check: if `networkConfiguration` is set to None, it
will be removed from the arguments of the API call.

This change will enable dagster users to use `bridge` network mode in
ECS Tasks which was requested in this
[issue](#16420) issue (3
requests including us).

## How I Tested These Changes
I ran dagster with this change and it successfully launched jobs as AWS
Tasks with `bridge` network mode.

---------

Co-authored-by: Levy Melnikov <[email protected]>

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

@@ -508,6 +508,13 @@ def launch_run(self, context: LaunchRunContext) -> None:
         if "launchType" in run_task_kwargs and run_task_kwargs.get("capacityProviderStrategy"):
             del run_task_kwargs["launchType"]
 
+        # Remove networkConfiguration if it is set to None
+        if (
+            "networkConfiguration" in run_task_kwargs
+            and run_task_kwargs.get("networkConfiguration") is None
+        ):
+            del run_task_kwargs["networkConfiguration"]
+
         # Run a task using the same network configuration as this processes's task.
         task = backoff(
             self._run_task,

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

@@ -1423,3 +1423,56 @@ def test_external_launch_type(
 
         assert task["taskDefinitionArn"] == task_definition["taskDefinitionArn"]
         assert task["launchType"] == "EXTERNAL"
+
+
+def test_removing_network_configuration(
+    ecs,
+    instance_cm,
+    workspace,
+    remote_job,
+    job,
+):
+    container_name = "external"
+
+    task_definition = ecs.register_task_definition(
+        family="external",
+        containerDefinitions=[{"name": container_name, "image": "dagster:first"}],
+        networkMode="bridge",
+        memory="512",
+        cpu="256",
+    )["taskDefinition"]
+
+    assert task_definition["networkMode"] == "bridge"
+    task_definition_arn = task_definition["taskDefinitionArn"]
+
+    # You can provide a family or a task definition ARN
+    with instance_cm(
+        {
+            "task_definition": task_definition_arn,
+            "container_name": container_name,
+            "run_task_kwargs": {"networkConfiguration": None},
+        }
+    ) 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"]
+
+        instance.launch_run(run.run_id, workspace)
+
+        # A new task definition is not created
+        assert ecs.list_task_definitions()["taskDefinitionArns"] == initial_task_definitions
+
+        # 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]
+
+        assert task["taskDefinitionArn"] == task_definition["taskDefinitionArn"]
+        assert "networkConfiguration" not in task