added parsing of cgroups file

elasticapm/base.py

@@ -26,7 +26,7 @@
 from elasticapm.conf.constants import ERROR
 from elasticapm.metrics.base_metrics import MetricsRegistry
 from elasticapm.traces import Tracer, get_transaction
-from elasticapm.utils import compat, is_master_process, stacks, varmap
+from elasticapm.utils import cgroup, compat, is_master_process, stacks, varmap
 from elasticapm.utils.encoding import keyword_field, shorten, transform
 from elasticapm.utils.module_import import import_string
 
@@ -97,9 +97,11 @@ def __init__(self, config=None, **inline):
             "max_flush_time": self.config.api_request_time / 1000.0,
             "max_buffer_size": self.config.api_request_size,
         }
-        self._transport = import_string(self.config.transport_class)(
-            compat.urlparse.urljoin(self.config.server_url, constants.EVENTS_API_PATH), **transport_kwargs
+        self._api_endpoint_url = compat.urlparse.urljoin(
+            self.config.server_url if self.config.server_url.endswith("/") else self.config.server_url + "/",
+            constants.EVENTS_API_PATH,
         )
+        self._transport = import_string(self.config.transport_class)(self._api_endpoint_url, **transport_kwargs)
 
         for exc_to_filter in self.config.filter_exception_types or []:
             exc_to_filter_type = exc_to_filter.split(".")[-1]
@@ -249,11 +251,33 @@ def get_process_info(self):
         }
 
     def get_system_info(self):
