improve kinesis startup routine and add KINESIS_MOCK_FORCE_JAVA config option (localstack#4344)

thrau · web-flow · commit 77d873d7889f · 2021-07-22T17:38:47.000+02:00
diff --git a/localstack/services/install.py b/localstack/services/install.py
@@ -12,7 +12,7 @@
 import requests
 
 from localstack import config
-from localstack.config import KINESIS_PROVIDER
+from localstack.config import KINESIS_PROVIDER, is_env_true
 from localstack.constants import (
     DEFAULT_SERVICE_PORTS,
     DYNAMODB_JAR_URL,
@@ -224,7 +224,11 @@ def install_kinesis_mock():
     is_probably_m1 = system == "darwin" and ("arm64" in version or "arm32" in version)
 
     LOG.debug("getting kinesis-mock for %s %s", system, machine)
-    if (machine == "x86_64" or machine == "amd64") and not is_probably_m1:
+
+    if is_env_true("KINESIS_MOCK_FORCE_JAVA"):
+        # sometimes the static binaries may have problems, and we want to fal back to Java
+        bin_file = "kinesis-mock.jar"
+    elif (machine == "x86_64" or machine == "amd64") and not is_probably_m1:
         if system == "windows":
             bin_file = "kinesis-mock-mostly-static.exe"
         elif system == "linux":
diff --git a/localstack/services/kinesis/kinesis_starter.py b/localstack/services/kinesis/kinesis_starter.py
@@ -1,14 +1,18 @@
 import logging
+import threading
 
 from localstack import config
 from localstack.constants import MODULE_MAIN_PATH
 from localstack.services import install
 from localstack.services.infra import do_run, log_startup_message, start_proxy_for_service
 from localstack.utils.aws import aws_stack
-from localstack.utils.common import chmod_r, get_free_tcp_port, mkdir, replace_in_file
+from localstack.utils.common import chmod_r, get_free_tcp_port, mkdir, replace_in_file, start_thread
 
 LOGGER = logging.getLogger(__name__)
 
+# event to indicate that the kinesis backend service has stopped (the terminal command has returned)
+kinesis_stopped = threading.Event()
+
 
 def apply_patches_kinesalite():
     files = [
@@ -33,6 +37,31 @@ def start_kinesis(port=None, asynchronous=False, update_listener=None):
         raise Exception('Unsupported Kinesis provider "%s"' % config.KINESIS_PROVIDER)
 
 
+def _run_proxy_and_command(cmd, port, backend_port, update_listener, asynchronous):
+    log_startup_message("Kinesis")
+    start_proxy_for_service("kinesis", port, backend_port, update_listener)
+
+    # TODO: generalize into service manager once it is introduced
+    try:
+        kinesis_cmd = do_run(cmd, asynchronous)
+    finally:
+        if asynchronous:
+
+            def _return_listener(*_):
+                try:
+                    ret_code = kinesis_cmd.result_future.result()
+                    if ret_code != 0:
+                        LOGGER.error("kinesis terminated with return code %s", ret_code)
+                finally:
+                    kinesis_stopped.set()
+
+            start_thread(_return_listener)
+        else:
+            kinesis_stopped.set()
+
+    return kinesis_cmd
+
+
 def start_kinesis_mock(port=None, asynchronous=False, update_listener=None):
     kinesis_mock_bin = install.install_kinesis_mock()
 
@@ -95,9 +124,14 @@ def start_kinesis_mock(port=None, asynchronous=False, update_listener=None):
             initialize_streams_param,
             kinesis_mock_bin,
         )
-    LOGGER.info("starting kinesis-mock proxy %d:%d with cmd: %s", port, backend_port, cmd)
-    start_proxy_for_service("kinesis", port, backend_port, update_listener)
-    return do_run(cmd, asynchronous)
+
+    return _run_proxy_and_command(
+        cmd=cmd,
+        port=port,
+        backend_port=backend_port,
+        update_listener=update_listener,
+        asynchronous=asynchronous,
+    )
 
 
 def start_kinesalite(port=None, asynchronous=False, update_listener=None):
@@ -125,20 +159,30 @@ def start_kinesalite(port=None, asynchronous=False, update_listener=None):
         latency,
         kinesis_data_dir_param,
     )
-    log_startup_message("Kinesis")
-    start_proxy_for_service("kinesis", port, backend_port, update_listener)
-    return do_run(cmd, asynchronous)
+
+    return _run_proxy_and_command(
+        cmd=cmd,
+        port=port,
+        backend_port=backend_port,
+        update_listener=update_listener,
+        asynchronous=asynchronous,
+    )
 
 
 def check_kinesis(expect_shutdown=False, print_error=False):
+    if expect_shutdown is False and kinesis_stopped.is_set():
+        raise AssertionError("kinesis backend has stopped")
+
     out = None
     try:
         # check Kinesis
         out = aws_stack.connect_to_service(service_name="kinesis").list_streams()
     except Exception as e:
         if print_error:
             LOGGER.exception("Kinesis health check failed: %s", e)
+
     if expect_shutdown:
-        assert out is None
+        assert out is None or kinesis_stopped.is_set()
     else:
-        assert isinstance(out["StreamNames"], list)
+        assert not kinesis_stopped.is_set()
+        assert out and isinstance(out.get("StreamNames"), list)
diff --git a/localstack/services/plugins.py b/localstack/services/plugins.py
@@ -1,7 +1,6 @@
 import json
 import logging
 import time
-import traceback
 from concurrent.futures import ThreadPoolExecutor
 
 import requests
@@ -110,10 +109,7 @@ def check_infra(retries=10, expect_shutdown=False, apis=None, additional_checks=
             additional(expect_shutdown=expect_shutdown)
     except Exception as e:
         if retries <= 0:
-            LOG.error(
-                "Error checking state of local environment (after some retries): %s"
-                % traceback.format_exc()
-            )
+            LOG.exception("Error checking state of local environment (after some retries)")
             raise e
         time.sleep(3)
         check_infra(
diff --git a/localstack/utils/bootstrap.py b/localstack/utils/bootstrap.py
@@ -789,6 +789,7 @@ def run(self):
                 kwargs["_thread"] = self
             result = self.func(self.params, **kwargs)
         except Exception as e:
+            self.result_future.set_exception(e)
             result = e
             if not self.quiet:
                 LOG.info(
diff --git a/localstack/utils/common.py b/localstack/utils/common.py
@@ -142,7 +142,7 @@ def run_cmd(self, params):
                 or not self.process
                 or self.process.returncode == 0
             ):
-                return
+                return self.process.returncode if self.process else None
             LOG.info(
                 "Restarting process (received exit code %s): %s"
                 % (self.process.returncode, self.cmd)
@@ -199,6 +199,7 @@ def filter_line(line):
             else:
                 self.process.communicate()
         except Exception as e:
+            self.result_future.set_exception(e)
             if self.process and not self.quiet:
                 LOG.warning('Shell command error "%s": %s' % (e, self.cmd))
         if self.process and not self.quiet and self.process.returncode != 0:
