def get_cgroup_file_paths()

in elasticapm/metrics/sets/cpu_linux.py [0:0]

• 38 lines of code
• 15 McCabe index (conditional complexity)

    def get_cgroup_file_paths(self, proc_self_cgroup, mount_info):
        """
        Try and find the paths for CGROUP memory limit files, first trying to find the root path
        in /proc/self/mountinfo, then falling back to the default location /sys/fs/cgroup
        :param proc_self_cgroup: path to "self" cgroup file, usually /proc/self/cgroup
        :param mount_info: path to "mountinfo" file, usually proc/self/mountinfo
        :return: a 3-tuple of memory info files, or None
        """
        line_cgroup = None
        try:
            with open(proc_self_cgroup, "r") as proc_self_cgroup_file:
                for line in proc_self_cgroup_file:
                    if line_cgroup is None and line.startswith("0:"):
                        line_cgroup = line
                    if MEMORY_CGROUP.match(line):
                        line_cgroup = line
                        break
        except IOError:
            logger.debug("Cannot read %s, skipping cgroup metrics", proc_self_cgroup, exc_info=True)
            return
        if line_cgroup is None:
            return
        try:
            with open(mount_info, "r") as mount_info_file:
                for line in mount_info_file:
                    # cgroup v2
                    matcher = CGROUP_V2_MOUNT_POINT.match(line)
                    if matcher is not None:
                        files = self._get_cgroup_v2_file_paths(line_cgroup, matcher.group(1))
                        if files:
                            return files
                    # cgroup v1
                    matcher = CGROUP_V1_MOUNT_POINT.match(line)
                    if matcher is not None:
                        files = self._get_cgroup_v1_file_paths(matcher.group(1))
                        if files:
                            return files
        except IOError:
            logger.debug("Cannot read %s, skipping cgroup metrics", mount_info, exc_info=True)
            return
        # discovery of cgroup path failed, try with default path
        files = self._get_cgroup_v2_file_paths(line_cgroup, SYS_FS_CGROUP)
        if files:
            return files
        files = self._get_cgroup_v1_file_paths(os.path.join(SYS_FS_CGROUP, "memory"))
        if files:
            return files
        logger.debug("Location of cgroup files failed, skipping cgroup metrics")