Feature/stacks container name (#35)

Send the container name for stacks from processes that are inside a Docker container

* Add container and hostname metadata to profiles

* Use absolute imports

* Update docs and DaemonSet file to add the Docker sock mount

* Remove redundant container code

* Add missing import

* Add missing Process import

* Strip container data from local stack file

* Separate long line into multiple lines

* Add option to disable the container names in profiles

* Reformat file

* Reformat using black

* Check if container name is not None before adding to container names list

* Update gprofiler/main.py

Co-authored-by: Yonatan Goldschmidt <yonatan.goldschmidt@granulate.io>

* Update gprofiler/main.py

Co-authored-by: Yonatan Goldschmidt <yonatan.goldschmidt@granulate.io>

* Update gprofiler/merge.py

Co-authored-by: Yonatan Goldschmidt <yonatan.goldschmidt@granulate.io>

* Fix argument renaming

* Ignore files in linting and activate venv on linting script

* Reordered imports

* Remove lxc from cgroup search

* Update gprofiler/main.py

Co-authored-by: Yonatan Goldschmidt <yonatan.goldschmidt@granulate.io>

* Remove redundant --log-file param

* Use absolute imports

* Reformat imports

Co-authored-by: Yonatan Goldschmidt <yonatan.goldschmidt@granulate.io>

Author: Alon Ben Tamar

.flake8

@@ -2,3 +2,4 @@
 max-line-length = 120
 # options for compatibility with black - see https://black.readthedocs.io/en/stable/compatible_configs.html#flake8
 extend-ignore = E203, W503
+exclude = .git,__pycache__,build,gprofiler.egg-info,venv

README.md

@@ -61,7 +61,8 @@ docker pull granulate/gprofiler:latest
 docker run --name gprofiler -d --restart=always \
     --network=host --pid=host --userns=host --privileged \
     -v /lib/modules:/lib/modules:ro -v /usr/src:/usr/src:ro \
-    granulate/gprofiler:latest -cu --token <token> --service-name <service> [options]
+    -v /var/run/docker.sock:/var/run/docker.sock \
+	granulate/gprofiler:latest -cu --token <token> --service-name <service> [options]
 ```
 
 For profiling with eBPF, kernel headers must be accessible from within the container at

deploy/k8s/gprofiler.yaml

@@ -26,6 +26,9 @@ spec:
         - name: usr-src
           hostPath:
             path: /usr/src
+        - name: docker-sock # Allow containers to fetch the names
+            hostPath:
+              path: /var/run/docker.sock
       hostPID: true
       securityContext:
         runAsUser: 0
@@ -51,6 +54,8 @@ spec:
             - name: usr-src
               mountPath: /usr/src
               readOnly: true
+            - name: docker-sock
+                mountPath: /var/run/docker.sock
           args:
             - -cu
             - --token

dev-requirements.txt

@@ -3,6 +3,5 @@ flake8
 black
 mypy
 isort
-docker
 pyinstaller==4.0
 staticx

gprofiler/__main__.py

@@ -2,7 +2,7 @@
 # Copyright (c) Granulate. All rights reserved.
 # Licensed under the AGPL3 License. See LICENSE.md in the project root for license information.
 #
-from .main import main
+from gprofiler.main import main
 
 if __name__ == "__main__":
     main()

gprofiler/client.py

@@ -12,7 +12,7 @@
 import requests
 from requests import Session
 
-from .utils import get_iso8061_format_time
+from gprofiler.utils import get_iso8061_format_time
 
 logger = logging.getLogger(__name__)
 
@@ -48,11 +48,18 @@ def _init_session(self, key: str, service: str):
         # Raises on failure
         self.get_health()
 
-    def get_base_url(self) -> str:
-        return "{}/{}/{}".format(self._host.rstrip("/"), self.BASE_PATH, self._version)
+    def get_base_url(self, api_version: str = None) -> str:
+        version = api_version if api_version is not None else self._version
+        return "{}/{}/{}".format(self._host.rstrip("/"), self.BASE_PATH, version)
 
     def _send_request(
-        self, method: str, path: str, data: Dict, files: Dict = None, timeout: float = DEFAULT_REQUEST_TIMEOUT
+        self,
+        method: str,
+        path: str,
+        data: Dict,
+        files: Dict = None,
+        timeout: float = DEFAULT_REQUEST_TIMEOUT,
+        api_version: str = None,
     ) -> Dict:
         opts: dict = {"headers": {}, "files": files, "timeout": timeout}
 
@@ -66,7 +73,7 @@ def _send_request(
                 json.dump(data, gzip_file, ensure_ascii=False)  # type: ignore
             opts["data"] = buffer.getvalue()
 
-        resp = self._session.request(method, "{}/{}".format(self.get_base_url(), path), **opts)
+        resp = self._session.request(method, "{}/{}".format(self.get_base_url(api_version), path), **opts)
         if 400 <= resp.status_code < 500:
             try:
                 data = resp.json()
@@ -107,4 +114,5 @@ def submit_profile(
                 "profile": profile,
             },
             timeout=self._upload_timeout,
+            api_version="v2",
         )

gprofiler/docker_client.py

@@ -0,0 +1,106 @@
+#
+# Copyright (c) Granulate. All rights reserved.
+# Licensed under the AGPL3 License. See LICENSE.md in the project root for license information.
+#
+import logging
+import re
+from typing import Dict, List, Optional, Set
+
+import docker
+
+DOCKER_SYSTEMD_CGROUPS = [re.compile(r"/system.slice/docker-([a-z0-9]{64})\.scope")]
+
+logger = logging.getLogger(__name__)
+
+
+class DockerClient:
+    def __init__(self):
+        try:
+            self._client = docker.from_env()
+        except Exception:
+            logger.warning(
+                'Could not initiate the Docker client, so the profiling data will not include the container'
+                ' names. If you are running the gProfiler in a container, please mount the Docker sock file'
+                'by running the Docker run command with the following argument: '
+                '"-v /var/run/docker.sock:/var/run/docker.sock". Otherwise, please open a new issue here: '
+                'https://github.com/Granulate/gprofiler/issues/new'
+            )
+            self._client = None
+
+        self._pid_to_container_name_cache: Dict[int, str] = {}
+        self._current_container_names: Set[str] = set()
+        self._container_id_to_name_cache: Dict[str, Optional[str]] = {}
+
+    def reset_cache(self):
+        self._pid_to_container_name_cache.clear()
+        self._current_container_names.clear()
+
+    @property
+    def container_names(self) -> List[str]:
+        return list(self._current_container_names)
+
+    def get_container_name(self, pid: int) -> str:
+        if self._client is None:
+            return ''
+        if pid in self._pid_to_container_name_cache:
+            return self._pid_to_container_name_cache[pid]
+        container_name: Optional[str] = self._safely_get_process_container_name(pid)
+        if container_name is None:
+            self._pid_to_container_name_cache[pid] = ''
+            return ''
+        self._pid_to_container_name_cache[pid] = container_name
+        return container_name
+
+    def _safely_get_process_container_name(self, pid: int) -> Optional[str]:
+        try:
+            container_id = self._get_process_container_id(pid)
+            if container_id is None:
+                return None
+            return self._get_container_name(container_id)
+        except Exception:
+            logger.warning(f'Could not get a container name for PID {pid}', exc_info=True)
+            return None
+
+    def _get_container_name(self, container_id) -> Optional[str]:
+        if container_id in self._container_id_to_name_cache:
+            container_name = self._container_id_to_name_cache[container_id]
+            if container_name is not None:
+                # Might happen a few times for the same container name, so we use a set to have unique values
+                self._current_container_names.add(container_name)
+            return container_name
+
+        self._refresh_container_names_cache()
+        if container_id not in self._container_id_to_name_cache:
+            self._container_id_to_name_cache[container_id] = None
+            return None
+        container_name = self._container_id_to_name_cache[container_id]
+        if container_name is not None:
+            self._current_container_names.add(container_name)
+        return container_name
+
+    def _refresh_container_names_cache(self):
+        # We re-fetch all of the currently running containers, so in order to keep the cache small we clear it
+        self._container_id_to_name_cache.clear()
+        running_containers = self._client.containers.list()
+        for container in running_containers:
+            self._container_id_to_name_cache[container.id] = container.name
+
+    @staticmethod
+    def _get_process_container_id(pid: int) -> Optional[str]:
+        # ECS uses /ecs/uuid/container-id
+        # standard Docker uses /docker/container-id
+        # k8s uses /kubepods/{burstable,besteffort}/uuid/container-id
+        try:
+            with open(f"/proc/{pid}/cgroup", 'r') as cgroup_file:
+                cgroup = cgroup_file.read()
+        except FileNotFoundError:
+            # The process died before we got to this point
+            return None
+        for line in cgroup.split():
+            if any(s in line for s in (':/docker/', ':/ecs/', ':/kubepods')):
+                return line.split("/")[-1]
+            for p in DOCKER_SYSTEMD_CGROUPS:
+                m = p.match(line)
+                if m is not None:
+                    return m.group(1)
+        return None

gprofiler/java.py

@@ -14,10 +14,10 @@
 import psutil
 from psutil import Process
 
-from .exceptions import StopEventSetException
-from .merge import parse_one_collapsed
-from .profiler_base import ProfilerBase
-from .utils import (
+from gprofiler.exceptions import StopEventSetException
+from gprofiler.merge import parse_one_collapsed
+from gprofiler.profiler_base import ProfilerBase
+from gprofiler.utils import (
     TEMPORARY_STORAGE_PATH,
     is_same_ns,
     pgrep_exe,

gprofiler/main.py

@@ -20,13 +20,14 @@
 import configargparse
 from requests import RequestException, Timeout
 
-from . import __version__, merge
-from .client import DEFAULT_UPLOAD_TIMEOUT, GRANULATE_SERVER_HOST, APIClient, APIError
-from .java import JavaProfiler
-from .perf import SystemProfiler
-from .profiler_base import NoopProfiler
-from .python import get_python_profiler
-from .utils import (
+from gprofiler import __version__, merge
+from gprofiler.client import DEFAULT_UPLOAD_TIMEOUT, GRANULATE_SERVER_HOST, APIClient, APIError
+from gprofiler.docker_client import DockerClient
+from gprofiler.java import JavaProfiler
+from gprofiler.perf import SystemProfiler
+from gprofiler.profiler_base import NoopProfiler
+from gprofiler.python import get_python_profiler
+from gprofiler.utils import (
     TEMPORARY_STORAGE_PATH,
     TemporaryDirectoryWithMode,
     atomically_symlink,
@@ -76,6 +77,7 @@ def __init__(
         rotating_output: bool,
         runtimes: Dict[str, bool],
         client: APIClient,
+        include_container_names=True,
     ):
         self._frequency = frequency
         self._duration = duration
@@ -101,6 +103,8 @@ def __init__(
             self._frequency, self._duration, self._stop_event, self._temp_storage_dir.name
         )
         self.initialize_python_profiler()
+        self._docker_client = DockerClient()
+        self._include_container_names = include_container_names
 
     def __enter__(self):
         self.start()
@@ -145,6 +149,7 @@ def _generate_output_files(
         base_filename = os.path.join(self._output_dir, "profile_{}".format(end_ts))
 
         collapsed_path = base_filename + ".col"
+        collapsed_data = self._strip_container_data(collapsed_data)
         Path(collapsed_path).write_text(collapsed_data)
 
         # point last_profile.col at the new file; and possibly, delete the previous one.
@@ -172,6 +177,15 @@ def _generate_output_files(
 
             logger.info(f"Saved flamegraph to {flamegraph_path}")
 
+    @staticmethod
+    def _strip_container_data(collapsed_data):
+        lines = []
+        for line in collapsed_data.splitlines():
+            if line.startswith("#"):
+                continue
+            lines.append(line[line.find(';') + 1 :])
+        return '\n'.join(lines)
+
     def start(self):
         self._stop_event.clear()
 
@@ -213,7 +227,9 @@ def _snapshot(self):
                 logger.exception(f"{future.name} profiling failed")
 
         local_end_time = local_start_time + datetime.timedelta(seconds=(time.monotonic() - monotonic_start_time))
-        merged_result = merge.merge_perfs(system_future.result(), process_perfs)
+        merged_result = merge.merge_perfs(
+            system_future.result(), process_perfs, self._docker_client, self._include_container_names
+        )
 
         if self._output_dir:
             self._generate_output_files(merged_result, local_start_time, local_end_time)
@@ -357,6 +373,13 @@ def parse_cmd_args():
         dest="log_rotate_backup_count",
         default=DEFAULT_LOG_BACKUP_COUNT,
     )
+    parser.add_argument(
+        "--disable-container-names",
+        action="store_true",
+        dest="disable_container_names",
+        default=False,
+        help="gProfiler won't gather the container names of processes that run in containers",
+    )
 
     continuous_command_parser = parser.add_argument_group("continuous")
     continuous_command_parser.add_argument(
@@ -459,7 +482,14 @@ def main():
 
         runtimes = {"java": args.java, "python": args.python}
         gprofiler = GProfiler(
-            args.frequency, args.duration, args.output_dir, args.flamegraph, args.rotating_output, runtimes, client
+            args.frequency,
+            args.duration,
+            args.output_dir,
+            args.flamegraph,
+            args.rotating_output,
+            runtimes,
+            client,
+            not args.disable_container_names,
         )
         logger.info("gProfiler initialized and ready to start profiling")
 

gprofiler/merge.py

@@ -2,11 +2,15 @@
 # Copyright (c) Granulate. All rights reserved.
 # Licensed under the AGPL3 License. See LICENSE.md in the project root for license information.
 #
+import json
 import logging
 import re
+import socket
 from collections import Counter, defaultdict
 from typing import Iterable, Mapping, MutableMapping
 
+from gprofiler.docker_client import DockerClient
+
 logger = logging.getLogger(__name__)
 
 SAMPLE_REGEX = re.compile(
@@ -63,7 +67,7 @@ def parse_many_collapsed(text: str) -> Mapping[int, Mapping[str, int]]:
     return results
 
 
-def collapse_stack(stack: str, comm: str) -> str:
+def collapse_stack(comm: str, stack: str) -> str:
     """
     Collapse a single stack from "perf".
     """
@@ -97,7 +101,12 @@ def parse_perf_script(script: str):
             logger.exception(f"Error processing sample: {sample}")
 
 
-def merge_perfs(perf_all: Iterable[Mapping[str, str]], process_perfs: Mapping[int, Mapping[str, int]]) -> str:
+def merge_perfs(
+    perf_all: Iterable[Mapping[str, str]],
+    process_perfs: Mapping[int, Mapping[str, int]],
+    docker_client: DockerClient,
+    should_determine_container_names: bool,
+) -> str:
     per_process_samples: MutableMapping[int, int] = Counter()
     new_samples: MutableMapping[str, int] = Counter()
     process_names = {}
@@ -108,7 +117,10 @@ def merge_perfs(perf_all: Iterable[Mapping[str, str]], process_perfs: Mapping[in
                 per_process_samples[pid] += 1
                 process_names[pid] = parsed["comm"]
             elif parsed["stack"] is not None:
-                new_samples[collapse_stack(parsed["stack"], parsed["comm"])] += 1
+                container_name = _get_container_name(pid, docker_client, should_determine_container_names)
+                collapsed_stack = collapse_stack(parsed["comm"], parsed["stack"])
+                stack_line = f'{container_name};{collapsed_stack}'
+                new_samples[stack_line] += 1
         except Exception:
             logger.exception(f"Error processing sample: {parsed}")
 
@@ -118,7 +130,20 @@ def merge_perfs(perf_all: Iterable[Mapping[str, str]], process_perfs: Mapping[in
         if process_perf_count > 0:
             ratio = perf_all_count / process_perf_count
             for stack, count in process_stacks.items():
-                full_stack = ";".join([process_names[pid], stack])
-                new_samples[full_stack] += round(count * ratio)
-
-    return "\n".join((f"{stack} {count}" for stack, count in new_samples.items()))
+                container_name = _get_container_name(pid, docker_client, should_determine_container_names)
+                stack_line = ";".join([container_name, process_names[pid], stack])
+                new_samples[stack_line] += round(count * ratio)
+    container_names = docker_client.container_names
+    docker_client.reset_cache()
+    profile_metadata = {
+        'containers': container_names,
+        'hostname': socket.gethostname(),
+        'container_names_enabled': should_determine_container_names,
+    }
+    output = [f"#{json.dumps(profile_metadata)}"]
+    output += [f"{stack} {count}" for stack, count in new_samples.items()]
+    return "\n".join(output)
+
+
+def _get_container_name(pid: int, docker_client: DockerClient, should_determine_container_names: bool):
+    return docker_client.get_container_name(pid) if should_determine_container_names else ""