Fixing support for python2 (#1218)

alesnovak-s1 · web-flow · commit 284dc549478c · 2023-11-16T14:51:08.000+01:00
* Fixing python2 smoke tests

* ThreadPool fallback for python2

* ThreadPool fallback for python2

* ThreadPool fallback for python2

* ThreadPool fallback for python2

* Added logging. Removed cancel_fututures param as it was introduced in python 3.9.

* Logging enabled

* Some logging

* Some logging

* Git added to the testing image

* test

* test

* agent log logging

* fixing python2 test

* fixing python2 test

* reenabling all tests

* reverting unused code

* Don't wait on ThreadPoolExecutor

* test

* test

* test

* test

* test

* test

* test

* Python 2.7 added to the unit tests

* Changing version format and grouping apt install packages into one line

* python2 not available in ubuntu 20.04 GHA anymore

* Changing back the version format
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-2.2.7
+2.2.7.pre8.1
diff --git a/scalyr_agent/builtin_monitors/mysql_monitor.py b/scalyr_agent/builtin_monitors/mysql_monitor.py
@@ -459,7 +459,7 @@ def _connect(self):
             self._cursor = None
             self._logger.error("Database connect failed: %s" % me)
         except Exception as ex:
-            self._logger.error("Exception trying to connect occured:  %s" % ex)
+            self._logger.error("Exception trying to connect occured:  %s" % ex, exc_info=ex)
             raise Exception("Exception trying to connect:  %s" % ex)
 
     def _close(self):
@@ -1314,6 +1314,7 @@ def _connect_to_db(self):
                 "Error establishing database connection: %s" % (six.text_type(e)),
                 limit_once_per_x_secs=300,
                 limit_key="mysql_connect_to_db",
+                exc_info=e
             )
 
     def gather_sample(self):
diff --git a/scalyr_agent/builtin_monitors/syslog_monitor.py b/scalyr_agent/builtin_monitors/syslog_monitor.py
@@ -26,8 +26,6 @@
 from __future__ import print_function
 
 import queue
-from concurrent.futures import ThreadPoolExecutor, CancelledError
-from socketserver import ThreadingMixIn
 
 from scalyr_agent import compat
 
@@ -56,7 +54,10 @@
 from six.moves import range
 import six.moves.socketserver
 
-from scalyr_agent.builtin_monitors.thread_pool import ExecutorMixIn
+if six.PY2:
+    from scalyr_agent.builtin_monitors.thread_pool_dummy import ExecutorMixIn
+else:
+    from scalyr_agent.builtin_monitors.thread_pool import ExecutorMixIn
 
 try:
     # Only available for python >= 3.6
@@ -591,6 +592,7 @@ def __init__(self, request_processing_executor, request, client_address, server)
         else:
             six.moves.socketserver.BaseRequestHandler.__init__(self, request, client_address, server)
 
+    @staticmethod
     def factory_method(request_processing_executor):
         def create_handler(request, client_address, server):
             return SyslogUDPHandler(request_processing_executor, request, client_address, server)
@@ -606,9 +608,13 @@ def handle(self):
             "destport": self.server.server_address[1],
         }
 
-        self.__request_processing_executor.submit(
-            self.server.syslog_handler.handle, data, extra
-        )
+        if self.__request_processing_executor:
+            self.__request_processing_executor.submit(
+                self.server.syslog_handler.handle, data, extra
+            )
+        else:
+            self.server.syslog_handler.handle(data, extra)
+
 
 
 class SocketNotReadyException(Exception):
@@ -636,7 +642,9 @@ def read(self):
             if not data:
                 self.is_closed = True
                 raise SocketClosed()
-        except (socket.timeout, *NON_BLOCKING_SOCKET_DATA_NOT_READY_EXCEPTIONS) as e:
+        except socket.timeout as e:
+            raise SocketNotReadyException(e)
+        except NON_BLOCKING_SOCKET_DATA_NOT_READY_EXCEPTIONS as e:
             raise SocketNotReadyException(e)
         except socket.error as e:
             if e.errno == errno.EAGAIN:
@@ -980,8 +988,6 @@ class SyslogTCPHandler(six.moves.socketserver.BaseRequestHandler):
 
     def __init__(self, *args, **kwargs):
         self.__request_processing_executor = kwargs.pop("request_processing_executor", None)
