Set up contract tests

.flake8

@@ -21,3 +21,6 @@ exclude =
   venv*/
   target
   __pycache__
+  mock_collector_service_pb2.py
+  mock_collector_service_pb2.pyi
+  mock_collector_service_pb2_grpc.py

.isort.cfg

@@ -13,7 +13,7 @@ profile=black
 ; )
 ; docs: https://github.com/timothycrosley/isort#multi-line-output-modes
 multi_line_output=3
-skip=target
+skip=target,mock_collector_service_pb2_grpc.py,mock_collector_service_pb2.py,mock_collector_service_pb2.pyi
 skip_glob=**/gen/*,.venv*/*,venv*/*,.tox/*
 known_first_party=opentelemetry,amazon
 known_third_party=psutil,pytest

.pylintrc

@@ -7,7 +7,7 @@ extension-pkg-whitelist=cassandra
 
 # Add list of files or directories to be excluded. They should be base names, not
 # paths.
-ignore=CVS,gen,Dockerfile,docker-compose.yml,README.md,requirements.txt
+ignore=CVS,gen,Dockerfile,docker-compose.yml,README.md,requirements.txt,mock_collector_service_pb2.py,mock_collector_service_pb2.pyi,mock_collector_service_pb2_grpc.py
 
 # Add files or directories matching the regex patterns to be excluded. The
 # regex matches against base names, not paths.

contract-tests/README.md

@@ -0,0 +1,42 @@
+# Introduction
+
+This directory contain contract tests that exist to prevent regressions. They cover:
+* [OpenTelemetry semantic conventions](https://github.com/open-telemetry/semantic-conventions/).
+* Application Signals-specific attributes.
+
+# How it works?
+
+The tests present here rely on the auto-instrumentation of a sample application which will send telemetry signals to a mock collector. The tests will use the data collected by the mock collector to perform assertions and validate that the contracts are being respected.
+
+# Types of tested frameworks
+
+The frameworks and libraries that are tested in the contract tests should fall in the following categories (more can be added on demand):
+* http-servers - applications meant to test http servers (e.g. Django).
+* http-clients - applications meant to test http clients (e.g. requests).
+* aws-sdk - Applications meant to test the AWS SDK (e.g. botocore).
+* database-clients - Applications meant to test database clients (e.g. psychopg2).
+
+When testing a framework, we will create a sample application. The sample applications are stored following this convention: `contract-tests/images/applications/<framework-name>`.
+
+# Adding tests for a new library or framework
+
+The steps to add a new test for a library or framework are:
+* Create a sample application.
+    * The sample application should be created in `contract-tests/images/applications/<framework-name>`.
+    * Implement a `pyproject.toml` (to ensure code style checks run), `Dockerfile`, and `requirements.txt` file. See the `requests` application for an example of this.
+* Add a test class for the sample application.
+    * The test class should be created in `contract-tests/tests/amazon/<framework-name>`.
+    * The test class should extend `contract_test_base.py`
+
+# How to run the tests locally?
+
+Pre-requirements:
+* Have `docker` installed and running - verify by running the `docker` command.
+* Ensure the `aws_opentelemetry_distro` wheel file exists in to `aws-otel-python-instrumentation/dist` folder
+
+From `aws-otel-python-instrumentation` dir, execute:
+
+```
+./contract-tests/set-up-contract-tests.sh
+pytest contract-tests/tests
+```

contract-tests/images/applications/requests/Dockerfile

@@ -0,0 +1,14 @@
+# Meant to be run from aws-otel-python-instrumentation/contract-tests.
+# Assumes existence of dist/aws_opentelemetry_distro-<pkg_version>-py3-none-any.whl.
+# Assumes filename of aws_opentelemetry_distro-<pkg_version>-py3-none-any.whl is passed in as "DISTRO" arg.
+FROM public.ecr.aws/docker/library/python:3.11-slim
+WORKDIR /requests
+COPY ./dist/$DISTRO /requests
+COPY ./contract-tests/images/applications/requests /requests
+
+ARG DISTRO
+RUN pip install -r requirements.txt && pip install ${DISTRO} --force-reinstall
+RUN opentelemetry-bootstrap -a install
+
+# Without `-u`, logs will be buffered and `wait_for_logs` will never return.
+CMD ["opentelemetry-instrument", "python", "-u", "./requests_server.py"]

contract-tests/images/applications/requests/pyproject.toml

@@ -0,0 +1,6 @@
+[project]
+name = "requests-server"
+description = "Simple server that relies on requests library"
+version = "1.0.0"
+license = "Apache-2.0"
+requires-python = ">=3.8"

contract-tests/images/applications/requests/requests_server.py

@@ -0,0 +1,62 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import atexit
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from threading import Thread
+
+from requests import Response, request
+from typing_extensions import override
+
+_PORT: int = 8080
+_NETWORK_ALIAS: str = "backend"
+_SUCCESS: str = "success"
+_ERROR: str = "error"
+_FAULT: str = "fault"
+
+
+class RequestHandler(BaseHTTPRequestHandler):
+    @override
+    # pylint: disable=invalid-name
+    def do_GET(self):
+        self.handle_request("GET")
+
+    @override
+    # pylint: disable=invalid-name
+    def do_POST(self):
+        self.handle_request("POST")
+
+    def handle_request(self, method: str):
+        status_code: int
+        if self.in_path(_NETWORK_ALIAS):
+            if self.in_path(_SUCCESS):
+                status_code = 200
+            elif self.in_path(_ERROR):
+                status_code = 400
+            elif self.in_path(_FAULT):
+                status_code = 500
+            else:
+                status_code = 404
+        else:
+            url: str = f"http://{_NETWORK_ALIAS}:{_PORT}/{_NETWORK_ALIAS}{self.path}"
+            response: Response = request(method, url, timeout=20)
+            status_code = response.status_code
+        self.send_response_only(status_code)
+        self.end_headers()
+
+    def in_path(self, sub_path: str):
+        return sub_path in self.path
+
+
+def main() -> None:
+    server_address: tuple[str, int] = ("0.0.0.0", _PORT)
+    request_handler_class: type = RequestHandler
+    requests_server: ThreadingHTTPServer = ThreadingHTTPServer(server_address, request_handler_class)
+    atexit.register(requests_server.shutdown)
+    server_thread: Thread = Thread(target=requests_server.serve_forever)
+    server_thread.start()
+    print("Ready")
+    server_thread.join()
+
+
+if __name__ == "__main__":
+    main()

contract-tests/images/applications/requests/requirements.txt

@@ -0,0 +1,4 @@
+opentelemetry-distro==0.43b0
+opentelemetry-exporter-otlp-proto-grpc==1.22.0
+typing-extensions==4.9.0
+requests==2.31.0

contract-tests/images/mock-collector/Dockerfile

@@ -0,0 +1,8 @@
+FROM public.ecr.aws/docker/library/python:3.11-slim
+WORKDIR /mock-collector
+COPY . /mock-collector
+
+RUN pip install -r requirements.txt
+
+# Without `-u`, logs will be buffered and `wait_for_logs` will never return.
+CMD ["python", "-u", "./mock_collector_server.py"]

contract-tests/images/mock-collector/README.md

@@ -0,0 +1,9 @@
+### Overview
+
+MockCollector mimics the behaviour of the actual OTEL collector, but stores export requests to be retrieved by contract tests. 
+
+### Protos
+To build protos:
+1. Run `pip install grpcio grpcio-tools`
+2. Change directory to `aws-otel-python-instrumentation/contract-tests/images/mock-collector/` 
+3. Run: `python -m grpc_tools.protoc -I./protos --python_out=. --pyi_out=. --grpc_python_out=. ./protos/mock_collector_service.proto`

contract-tests/images/mock-collector/mock_collector_client.py

@@ -0,0 +1,143 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from datetime import datetime, timedelta
+from logging import Logger, getLogger
+from time import sleep
+from typing import Callable, List, Set, TypeVar
+
+from google.protobuf.internal.containers import RepeatedScalarFieldContainer
+from grpc import Channel, insecure_channel
+from mock_collector_service_pb2 import (
+    ClearRequest,
+    GetMetricsRequest,
+    GetMetricsResponse,
+    GetTracesRequest,
+    GetTracesResponse,
+)
+from mock_collector_service_pb2_grpc import MockCollectorServiceStub
+
+from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ExportMetricsServiceRequest
+from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
+from opentelemetry.proto.metrics.v1.metrics_pb2 import Metric, ResourceMetrics, ScopeMetrics
+from opentelemetry.proto.trace.v1.trace_pb2 import ResourceSpans, ScopeSpans, Span
+
+_logger: Logger = getLogger(__name__)
+_TIMEOUT_DELAY: timedelta = timedelta(seconds=20)
+_WAIT_INTERVAL_SEC: float = 0.1
+T: TypeVar = TypeVar("T")
+
+
+class ResourceScopeSpan:
+    """Data class used to correlate resources, scope and telemetry signals.
+
+    Correlate resource, scope and span
+    """
+
+    def __init__(self, resource_spans: ResourceSpans, scope_spans: ScopeSpans, span: Span):
+        self.resource_spans: ResourceSpans = resource_spans
+        self.scope_spans: ScopeSpans = scope_spans
+        self.span: Span = span
+
+
+class ResourceScopeMetric:
+    """Data class used to correlate resources, scope and telemetry signals.
+
+    Correlate resource, scope and metric
+    """
+
+    def __init__(self, resource_metrics: ResourceMetrics, scope_metrics: ScopeMetrics, metric: Metric):
+        self.resource_metrics: ResourceMetrics = resource_metrics
+        self.scope_metrics: ScopeMetrics = scope_metrics
+        self.metric: Metric = metric
+
+
+class MockCollectorClient:
+    """The mock collector client is used to interact with the Mock collector image, used in the tests."""
+
+    def __init__(self, mock_collector_address: str, mock_collector_port: str):
+        channel: Channel = insecure_channel(f"{mock_collector_address}:{mock_collector_port}")
+        self.client: MockCollectorServiceStub = MockCollectorServiceStub(channel)
+
+    def clear_signals(self) -> None:
+        """Clear all the signals in the backend collector"""
+        self.client.clear(ClearRequest())
+
+    def get_traces(self) -> List[ResourceScopeSpan]:
+        """Get all traces that are currently stored in the collector
+
+        Returns:
+            List of `ResourceScopeSpan` which is essentially a flat list containing all the spans and their related
+            scope and resources.
+        """
+
+        def get_export() -> List[ExportTraceServiceRequest]:
+            response: GetTracesResponse = self.client.get_traces(GetTracesRequest())
+            serialized_traces: RepeatedScalarFieldContainer[bytes] = response.traces
+            return list(map(ExportTraceServiceRequest.FromString, serialized_traces))
+
+        def wait_condition(exported: List[ExportTraceServiceRequest], current: List[ExportTraceServiceRequest]) -> bool:
+            return 0 < len(exported) == len(current)
+
+        exported_traces: List[ExportTraceServiceRequest] = _wait_for_content(get_export, wait_condition)
+        spans: List[ResourceScopeSpan] = []
+        for exported_trace in exported_traces:
+            for resource_span in exported_trace.resource_spans:
+                for scope_span in resource_span.scope_spans:
+                    for span in scope_span.spans:
+                        spans.append(ResourceScopeSpan(resource_span, scope_span, span))
+        return spans
+
+    def get_metrics(self, present_metrics: Set[str]) -> List[ResourceScopeMetric]:
+        """Get all metrics that are currently stored in the mock collector.
+
+        Returns:
+             List of `ResourceScopeMetric` which is a flat list containing all metrics and their related scope and
+             resources.
+        """
+
+        present_metrics_lower: Set[str] = {s.lower() for s in present_metrics}
+
+        def get_export() -> List[ExportMetricsServiceRequest]:
+            response: GetMetricsResponse = self.client.get_metrics(GetMetricsRequest())
+            serialized_metrics: RepeatedScalarFieldContainer[bytes] = response.metrics
+            return list(map(ExportMetricsServiceRequest.FromString, serialized_metrics))
+
+        def wait_condition(
+            exported: List[ExportMetricsServiceRequest], current: List[ExportMetricsServiceRequest]
+        ) -> bool:
+            received_metrics: Set[str] = set()
+            for exported_metric in current:
+                for resource_metric in exported_metric.resource_metrics:
+                    for scope_metric in resource_metric.scope_metrics:
+                        for metric in scope_metric.metrics:
+                            received_metrics.add(metric.name.lower())
+            return 0 < len(exported) == len(current) and present_metrics_lower.issubset(received_metrics)
+
+        exported_metrics: List[ExportMetricsServiceRequest] = _wait_for_content(get_export, wait_condition)
+        metrics: List[ResourceScopeMetric] = []
+        for exported_metric in exported_metrics:
+            for resource_metric in exported_metric.resource_metrics:
+                for scope_metric in resource_metric.scope_metrics:
+                    for metric in scope_metric.metrics:
+                        metrics.append(ResourceScopeMetric(resource_metric, scope_metric, metric))
+        return metrics
+
+
+def _wait_for_content(get_export: Callable[[], List[T]], wait_condition: Callable[[List[T], List[T]], bool]) -> List[T]:
+    # Verify that there is no more data to be received
+    deadline: datetime = datetime.now() + _TIMEOUT_DELAY
+    exported: List[T] = []
+
+    while deadline > datetime.now():
+        try:
+            current_exported: List[T] = get_export()
+            if wait_condition(exported, current_exported):
+                return current_exported
+            exported = current_exported
+
+            sleep(_WAIT_INTERVAL_SEC)
+        # pylint: disable=broad-exception-caught
+        except Exception:
+            _logger.exception("Error while reading content")
+
+    raise RuntimeError("Timeout waiting for content")

contract-tests/images/mock-collector/mock_collector_metrics_service.py

@@ -0,0 +1,31 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from queue import Queue
+from typing import List
+
+from grpc import ServicerContext
+from typing_extensions import override
+
+from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import (
+    ExportMetricsServiceRequest,
+    ExportMetricsServiceResponse,
+)
+from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2_grpc import MetricsServiceServicer
+
+
+class MockCollectorMetricsService(MetricsServiceServicer):
+    _export_requests: Queue = Queue(maxsize=-1)
+
+    def get_requests(self) -> List[ExportMetricsServiceRequest]:
+        with self._export_requests.mutex:
+            return list(self._export_requests.queue)
+
+    def clear_requests(self) -> None:
+        with self._export_requests.mutex:
+            self._export_requests.queue.clear()
+
+    @override
+    # pylint: disable=invalid-name
+    def Export(self, request: ExportMetricsServiceRequest, context: ServicerContext) -> ExportMetricsServiceResponse:
+        self._export_requests.put(request)
+        return ExportMetricsServiceResponse()

contract-tests/images/mock-collector/mock_collector_server.py

@@ -0,0 +1,35 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import atexit
+from concurrent.futures import ThreadPoolExecutor
+
+from grpc import server
+from mock_collector_metrics_service import MockCollectorMetricsService
+from mock_collector_service import MockCollectorService
+from mock_collector_service_pb2_grpc import add_MockCollectorServiceServicer_to_server
+from mock_collector_trace_service import MockCollectorTraceService
+
+from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2_grpc import add_MetricsServiceServicer_to_server
+from opentelemetry.proto.collector.trace.v1.trace_service_pb2_grpc import add_TraceServiceServicer_to_server
+
+
+def main() -> None:
+    mock_collector_server: server = server(thread_pool=ThreadPoolExecutor(max_workers=10))
+    mock_collector_server.add_insecure_port("0.0.0.0:4315")
+
+    trace_collector: MockCollectorTraceService = MockCollectorTraceService()
+    metrics_collector: MockCollectorMetricsService = MockCollectorMetricsService()
+    mock_collector: MockCollectorService = MockCollectorService(trace_collector, metrics_collector)
+
+    add_TraceServiceServicer_to_server(trace_collector, mock_collector_server)
+    add_MetricsServiceServicer_to_server(metrics_collector, mock_collector_server)
+    add_MockCollectorServiceServicer_to_server(mock_collector, mock_collector_server)
+
+    mock_collector_server.start()
+    atexit.register(mock_collector_server.stop, None)
+    print("Ready")
+    mock_collector_server.wait_for_termination(None)
+
+
+if __name__ == "__main__":
+    main()