tests: integration: clean up valgrind services

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

tests/integration/conftest.py

@@ -19,6 +19,8 @@
 import logging
 import pytest
 
+from utils.test_service import stop_active_services
+
 # Configure logging
 def configure_logging():
     logger = logging.getLogger(__name__)
@@ -52,6 +54,10 @@ def pytest_sessionstart(session):
     #flb = FluentBitManager(GLOBAL_CONFIG['fluent_bit']['config_path'])
     #flb = FluentBitManager()
 
+@pytest.hookimpl(trylast=True)
+def pytest_runtest_teardown(item, nextitem):
+    stop_active_services()
+
 @pytest.hookimpl(trylast=True)
 def pytest_sessionfinish(session, exitstatus):
     pass #logger.info("Finishing pytest session")

tests/integration/scenarios/out_http/tests/test_out_http_001.py

@@ -16,6 +16,13 @@
 
 logger = logging.getLogger(__name__)
 
+def _valgrind_timeout(timeout):
+    if os.environ.get("VALGRIND"):
+        return max(timeout * 3, 30)
+
+    return timeout
+
+
 def _wait_for_http_server(port, timeout=5):
     deadline = time.time() + timeout
 
@@ -128,7 +135,7 @@ def stop(self):
     def wait_for_requests(self, minimum_count, timeout=10):
         return self.service.wait_for_condition(
             lambda: data_storage["requests"] if len(data_storage["requests"]) >= minimum_count else None,
-            timeout=timeout,
+            timeout=_valgrind_timeout(timeout),
             interval=0.5,
             description=f"{minimum_count} outbound HTTP requests",
         )
@@ -148,7 +155,7 @@ def _read_log():
 
         return self.service.wait_for_condition(
             _read_log,
-            timeout=timeout,
+            timeout=_valgrind_timeout(timeout),
             interval=0.25,
             description=f"log message '{pattern}'",
         )

tests/integration/src/utils/test_service.py

@@ -7,6 +7,22 @@
 from utils.network import find_available_port
 
 
+_active_services = set()
+
+
+def stop_active_services():
+    errors = []
+
+    for service in list(_active_services):
+        try:
+            service.stop()
+        except Exception as exc:
+            errors.append(exc)
+
+    if errors:
+        raise errors[0]
+
+
 class FluentBitTestService:
     def __init__(
         self,
@@ -69,20 +85,29 @@ def start(self):
         for key, value in self.extra_env.items():
             self._set_env(key, str(value))
 
-        if self.pre_start:
-            self.pre_start(self)
+        _active_services.add(self)
 
-        self.flb.start()
+        try:
+            if self.pre_start:
+                self.pre_start(self)
+
+            self.flb.start()
+        except Exception:
+            self.stop()
+            raise
 
     def stop(self):
+        had_allocated_ports = bool(self._allocated_ports)
+
         try:
             if self.flb:
                 self.flb.stop()
         finally:
-            if self.post_stop:
+            if self.post_stop and had_allocated_ports:
                 self.post_stop(self)
             self._restore_env()
             self._allocated_ports.clear()
+            _active_services.discard(self)
 
     def wait_for_http_endpoint(self, url, *, timeout=10, interval=0.5):
         deadline = time.time() + timeout