import copy import importlib.util from pathlib import Path import os import tempfile import io import sys import threading import time import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location('memory_measure', Path(__file__).with_name('measure-memory.py')) measure = importlib.util.module_from_spec(spec) spec.loader.exec_module(measure) class MemoryEvidenceTests(unittest.TestCase): def test_host_units_and_missing_fields(self): raw = 'MemTotal: 2048 kB\nMemAvailable: 1024 kB\nSwapTotal: 4096 kB\nSwapFree: 3072 kB\n' parsed = measure.parse_host(raw, 'pswpin 20\npswpout 30\noom_kill 1\n') self.assertEqual(parsed['MemTotalBytes'], 2048 * 1024) self.assertEqual(parsed['pswpin'], 20) for invalid in (raw.replace('kB', 'MB'), raw.replace('MemAvailable: 1024 kB\n', ''), raw.replace('MemAvailable: 1024', 'MemAvailable: 4096')): with self.assertRaises(ValueError): measure.parse_host(invalid, '') def test_cache_and_no_configured_limit_are_distinct(self): item = {'Id': 'a' * 64, 'Image': 'sha256:fixed', 'State': {'StartedAt': 't0', 'Running': True}, 'HostConfig': {'Memory': 0}} stats = {'id': 'a' * 64, 'memory_stats': {'usage': 100, 'limit': 2048, 'stats': {'inactive_file': 30}}} result = measure.project_container(item, stats, 'a' * 64) self.assertEqual(result['usageIncludingCacheBytes'], 100) self.assertEqual(result['usageMinusInactiveFileBytes'], 70) self.assertEqual(result['configuredLimitBytes'], 0) with self.assertRaises(ValueError): measure.project_container(item, stats, 'b' * 64) def test_simultaneous_peak_is_not_sum_of_separate_peaks(self): host = {'MemTotalBytes': 2048, 'MemAvailableBytes': 1024, 'pswpin': 3, 'pswpout': 5, 'oom_kill': 1} def container(identifier, usage): return {'id': identifier, 'imageId': 'i', 'startedAt': 't0', 'restartCount': 0, 'running': True, 'oomKilled': False, 'usageIncludingCacheBytes': usage} samples = [{'elapsedMs': 0, 'host': host, 'containers': [container('a', 100), container('b', 20)]}, {'elapsedMs': 1000, 'host': {**host, 'pswpin': 4}, 'containers': [container('a', 20), container('b', 100)]}] result = measure.summarize(samples) self.assertEqual(result['maxObservedContainerUsageIncludingCacheBytes'], 120) self.assertEqual(result['pswpinDelta'], 1) self.assertTrue(result['containerIdentityAndStartStable']) self.assertFalse(result['acceptancePassed']) changed = copy.deepcopy(samples) changed[-1]['containers'][0]['startedAt'] = 't1' changed[-1]['containers'][1]['oomKilled'] = True self.assertFalse(measure.summarize(changed)['containerIdentityAndStartStable']) self.assertTrue(measure.summarize(changed)['anyObservedOomKilled']) def test_counter_rollback_is_unknown(self): samples = [{'elapsedMs': i, 'host': {'MemTotalBytes': 2, 'MemAvailableBytes': 1, 'pswpin': value}, 'containers': []} for i, value in enumerate([100, 2])] self.assertIsNone(measure.summarize(samples)['pswpinDelta']) class CgroupEvidenceTests(unittest.TestCase): identifier = 'a' * 64 path = '/sys/fs/cgroup/system.slice/docker-' + identifier + '.scope' def setUp(self): self.before = {'Id': self.identifier, 'Image': 'sha256:fixed', 'RestartCount': 0, 'State': {'StartedAt': 'original-start', 'Pid': 123, 'Running': True, 'OOMKilled': False}, 'HostConfig': {'Memory': 512, 'MemorySwap': 1024}} self.stats = {'id': self.identifier, 'memory_stats': {'usage': 100, 'limit': 512, 'stats': {'inactive_file': 30}}} self.files = { '/proc/self/mountinfo': '1 0 0:1 / /sys/fs/cgroup rw - cgroup2 cgroup rw\n', '/proc/123/cgroup': '0::/system.slice/docker-' + self.identifier + '.scope\n', '/proc/123/stat': '123 (a name with ) parentheses) S ' + ' '.join(['0'] * 18 + ['987'] + ['0'] * 8), self.path + '/memory.max': '512\n', self.path + '/memory.swap.max': '512\n', self.path + '/cpu.stat': 'nr_periods 10\nnr_throttled 2\nthrottled_usec 1234\n', self.path + '/memory.current': '100\n', self.path + '/memory.peak': '200\n', self.path + '/memory.events': 'low 0\nhigh 1\nmax 2\noom 3\noom_kill 4\n', self.path + '/memory.swap.current': '5\n', self.path + '/memory.swap.peak': '6\n', self.path + '/memory.swap.events': 'high 0\nmax 2\nfail 1\n', } def observe(self, after=None, reader=None, resolver=None, statter=None): def read(path, _expected_device=None, deadline=None): if path not in self.files: raise FileNotFoundError() return self.files[path] queries = [] def get(path, deadline=None): queries.append(path) if '/stats?' in path: return self.stats return self.before if len(queries) == 1 else (after or self.before) with patch.object(measure, 'docker_get', side_effect=get), \ patch.object(measure, 'read_bounded', side_effect=reader or read), \ patch.object(measure.os.path, 'realpath', side_effect=resolver or (lambda path: path)), \ patch.object(measure.os, 'stat', side_effect=statter or (lambda path: os.stat_result((0, 7, 8, 0, 0, 0, 0, 0, 0, 0)))), \ patch.object(measure.os, 'sep', '/'): result = measure.collect_container(self.identifier) self.assertEqual(queries, ['/containers/' + self.identifier + '/json', '/containers/' + self.identifier + '/stats?stream=false&one-shot=true', '/containers/' + self.identifier + '/json']) return result def test_exact_identity_metrics_and_docker_cache_are_distinct(self): result = self.observe() self.assertTrue(result['identityStableDuringSample']) self.assertEqual(result['usageMinusInactiveFileBytes'], 70) self.assertEqual(result['cgroupV2']['status'], 'available') self.assertEqual(result['cgroupV2']['metrics']['memory.current'], 100) self.assertEqual(result['cgroupV2']['metrics']['memory.peak'], 200) self.assertEqual(result['cgroupV2']['metrics']['memory.events']['oom_kill'], 4) self.assertEqual(result['cgroupV2']['metrics']['memory.swap.current'], 5) self.assertEqual(result['cgroupV2']['metrics']['memory.swap.peak'], 6) self.assertEqual(result['cgroupV2']['metrics']['memory.swap.events']['fail'], 1) def test_optional_swap_peak_missing_is_unknown_not_zero(self): del self.files[self.path + '/memory.swap.peak'] result = self.observe()['cgroupV2'] self.assertEqual(result['status'], 'partial') self.assertIsNone(result['metrics']['memory.swap.peak']) self.assertEqual(result['metrics']['memory.peak'], 200) self.assertIn('memory.swap.peak', result['unavailable']) def test_each_missing_metric_is_explicit_unknown(self): for name in measure.CGROUP_FILES: with self.subTest(name=name): value = self.files.pop(self.path + '/' + name) result = self.observe()['cgroupV2'] self.assertIsNone(result['metrics'][name]) self.assertEqual(result['status'], 'partial') self.files[self.path + '/' + name] = value def test_malformed_metrics_never_coerce_to_zero(self): for value in ('-1\n', '+1\n', ' 1\n', '01\n', '1\n2\n', '18446744073709551616\n', 'max\n'): with self.subTest(value=value): self.files[self.path + '/memory.current'] = value self.assertIsNone(self.observe()['cgroupV2']['metrics']['memory.current']) for value in ('oom 1\noom 2\n', 'low 0\nhigh -1\nmax 0\noom 0\noom_kill 0\n', 'low 0\n', 'low 0\n'): with self.subTest(value=value): self.files[self.path + '/memory.events'] = value self.assertIsNone(self.observe()['cgroupV2']['metrics']['memory.events']) def test_docker_identity_drift_invalidates_entire_cgroup_sample(self): for field, value in [('Pid', 124), ('StartedAt', 'new'), ('OOMKilled', True), ('Running', False)]: after = copy.deepcopy(self.before); after['State'][field] = value result = self.observe(after) self.assertFalse(result['identityStableDuringSample']) self.assertTrue(all(value is None for value in result['cgroupV2']['metrics'].values())) for field, value in [('Id', 'b' * 64), ('RestartCount', 1), ('Image', 'sha256:new')]: after = copy.deepcopy(self.before); after[field] = value self.assertEqual(self.observe(after)['cgroupV2']['status'], 'unavailable') def test_process_reuse_rootfs_swap_and_membership_move_are_unknown(self): for changed in ('/proc/123/stat', '/proc/123/cgroup'): count = 0 def read(path, _expected_device=None, deadline=None): nonlocal count if path == changed: count += 1 if count == 2: return self.files[path].replace('987', '988') if path.endswith('/stat') else '0::/docker/' + 'b' * 64 + '\n' return self.files[path] self.assertEqual(self.observe(reader=read)['cgroupV2']['status'], 'unavailable') count = 0 def statter(path): nonlocal count if path == '/proc/123/root': count += 1 return os.stat_result((0, count, 8, 0, 0, 0, 0, 0, 0, 0)) self.assertEqual(self.observe(statter=statter)['cgroupV2']['status'], 'unavailable') def test_unsupported_v1_namespace_subtree_and_unbound_membership(self): original = self.files['/proc/123/cgroup'] for value in ('2:memory:/docker/' + self.identifier + '\n', '0::/\n', '0::/docker/' + 'b' * 64 + '\n', '0::/docker/' + self.identifier + '/../other\n', original.rstrip() + '/child\n', original + original): self.files['/proc/123/cgroup'] = value self.assertEqual(self.observe()['cgroupV2']['status'], 'unavailable') self.files['/proc/123/cgroup'] = original self.files['/proc/self/mountinfo'] = '1 0 0:1 /subtree /sys/fs/cgroup rw - cgroup2 cgroup rw\n' self.assertEqual(self.observe()['cgroupV2']['status'], 'unavailable') def test_symlink_outside_or_to_other_container_is_refused(self): for destination in ('/etc', '/sys/fs/cgroup/docker/' + 'b' * 64): self.assertEqual(self.observe(resolver=lambda path: destination if path == self.path else path)['cgroupV2']['status'], 'unavailable') result = self.observe(resolver=lambda path: '/etc/passwd' if path.endswith('/memory.current') else path) self.assertIsNone(result['cgroupV2']['metrics']['memory.current']) def test_cgroup_directory_replacement_or_nested_filesystem_is_unknown(self): for nested in (False, True): calls = 0 def statter(path): nonlocal calls if path == self.path: calls += 1 inode = 8 if path == self.path and calls > 1 else 7 device = 9 if nested and path == self.path else 8 return os.stat_result((0, inode, device, 0, 0, 0, 0, 0, 0, 0)) result = self.observe(statter=statter)['cgroupV2'] self.assertEqual(result['status'], 'unavailable') self.assertTrue(all(value is None for value in result['metrics'].values())) def test_pid_reuse_unreadable_process_and_after_oom_stay_unknown(self): after = copy.deepcopy(self.before); after['State']['OOMKilled'] = True result = self.observe(after) host = {'MemTotalBytes': 2048, 'MemAvailableBytes': 1024} summary = measure.summarize([{'elapsedMs': 0, 'host': host, 'containers': [result]}]) self.assertTrue(summary['anyObservedOomKilled']) self.assertFalse(summary['containerIdentityAndStartStable']) self.assertFalse(summary['allCgroupV2MetricsAvailable']) self.assertFalse(summary['acceptancePassed']) def read(path, _expected_device=None, deadline=None): if path == '/proc/123/stat': raise PermissionError() return self.files[path] self.assertEqual(self.observe(reader=read)['cgroupV2']['status'], 'unavailable') def test_real_temp_files_are_bounded_and_strictly_ascii(self): with tempfile.TemporaryDirectory(dir=Path(__file__).resolve().parent) as directory: path = Path(directory) / 'memory.current' path.write_bytes(b'123\n') self.assertEqual(measure.parse_counter(measure.read_bounded(path)), 123) with self.assertRaises(ValueError): measure.read_bounded(path, path.stat().st_dev + 1) for raw in (b'0' * (measure.MAX_FILE_BYTES + 1), b'0\n' * (measure.MAX_FILE_LINES + 1), b'\xff'): path.write_bytes(raw) with self.assertRaises((ValueError, UnicodeError)): measure.read_bounded(path) def test_after_stopped_or_unknown_identity_cannot_summarize_running(self): for value in (False, None): after = copy.deepcopy(self.before); after['State']['Running'] = value result = self.observe(after) self.assertIs(result['runningAfter'], value) summary = measure.summarize([{'elapsedMs': 0, 'host': {'MemTotalBytes': 2048, 'MemAvailableBytes': 1024}, 'containers': [result]}]) self.assertFalse(summary['allObservedRunning']) result = self.observe() summary = measure.summarize([{'elapsedMs': 0, 'host': {'MemTotalBytes': 2048, 'MemAvailableBytes': 1024}, 'containers': [result]}]) self.assertTrue(summary['allObservedRunning']) def test_event_future_counters_retained_but_duplicates_and_overflow_refuse(self): self.assertEqual(measure.parse_events('high 0\nmax 1\nfail 2\nfuture 3\n', ('high', 'max', 'fail'))['future'], 3) for value in ('x 1\nx 2\n', 'x 18446744073709551616\n', ''.join('x_' + str(i) + ' 0\n' for i in range(33))): with self.assertRaises(ValueError): measure.parse_events(value, ()) class DeadlineEvidenceTests(unittest.TestCase): def test_expired_http_never_opens_socket_or_schedules_watchdog(self): with patch.object(measure.time, 'monotonic', return_value=10), \ patch.object(measure.socket, 'socket') as own_socket, patch.object(measure.threading, 'Timer') as timer: with self.assertRaises(measure.ObservationDeadlineExceeded): measure.docker_get('/info', 10) own_socket.assert_not_called() timer.assert_not_called() def test_expired_container_read_never_starts_io(self): with patch.object(measure.time, 'monotonic', return_value=10), patch.object(measure, 'docker_get') as get: with self.assertRaises(measure.ObservationDeadlineExceeded): measure.collect_container('a' * 64, 10) get.assert_not_called() def test_near_cutoff_after_stats_stops_before_proc_or_more_gets(self): clock, queries = [9.9], [] def get(path, deadline): self.assertEqual(deadline, 10) queries.append(path) if len(queries) == 2: clock[0] = 10 return {} with patch.object(measure.time, 'monotonic', side_effect=lambda: clock[0]), \ patch.object(measure, 'docker_get', side_effect=get), patch.object(measure, 'read_bounded') as read: with self.assertRaises(measure.ObservationDeadlineExceeded): measure.collect_container('a' * 64, 10) self.assertEqual(len(queries), 2) read.assert_not_called() def test_metric_deadline_is_not_downgraded_to_partial_or_followed_by_io(self): fixture = CgroupEvidenceTests(); fixture.setUp() clock, reads, queries = [9.9], [], [] def read(path, _device=None, deadline=None): self.assertEqual(deadline, 10) reads.append(path) if path.endswith('/memory.current'): clock[0] = 10 return fixture.files[path] def get(path, deadline): queries.append(path) return fixture.stats if '/stats?' in path else fixture.before with patch.object(measure.time, 'monotonic', side_effect=lambda: clock[0]), \ patch.object(measure, 'docker_get', side_effect=get), patch.object(measure, 'read_bounded', side_effect=read), \ patch.object(measure.os.path, 'realpath', side_effect=lambda path: path), \ patch.object(measure.os, 'stat', return_value=os.stat_result((0, 7, 8, 0, 0, 0, 0, 0, 0, 0))), \ patch.object(measure.os, 'sep', '/'): with self.assertRaises(measure.ObservationDeadlineExceeded): measure.collect_container(fixture.identifier, 10) self.assertEqual(reads[-1], fixture.path + '/memory.current') self.assertEqual(len(queries), 2) def test_final_daemon_check_crosses_deadline_without_writing_report(self): clock, calls = [0], [] def get(path, deadline): self.assertEqual(deadline, 360) calls.append(path) if len(calls) == 2: clock[0] = 361 return {'ID': 'owned-daemon'} host = 'MemTotal: 2048 kB\nMemAvailable: 1024 kB\nSwapTotal: 0 kB\nSwapFree: 0 kB\n' args = ['measure-memory.py', '--daemon-id', 'owned-daemon', '--container', 'a' * 64, '--samples', '1', '--output', 'NEVER_CREATED'] with patch.object(sys, 'argv', args), patch.object(measure.os, 'name', 'posix'), \ patch.object(measure.socket, 'AF_UNIX', 1, create=True), \ patch.object(measure.time, 'monotonic', side_effect=lambda: clock[0]), patch.object(measure, 'docker_get', side_effect=get), \ patch.object(measure, 'read_bounded', side_effect=[host, '', 'some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n']), patch.object(measure, 'collect_container', return_value={}), \ patch.object(measure, 'summarize', return_value={}), patch('builtins.open') as output: with self.assertRaises(measure.ObservationDeadlineExceeded): measure.main() output.assert_not_called() def test_slow_body_absolute_deadline_closes_only_own_socket_and_joins(self): # A real HTTPResponse/BufferedReader and real watchdog timer, with an # in-memory socket transport. No AF_UNIX availability or local network # permissions are needed on Windows. Inactivity timeouts never fire: # the body keeps producing bytes until shutdown interrupts the read. class DripBody(io.RawIOBase): def __init__(self, stopped): self.header = b'HTTP/1.1 200 OK\r\nContent-Length: 1000\r\nConnection: close\r\n\r\n' self.stopped = stopped self.body_reads = 0 def readable(self): return True def readinto(self, buffer): if self.header: value, self.header = self.header[:len(buffer)], self.header[len(buffer):] buffer[:len(value)] = value return len(value) if self.stopped.wait(0.01): return 0 self.body_reads += 1 buffer[0] = 32 return 1 class SocketTransport: def __init__(self): self.stopped = threading.Event() self.body = DripBody(self.stopped) self.closed = False self.request = b'' def connect(self, path): self.path = path def settimeout(self, value): self.timeout = value def sendall(self, value): self.request += value def makefile(self, _mode): return io.BufferedReader(self.body) def shutdown(self, _how): self.stopped.set() def close(self): self.closed = True client, unrelated = SocketTransport(), SocketTransport() started = time.monotonic() original_timer = measure.threading.Timer timers = [] def timer(*args, **kwargs): value = original_timer(*args, **kwargs) timers.append(value) return value with patch.object(measure.socket, 'AF_UNIX', 1, create=True), \ patch.object(measure.socket, 'socket', return_value=client), \ patch.object(measure.threading, 'Timer', side_effect=timer): with self.assertRaises(measure.ObservationDeadlineExceeded): measure.docker_get('/info', started + 0.12) self.assertLess(time.monotonic() - started, 1.5) self.assertTrue(client.closed) self.assertTrue(client.stopped.is_set()) self.assertGreater(client.body.body_reads, 0) self.assertFalse(unrelated.closed) self.assertFalse(unrelated.stopped.is_set()) self.assertEqual(client.path, '/var/run/docker.sock') self.assertIn(b'GET /v1.45/info ', client.request) self.assertEqual(len(timers), 1) self.assertFalse(timers[0].is_alive(), 'No deadline callback can perform late I/O after return') if __name__ == '__main__': unittest.main()