Fix conflict handling for docker container naming (PrefectHQ#4514)

* Fix conflict handling for docker container naming

* Update test for name conflict handling

src/prefect/agent/docker/agent.py

@@ -458,25 +458,29 @@ def deploy_flow(self, flow_run: GraphQLResult) -> str:
             # would be empty after cleaning
             or flow_run.id
         )
-        # Checking for container names here is a slight deploy performance hit. If
-        # it ends up being an issue we can move to a try/except model
-        existing_names = {c["name"] for c in self.docker_client.containers()}
-        index = 0
-        while container_name in existing_names:
-            index += 1
-            container_name = f"{slugified_name}-{index}"
-
-        # Create the container
-        container = self.docker_client.create_container(
-            image,
-            command=get_flow_run_command(flow_run),
-            environment=env_vars,
-            name=container_name,
-            volumes=container_mount_paths,
-            host_config=self.docker_client.create_host_config(**host_config),
-            networking_config=networking_config,
-            labels=labels,
-        )
+
+        # Create the container with retries on name conflicts
+        index = 0  # will be bumped on name colissions
+        while True:
+            try:
+                container = self.docker_client.create_container(
+                    image,
+                    command=get_flow_run_command(flow_run),
+                    environment=env_vars,
+                    name=container_name,
+                    volumes=container_mount_paths,
+                    host_config=self.docker_client.create_host_config(**host_config),
+                    networking_config=networking_config,
+                    labels=labels,
+                )
+            except docker.errors.APIError as exc:
+                if "Conflict" in str(exc) and "container name" in str(exc):
+                    index += 1
+                    container_name = f"{slugified_name}-{index}"
+                else:
+                    raise
+            else:
+                break
 
         # Connect the rest of the networks
         if self.networks:
@@ -486,7 +490,8 @@ def deploy_flow(self, flow_run: GraphQLResult) -> str:
                 )
         # Start the container
         self.logger.debug(
-            "Starting Docker container with ID {}".format(container.get("Id"))
+            f"Starting Docker container with ID {container.get('Id')} and "
+            f"name {container_name!r}"
         )
         if self.networks:
             self.logger.debug(

tests/agent/test_docker_agent.py

@@ -315,14 +315,20 @@ def test_docker_agent_deploy_flow_sets_container_name_with_index(api, collision_
 
     if collision_count:
         # Add the basic name first
-        existing_containers = [{"name": "flow-run-name"}]
+        existing_names = ["flow-run-name"]
         for i in range(1, collision_count):
-            existing_containers.append({"name": f"flow-run-name-{i}"})
+            existing_names.append(f"flow-run-name-{i}")
     else:
-        # Ensure a random container doesn't matter
-        existing_containers = [{"name": "foobar"}]
+        existing_names = []
 
-    api.containers.return_value = existing_containers
+    def fail_if_name_exists(*args, **kwargs):
+        if kwargs.get("name") in existing_names:
+            raise docker.errors.APIError(
+                "Conflict. The container name 'foobar' is already in use"
+            )
+        return {}
+
+    api.create_container = MagicMock(side_effect=fail_if_name_exists)
 
     agent = DockerAgent()
     agent.deploy_flow(
@@ -345,10 +351,9 @@ def test_docker_agent_deploy_flow_sets_container_name_with_index(api, collision_
         )
     )
 
-    expected_name = "flow-run-name"
-    if collision_count:
-        expected_name += f"-{collision_count}"
-
+    expected_name = (
+        "flow-run-name" if not collision_count else f"flow-run-name-{collision_count}"
+    )
     assert api.create_container.call_args[1]["name"] == expected_name