Revert "Add events method to docker clients (#9932)"

dfangl · dfangl · commit ec272416ab82 · 2024-01-09T17:13:02.000+01:00
This reverts commit 94f9ddf.
diff --git a/localstack/utils/container_utils/container_client.py b/localstack/utils/container_utils/container_client.py
@@ -9,21 +9,9 @@
 import tarfile
 import tempfile
 from abc import ABCMeta, abstractmethod
-from datetime import datetime
 from enum import Enum, unique
 from pathlib import Path
-from typing import (
-    Dict,
-    Generator,
-    List,
-    Literal,
-    NamedTuple,
-    Optional,
-    Protocol,
-    Tuple,
-    Union,
-    get_args,
-)
+from typing import Dict, List, Literal, NamedTuple, Optional, Protocol, Tuple, Union, get_args
 
 from localstack import config
 from localstack.utils.collections import HashableList, ensure_list
@@ -682,18 +670,6 @@ def get_container_logs(self, container_name_or_id: str, safe: bool = False) -> s
     def stream_container_logs(self, container_name_or_id: str) -> CancellableStream:
         """Returns a blocking generator you can iterate over to retrieve log output as it happens."""
 
-    @abstractmethod
-    def events(
-        self,
-        since: Optional[Union[datetime, int]] = None,
-        until: Optional[Union[datetime, int]] = None,
-        filters: Optional[Dict] = None,
-    ) -> Generator[Dict, None, None]:
-        """Returns a generator over events generated by the container runtime.
-
-        More information: https://docs.docker.com/engine/reference/commandline/events
-        """
-
     @abstractmethod
     def inspect_container(self, container_name_or_id: str) -> Dict[str, Union[Dict, str]]:
         """Get detailed attributes of a container.
diff --git a/localstack/utils/container_utils/docker_cmd_client.py b/localstack/utils/container_utils/docker_cmd_client.py
@@ -6,8 +6,7 @@
 import re
 import shlex
 import subprocess
-from datetime import datetime
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import Dict, List, Optional, Tuple, Union
 
 from localstack import config
 from localstack.utils.collections import ensure_list
@@ -38,14 +37,10 @@
 
 class CancellableProcessStream(CancellableStream):
     process: subprocess.Popen
-    map_fn: Callable[[str], Any]
 
-    def __init__(
-        self, process: subprocess.Popen, map_fn: Union[Callable[[str], Any], None] = None
-    ) -> None:
+    def __init__(self, process: subprocess.Popen) -> None:
         super().__init__()
         self.process = process
-        self.map_fn = map_fn
 
     def __iter__(self):
         return self
@@ -54,11 +49,7 @@ def __next__(self):
         line = self.process.stdout.readline()
         if not line:
             raise StopIteration
-
-        if self.map_fn is not None:
-            return self.map_fn(line)
-        else:
-            return line
+        return line
 
     def close(self):
         return self.process.terminate()
@@ -417,41 +408,6 @@ def stream_container_logs(self, container_name_or_id: str) -> CancellableStream:
 
         return CancellableProcessStream(process)
 
-    def events(
-        self,
-        since: Optional[Union[datetime, int]] = None,
-        until: Optional[Union[datetime, int]] = None,
-        filters: Optional[Dict] = None,
-    ) -> CancellableStream:
-        cmd = self._docker_cmd()
-        cmd += ["events", "--format", "{{json .}}"]
-
-        def _to_timestamp(value: Union[datetime, int]):
-            if isinstance(value, datetime):
-                return value.timestamp()
-            else:
-                return value
-
-        if since:
-            cmd += [_to_timestamp(since)]
-        if until:
-            cmd += ["--until", _to_timestamp(until)]
-        if filters:
-            filter_pairs = [f"{key}={value}" for (key, value) in filters.items()]
-            cmd += ["--filter", ",".join(filter_pairs)]
-
-        process: subprocess.Popen = run(
-            cmd, asynchronous=True, outfile=subprocess.PIPE, stderr=subprocess.STDOUT
-        )
-
-        def decode_fn(line: str):
-            try:
-                return json.loads(line)
-            except json.JSONDecodeError as e:
-                LOG.warning("Error decoding docker event %s: %s", line, e)
-
-        return CancellableProcessStream(process, map_fn=decode_fn)
-
     def _inspect_object(self, object_name_or_id: str) -> Dict[str, Union[dict, list, str]]:
         cmd = self._docker_cmd()
         cmd += ["inspect", "--format", "{{json .}}", object_name_or_id]
diff --git a/localstack/utils/container_utils/docker_sdk_client.py b/localstack/utils/container_utils/docker_sdk_client.py
@@ -6,9 +6,8 @@
 import re
 import socket
 import threading
-from datetime import datetime
 from time import sleep
-from typing import Dict, Generator, List, Optional, Tuple, Union, cast
+from typing import Dict, List, Optional, Tuple, Union, cast
 from urllib.parse import quote
 
 import docker
@@ -362,14 +361,6 @@ def stream_container_logs(self, container_name_or_id: str) -> CancellableStream:
         except APIError as e:
             raise ContainerException() from e
 
-    def events(
-        self,
-        since: Optional[Union[datetime, int]] = None,
-        until: Optional[Union[datetime, int]] = None,
-        filters: Optional[Dict] = None,
-    ) -> Generator[Dict, None, None]:
-        yield from self.client().events(since=since, until=until, filters=filters, decode=True)
-
     def inspect_container(self, container_name_or_id: str) -> Dict[str, Union[Dict, str]]:
         try:
             return self.client().containers.get(container_name_or_id).attrs
diff --git a/tests/integration/docker_utils/test_docker.py b/tests/integration/docker_utils/test_docker.py
@@ -2,7 +2,6 @@
 import ipaddress
 import logging
 import os
-import queue
 import re
 import time
 from typing import NamedTuple, Type
@@ -40,7 +39,7 @@
 )
 from localstack.utils.net import Port, PortNotAvailableException, get_free_tcp_port
 from localstack.utils.strings import to_bytes
-from localstack.utils.threads import FuncThread, start_thread
+from localstack.utils.threads import FuncThread
 from tests.integration.docker_utils.conftest import is_podman_test, skip_for_podman
 
 ContainerInfo = NamedTuple(
@@ -1049,69 +1048,6 @@ def test_stream_logs(self, docker_client: ContainerClient):
         finally:
             docker_client.remove_container(container_name)
 
-    def test_events(self, docker_client: ContainerClient):
-        # create background thread watching for events
-        q = queue.Queue()
-
-        should_stop = False
-        container_name = _random_container_name()
-
-        def stream_messages(*_):
-            for event in docker_client.events(filters={"id": "container_name"}):
-                if should_stop:
-                    break
-                q.put(event)
-
-        start_thread(stream_messages, name="docker-events-poller")
-
-        # run a container to generate some events
-        id = docker_client.create_container(
-            "alpine",
-            name=container_name,
-            detach=False,
-            command=["sh", "-c", "sleep 1"],
-        )
-        try:
-            docker_client.start_container(
-                id,
-                attach=True,
-            )
-        finally:
-            docker_client.remove_container(container_name)
-
-        should_stop = True
-
-        # flags to indicate that expected messages have been observed
-        # running a container and then removing the container should at least
-        # contain the following:
-        #
-        # - {"status": "create", ...}
-        # - {"status": "destroy", ...}
-        received_create = False
-        received_destroy = False
-
-        max_messages = 50
-        for _ in range(max_messages):
-            if received_create and received_destroy:
-                break
-
-            msg = q.get()
-
-            # filter out only events for this container
-            if msg.get("id") != id:
-                continue
-
-            # update test state based on message content
-            match msg.get("status"):
-                case "create":
-                    received_create = True
-                case "destroy":
-                    received_destroy = True
-
-            q.task_done()
-
-        assert received_create and received_destroy
-
     @markers.skip_offline
     def test_pull_docker_image(self, docker_client: ContainerClient):
         try:
