资源限制约束运行，请求参与放置

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-memory-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    allocator = '''from pathlib import Path
import json,time
group=Path('/sys/fs/cgroup')
print(json.dumps({'memory_max':(group/'memory.max').read_text().strip(),
                  'initial_memory_current':int((group/'memory.current').read_text())}),flush=True)
while not Path('/tmp/go').exists(): time.sleep(0.1)
chunks=[]
for i in range(16):
 chunks.append(bytearray(8*1024*1024))
 print('allocated_chunks='+str(len(chunks)),flush=True)
 time.sleep(0.05)
raise RuntimeError('bounded allocation unexpectedly escaped the expected limit')
'''
    constrained = spec(allocator)
    constrained['restartPolicy'] = 'Never'
    constrained['containers'][0]['resources'] = {
        'requests': {'cpu': '10m', 'memory': '32Mi'}, 'limits': {'cpu': '200m', 'memory': '64Mi'}}
    for name, podspec in [('limited', constrained), ('neighbor', spec())]:
        apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': name}, 'spec': podspec})
        wait_for(lambda name=name: obj(namespace, 'pod', name), pod_ready)
    first_log = wait_for(lambda: k('logs', '-n', namespace, 'limited').stdout.strip(), bool)
    before = json.loads(first_log)
    assert before['memory_max'] == str(64 * 1024 * 1024)
    k('exec', '-n', namespace, 'limited', '--', 'python3', '-c',
      'from pathlib import Path; Path("/tmp/go").touch()')
    failed = wait_for(lambda: obj(namespace, 'pod', 'limited'), lambda p: p.get('status', {}).get('phase') == 'Failed')
    status = failed['status']['containerStatuses'][0]
    terminated = status['state']['terminated']
    assert terminated['reason'] == 'OOMKilled' and terminated['exitCode'] == 137
    assert status['restartCount'] == 0
    assert failed['status'].get('reason') != 'Evicted'
    neighbor = obj(namespace, 'pod', 'neighbor')
    assert pod_ready(neighbor) and neighbor['status']['containerStatuses'][0]['restartCount'] == 0
    output = k('logs', '-n', namespace, 'limited').stdout.splitlines()
    chunks = [int(line.split('=')[1]) for line in output if line.startswith('allocated_chunks=')]
    assert chunks and max(chunks) < 16
    print(json.dumps({'before': before, 'last_reported_allocated_chunks': max(chunks),
                      'termination_reason': terminated['reason'], 'exit_code': terminated['exitCode'],
                      'container_restarts_with_never_policy': 0,
                      'pod_was_not_evicted': True, 'neighbor_still_ready': True}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')