-        return {
+        system_data = {
             "hostname": keyword_field(socket.gethostname()),
             "architecture": platform.machine(),
             "platform": platform.system().lower(),
         }
+        system_data.update(cgroup.get_cgroup_container_metadata())
+        pod_name = os.environ.get("KUBERNETES_POD_NAME") or system_data["hostname"]
+        changed = False
+        if "kubernetes" in system_data:
+            k8s = system_data["kubernetes"]
+            k8s["pod"]["name"] = pod_name
+        else:
+            k8s = {"pod": {"name": pod_name}}
+        # get kubernetes metadata from environment
+        if "KUBERNETES_NODE_NAME" in os.environ:
+            k8s["node"] = {"name": os.environ["KUBERNETES_NODE_NAME"]}
+            changed = True
+        if "KUBERNETES_NAMESPACE" in os.environ:
+            k8s["namespace"] = os.environ["KUBERNETES_NAMESPACE"]
+            changed = True
+        if "KUBERNETES_POD_UID" in os.environ:
+            # this takes precedence over any value from /proc/self/cgroup
+            k8s["pod"]["uid"] = os.environ["KUBERNETES_POD_UID"]
+            changed = True
+        if changed:
+            system_data["kubernetes"] = k8s
+        return system_data
 
     def _build_metadata(self):
         return {

elasticapm/conf/constants.py

@@ -1,4 +1,4 @@
-EVENTS_API_PATH = "/intake/v2/events"
+EVENTS_API_PATH = "intake/v2/events"
 
 TRACE_CONTEXT_VERSION = 0
 TRACEPARENT_HEADER_NAME = "elastic-apm-traceparent"

elasticapm/utils/cgroup.py

@@ -0,0 +1,75 @@
+import os
+import re
+
+CGROUP_PATH = "/proc/self/cgroup"
+
+SYSTEMD_SCOPE_SUFFIX = ".scope"
+
+kubepods_regexp = re.compile(
+    r"(?:^/kubepods/[^/]+/pod([^/]+)$)|(?:^/kubepods\.slice/kubepods-[^/]+\.slice/kubepods-[^/]+-pod([^/]+)\.slice$)"
+)
+
+container_id_regexp = re.compile("^[0-9A-Fa-f]{64}$")
+
+
+def get_cgroup_container_metadata():
+    """
+    Reads docker/kubernetes metadata (container id, pod id) from /proc/self/cgroup
+
+    The result is a nested dictionary with the detected IDs, e.g.
+
+        {
+            "container": {"id": "2227daf62df6694645fee5df53c1f91271546a9560e8600a525690ae252b7f63"},
+            "pod": {"uid": "90d81341_92de_11e7_8cf2_507b9d4141fa"}
+        }
+
+    :return: a dictionary with the detected ids or {}
+    """
+    if not os.path.exists(CGROUP_PATH):
+        return {}
+    with open(CGROUP_PATH) as f:
+        return parse_cgroups(f) or {}
+
+
+def parse_cgroups(filehandle):
+    """
+    Reads lines from a file handle and tries to parse docker container IDs and kubernetes Pod IDs.
+
+    See tests.utils.docker_tests.test_cgroup_parsing for a set of test cases
+
+    :param filehandle:
+    :return: nested dictionary or None
+    """
+    for line in filehandle:
+        parts = line.strip().split(":")
+        if len(parts) != 3:
+            continue
+        cgroup_path = parts[2]
+
+        # Depending on the filesystem driver used for cgroup
+        # management, the paths in /proc/pid/cgroup will have
+        # one of the following formats in a Docker container:
+        #
+        #   systemd: /system.slice/docker-<container-ID>.scope
+        #   cgroupfs: /docker/<container-ID>
+        #
+        # In a Kubernetes pod, the cgroup path will look like:
+        #
+        #   systemd:
+        #      /kubepods.slice/kubepods-<QoS-class>.slice/kubepods-<QoS-class>-pod<pod-UID>.slice/<container-iD>.scope
+        #   cgroupfs:
+        #      /kubepods/<QoS-class>/pod<pod-UID>/<container-iD>
+
+        directory, container_id = os.path.split(cgroup_path)
+        if container_id.endswith(SYSTEMD_SCOPE_SUFFIX):
+            container_id = container_id[: -len(SYSTEMD_SCOPE_SUFFIX)]
+            if "-" in container_id:
+                container_id = container_id.split("-", 1)[1]
+        kubepods_match = kubepods_regexp.match(directory)
+        if kubepods_match:
+            pod_id = kubepods_match.group(1)
+            if not pod_id:
+                pod_id = kubepods_match.group(2)
+            return {"container": {"id": container_id}, "pod": {"uid": pod_id}}
+        elif container_id_regexp.match(container_id):
+            return {"container": {"id": container_id}}

tests/client/client_tests.py

@@ -39,15 +39,97 @@ def test_process_info(elasticapm_client):
 
 
 def test_system_info(elasticapm_client):
-    system_info = elasticapm_client.get_system_info()
+    # mock docker/kubernetes data here to get consistent behavior if test is run in docker
+    with mock.patch("elasticapm.utils.cgroup.get_cgroup_container_metadata") as mocked:
+        mocked.return_value = {}
+        system_info = elasticapm_client.get_system_info()
     assert {"hostname", "architecture", "platform"} == set(system_info.keys())
 
 
+def test_docker_kubernetes_system_info(elasticapm_client):
+    # mock docker/kubernetes data here to get consistent behavior if test is run in docker
+    with mock.patch("elasticapm.utils.cgroup.get_cgroup_container_metadata") as mock_metadata, mock.patch(
+        "socket.gethostname"
+    ) as mock_gethostname:
+        mock_metadata.return_value = {"container": {"id": "123"}, "kubernetes": {"pod": {"uid": "456"}}}
+        mock_gethostname.return_value = "foo"
+        system_info = elasticapm_client.get_system_info()
+    assert system_info["container"] == {"id": "123"}
+    assert system_info["kubernetes"] == {"pod": {"uid": "456", "name": "foo"}}
+
+
+@mock.patch.dict(
+    "os.environ",
+    {
+        "KUBERNETES_NODE_NAME": "node",
+        "KUBERNETES_NAMESPACE": "namespace",
+        "KUBERNETES_POD_NAME": "pod",
+        "KUBERNETES_POD_UID": "podid",
+    },
+)
+def test_docker_kubernetes_system_info_from_environ():
+    # initialize agent only after overriding environment
+    elasticapm_client = Client(metrics_interval="0ms")
+    # mock docker/kubernetes data here to get consistent behavior if test is run in docker
+    with mock.patch("elasticapm.utils.cgroup.get_cgroup_container_metadata") as mock_metadata:
+        mock_metadata.return_value = {}
+        system_info = elasticapm_client.get_system_info()
+    assert "kubernetes" in system_info
+    assert system_info["kubernetes"] == {
+        "pod": {"uid": "podid", "name": "pod"},
+        "node": {"name": "node"},
+        "namespace": "namespace",
+    }
+
+
+@mock.patch.dict(
+    "os.environ",
+    {
+        "KUBERNETES_NODE_NAME": "node",
+        "KUBERNETES_NAMESPACE": "namespace",
+        "KUBERNETES_POD_NAME": "pod",
+        "KUBERNETES_POD_UID": "podid",
+    },
+)
+def test_docker_kubernetes_system_info_from_environ_overrides_cgroups():
+    # initialize agent only after overriding environment
+    elasticapm_client = Client(metrics_interval="0ms")
+    # mock docker/kubernetes data here to get consistent behavior if test is run in docker
+    with mock.patch("elasticapm.utils.cgroup.get_cgroup_container_metadata") as mock_metadata, mock.patch(
+        "socket.gethostname"
+    ) as mock_gethostname:
+        mock_metadata.return_value = {"container": {"id": "123"}, "kubernetes": {"pod": {"uid": "456"}}}
+        mock_gethostname.return_value = "foo"
+        system_info = elasticapm_client.get_system_info()
+    assert "kubernetes" in system_info
+    assert system_info["kubernetes"] == {
+        "pod": {"uid": "podid", "name": "pod"},
+        "node": {"name": "node"},
+        "namespace": "namespace",
+    }
+    assert system_info["container"] == {"id": "123"}
+
+
+@mock.patch.dict("os.environ", {"KUBERNETES_NAMESPACE": "namespace"})
+def test_docker_kubernetes_system_info_except_hostname_from_environ():
+    # initialize agent only after overriding environment
+    elasticapm_client = Client(metrics_interval="0ms")
+    # mock docker/kubernetes data here to get consistent behavior if test is run in docker
+    with mock.patch("elasticapm.utils.cgroup.get_cgroup_container_metadata") as mock_metadata, mock.patch(
+        "socket.gethostname"
+    ) as mock_gethostname:
+        mock_metadata.return_value = {}
+        mock_gethostname.return_value = "foo"
+        system_info = elasticapm_client.get_system_info()
+    assert "kubernetes" in system_info
+    assert system_info["kubernetes"] == {"pod": {"name": "foo"}, "namespace": "namespace"}
+
+
 def test_config_by_environment():
-    with mock.patch.dict("os.environ", {"ELASTIC_APM_SERVICE_NAME": "app", "ELASTIC_APM_SECRET_TOKEN": "token"}):
+    with mock.patch.dict("os.environ", {"ELASTIC_APM_SERVICE_NAME": "envapp", "ELASTIC_APM_SECRET_TOKEN": "envtoken"}):
         client = Client(metrics_interval="0ms")
-        assert client.config.service_name == "app"
-        assert client.config.secret_token == "token"
+        assert client.config.service_name == "envapp"
+        assert client.config.secret_token == "envtoken"
         assert client.config.disable_send is False
     with mock.patch.dict("os.environ", {"ELASTIC_APM_DISABLE_SEND": "true"}):
         client = Client(metrics_interval="0ms")
@@ -182,7 +264,7 @@ def test_send(sending_elasticapm_client):
     for k, v in expected_headers.items():
         assert seen_headers[k] == v
 
-    assert 250 < request.content_length < 350
+    assert 250 < request.content_length < 400
 
 
 @pytest.mark.parametrize("sending_elasticapm_client", [{"disable_send": True}], indirect=True)
@@ -805,3 +887,21 @@ def test_ensure_parent_doesnt_change_existing_id(elasticapm_client):
     span_id = transaction.ensure_parent_id()
     span_id_2 = transaction.ensure_parent_id()
     assert span_id == span_id_2
+
+
+@pytest.mark.parametrize(
+    "elasticapm_client,expected",
+    [
+        ({"server_url": "http://localhost"}, "http://localhost/intake/v2/events"),
+        ({"server_url": "http://localhost/"}, "http://localhost/intake/v2/events"),
+        ({"server_url": "http://localhost:8200"}, "http://localhost:8200/intake/v2/events"),
+        ({"server_url": "http://localhost:8200/"}, "http://localhost:8200/intake/v2/events"),
+        ({"server_url": "http://localhost/a"}, "http://localhost/a/intake/v2/events"),
+        ({"server_url": "http://localhost/a/"}, "http://localhost/a/intake/v2/events"),
+        ({"server_url": "http://localhost:8200/a"}, "http://localhost:8200/a/intake/v2/events"),
+        ({"server_url": "http://localhost:8200/a/"}, "http://localhost:8200/a/intake/v2/events"),
+    ],
+    indirect=["elasticapm_client"],
+)
+def test_server_url_joining(elasticapm_client, expected):
+    assert elasticapm_client._api_endpoint_url == expected

tests/utils/cgroup_tests.py

@@ -0,0 +1,37 @@
+import mock
+import pytest
+
+from elasticapm.utils import cgroup, compat
+
+
+@pytest.mark.parametrize(
+    "test_input,expected",
+    [
+        (
+            "12:devices:/docker/051e2ee0bce99116029a13df4a9e943137f19f957f38ac02d6bad96f9b700f76",
+            {"container": {"id": "051e2ee0bce99116029a13df4a9e943137f19f957f38ac02d6bad96f9b700f76"}},
+        ),
+        (
+            "1:name=systemd:/system.slice/docker-cde7c2bab394630a42d73dc610b9c57415dced996106665d427f6d0566594411.scope",
+            {"container": {"id": "cde7c2bab394630a42d73dc610b9c57415dced996106665d427f6d0566594411"}},
+        ),
+        (
+            "1:name=systemd:/kubepods/besteffort/pode9b90526-f47d-11e8-b2a5-080027b9f4fb/15aa6e53-b09a-40c7-8558-c6c31e36c88a",
+            {
+                "container": {"id": "15aa6e53-b09a-40c7-8558-c6c31e36c88a"},
+                "pod": {"uid": "e9b90526-f47d-11e8-b2a5-080027b9f4fb"},
+            },
+        ),
+        (
+            "1:name=systemd:/kubepods.slice/kubepods-burstable.slice/kubepods-burstable-pod90d81341_92de_11e7_8cf2_507b9d4141fa.slice/crio-2227daf62df6694645fee5df53c1f91271546a9560e8600a525690ae252b7f63.scope",
+            {
+                "container": {"id": "2227daf62df6694645fee5df53c1f91271546a9560e8600a525690ae252b7f63"},
+                "pod": {"uid": "90d81341_92de_11e7_8cf2_507b9d4141fa"},
+            },
+        ),
+    ],
+)
+def test_cgroup_parsing(test_input, expected):
+    f = compat.StringIO(test_input)
+    result = cgroup.parse_cgroups(f)
+    assert result == expected