def parse_cgroups()

in elasticapm/utils/cgroup.py [0:0]

• 21 lines of code
• 12 McCabe index (conditional complexity)

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 first part of kubepods_regexp matched
            if not pod_id:
                pod_id = kubepods_match.group(3)  # if second part of kubepods_regexp matched
                if pod_id:
                    pod_id = pod_id.replace("_", "-")
            return {"container": {"id": container_id}, "kubernetes": {"pod": {"uid": pod_id}}}
        elif container_id_regexp.match(container_id):
            return {"container": {"id": container_id}}