Implementation of functionality

.circleci/config.yml

@@ -0,0 +1,28 @@
+version: 2.1
+
+orbs:
+  # The python orb contains a set of prepackaged CircleCI configuration you can use repeatedly in your configuration files
+  # Orb commands and jobs help you with common scripting around a language/tool
+  # so you dont have to copy and paste it everywhere.
+  # See the orb documentation here: https://circleci.com/developer/orbs/orb/circleci/python
+  python: circleci/python@1.2
+
+workflows:
+  sample:
+    jobs:
+      - build-and-test
+
+
+jobs:
+  build-and-test:
+    docker:
+      - image: cimg/python:3.8
+    steps:
+      - checkout
+      - run:
+          name: installpackages
+          command: |
+            make testsuite.install
+      - run:
+          name: Run tests
+          command: pytest

Makefile

@@ -0,0 +1,6 @@
+testsuite.install:
+	pip install pytest pytest-mock responses
+	python setup.py develop
+
+testsuite.run:
+	python -m pytest .

README.md

@@ -0,0 +1,3 @@
+# opentelem-python
+
+Open Telemetry Codecov Python Prototype

codecovopentelem/__init__.py

@@ -0,0 +1,157 @@
+import json
+import logging
+import random
+import re
+import urllib.parse
+from base64 import b64encode
+from decimal import Decimal
+from io import StringIO
+from typing import Optional, Tuple
+
+import coverage
+import requests
+from coverage.xmlreport import XmlReporter
+from opentelemetry.sdk.trace import SpanProcessor
+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+
+log = logging.getLogger("codecovopentelem")
+
+
+class CodecovCoverageStorageManager(object):
+    def __init__(self, writeable_folder: str):
+        if writeable_folder is None:
+            writeable_folder = "/home/codecov"
+        self._writeable_folder = writeable_folder
+        self.inner = {}
+
+    def start_cov_for_span(self, span_id):
+        cov = coverage.Coverage(data_file=f"{self._writeable_folder}/.{span_id}file")
+        self.inner[span_id] = cov
+        cov.start()
+
+    def stop_cov_for_span(self, span_id):
+        cov = self.inner.get(span_id)
+        if cov is not None:
+            cov.stop()
+
+    def pop_cov_for_span(self, span_id):
+        return self.inner.pop(span_id, None)
+
+
+class CodecovCoverageGenerator(SpanProcessor):
+    def __init__(
+        self,
+        cov_storage: CodecovCoverageStorageManager,
+        sample_rate: Decimal,
+        name_regex: re.Pattern = None,
+    ):
+        self._cov_storage = cov_storage
+        self._sample_rate = sample_rate
+        self._name_regex = name_regex
+
+    def _should_profile_span(self, span, parent_context):
+        return random.random() < self._sample_rate and (
+            self._name_regex is None or self._name_regex.match(span.name)
+        )
+
+    def on_start(self, span, parent_context=None):
+        if self._should_profile_span(span, parent_context):
+            span_id = span.context.span_id
+            self._cov_storage.start_cov_for_span(span_id)
+
+    def on_end(self, span):
+        span_id = span.context.span_id
+        self._cov_storage.stop_cov_for_span(span_id)
+
+
+class CoverageExporter(SpanExporter):
+    def __init__(
+        self,
+        cov_storage: CodecovCoverageStorageManager,
+        repository_token: str,
+        profiling_identifier: str,
+        codecov_endpoint: str,
+    ):
+        self._cov_storage = cov_storage
+        self._repository_token = repository_token
+        self._profiling_identifier = profiling_identifier
+        self._codecov_endpoint = codecov_endpoint
+
+    def _load_codecov_dict(self, span, cov):
+        k = StringIO()
+        coverage_dict = {}
+        try:
+            reporter = XmlReporter(cov)
+            reporter.report(None, outfile=k)
+            k.seek(0)
+            d = k.read().encode()
+            coverage_dict["type"] = "bytes"
+            coverage_dict["coverage"] = b64encode(d).decode()
+        except coverage.CoverageException:
+            pass
+        return coverage_dict
+
+    def export(self, spans):
+        data = []
+        untracked_spans = []
+        for span in spans:
+            span_id = span.context.span_id
+            cov = self._cov_storage.pop_cov_for_span(span_id)
+            s = json.loads(span.to_json())
+            if cov is not None:
+                s["codecov"] = self._load_codecov_dict(span, cov)
+                data.append(s)
+            else:
+                untracked_spans.append(s)
+        url = urllib.parse.urljoin(self._codecov_endpoint, "/profiling/uploads")
+        res = requests.post(
+            url,
+            headers={"Authorization": f"repotoken {self._repository_token}"},
+            json={"profiling": self._profiling_identifier},
+        )
+        try:
+            res.raise_for_status()
+        except requests.HTTPError:
+            log.warning("Unable to send profiling data to codecov")
+            return SpanExportResult.FAILURE
+        location = res.json()["raw_upload_location"]
+        requests.put(
+            location,
+            headers={"Content-Type": "application/txt"},
+            data=json.dumps({"spans": data, "untracked": untracked_spans}).encode(),
+        )
+        return SpanExportResult.SUCCESS
+
+
+def get_codecov_opentelemetry_instances(
+    repository_token: str,
+    profiling_identifier: str,
+    sample_rate: float,
+    name_regex: Optional[re.Pattern],
+    codecov_endpoint: str = None,
+    writeable_folder: str = None,
+) -> Tuple[CodecovCoverageGenerator, CoverageExporter]:
+    """
+    Entrypoint for getting a span processor/span exporter
+        pair for getting profiling data into codecov
+
+    Args:
+        repository_token (str): The profiling-capable authentication token
+        profiling_identifier (str): The identifier for what profiling one is doing
+        sample_rate (float): The sampling rate for codecov
+        name_regex (Optional[re.Pattern]): A regex to filter which spans should be
+            sampled
+        codecov_endpoint (str, optional): For configuring the endpoint in case
+            the user is in enterprise (not supported yet). Default is "https://api.codecov.io/"
+        writeable_folder (str, optional): A folder that is guaranteed to be write-able
+            in the system. It's only used for temporary files, and nothing is expected
+            to live very long in there.
+    """
+    if codecov_endpoint is None:
+        codecov_endpoint = "https://api.codecov.io"
+    manager = CodecovCoverageStorageManager(writeable_folder)
+    generator = CodecovCoverageGenerator(manager, sample_rate, name_regex)
+    exporter = CoverageExporter(
+        manager, repository_token, profiling_identifier, codecov_endpoint
+    )
+    return (generator, exporter)

