普通初始化容器建立启动前置条件

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-init-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    prepare = r'''from pathlib import Path
import sys,time
root=Path('/work'); counter=root/'attempts'
attempt=int(counter.read_text())+1 if counter.exists() else 1
counter.write_text(str(attempt))
with (root/'order').open('a') as f: f.write('prepare'+str(attempt)+'\n')
if attempt == 1: sys.exit(7)
print('waiting for release',flush=True)
while not (root/'go').exists(): time.sleep(0.1)
(root/'input').write_text('ready')
'''
    transform = r'''from pathlib import Path
root=Path('/work'); assert (root/'input').read_text()=='ready'
(root/'output').write_text('READY')
with (root/'order').open('a') as f: f.write('transform\n')
'''
    main = r'''from pathlib import Path
import json,time
root=Path('/work'); assert (root/'output').read_text()=='READY'
with (root/'order').open('a') as f: f.write('app\n')
print(json.dumps({'order':(root/'order').read_text().splitlines(),
                  'artifact':(root/'output').read_text()}),flush=True)
time.sleep(600)
'''
    podspec = spec(main)
    mount = [{'name': 'work', 'mountPath': '/work'}]
    podspec['volumes'] = [{'name': 'work', 'emptyDir': {}}]
    podspec['containers'][0]['volumeMounts'] = mount
    podspec['initContainers'] = []
    for name, script in [('prepare', prepare), ('transform', transform)]:
        container = spec(script)['containers'][0]
        container['name'] = name
        container['volumeMounts'] = mount
        podspec['initContainers'].append(container)
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'ordered'}, 'spec': podspec})
    def retry_running(p):
        statuses = p.get('status', {}).get('initContainerStatuses', [])
        return statuses and statuses[0]['restartCount'] >= 1 and 'running' in statuses[0]['state']
    waiting = wait_for(lambda: obj(namespace, 'pod', 'ordered'), retry_running)
    first = waiting['status']['initContainerStatuses'][0]
    assert first['lastState']['terminated']['exitCode'] == 7
    assert waiting['status']['phase'] == 'Pending'
    assert all('running' not in c['state'] for c in waiting['status'].get('containerStatuses', []))
    assert 'running' not in waiting['status']['initContainerStatuses'][1]['state']
    k('exec', '-n', namespace, 'ordered', '-c', 'prepare', '--', 'python3', '-c',
      'from pathlib import Path; Path("/work/go").touch()')
    ready = wait_for(lambda: obj(namespace, 'pod', 'ordered'), pod_ready)
    assert ready['metadata']['uid'] == waiting['metadata']['uid']
    output = wait_for(lambda: k('logs', '-n', namespace, 'ordered', '-c', 'app').stdout.strip(), bool)
    observed = json.loads(output)
    assert observed == {'order': ['prepare1', 'prepare2', 'transform', 'app'], 'artifact': 'READY'}
    assert ready['status']['initContainerStatuses'][0]['restartCount'] == 1
    assert all(c['state']['terminated']['exitCode'] == 0 for c in ready['status']['initContainerStatuses'])
    never = spec()
    never['restartPolicy'] = 'Never'
    failed_init = spec('raise SystemExit(7)')['containers'][0]
    failed_init['name'] = 'fails'
    never['initContainers'] = [failed_init]
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'never'}, 'spec': never})
    failed = wait_for(lambda: obj(namespace, 'pod', 'never'), lambda p: p.get('status', {}).get('phase') == 'Failed')
    status = failed['status']['initContainerStatuses'][0]
    assert status['restartCount'] == 0 and status['state']['terminated']['exitCode'] == 7
    assert not any('running' in c['state'] for c in failed['status'].get('containerStatuses', []))
    print(json.dumps({'ordered_artifact': observed, 'retried_init_count': 1,
                      'app_waited_for_both_regular_init_containers': True,
                      'never_policy_failed_pod_without_retry': True}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')