none
  $   ˆl–ßi~”jn-jqA
àAq·TÅ£Ê„ÏÉÀÇ ?÷     #
# Copyright (C) 2022 Red Hat, Inc.
# SPDX-License-Identifier: GPL-3.0-or-later


import errno
import logging
import os
import re
from typing import Any, Callable, DefaultDict, Iterable, List, NamedTuple, Optional, Tuple

from cockpit._vendor.systemd_ctypes import Handle

USER_HZ = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
MS_PER_JIFFY = 1000 / (USER_HZ if (USER_HZ > 0) else 100)
HWMON_PATH = '/sys/class/hwmon'

# we would like to do this, but mypy complains; https://github.com/python/mypy/issues/2900
# Samples = collections.defaultdict[str, Union[float, Dict[str, Union[float, None]]]]
Samples = DefaultDict[str, Any]

logger = logging.getLogger(__name__)


def read_int_file(rootfd: int, statfile: str, default: Optional[int] = None, key: bytes = b'') -> Optional[int]:
    # Not every stat is available, such as cpu.weight
    try:
        fd = os.open(statfile, os.O_RDONLY, dir_fd=rootfd)
    except FileNotFoundError:
        return None

    try:
        data = os.read(fd, 1024)
    except OSError as e:
        # cgroups can disappear between the open and read
        if e.errno != errno.ENODEV:
            logger.warning('Failed to read %s: %s', statfile, e)
        return None
    finally:
        os.close(fd)

    if key:
        start = data.index(key) + len(key)
        end = data.index(b'\n', start)
        data = data[start:end]

    try:
        # 0 often means "none", so replace it with default value
        return int(data) or default
    except ValueError:
        # Some samples such as "memory.max" contains "max" when there is a no limit
        return None


class SampleDescription(NamedTuple):
    name: str
    units: str
    semantics: str
    instanced: bool


class Sampler:
    descriptions: List[SampleDescription]

    def sample(self, samples: Samples) -> None:
        raise NotImplementedError


class CPUSampler(Sampler):
    descriptions = [
        SampleDescription('cpu.basic.nice', 'millisec', 'counter', instanced=False),
        SampleDescription('cpu.basic.user', 'millisec', 'counter', instanced=False),
        SampleDescription('cpu.basic.system', 'millisec', 'counter', instanced=False),
        SampleDescription('cpu.basic.iowait', 'millisec', 'counter', instanced=False),

        SampleDescription('cpu.core.nice', 'millisec', 'counter', instanced=True),
        SampleDescription('cpu.core.user', 'millisec', 'counter', instanced=True),
        SampleDescription('cpu.core.system', 'millisec', 'counter', instanced=True),
        SampleDescription('cpu.core.iowait', 'millisec', 'counter', instanced=True),
    ]

    def sample(self, samples: Samples) -> None:
        with open('/proc/stat') as stat:
            for line in stat:
                if not line.startswith('cpu'):
                    continue
                cpu, user, nice, system, _idle, iowait = line.split()[:6]
                core = cpu[3:] or None
                if core:
                    prefix = 'cpu.core'
                    samples[f'{prefix}.nice'][core] = int(nice) * MS_PER_JIFFY
                    samples[f'{prefix}.user'][core] = int(user) * MS_PER_JIFFY
                    samples[f'{prefix}.system'][core] = int(system) * MS_PER_JIFFY
                    samples[f'{prefix}.iowait'][core] = int(iowait) * MS_PER_JIFFY
                else:
                    prefix = 'cpu.basic'
                    samples[f'{prefix}.nice'] = int(nice) * MS_PER_JIFFY
                    samples[f'{prefix}.user'] = int(user) * MS_PER_JIFFY
                    samples[f'{prefix}.system'] = int(system) * MS_PER_JIFFY
                    samples[f'{prefix}.iowait'] = int(iowait) * MS_PER_JIFFY


class MemorySampler(Sampler):
    descriptions = [
        SampleDescription('memory.free', 'bytes', 'instant', instanced=False),
        SampleDescription('memory.used', 'bytes', 'instant', instanced=False),
        SampleDescription('memory.cached', 'bytes', 'instant', instanced=False),
        SampleDescription('memory.swap-used', 'bytes', 'instant', instanced=False),
    ]

    def sa