tests/test_exporter.py

@@ -0,0 +1,56 @@
+import json
+
+import pytest
+import responses
+from coverage import Coverage
+from coverage.xmlreport import XmlReporter
+
+from codecovopentelem import CoverageExporter
+
+
+@pytest.fixture
+def mocked_responses():
+    with responses.RequestsMock() as rsps:
+        yield rsps
+
+
+def test_export_span(mocker, mocked_responses):
+    mocker.patch.object(
+        XmlReporter,
+        "report",
+        side_effect=lambda a, outfile: outfile.write("somedatahere"),
+    )
+    cov = Coverage()
+    cov_storage, repository_token, profiling_identifier, codecov_endpoint = (
+        mocker.MagicMock(pop_cov_for_span=mocker.MagicMock(return_value=cov)),
+        "repository_token",
+        "identifier",
+        "http://codecov.test/endpoint",
+    )
+    mocked_responses.add(
+        responses.POST,
+        "http://codecov.test/profiling/uploads",
+        json={"raw_upload_location": "http://storage.test/endpoint"},
+        status=200,
+        content_type="application/json",
+    )
+    mocked_responses.add(
+        responses.PUT,
+        "http://storage.test/endpoint",
+        status=200,
+        content_type="application/json",
+    )
+    exporter = CoverageExporter(
+        cov_storage, repository_token, profiling_identifier, codecov_endpoint
+    )
+    span = mocker.MagicMock(to_json=mocker.MagicMock(return_value="{}"))
+    assert exporter.export([span])
+    assert len(mocked_responses.calls) == 2
+    assert (
+        mocked_responses.calls[0].request.url == "http://codecov.test/profiling/uploads"
+    )
+    print(mocked_responses.calls[1].request.body)
+    assert json.loads(mocked_responses.calls[1].request.body) == {
+        "spans": [{"codecov": {"type": "bytes", "coverage": "c29tZWRhdGFoZXJl"}}],
+        "untracked": [],
+    }