import asyncio
import ctypes
import glob
import json
import logging
import platform
import sys
import time
from math import isclose
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, NamedTuple, Optional, Sequence, Union
from cockpit.protocol import CockpitProblem
from ..channel import AsyncChannel, ChannelError
from ..jsonutil import JsonObject, JsonValue, get_int, get_objv, get_str, get_strv
if TYPE_CHECKING:
import cpmapi as c_api
from pcp import pmapi
else:
pmapi = None
c_api = None
logger = logging.getLogger(__name__)
class MetricNotFoundError(CockpitProblem):
pass
class PcpMetricInfo(Dict[str, JsonValue]):
def __init__(self, value: JsonObject) -> None:
self.name = get_str(value, 'name')
self.derive = get_str(value, 'derive', '')
self.units = get_str(value, 'units', '')
super().__init__(name=self.name, derive=self.derive, units=self.units)
class MetricInfo(NamedTuple):
pmid: int
derive: str
desc: Any
name: str
factor: float
units: Any
instanced: bool
def try_import_pcp() -> None:
global c_api, pmapi
if c_api is None or pmapi is None:
try:
import cpmapi as c_api
from pcp import pmapi
except ImportError as exc:
raise ChannelError('not-supported', message='python3-pcp not installed') from exc
class ArchiveInfo:
metric_descriptions: List[MetricInfo]
def __init__(self, context: 'pmapi.pmContext', start: float, path: str) -> None:
self.context = context
self.start = start
self.path = path
self.metric_descriptions = []
def sort_key(self) -> float:
return self.start
class PcpMetricsChannel(AsyncChannel):
payload = 'metrics1'
pcp_dir: str
archive_batch: int
context: 'pmapi.pmContext'
source: str
interval: int
start_timestamp: int
last_timestamp: float
next_timestamp: float
limit: int
last_samples: Any = None
last_results: 'pmapi.pmResult | None' = None
metric_descriptions: List[MetricInfo]
def parse_options(self, options: JsonObject) -> None:
self.archive_batch = 60
self.last_timestamp = 0
self.next_timestamp = 0
max_size = sys.maxsize
min_size = -sys.maxsize - 1
self.interval = get_int(options, 'interval', 1000)
if self.interval <= 0 or self.interval > max_size:
raise ChannelError('protocol-error', message=f'invalid "interval" value: {self.interval}')
self.start_timestamp = get_int(options, 'timestamp', 0)
if self.start_timestamp / 1000 < min_size or self.start_timestamp / 1000 > max_size:
raise ChannelError('protocol-error', message=f'invalid "timestamp" value: {self.start_timestamp}')
if self.start_timestamp < 0:
self.start_timestamp = int((time.time() * 1000) + self.start_timestamp)
self.metrics = get_objv(options, 'metrics', PcpMetricInfo)
self.limit = get_int(options, 'limit', max_size)
if self.limit <= 0 or self.limit > max_size:
raise ChannelError('protocol-error', message=f'invalid "limit" value: {self.limit}')
self.instances = get_strv(options, 'instances', '')
self.omit_instances = get_strv(options, 'omit-instances', [])
self.source = get_str(options, 'source', '')
if self.source == '':
raise ChannelError('protocol-error', message='no "source" option specified for metrics channel')
@staticmethod
def float_to_timeval(timestamp: float) -> 'pmapi.timeval':
sec = int(timestamp / 1000)
usec = int((timestamp % 1000) * 1000)
return pmapi.timeval(sec, usec)
@staticmethod
def float_to_timespec(timestamp: float) -> 'pmapi.timespec':
ts = pmapi.timespec()
ts.tv_sec = int(timestamp / 1000)
ts.tv_nsec = int((timestamp % 1000) * 1000000)
return ts
@staticmethod
def get_context_and_name(source: str) -> 'tuple[str, str]':
if source == "":
raise ChannelError('protocol-error', message='no "source" option specified for metrics channel')
elif source.startswith('/'):
name = source
context_type = c_api.PM_CONTEXT_ARCHIVE
elif source == 'pcp-archive':
hostname = platform.node()
archive_dir = pmapi.pmContext.pmGetConfig("PCP_LOG_DIR")
name = f'{archive_dir}/pmlogger/{hostname}'
context_type = c_api.PM_CONTEXT_ARCHIVE
elif source == 'direct':
name = source
context_type = c_api.PM_CONTEXT_LOCAL
elif source == 'pmcd':
name = 'local:'
context_type = c_api.PM_CONTEXT_HOST
else:
raise ChannelError('not-supported',
message=f'unsupported "source" option specified for metrics: {source}')
return (name, context_type)
def get_archives(self, name: str) -> Sequence[ArchiveInfo]:
archives = sorted(self.prepare_archives(name), key=ArchiveInfo.sort_key)
if len(archives) == 0:
raise ChannelError('not-found')
for archive in archives:
for metric in self.metrics:
metric_desc = None
try:
metric_desc = self.convert_metric_description(archive.context, metric)
except MetricNotFoundError:
raise ChannelError('') from None
assert metric_desc is not None
archive.metric_descriptions.append(metric_desc)
return archives
def convert_metric_description(self, context: 'pmapi.pmContext', metric: JsonObject) -> MetricInfo:
name = get_str(metric, 'name', '')
if name == '':
raise ChannelError('protocol-error',
message='invalid "metrics" option was specified (no name for metric)')
units = get_str(metric, 'units', '')
derive = get_str(metric, 'derive', '')
try:
pm_ids = context.pmLookupName(name)
except pmapi.pmErr as exc:
if exc.errno() == c_api.PM_ERR_NAME:
logger.error("no such metric: %s", name)
raise MetricNotFoundError('error', message=f'no such metric: {name}') from None
else:
raise ChannelError('internal-error', message=str(exc)) from None
try:
pm_desc = context.pmLookupDesc(pm_ids[0])
except pmapi.pmErr as exc:
if exc.errno() == c_api.PM_ERR_NAME:
raise ChannelError('not-found', message=f'no such metric: {name}') from None
else:
raise ChannelError('internal-error', message=str(exc)) from None
instanced = pm_desc.indom != c_api.PM_INDOM_NULL
if instanced:
if len(self.instances) > 0:
context.pmDelProfile(pm_desc, None)
for instance in self.instances:
try:
instid = context.pmLookupInDom(pm_desc, instance)
context.pmAddProfile(pm_desc, instid)
except pmapi.pmErr as exc:
logger.debug("Unable to add profile: instance=%s err=%s", instance, exc)
if len(self.omit_instances) > 0:
context.pmAddProfile(pm_desc, None)
for omit_instance in self.omit_instances:
try:
instid = context.pmLookupInDom(pm_desc, omit_instance)
context.pmDelProfile(pm_desc, [instid])
except pmapi.pmErr as exc:
logger.debug("Unable to remove profile: instance=%s err=%s", omit_instance, exc)
factor = 1.0
pm_units = pm_desc.units
if units:
try:
[parsed_units, factor] = context.pmParseUnitsStr(units)
except pmapi.pmErr as exc:
if exc.errno() == c_api.PM_ERR_NAME:
raise ChannelError('not-found', message=f'no such metric: {name}') from None
else:
raise ChannelError('internal-error', message=str(exc)) from None
self.try_convert_unit(context, pm_desc, pm_units)
if units != parsed_units or not isclose(factor, 1.0):
pm_units = parsed_units
return MetricInfo(pmid=pm_ids[0],
name=name,
derive=derive,
desc=pm_desc,
factor=factor,
units=pm_units,
instanced=instanced)
@staticmethod
def try_convert_unit(context: 'pmapi.pmContext', pm_desc: 'pmapi.pmDesc', pm_units: 'pmapi.pmUnits') -> None:
"""Try to convert a dummy value to validate that the metric is convertible"""
dummy = pmapi.pmAtomValue()
dummy.d = 0.0
try:
context.pmConvScale(c_api.PM_TYPE_DOUBLE, dummy, [pm_desc], 0, pm_units)
except pmapi.pmErr as exc:
raise ChannelError('internal-error', message=str(exc)) from None
@staticmethod
def prepare_archives(archive_dir: str) -> Iterable[ArchiveInfo]:
indexes = glob.glob(glob.escape(archive_dir) + '/*.index')
for archive_path in indexes:
logger.debug('opening archive: %r', archive_path)
try:
context = pmapi.pmContext(c_api.PM_CONTEXT_ARCHIVE, archive_path)
log_label = context.pmGetArchiveLabel()
archive_start = float(log_label.start) * 1000
yield ArchiveInfo(context, archive_start, archive_path)
except pmapi.pmErr as exc:
if exc.errno() != c_api.PM_ERR_LOGFILE:
raise ChannelError('not-found', message=f'could not read archive {archive_path}') from None
@staticmethod
def semantic_val(sem_id: int) -> str:
if sem_id == c_api.PM_SEM_COUNTER:
return "counter"
elif sem_id == c_api.PM_SEM_INSTANT:
return "instant"
elif sem_id == c_api.PM_SEM_DISCRETE:
return "discrete"
return ""
def send_meta(self, results: 'pmapi.pmResult', context: 'pmapi.pmContext') -> None:
metrics = []
for metric_desc in self.metric_descriptions:
desc: Dict[str, Union[str, List[str]]] = {"name": metric_desc.name}
if metric_desc.derive:
desc['derive'] = metric_desc.derive
if isclose(metric_desc.factor, 1.0):
desc['units'] = str(metric_desc.units)
else:
desc['units'] = f"{context.pmUnitsStr(metric_desc.units)}*{1.0 / metric_desc.factor}"
semantics = self.semantic_val(metric_desc.desc.sem)
if semantics != "":
desc['semantics'] = self.semantic_val(metric_desc.desc.sem)
if metric_desc.instanced:
insts: List[str] = []
for i in range(results.contents.numpmid):
pmid = results.contents.get_pmid(i)
if metric_desc.pmid != pmid:
continue
for j in range(results.contents.get_numval(i)):
value = results.contents.get_vlist(i, j)
instance_desc = context.pmNameInDom(metric_desc.desc, value.inst)
insts.append(instance_desc)
desc['instances'] = insts
metrics.append(desc)
now = int(time.time()) * 1000
ms = (results.contents.timestamp.tv_nsec / 1000000
if hasattr(results.contents.timestamp, 'tv_nsec')
else results.contents.timestamp.tv_usec / 1000)
timestamp = int(results.contents.timestamp.tv_sec * 1000 + ms)
self.send_json(source=self.source,
interval=self.interval,
timestamp=timestamp,
metrics=metrics,
now=now)
def needs_meta_update(self, results: 'pmapi.pmResult') -> bool:
"""
If a multi-instance metric changes its instances we need to send a new
meta message when these change. For example when an drive or ethernet
card is removed out.
"""
last_results = self.last_results
if last_results is None:
return True
for i in range(results.contents.numpmid):
if not self.metric_descriptions[i].instanced:
continue
numval1 = results.contents.get_numval(i)
numval2 = last_results.contents.get_numval(i)
if numval1 != numval2:
return True
for j in range(numval1):
if results.contents.get_inst(i, j) != last_results.contents.get_inst(i, j):
return True
return False
async def sample(self, context: 'pmapi.pmContext', archive_batch: int, limit: int, total_fetched: int) -> int:
pmids = (ctypes.c_uint * len(self.metric_descriptions))()
for i, metric in enumerate(self.metric_descriptions):
pmids[i] = metric.pmid
while True:
fetched: List[Any] = []
try:
for _ in range(archive_batch):
if total_fetched == limit:
if context.type != c_api.PM_CONTEXT_ARCHIVE:
return total_fetched
self.send_updates(fetched)
logger.debug('Reached limit "%s", stopping', self.limit)
return total_fetched
results = context.pmFetch(pmids)
if self.needs_meta_update(results):
if fetched:
self.send_updates(fetched)
fetched.clear()
self.send_meta(results, context)
fetched.append(self.parse_fetched_results(context, results))
self.last_results = results
total_fetched += 1
self.send_updates(fetched)
fetched.clear()
await asyncio.sleep(0)
except pmapi.pmErr as exc:
logger.debug('Fetching error: %r, fetched %r', exc, fetched)
if exc.errno() != c_api.PM_ERR_EOL:
raise ChannelError('internal-error', message=str(exc)) from None
if len(fetched) > 0:
self.send_updates(fetched)
break
return total_fetched
def parse_fetched_results(self, context: 'pmapi.pmContext', results: Any) -> Any:
metrics = list(self.metrics)
samples: Dict[str, Optional[float] | list[Optional[float]]] = {}
samples['timestamp'] = float(results.contents.timestamp)
for i in range(results.contents.numpmid):
values: Optional[List[Optional[float]] | Optional[float]] = None
numval: int = results.contents.get_numval(i)
metric_desc = self.metric_descriptions[i]
content_type = metric_desc.desc.type
if numval < 0:
samples[metrics[i].name] = None
continue
if content_type == c_api.PM_TYPE_AGGREGATE or \
content_type == c_api.PM_TYPE_EVENT or \
content_type == c_api.PM_TYPE_STRING:
samples[metrics[i].name] = None
continue
if not metric_desc.instanced:
values = self.build_sample(context, results, metric_desc, i, 0)
else:
vals: List[Optional[float]] = []
for j in range(numval):
vals.append(self.build_sample(context, results, metric_desc, i, j))
values = vals
samples[metrics[i].name] = values
return samples
def build_sample(self, context: 'pmapi.pmContext', results: Any, metric_desc: MetricInfo,
metric: int, instance: int) -> 'float | None':
pmid = results.contents.get_pmid(metric)
logger.debug("build_sample pmid=%d, metric=%d, instance=%d", pmid, metric, instance)
valueset = results.contents.get_vset(metric)
if valueset.numval <= instance:
return None
valfmt = results.contents.get_valfmt(metric)
value = results.contents.get_vlist(metric, instance)
content_type = metric_desc.desc.type
sample_value = None
atom = None
if content_type == c_api.PM_TYPE_64:
try:
atom = context.pmExtractValue(valfmt,
value,
c_api.PM_TYPE_64,
c_api.PM_TYPE_64)
sample_value = atom.ll & ((1 << 48) - 1)
except Exception as exc:
logger.exception("Unable to extract PCP TYPE_64 value %s", exc)
elif content_type == c_api.PM_TYPE_U64:
try:
atom = context.pmExtractValue(valfmt,
value,
c_api.PM_TYPE_U64,
c_api.PM_TYPE_U64)
sample_value = (atom.ull << 16) >> 16
except Exception as exc:
logger.exception("Unable to extract PCP TYPE_U64 value %s", exc)
else:
try:
atom = context.pmExtractValue(valfmt,
value,
content_type,
c_api.PM_TYPE_DOUBLE)
sample_value = atom.d
except Exception as exc:
logger.exception("Unable to extract PCP value %s", exc)
if metric_desc.desc.units != metric_desc.units:
try:
dummy = pmapi.pmAtomValue()
dummy.d = sample_value
converted_atom = context.pmConvScale(c_api.PM_TYPE_DOUBLE,
dummy,
[metric_desc.desc],
0,
metric_desc.units)
sample_value = converted_atom.d * metric_desc.factor
except pmapi.pmErr as exc:
raise ChannelError('internal-error', message=str(exc)) from None
return sample_value
def calculate_sample_rate(self, value: float, old_value: Optional[float]) -> Any:
if old_value is not None:
return (value - old_value) / (self.next_timestamp - self.last_timestamp)
else:
return False
def send_updates(self, samples: Sequence[Any]) -> None:
data: list[list[float | list[float]]] = []
last_samples = self.last_samples or {}
for sample in samples:
assert isinstance(sample['timestamp'], float)
self.next_timestamp = sample['timestamp']
sampled_values: list[float | list[float]] = []
for metricinfo in self.metrics:
value = sample[metricinfo.name]
old_value = last_samples.get(metricinfo.name, None)
if isinstance(value, list):
if metricinfo.derive == 'rate':
tmp = []
for index, val in enumerate(value):
old_val = None
if old_value is not None:
try:
old_val = old_value[index]
except IndexError:
pass
tmp.append(self.calculate_sample_rate(val, old_val))
sampled_values.append(tmp)
else:
sampled_values.append(value)
else:
if metricinfo.derive == 'rate' and value is not None:
sampled_values.append(self.calculate_sample_rate(value, old_value))
else:
if value is None:
sampled_values.append(False)
else:
sampled_values.append(value)
data.append(sampled_values)
self.last_timestamp = self.next_timestamp
last_samples = sample
self.last_samples = last_samples
self.send_data(json.dumps(data).encode())
async def sample_archives(self, archives: 'Sequence[ArchiveInfo]') -> None:
total_fetched = 0
for i, archive in enumerate(archives):
self.metric_descriptions = archive.metric_descriptions
self.last_results = None
self.last_samples = None
timestamp = self.start_timestamp
if i != len(archives) - 1 and archives[i + 1].start < timestamp:
continue
if timestamp < archive.start:
timestamp = int(archive.start)
context = archive.context
try:
if hasattr(c_api, 'PM_XTB_SET'):
context.pmSetMode(c_api.PM_MODE_INTERP | c_api.PM_XTB_SET(c_api.PM_TIME_MSEC),
self.float_to_timeval(timestamp), self.interval)
else:
context.pmSetMode(c_api.PM_MODE_INTERP,
self.float_to_timespec(timestamp),
self.float_to_timespec(self.interval))
except pmapi.pmErr as exc:
raise ChannelError('internal-error', message=str(exc)) from None
total_fetched = await self.sample(archive.context, self.archive_batch, self.limit, total_fetched)
def prepare_direct_context(self, name: str, context_type: str) -> 'pmapi.pmContext':
try:
direct_context = pmapi.pmContext(context_type, name)
except pmapi.pmErr as exc:
raise ChannelError('internal-error', message=str(exc)) from None
for metric in self.metrics:
metric_desc = None
try:
metric_desc = self.convert_metric_description(direct_context, metric)
except MetricNotFoundError:
raise ChannelError('') from None
assert metric_desc is not None
self.metric_descriptions.append(metric_desc)
return direct_context
async def run(self, options: JsonObject) -> None:
self.metric_descriptions = []
logger.debug('metrics pcp-archive open: %r, channel: %r', options, self.channel)
self.parse_options(options)
try_import_pcp()
if pmapi is True:
raise ChannelError('not-supported')
name, context_type = self.get_context_and_name(self.source)
if context_type == c_api.PM_CONTEXT_ARCHIVE:
archives = self.get_archives(name)
self.ready()
await self.sample_archives(archives)
else:
direct_context = self.prepare_direct_context(name, context_type)
self.ready()
while True:
await self.sample(direct_context, 1, 1, 0)
await asyncio.sleep(self.interval / 1000)
