cgroup 在层级中执行容器资源预算

import json
import subprocess
import sys
import time
from uuid import uuid4

assert len(sys.argv) == 3, 'usage: python3 example.py DISPOSABLE_KUBECONFIG CONTEXT'
base = ['kubectl', '--kubeconfig', sys.argv[1], '--context', sys.argv[2], '--request-timeout=30s']
image = 'python:3.11-slim-bookworm'


def k(*args, payload=None, expected=0):
    result = subprocess.run(base + list(args), input=json.dumps(payload) if payload is not None else None,
                            capture_output=True, text=True, timeout=150)
    if expected is not None:
        assert result.returncode == expected, result.stderr
    return result


def obj(namespace, kind, name=None, selector=None):
    args = ['get', kind, '-n', namespace, '-o', 'json']
    if name:
        args.append(name)
    if selector:
        args += ['-l', selector]
    return json.loads(k(*args).stdout)


def apply(namespace, data):
    return k('apply', '-n', namespace, '-f', '-', payload=data)


def wait_for(read, ready, seconds=120):
    until = time.monotonic() + seconds
    while time.monotonic() < until:
        value = read()
        if ready(value):
            return value
        time.sleep(0.5)
    raise AssertionError('expected Kubernetes condition did not arrive')


def pod_ready(pod):
    return any(c['type'] == 'Ready' and c['status'] == 'True'
               for c in pod.get('status', {}).get('conditions', []))


def spec(script='import time; time.sleep(600)'):
    return {'automountServiceAccountToken': False, 'terminationGracePeriodSeconds': 1,
            'containers': [{'name': 'app', 'image': image, 'imagePullPolicy': 'IfNotPresent',
                            'command': ['python3', '-c', script],
                            'resources': {'requests': {'cpu': '10m', 'memory': '24Mi'},
                                          'limits': {'cpu': '200m', 'memory': '96Mi'}}}]}

namespace = 'notes-cgroup-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    podspec = spec()
    podspec['containers'] = []
    for name, request, limit, memory in [('small', '25m', '100m', '64Mi'), ('large', '50m', '200m', '96Mi')]:
        container = spec()['containers'][0]
        container['name'] = name
        container['resources'] = {'requests': {'cpu': request, 'memory': '24Mi'},
                                  'limits': {'cpu': limit, 'memory': memory}}
        podspec['containers'].append(container)
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'budgets'}, 'spec': podspec})
    wait_for(lambda: obj(namespace, 'pod', 'budgets'), pod_ready)
    snapshot = '''from pathlib import Path
import json,os
p=Path('/sys/fs/cgroup')
assert (p/'cgroup.controllers').exists(), 'this example requires cgroup v2'
print(json.dumps({'cpu_max':(p/'cpu.max').read_text().strip(),
                  'cpu_weight':int((p/'cpu.weight').read_text()),
                  'memory_max':int((p/'memory.max').read_text()),
                  'group_inode':os.stat(p).st_ino}))
'''
    observations = {}
    for name, expected_cpu, expected_memory in [('small', 0.1, 64), ('large', 0.2, 96)]:
        data = json.loads(k('exec', '-n', namespace, 'budgets', '-c', name, '--', 'python3', '-c', snapshot).stdout)
        quota, period = map(int, data['cpu_max'].split())
        assert quota / period == expected_cpu
        assert data['memory_max'] == expected_memory * 1024 * 1024
        observations[name] = data
    assert observations['small']['group_inode'] != observations['large']['group_inode']
    load = '''from pathlib import Path
import json,time
p=Path('/sys/fs/cgroup/cpu.stat')
def counters(): return {k:int(v) for k,v in (line.split() for line in p.read_text().splitlines())}
before=counters(); begin=time.monotonic(); cpu=time.process_time(); value=0
while time.monotonic()-begin < 3: value=(value+1)%1000003
after=counters()
print(json.dumps({'wall_seconds':time.monotonic()-begin,'process_cpu_seconds':time.process_time()-cpu,
                  'nr_throttled_delta':after['nr_throttled']-before['nr_throttled'],
                  'throttled_usec_delta':after['throttled_usec']-before['throttled_usec']}))
'''
    throttling = json.loads(k('exec', '-n', namespace, 'budgets', '-c', 'small', '--', 'python3', '-c', load).stdout)
    assert throttling['nr_throttled_delta'] > 0 and throttling['throttled_usec_delta'] > 0
    final = obj(namespace, 'pod', 'budgets')
    assert pod_ready(final) and all(c['restartCount'] == 0 for c in final['status']['containerStatuses'])
    assert 'resources' not in final['spec']
    print(json.dumps({'container_groups': observations, 'measured_throttling': throttling,
                      'container_restarts': 0, 'pod_level_resource_field_used': False}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')