ConfigMap 文件与环境变量有不同生命周期

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-lifetime-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    apply(namespace, {'apiVersion': 'v1', 'kind': 'ConfigMap', 'metadata': {'name': 'settings'},
                      'data': {'mode': 'blue'}})
    script = ('from pathlib import Path; import time; '
              'held=open("/etc/lab/config/mode"); '
              'Path("/tmp/startup").write_text(held.read()); '
              'Path("/tmp/held-fd").write_text(str(held.fileno())); time.sleep(600)')
    podspec = spec(script)
    podspec['containers'][0]['env'] = [{'name': 'MODE', 'valueFrom':
        {'configMapKeyRef': {'name': 'settings', 'key': 'mode'}}}]
    podspec['containers'][0]['volumeMounts'] = [
        {'name': 'settings', 'mountPath': '/etc/lab/config', 'readOnly': True},
        {'name': 'settings', 'mountPath': '/etc/lab/single-mode', 'subPath': 'mode', 'readOnly': True}]
    podspec['volumes'] = [{'name': 'settings', 'configMap': {'name': 'settings'}}]
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'reader'}, 'spec': podspec})
    initial_pod = wait_for(lambda: obj(namespace, 'pod', 'reader'), pod_ready)
    wait_for(lambda: k('exec', '-n', namespace, 'reader', '--', 'python3', '-c',
                      'from pathlib import Path; raise SystemExit(not Path("/tmp/held-fd").exists())',
                      expected=None), lambda result: result.returncode == 0)
    probe = '''from pathlib import Path
import json,os
fd=Path('/tmp/held-fd').read_text()
print(json.dumps({'environment':os.environ['MODE'],
                  'full_volume':Path('/etc/lab/config/mode').read_text(),
                  'subpath':Path('/etc/lab/single-mode').read_text(),
                  'startup_copy':Path('/tmp/startup').read_text(),
                  'held_inode':Path('/proc/1/fd/'+fd).read_text()}))'''
    def observe():
        return json.loads(k('exec', '-n', namespace, 'reader', '--', 'python3', '-c', probe).stdout)
    before = observe()
    assert set(before.values()) == {'blue'}
    k('patch', 'configmap', 'settings', '-n', namespace, '--type=merge',
      '-p', json.dumps({'data': {'mode': 'green'}}))
    after = wait_for(observe, lambda values: values['full_volume'] == 'green', seconds=180)
    assert {v for key, v in after.items() if key != 'full_volume'} == {'blue'}
    final_pod = obj(namespace, 'pod', 'reader')
    assert final_pod['metadata']['uid'] == initial_pod['metadata']['uid']
    assert final_pod['status']['containerStatuses'][0]['restartCount'] == 0
    print(json.dumps({'before': before, 'after': after, 'same_pod_uid': True,
                      'container_restart_count': 0, 'automatic_app_reload_tested': False}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')