Merge pull request #23517 from def-/pr-docker-weirdness

mzcompose: Safer docker compose file

Author: def-

misc/python/materialize/mzcompose/composition.py

@@ -163,6 +163,10 @@ def __init__(
         if munge_services:
             self.dependencies = self._munge_services(self.compose["services"].items())
 
+        # Emit the munged configuration to a temporary file so that we can later
+        # pass it to Docker Compose.
+        self._write_compose()
+
     def _munge_services(
         self, services: list[tuple[str, dict]]
     ) -> mzbuild.DependencySet:
@@ -229,6 +233,12 @@ def _munge_services(
 
         return deps
 
+    def _write_compose(self) -> None:
+        new_file = TemporaryFile(mode="w")
+        os.set_inheritable(new_file.fileno(), True)
+        yaml.dump(self.compose, new_file)
+        self.file = new_file
+
     def invoke(
         self,
         *args: str,
@@ -263,12 +273,8 @@ def invoke(
             ("--project-name", self.project_name) if self.project_name else ()
         )
 
-        # Emit the munged configuration to a temporary file so that we can later
-        # pass it to Docker Compose.
-        file = TemporaryFile(mode="w")
-        os.set_inheritable(file.fileno(), True)
-        yaml.dump(self.compose, file)
-        file.flush()
+        # Make sure file doesn't get changed/deleted while we use it
+        file = self.file
 
         ret = None
         for retry in range(1, max_tries + 1):
@@ -393,6 +399,8 @@ def override(self, *services: "Service") -> Iterator[None]:
         # config for an `mzbuild` config.
         deps.acquire()
 
+        self._write_compose()
+
         # Ensure image freshness
         self.pull_if_variable([service.name for service in services])
 
@@ -421,6 +429,7 @@ def override(self, *services: "Service") -> Iterator[None]:
 
             # Restore the old composition.
             self.compose = old_compose
+            self._write_compose()
 
     @contextmanager
     def test_case(self, name: str) -> Iterator[None]:
@@ -687,6 +696,7 @@ def up(
             for service in self.compose["services"].values():
                 service["entrypoint"] = ["sleep", "infinity"]
                 service["command"] = []
+            self._write_compose()
 
         self.invoke(
             "up",
@@ -698,6 +708,7 @@ def up(
 
         if persistent:
             self.compose = old_compose  # type: ignore
+            self._write_compose()
 
     def validate_sources_sinks_clusters(self) -> str | None:
         """Validate that all sources, sinks & clusters are in a good state"""

test/cluster/mzcompose.py

@@ -74,7 +74,8 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
 
     for i, name in enumerate(c.workflows):
         # incident-70 requires more memory, runs in separate CI step
-        if name in ("default", "test-incident-70"):
+        # concurrent-connections is too flaky
+        if name in ("default", "test-incident-70", "test-concurrent-connections"):
             continue
         if shard is None or shard_count is None or i % int(shard_count) == shard:
             with c.test_case(name):
@@ -2570,7 +2571,7 @@ def workflow_test_concurrent_connections(c: Composition) -> None:
 
     def worker(c: Composition, i: int) -> None:
         start_time = time.time()
-        c.sql("SELECT 1")
+        c.sql("SELECT 1", print_statement=False)
         end_time = time.time()
         runtimes[i] = end_time - start_time