ci: wait for dependent services before running tests (#11780)

.gitlab/tests.yml

@@ -1,7 +1,7 @@
 stages:
   - precheck
-  - hatch
   - riot
+  - hatch
 
 variables:
   RIOT_RUN_CMD: riot -P -v run --exitfirst --pass-env -s
@@ -30,6 +30,9 @@ variables:
   parallel: 4
   # DEV: This is the max retries that GitLab currently allows for
   retry: 2
+  before_script:
+    - !reference [.testrunner, before_script]
+    - pip install riot==0.20.1
   script:
     - export PYTEST_ADDOPTS="${PYTEST_ADDOPTS} --ddtrace"
     - export _DD_CIVISIBILITY_USE_CI_CONTEXT_PROVIDER=true
@@ -51,7 +54,7 @@ variables:
   services:
     - !reference [.services, testagent]
   before_script:
-    - !reference [.testrunner, before_script]
+    - !reference [.test_base_hatch, before_script]
     # DEV: All job variables get shared with services, setting `DD_TRACE_AGENT_URL` on the testagent will tell it to forward all requests to the
     # agent at that host. Therefore setting this as a variable will cause recursive requests to the testagent
     - export DD_TRACE_AGENT_URL="http://testagent:9126"
@@ -88,12 +91,14 @@ build_base_venvs:
     - !reference [.services, ddagent]
   # DEV: This is the max retries that GitLab currently allows for
   retry: 2
-  script:
+  before_script:
+    - !reference [.testrunner, before_script]
     - pip install riot==0.20.1
     - unset DD_SERVICE
     - unset DD_ENV
     - unset DD_TAGS
     - unset DD_TRACE_REMOVE_INTEGRATION_SERVICE_NAMES_ENABLED
+  script:
     - |
       hashes=( $(riot list --hash-only "${SUITE_NAME}" | sort | ./.gitlab/ci-split-input.sh) )
       if [[ ${#hashes[@]} -eq 0 ]]; then
@@ -116,7 +121,7 @@ build_base_venvs:
     - !reference [.test_base_riot, services]
     - !reference [.services, testagent]
   before_script:
-    - !reference [.testrunner, before_script]
+    - !reference [.test_base_riot, before_script]
     # DEV: All job variables get shared with services, setting `DD_TRACE_AGENT_URL` on the testagent will tell it to forward all requests to the
     # agent at that host. Therefore setting this as a variable will cause recursive requests to the testagent
     - export DD_TRACE_AGENT_URL="http://testagent:9126"

.riot/requirements/151d7b0.txt

@@ -0,0 +1,41 @@
+#
+# This file is autogenerated by pip-compile with Python 3.9
+# by the following command:
+#
+#    pip-compile --allow-unsafe --no-annotate .riot/requirements/151d7b0.in
+#
+amqp==2.6.1
+attrs==24.3.0
+cassandra-driver==3.29.2
+certifi==2024.12.14
+charset-normalizer==3.4.0
+click==8.1.7
+coverage[toml]==7.6.9
+exceptiongroup==1.2.2
+future==1.0.0
+geomet==0.2.1.post1
+hypothesis==6.45.0
+idna==3.10
+importlib-metadata==8.5.0
+iniconfig==2.0.0
+kombu==4.2.2.post1
+mock==5.1.0
+mysql-connector-python==9.1.0
+opentracing==2.4.0
+packaging==24.2
+pluggy==1.5.0
+psycopg2-binary==2.9.10
+pytest==8.3.4
+pytest-cov==6.0.0
+pytest-mock==3.14.0
+pytest-randomly==3.16.0
+python-dateutil==2.9.0.post0
+pytz==2024.2
+requests==2.32.3
+six==1.17.0
+sortedcontainers==2.4.0
+tomli==2.2.1
+urllib3==2.2.3
+vertica-python==0.6.14
+vine==1.3.0
+zipp==3.21.0

hatch.toml

@@ -124,7 +124,7 @@ extra-dependencies = [
 ]
 
 [envs.slotscheck.scripts]
-_ = [
+test = [
     "python -m slotscheck -v ddtrace/",
 ]
 

riotfile.py

@@ -586,6 +586,7 @@ def select_pys(min_version: str = MIN_PYTHON_VERSION, max_version: str = MAX_PYT
                 "vertica-python": ">=0.6.0,<0.7.0",
                 "kombu": ">=4.2.0,<4.3.0",
                 "pytest-randomly": latest,
+                "requests": latest,
             },
         ),
         Venv(

scripts/gen_gitlab_config.py

@@ -15,7 +15,7 @@ class JobSpec:
     runner: str
     pattern: t.Optional[str] = None
     snapshot: bool = False
-    services: t.Optional[t.Set[str]] = None
+    services: t.Optional[t.List[str]] = None
     env: t.Optional[t.Dict[str, str]] = None
     parallelism: t.Optional[int] = None
     retry: t.Optional[int] = None
@@ -32,16 +32,25 @@ def __str__(self) -> str:
         lines.append(f"{self.name}:")
         lines.append(f"  extends: {base}")
 
-        if self.services:
+        services = set(self.services or [])
+        if services:
             lines.append("  services:")
 
-            _services = [f"!reference [.services, {_}]" for _ in self.services]
+            _services = [f"!reference [.services, {_}]" for _ in services]
             if self.snapshot:
                 _services.insert(0, f"!reference [{base}, services]")
 
             for service in _services:
                 lines.append(f"    - {service}")
 
+        wait_for: t.Set[str] = services.copy()
+        if self.snapshot:
+            wait_for.add("testagent")
+        if wait_for:
+            lines.append("  before_script:")
+            lines.append(f"    - !reference [{base}, before_script]")
+            lines.append(f"    - riot -v run -s --pass-env wait -- {' '.join(wait_for)}")
+
         env = self.env
         if not env or "SUITE_NAME" not in env:
             env = env or {}
@@ -89,7 +98,6 @@ def gen_required_suites() -> None:
     TESTS_GEN.write_text(
         (GITLAB / "tests.yml").read_text().replace(r"{{services.yml}}", (GITLAB / "services.yml").read_text())
     )
-
     # Generate the list of suites to run
     with TESTS_GEN.open("a") as f:
         for suite in required_suites:
@@ -159,11 +167,6 @@ def check(name: str, command: str, paths: t.Set[str]) -> None:
         command="hatch run meta-testing:meta-testing",
         paths={"**conftest.py"},
     )
-    check(
-        name="slotscheck",
-        command="hatch run slotscheck:_",
-        paths={"**.py"},
-    )
 
 
 # -----------------------------------------------------------------------------

tests/suitespec.yml

@@ -195,6 +195,12 @@ suites:
       - tests/cache/*
     runner: riot
     snapshot: true
+  slotscheck:
+    parallelism: 1
+    paths:
+      - 'ddtrace/**/*.py'
+    runner: hatch
+    snapshot: false
   profile:
     env:
       DD_TRACE_AGENT_URL: ''

tests/wait-for-services.py

@@ -1,101 +1,153 @@
+import logging
+import os
 import sys
 import time
+import typing as t
 
 from cassandra.cluster import Cluster
 from cassandra.cluster import NoHostAvailable
 from contrib.config import CASSANDRA_CONFIG
+from contrib.config import ELASTICSEARCH_CONFIG
+from contrib.config import HTTPBIN_CONFIG
 from contrib.config import MYSQL_CONFIG
+from contrib.config import OPENSEARCH_CONFIG
 from contrib.config import POSTGRES_CONFIG
 from contrib.config import RABBITMQ_CONFIG
 from contrib.config import VERTICA_CONFIG
 import kombu
 import mysql.connector
 from psycopg2 import OperationalError
 from psycopg2 import connect
+import requests
 import vertica_python
 
 
-def try_until_timeout(exception):
+logging.basicConfig(level=logging.INFO)
+log = logging.getLogger(__name__)
+
+
+def try_until_timeout(exception, tries: int = 100, timeout: float = 0.2, args: t.Optional[t.Dict[str, t.Any]] = None):
     """Utility decorator that tries to call a check until there is a
     timeout.  The default timeout is about 20 seconds.
 
     """
+    if not args:
+        args = {}
 
     def wrap(fn):
-        def wrapper(*args, **kwargs):
+        def wrapper(**kwargs):
             err = None
 
-            for _ in range(100):
+            _kwargs = args.copy()
+            _kwargs.update(kwargs)
+
+            for i in range(tries):
                 try:
-                    fn()
+                    log.info("Attempt %d: %s(%r)", i, fn.__name__, _kwargs)
+                    fn(**_kwargs)
                 except exception as e:
                     err = e
-                    time.sleep(0.2)
+                    time.sleep(timeout)
                 else:
                     break
             else:
                 if err:
                     raise err
+            log.info("Succeeded: %s", fn.__name__)
 
         return wrapper
 
     return wrap
 
 
-@try_until_timeout(OperationalError)
-def check_postgres():
-    conn = connect(**POSTGRES_CONFIG)
+@try_until_timeout(OperationalError, args={"pg_config": POSTGRES_CONFIG})
+def check_postgres(pg_config):
+    conn = connect(**pg_config)
     try:
         conn.cursor().execute("SELECT 1;")
     finally:
         conn.close()
 
 
-@try_until_timeout(NoHostAvailable)
-def check_cassandra():
-    with Cluster(**CASSANDRA_CONFIG).connect() as conn:
+@try_until_timeout(NoHostAvailable, args={"cassandra_config": CASSANDRA_CONFIG})
+def check_cassandra(cassandra_config):
+    with Cluster(**cassandra_config).connect() as conn:
         conn.execute("SELECT now() FROM system.local")
 
 
-@try_until_timeout(Exception)
-def check_mysql():
-    conn = mysql.connector.connect(**MYSQL_CONFIG)
+@try_until_timeout(Exception, args={"mysql_config": MYSQL_CONFIG})
+def check_mysql(mysql_config):
+    conn = mysql.connector.connect(**mysql_config)
     try:
         conn.cursor().execute("SELECT 1;")
     finally:
         conn.close()
 
 
-@try_until_timeout(Exception)
-def check_vertica():
-    conn = vertica_python.connect(**VERTICA_CONFIG)
+@try_until_timeout(Exception, args={"vertica_config": VERTICA_CONFIG})
+def check_vertica(vertica_config):
+    conn = vertica_python.connect(**vertica_config)
     try:
         conn.cursor().execute("SELECT 1;")
     finally:
         conn.close()
 
 
-@try_until_timeout(Exception)
-def check_rabbitmq():
-    url = "amqp://{user}:{password}@{host}:{port}//".format(**RABBITMQ_CONFIG)
+@try_until_timeout(Exception, args={"url": "amqp://{user}:{password}@{host}:{port}//".format(**RABBITMQ_CONFIG)})
+def check_rabbitmq(url):
     conn = kombu.Connection(url)
     try:
         conn.connect()
     finally:
         conn.release()
 
 
+@try_until_timeout(Exception, args={"url": os.environ.get("DD_TRACE_AGENT_URL", "http://localhost:8126")})
+def check_agent(url):
+    if not url.endswith("/"):
+        url += "/"
+
+    res = requests.get(url)
+    if res.status_code not in (404, 200):
+        raise Exception("Agent not ready")
+
+
+@try_until_timeout(Exception, args={"url": "http://{host}:{port}/".format(**ELASTICSEARCH_CONFIG)})
+def check_elasticsearch(url):
+    requests.get(url).raise_for_status()
+
+
+@try_until_timeout(
+    Exception, tries=120, timeout=1, args={"url": "http://{host}:{port}/".format(**OPENSEARCH_CONFIG)}
+)  # 2 minutes, OpenSearch is slow to start
+def check_opensearch(url):
+    requests.get(url).raise_for_status()
+
+
+@try_until_timeout(Exception, args={"url": "http://{host}:{port}/".format(**HTTPBIN_CONFIG)})
+def check_httpbin(url):
+    requests.get(url).raise_for_status()
+
+
 if __name__ == "__main__":
     check_functions = {
         "cassandra": check_cassandra,
-        "postgres": check_postgres,
+        "ddagent": check_agent,
+        "elasticsearch": check_elasticsearch,
+        "httpbin_local": check_httpbin,
         "mysql": check_mysql,
-        "vertica": check_vertica,
+        "opensearch": check_opensearch,
+        "postgres": check_postgres,
         "rabbitmq": check_rabbitmq,
+        "testagent": check_agent,
+        "vertica": check_vertica,
     }
     if len(sys.argv) >= 2:
         for service in sys.argv[1:]:
-            check_functions[service]()
+            if service not in check_functions:
+                log.warning("Unknown service: %s", service)
+            else:
+                check_functions[service]()
     else:
         print("usage: python {} SERVICE_NAME".format(sys.argv[0]))
         sys.exit(1)