import argparse import datetime import json import sys import time from typing import AsyncGenerator import pytest import pytest_asyncio # Skip import when PCP is not available (for example in our tox env without system packages) try: from cpmapi import ( PM_ID_NULL, PM_INDOM_NULL, PM_SEM_COUNTER, PM_SEM_DISCRETE, PM_SEM_INSTANT, PM_TYPE_64, PM_TYPE_DOUBLE, PM_TYPE_STRING, PM_TYPE_U32, PM_TYPE_U64, ) from pcp import pmi # This comes from inside of the PCP code. It's fixed upstream but not yet # released. See https://github.com/performancecopilot/pcp/pull/2330 pytestmark = pytest.mark.filterwarnings( r"ignore:datetime\.datetime\.utcfromtimestamp\(\) is deprecated:DeprecationWarning" ) except ImportError: pytestmark = pytest.mark.skip("PCP not available") from cockpit.bridge import Bridge from .mocktransport import MockTransport @pytest_asyncio.fixture async def no_init_transport() -> AsyncGenerator[MockTransport, None]: bridge = Bridge(argparse.Namespace(privileged=False, beipack=False)) transport = MockTransport(bridge) try: yield transport finally: await transport.stop() @pytest.fixture def transport(no_init_transport: MockTransport) -> MockTransport: no_init_transport.init() return no_init_transport @pytest.fixture def broken_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('mock-archives') with open(pcp_dir / '0.index', 'w') as f: f.write("not a pcp index file") with open(pcp_dir / '0.meta', 'w') as f: f.write("not a pcp meta file") with open(pcp_dir / '0.0', 'w') as f: f.write("not a pcp sample file") return pcp_dir @pytest.fixture def discrete_metric_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('discrete-archive') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") archive_1.pmiAddMetric("mock.value", PM_ID_NULL, PM_TYPE_64, PM_INDOM_NULL, PM_SEM_DISCRETE, archive_1.pmiUnits(0, 0, 0, 0, 0, 0)) archive_1.pmiPutValue("mock.value", None, "4") archive_1.pmiWrite(0, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def big_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('big-archive') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") archive_1.pmiAddMetric("mock.value", PM_ID_NULL, PM_TYPE_64, PM_INDOM_NULL, PM_SEM_INSTANT, archive_1.pmiUnits(0, 0, 0, 0, 0, 0)) for i in range(1000): archive_1.pmiPutValue("mock.value", None, str(i)) archive_1.pmiWrite(i, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def multi_file_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('mock-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") archive_1.pmiAddMetric("mock.value", PM_ID_NULL, PM_TYPE_U32, PM_INDOM_NULL, PM_SEM_INSTANT, archive_1.pmiUnits(0, 0, 0, 0, 0, 0)) archive_1.pmiPutValue("mock.value", None, "10") archive_1.pmiWrite(0, 0) archive_1.pmiPutValue("mock.value", None, "11") archive_1.pmiWrite(1, 0) archive_1.pmiPutValue("mock.value", None, "12") archive_1.pmiWrite(2, 0) archive_1.pmiEnd() archive_2 = pmi.pmiLogImport(f"{pcp_dir}/1") archive_2.pmiAddMetric("mock.value", PM_ID_NULL, PM_TYPE_U32, PM_INDOM_NULL, PM_SEM_INSTANT, archive_2.pmiUnits(0, 0, 0, 0, 0, 0)) archive_2.pmiAddMetric("mock.late", PM_ID_NULL, PM_TYPE_U32, PM_INDOM_NULL, PM_SEM_INSTANT, archive_2.pmiUnits(0, 0, 0, 0, 0, 0)) archive_2.pmiPutValue("mock.value", None, "13") archive_2.pmiPutValue("mock.late", None, "30") archive_2.pmiWrite(3, 0) archive_2.pmiPutValue("mock.value", None, "14") archive_2.pmiPutValue("mock.late", None, "31") archive_2.pmiWrite(4, 0) archive_2.pmiPutValue("mock.value", None, "15") archive_2.pmiPutValue("mock.late", None, "32") archive_2.pmiWrite(5, 0) archive_2.pmiEnd() return pcp_dir @pytest.fixture def timestamps_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('timestamps-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") archive_1.pmiAddMetric("mock.value", PM_ID_NULL, PM_TYPE_U32, PM_INDOM_NULL, PM_SEM_INSTANT, archive_1.pmiUnits(0, 0, 0, 0, 0, 0)) timestamp = int(datetime.datetime.fromisoformat('2023-01-01').timestamp()) archive_1.pmiPutValue("mock.value", None, "10") archive_1.pmiWrite(timestamp, 0) timestamp = int(datetime.datetime.fromisoformat('2023-06-01').timestamp()) archive_1.pmiPutValue("mock.value", None, "11") archive_1.pmiWrite(timestamp, 0) timestamp = int(datetime.datetime.fromisoformat('2023-12-01').timestamp()) archive_1.pmiPutValue("mock.value", None, "12") archive_1.pmiWrite(timestamp, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def instances_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('instances-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") domain = 60 # Linux kernel pmid = archive_1.pmiID(domain, 2, 0) indom = archive_1.pmiInDom(domain, 2) units = archive_1.pmiUnits(0, 0, 0, 0, 0, 0) archive_1.pmiAddMetric("kernel.all.load", pmid, PM_TYPE_DOUBLE, indom, PM_SEM_INSTANT, units) archive_1.pmiAddInstance(indom, "1 minute", 1) archive_1.pmiAddInstance(indom, "5 minute", 5) archive_1.pmiAddInstance(indom, "15 minute", 15) archive_1.pmiPutValue("kernel.all.load", "1 minute", "1.0") archive_1.pmiPutValue("kernel.all.load", "5 minute", "5.0") archive_1.pmiPutValue("kernel.all.load", "15 minute", "15.0") archive_1.pmiWrite(0, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def instances_change_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('instances-change-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") domain = 60 # Linux kernel pmid = archive_1.pmiID(domain, 2, 0) indom = archive_1.pmiInDom(domain, 2) units = archive_1.pmiUnits(0, 0, 0, 0, 0, 0) archive_1.pmiAddMetric("network.interface.total.bytes", pmid, PM_TYPE_U32, indom, PM_SEM_COUNTER, units) archive_1.pmiAddInstance(indom, "lo", 1) archive_1.pmiAddInstance(indom, "eth0", 2) archive_1.pmiAddInstance(indom, "eth1", 3) archive_1.pmiPutValue("network.interface.total.bytes", "lo", "1") archive_1.pmiPutValue("network.interface.total.bytes", "eth0", "0") archive_1.pmiPutValue("network.interface.total.bytes", "eth1", "1") archive_1.pmiWrite(1597663539413, 0) archive_1.pmiPutValue("network.interface.total.bytes", "lo", "2") archive_1.pmiPutValue("network.interface.total.bytes", "eth0", "1") archive_1.pmiWrite(1597663539483, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def instances_rate_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('instances-rate-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") domain = 60 # Linux kernel pmid = archive_1.pmiID(domain, 2, 0) indom = archive_1.pmiInDom(domain, 2) units = archive_1.pmiUnits(0, 0, 0, 0, 0, 0) archive_1.pmiAddMetric("disk.value", pmid, PM_TYPE_U32, indom, PM_SEM_INSTANT, units) archive_1.pmiAddInstance(indom, "sda", 1) archive_1.pmiAddInstance(indom, "sdb", 2) archive_1.pmiAddInstance(indom, "sdc", 3) archive_1.pmiPutValue("disk.value", "sda", "1") archive_1.pmiPutValue("disk.value", "sdb", "1") archive_1.pmiPutValue("disk.value", "sdc", "1") archive_1.pmiWrite(0, 0) archive_1.pmiPutValue("disk.value", "sda", "1") archive_1.pmiPutValue("disk.value", "sdb", "1") archive_1.pmiPutValue("disk.value", "sdc", "2") archive_1.pmiWrite(1, 0) archive_1.pmiPutValue("disk.value", "sda", "1") archive_1.pmiPutValue("disk.value", "sdb", "10") archive_1.pmiPutValue("disk.value", "sdc", "3") archive_1.pmiWrite(2, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def unsupported_metric_archive(tmpdir_factory): """Cockpit only support float/integer type metrics""" pcp_dir = tmpdir_factory.mktemp('unsupported-metric-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") archive_1.pmiAddMetric("mock.value", PM_ID_NULL, PM_TYPE_STRING, PM_INDOM_NULL, PM_SEM_INSTANT, archive_1.pmiUnits(0, 0, 0, 0, 0, 0)) archive_1.pmiPutValue("mock.value", None, b"10") archive_1.pmiWrite(0, 0) archive_1.pmiEnd() return pcp_dir @pytest.fixture def mem_avail_archive(tmpdir_factory): pcp_dir = tmpdir_factory.mktemp('mem-avail-archives') archive_1 = pmi.pmiLogImport(f"{pcp_dir}/0") # https://github.com/performancecopilot/pcp/blob/766a78e631998e97196eeed9cc36631f30add74b/src/collectl2pcp/metrics.c#L339 # pminfo -m -f "mem.util.available" domain = 60 # Linux kernel pmid = archive_1.pmiID(domain, 1, 58) units = archive_1.pmiUnits(1, 0, 0, 1, 0, 0) archive_1.pmiAddMetric("mem.util.available", pmid, PM_TYPE_U64, PM_INDOM_NULL, PM_SEM_INSTANT, units) archive_1.pmiPutValue("mem.util.available", None, "19362828") archive_1.pmiWrite(0, 0) archive_1.pmiEnd() return pcp_dir def assert_metrics_meta(meta, source, timestamp=0, interval=1000): assert meta['timestamp'] == timestamp assert meta['interval'] == interval assert meta['source'] == source @pytest.mark.asyncio async def test_pcp_open_error(transport, multi_file_archive): await transport.check_open('metrics1', source=str(multi_file_archive), interval=-10, problem='protocol-error', reply_keys={'message': 'invalid "interval" value: -10'}) await transport.check_open('metrics1', problem='protocol-error', reply_keys={'message': 'no "source" option specified for metrics channel'}) await transport.check_open('metrics1', source="bazinga", problem='not-supported', reply_keys={'message': 'unsupported "source" option specified for metrics: bazinga'}) await transport.check_open('metrics1', source="/non-existent", problem='not-found') await transport.check_open('metrics1', source=str(multi_file_archive), metrics=[{"name": ""}], problem='protocol-error', reply_keys={'message': 'invalid "metrics" option was specified (no name for metric)'}) await transport.check_open('metrics1', source=str(multi_file_archive), limit=-10, problem='protocol-error', reply_keys={'message': 'invalid "limit" value: -10'}) invalid_timestamp_size = sys.maxsize * 1000 + 3 await transport.check_open('metrics1', source=str(multi_file_archive), timestamp=invalid_timestamp_size, problem='protocol-error', reply_keys={'message': f'invalid "timestamp" value: {invalid_timestamp_size}'}) await transport.check_open('metrics1', problem='protocol-error', reply_keys={'message': 'no "source" option specified for metrics channel'}) @pytest.mark.asyncio async def test_pcp_open(transport, multi_file_archive): await transport.check_open('metrics1', source=str(multi_file_archive), metrics=[{"name": "mock.value"}]) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(multi_file_archive)) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'mock.value' assert 'derive' not in metric assert metric['semantics'] == 'instant' # assert_sample (tc, "[[10],[11],[12]]"); _, data = await transport.next_frame() data = json.loads(data) assert data == [[10], [11], [12]] _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'mock.value' assert 'derive' not in metric assert metric['semantics'] == 'instant' _, data = await transport.next_frame() data = json.loads(data) assert data == [[13], [14], [15]] @pytest.mark.asyncio async def test_pcp_big_archive(transport, big_archive): await transport.check_open('metrics1', source=str(big_archive), metrics=[{"name": "mock.value"}]) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(big_archive)) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'mock.value' assert 'derive' not in metric assert metric['semantics'] == 'instant' _, data = await transport.next_frame() data = json.loads(data) # archives batch size is hardcoded to 60 assert data == [[i] for i in range(60)] @pytest.mark.asyncio async def test_pcp_instances_option(transport, instances_archive): await transport.check_open('metrics1', source=str(instances_archive), metrics=[{"name": "kernel.all.load"}], instances=["1 minute"]) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(instances_archive)) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'kernel.all.load' assert 'derive' not in metric assert metric['semantics'] == 'instant' assert metric['instances'] == ['1 minute'] _, data = await transport.next_frame() data = json.loads(data) assert data == [[[1.0]]] @pytest.mark.asyncio async def test_pcp_omit_instances_option(transport, instances_archive): await transport.check_open('metrics1', source=str(instances_archive), metrics=[{"name": "kernel.all.load"}], omit_instances=["1 minute"]) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(instances_archive)) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'kernel.all.load' assert 'derive' not in metric assert metric['semantics'] == 'instant' assert metric['instances'] == ['15 minute', '5 minute'] _, data = await transport.next_frame() data = json.loads(data) assert data == [[[15.0, 5.0]]] @pytest.mark.asyncio async def test_pcp_instances(transport, instances_archive): await transport.check_open('metrics1', source=str(instances_archive), metrics=[{"name": "kernel.all.load"}]) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(instances_archive)) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'kernel.all.load' assert 'derive' not in metric assert metric['semantics'] == 'instant' assert metric['instances'] == ['15 minute', '1 minute', '5 minute'] _, data = await transport.next_frame() data = json.loads(data) assert data == [[[15.0, 1.0, 5.0]]] @pytest.mark.asyncio async def test_pcp_timestamp(transport, timestamps_archive): timestamp = int(datetime.datetime.fromisoformat('2023-07-01').timestamp()) * 1000 await transport.check_open('metrics1', source=str(timestamps_archive), metrics=[{"name": "mock.value"}], limit=1, timestamp=timestamp) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(timestamps_archive), timestamp=timestamp) metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'mock.value' _, data = await transport.next_frame() data = json.loads(data) assert data == [[11.0]] @pytest.mark.asyncio async def test_pcp_negative_timestamp(transport, timestamps_archive): """ Given a negative timestamp the current time is taken and subtracted with the given timestamp """ timestamp = int(datetime.datetime.fromisoformat('2023-07-01').timestamp()) * 1000 relative_timestamp = int(time.time() * 1000) - timestamp await transport.check_open('metrics1', source=str(timestamps_archive), metrics=[{"name": "mock.value"}], limit=1, timestamp=-relative_timestamp) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) # time.time() is not exact assert (meta['timestamp'] - timestamp) < 1000 # 1s metrics = meta['metrics'] assert len(metrics) == 1 metric = metrics[0] assert metric['name'] == 'mock.value' _, data = await transport.next_frame() data = json.loads(data) assert data == [[11.0]] @pytest.mark.asyncio async def test_pcp_limit_archive(transport, big_archive): await transport.check_open('metrics1', source=str(big_archive), limit=30, metrics=[{"name": "mock.value"}]) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(big_archive)) _, data = await transport.next_frame() data = json.loads(data) assert data == [[i] for i in range(30)] _, data = await transport.next_frame() data = json.loads(data) assert data['command'] == 'close' @pytest.mark.asyncio async def test_pcp_broken_archive(transport, broken_archive): await transport.check_open('metrics1', source=str(broken_archive), metrics=[{"name": "mock.value", "derive": "rate"}], problem='not-found', reply_keys={'message': f'could not read archive {broken_archive}/0.index'}) @pytest.mark.asyncio async def test_pcp_instances_change(transport, instances_change_archive): await transport.check_open('metrics1', source=str(instances_change_archive), metrics=[{"name": "network.interface.total.bytes"}], limit=2) _, data = await transport.next_frame() # first message is always the meta message meta = json.loads(data) assert_metrics_meta(meta, str(instances_change_archive), 1597663539413000, 1000) metrics = meta['metrics'][0] assert metrics['instances'] == ['eth1', 'lo', 'eth0'] _, data = await transport.next_frame() data = json.loads(data) assert data == [[[1, 1, 0]]] # eth1 is unplugged, new meta message _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(instances_change_archive), 1597663539414000, 1000) metrics = meta['metrics'][0] assert metrics['instances'] == ['lo', 'eth0'] _, data = await transport.next_frame() data = json.loads(data) assert data == [[[1, 0]]] @pytest.mark.asyncio async def test_pcp_scale_memory_unit(transport, mem_avail_archive): await transport.check_open('metrics1', source=str(mem_avail_archive), metrics=[{"name": "mem.util.available", "units": "bytes"}], limit=1) _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(mem_avail_archive), 0, 1000) metric = meta['metrics'][0] assert metric['name'] == "mem.util.available" assert metric['units'] == "byte" assert metric['semantics'] == "instant" _, data = await transport.next_frame() data = json.loads(data) assert data == [[19827535872]] @pytest.mark.asyncio async def test_pcp_discrete_metric(transport, discrete_metric_archive): await transport.check_open('metrics1', source=str(discrete_metric_archive), metrics=[{"name": "mock.value"}]) _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(discrete_metric_archive), 0, 1000) metric = meta['metrics'][0] assert metric['name'] == "mock.value" assert metric['units'] == "" assert metric['semantics'] == "discrete" _, data = await transport.next_frame() data = json.loads(data) assert data == [[4]] @pytest.mark.asyncio async def test_pcp_rate_metric(transport, multi_file_archive): await transport.check_open('metrics1', source=str(multi_file_archive), metrics=[{"name": "mock.value", "derive": "rate"}]) _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(multi_file_archive)) metric = meta['metrics'][0] assert metric['name'] == "mock.value" assert metric['units'] == "" assert metric['semantics'] == "instant" _, data = await transport.next_frame() data = json.loads(data) assert data == [[False], [1.0], [1.0]] # new archive, new meta _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(multi_file_archive), timestamp=3000) _, data = await transport.next_frame() data = json.loads(data) assert data == [[False], [1.0], [1.0]] @pytest.mark.asyncio async def test_pcp_rate_instances(transport, instances_rate_archive): await transport.check_open('metrics1', source=str(instances_rate_archive), metrics=[{"name": "disk.value", "derive": "rate"}]) _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(instances_rate_archive)) metric = meta['metrics'][0] assert metric['name'] == "disk.value" assert metric['derive'] == "rate" assert metric['semantics'] == "instant" _, data = await transport.next_frame() data = json.loads(data) assert data == [[[False, False, False]], [[1.0, 0.0, 0.0]], [[1.0, 0.0, 9.0]]] _, data = await transport.next_frame() data = json.loads(data) print(data) @pytest.mark.asyncio async def test_pcp_unsupported_metric(transport, unsupported_metric_archive): """Verify that the PCP channel does not crash on an unknown metric type""" await transport.check_open('metrics1', source=str(unsupported_metric_archive), metrics=[{"name": "mock.value"}]) _, data = await transport.next_frame() meta = json.loads(data) assert_metrics_meta(meta, str(unsupported_metric_archive), 0, 1000)