-        if not self.__request_processing_executor:
-            raise ValueError("request_processing_executor is required")
 
         self.request_parser = kwargs.pop("request_parser", "default")
         self.incomplete_frame_timeout = kwargs.pop("incomplete_frame_timeout", None)
@@ -1077,25 +1083,30 @@ def handle(self):
         )
 
         try:
-            # Worker is responsible for processing the data read from one request.
-            # Using the queue ensures the data is processed in the order it was read.
-            DONE = "DONE"
-            work_queue = queue.Queue()
-            def worker(queue, is_shutdown):
-                data = queue.get(block=True)
-                while data != DONE:
-                    if is_shutdown():
-                        global_log.info("ThreadPool shutting down, skipping further request processing.")
-                        break
-                    self.__request_data_process(syslog_parser, data)
+            if self.__request_processing_executor:
+                # Worker is responsible for processing the data read from one request.
+                # Using the queue ensures the data is processed in the order it was read.
+                DONE = "DONE"
+                work_queue = queue.Queue()
+                def worker(queue, is_shutdown):
                     data = queue.get(block=True)
+                    while data != DONE:
+                        if is_shutdown():
+                            global_log.info("ThreadPool shutting down, skipping further request processing.")
+                            break
+                        self.__request_data_process(syslog_parser, data)
+                        data = queue.get(block=True)
 
-            self.__request_processing_executor.submit(worker, work_queue, lambda: self.__request_processing_executor._shutdown)
+                self.__request_processing_executor.submit(worker, work_queue, lambda: self.__request_processing_executor._shutdown)
 
-            for data in self.__request_stream_read(syslog_request, self.server.is_running):
-                work_queue.put(data)
+                for data in self.__request_stream_read(syslog_request, self.server.is_running):
+                    work_queue.put(data)
+
+                work_queue.put(DONE)
+            else:
+                for data in self.__request_stream_read(syslog_request, self.server.is_running):
+                    self.__request_data_process(syslog_parser, data)
 
-            work_queue.put(DONE)
 
         except Exception as e:
             global_log.warning(
diff --git a/scalyr_agent/builtin_monitors/thread_pool.py b/scalyr_agent/builtin_monitors/thread_pool.py
@@ -76,11 +76,11 @@ def get_singleton(cls, name, max_workers=None):
         return cls.__instances[name]
 
     @classmethod
-    def shutdown(cls, wait=True, cancel_futures=False):
+    def shutdown(cls, wait=True):
         wait_on_futures_str = " and waiting on futures" if wait else ""
         for name, executor in cls.__instances.items():
             global_log.info("Shutting down ThreadPoolExecutor%s: %s", wait_on_futures_str, name)
             try:
-                executor.shutdown(wait=wait, cancel_futures=cancel_futures)
+                executor.shutdown(wait=wait)
             except Exception as e:
                 global_log.error("Failed shutting down the ThreadPoolExecutor %s", executor, exc_info=e)
diff --git a/scalyr_agent/builtin_monitors/thread_pool_dummy.py b/scalyr_agent/builtin_monitors/thread_pool_dummy.py
@@ -0,0 +1,11 @@
+from six.moves.socketserver import ThreadingMixIn
+
+class ExecutorMixIn(ThreadingMixIn):
+    def __init__(self, global_config):
+        self._request_reading_executor = None
+        self._request_processing_executor = None
+
+class ThreadPoolExecutorFactory():
+    @classmethod
+    def shutdown(cls, wait=True):
+        pass
diff --git a/scalyr_agent/monitors_manager.py b/scalyr_agent/monitors_manager.py
@@ -17,8 +17,6 @@
 from __future__ import unicode_literals
 from __future__ import absolute_import
 
-from scalyr_agent.builtin_monitors.thread_pool import ThreadPoolExecutorFactory
-
 if False:  # NOSONAR
     from typing import List
 
@@ -41,6 +39,13 @@
 
 import six
 
+if six.PY2:
+    from scalyr_agent.builtin_monitors.thread_pool_dummy import ThreadPoolExecutorFactory
+else:
+    from scalyr_agent.builtin_monitors.thread_pool import ThreadPoolExecutorFactory
+
+
+
 log = scalyr_logging.getLogger(__name__)
 
 # Holds reference to the currently active MonitorsManager instance (singleton). Reference to this
@@ -211,7 +216,7 @@ def stop_manager(self, wait_on_join=True, join_timeout=5):
         start_time = time.time()
 
         log.info("Stopping shared thread pools")
-        ThreadPoolExecutorFactory.shutdown(wait=False, cancel_futures=True)
+        ThreadPoolExecutorFactory.shutdown(wait=False)
 
 
         for monitor in self.__running_monitors:
@@ -235,8 +240,6 @@ def stop_manager(self, wait_on_join=True, join_timeout=5):
                 except Exception:
                     log.exception("Failed to stop the metric log due to an exception")
 
-                ThreadPoolExecutorFactory.shutdown(wait=True, cancel_futures=False)
-
 
 
         for monitor in self.__running_monitors:
diff --git a/tests/end_to_end_tests/managed_packages_tests/remote_machine_tests/tools/__init__.py b/tests/end_to_end_tests/managed_packages_tests/remote_machine_tests/tools/__init__.py
@@ -73,6 +73,8 @@ def download_stable_packages(
 
     output_dir.mkdir(parents=True, exist_ok=True)
     agent_package_path = output_dir / file_name
+    if agent_package_path.exists():
+        raise Exception("While downloading a stable package, found an existing one: " + str(agent_package_path))
     with requests.Session() as s:
         resp = s.get(url=package_url)
         resp.raise_for_status()
diff --git a/tests/image_builder/monitors/base/Dockerfile b/tests/image_builder/monitors/base/Dockerfile
@@ -2,9 +2,7 @@ FROM ubuntu:18.04
 
 ENV DEBIAN_FRONTEND=noninteractive
 RUN apt update -y
-RUN apt install -y mysql-server
-RUN apt install -y postgresql
-
+RUN apt install -y mysql-server postgresql git
 
 USER postgres
 RUN service postgresql start && createuser root && createdb root && psql -c "alter user root superuser;" && service postgresql stop
diff --git a/tests/smoke_tests/monitors_test/mysql_test.py b/tests/smoke_tests/monitors_test/mysql_test.py
@@ -34,26 +34,33 @@
 from tests.utils.log_reader import LogMetricReader, LogReaderError
 from tests.utils.log_reader import AgentLogReader
 
+from scalyr_agent import scalyr_logging
+
 import six
 
 HOST = "localhost"
 USERNAME = "scalyr_test_user"
 PASSWORD = "scalyr_test_password"
 DATABASE = "scalyr_test_db"
 
+global_log = scalyr_logging.getLogger(__name__)
 
 @pytest.fixture()
 def mysql_client():
     # we change owner of the mysql files to workaround the issue which happens with mysql server in docker.
     # see: https://serverfault.com/a/872576
     os.system("chown -R mysql:mysql /var/lib/mysql /var/run/mysqld")
 
+    global_log.info("Starting the mysql server")
     exit_code = os.system("service mysql start --ssl")
 
     # On failure include service logs for ease of debugging
     if exit_code != 0:
+        global_log.error("Mysql server start error, dumping logs:")
         os.system("cat /var/log/mysql/mysql.log || true")
 
+    assert exit_code == 0, "Mysql server start error"
+
     os.system("mysql < /init.sql")
 
     time.sleep(3)
@@ -111,7 +118,6 @@ def _agent_config(self):  # type: () -> Dict[six.text_type, Any]
 class MySqlLogReader(LogMetricReader):
     LINE_PATTERN = r"\s*(?P<timestamp>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}.\d+Z)\s\[mysql_monitor\((?P<instance_id>[^\]]+)\)\]\s(?P<metric_name>[^\s]+)\s(?P<metric_value>.+)"
 
-
 def _test(
     request,
     python_version,
@@ -205,11 +211,14 @@ def _test(
 
         agent_log_reader.go_to_end()
     except LogReaderError as e:
-        if not expected_exception or (
-            expected_exception not in str(e)
-            and not re.search(expected_exception, str(e))
-        ):
-            raise e
+        if not expected_exception:
+            msg = "AssertionError, not exception expected, got: " + str(e)
+            global_log.error(msg, exc_info=e)
+            raise AssertionError(msg)
+        if expected_exception not in str(e)and not re.search(expected_exception, str(e)):
+            msg = "Assertion error, expected: " + str(expected_exception) + "got exception: " + str(e)
+            global_log.error(msg, exc_info=e)
+            raise AssertionError(msg)
 
 
 @pytest.mark.usefixtures("agent_environment")
@@ -283,7 +292,7 @@ def test_mysql_python2_ssl_bad_hostname(request):
         use_socket=False,
         use_ssl=True,
         ca_file="/var/lib/mysql/ca.pem",
-        expected_exception=r"hostname '127.0.0.1' doesn't match 'MySQL_Server_5.7.\d+_Auto_Generated_Server_Certificate'",
+        expected_exception=r"hostname u'127.0.0.1' doesn't match u'MySQL_Server_5.7.\d+_Auto_Generated_Server_Certificate'",
     )
 
 
diff --git a/tests/smoke_tests/monitors_test/syslog_test.py b/tests/smoke_tests/monitors_test/syslog_test.py
@@ -91,8 +91,12 @@ def _test(python_version):
     finally:
         runner.stop(executable=python_version)
 
-
 @pytest.mark.usefixtures("agent_environment")
 @dockerized_case(CommonMonitorBuilder, __file__)
 def test_syslog_python3(request):
     _test(python_version="python3")
+
+@pytest.mark.usefixtures("agent_environment")
+@dockerized_case(CommonMonitorBuilder, __file__)
+def test_syslog_python2(request):
+    _test(python_version="python2")
diff --git a/tests/utils/agent_runner.py b/tests/utils/agent_runner.py
@@ -218,6 +218,8 @@ def start(self, executable="python"):
             if self._enable_coverage:
                 # NOTE: We need to pass in command string as a single argument to coverage run
                 args = [
+                    executable,
+                    "-m",
                     "coverage",
                     "run",
                     "--branch",
diff --git a/tests/utils/dockerized.py b/tests/utils/dockerized.py
@@ -25,13 +25,14 @@
 
 from tests.utils.compat import Path
 
-from scalyr_agent import compat
+from scalyr_agent import compat, scalyr_logging
 
 DEFAULT_FILE_PATHS_TO_COPY = [
     "/var/log/scalyr-agent-2/agent.log",
     "/root/scalyr-agent-dev/log/agent.log",
 ]
 
+global_log = scalyr_logging.getLogger(__name__)
 
 def dockerized_case(
     builder_cls,
@@ -133,7 +134,8 @@ def wrapper(request, *args, **kwargs):
             exit_code = container.wait()["StatusCode"]
 
             logs = six.ensure_text(container.logs(follow=True))
-            print(logs)
+            global_log.info("Container logs")
+            global_log.info(logs)
 
             # save logs if artifacts path is specified.
             artifacts_path = request.config.getoption("--artifacts-path", None)
@@ -157,7 +159,7 @@ def wrapper(request, *args, **kwargs):
 
             if remove_container:
                 container.remove(force=True)
-                print("Container '{0}' removed.".format(builder.image_tag))
+                global_log.info("Container '{0}' removed.".format(builder.image_tag))
 
             # raise failed assertion, due to non-zero result from container.
             if exit_code:
diff --git a/tests/utils/image_builder.py b/tests/utils/image_builder.py
@@ -36,6 +36,9 @@
 from tests.utils.common import create_tmp_directory
 from tests.utils.compat import Path
 
+from scalyr_agent import scalyr_logging
+
+global_log = scalyr_logging.getLogger(__name__)
 
 def _copy_agent_source(src_path, dest_path):
     gitignore_path = src_path / ".gitignore"
@@ -186,8 +189,7 @@ def build(self, image_cache_path=None, skip_requirements=False):
         dockerfile_path.write_text(self.get_dockerfile_content())
         self._copy_to_build_context(build_context_path)
 
-        subprocess.check_call(
-            [
+        cmd = [
                 "docker",
                 "build",
                 "-t",
@@ -197,6 +199,11 @@ def build(self, image_cache_path=None, skip_requirements=False):
                 "--rm",
                 six.text_type(build_context_path),
             ]
+
+        global_log.info(" ".join(cmd))
+
+        subprocess.check_output(
+            cmd
         )
 
         shutil.rmtree(six.text_type(build_context_path), ignore_errors=True)
