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'}}}]}


def deployment(namespace, name, podspec):
    return apply(namespace, {'apiVersion': 'apps/v1', 'kind': 'Deployment', 'metadata': {'name': name},
        'spec': {'replicas': 1, 'selector': {'matchLabels': {'app': name}},
                 'template': {'metadata': {'labels': {'app': name}}, 'spec': podspec}}})


def deployment_pod(namespace, name):
    wait_for(lambda: obj(namespace, 'deployment', name), lambda d:
             d.get('status', {}).get('observedGeneration', 0) >= d['metadata']['generation']
             and d.get('status', {}).get('updatedReplicas') == 1
             and d.get('status', {}).get('availableReplicas') == 1
             and d.get('status', {}).get('replicas') == 1)
    def pods():
        return [p for p in obj(namespace, 'pods', selector='app=' + name)['items']
                if not p['metadata'].get('deletionTimestamp')]
    return wait_for(pods, lambda p: len(p) == 1 and pod_ready(p[0]))[0]

namespace = 'notes-config-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    for version, mode in [('v1', 'blue'), ('v2', 'green')]:
        apply(namespace, {'apiVersion': 'v1', 'kind': 'ConfigMap',
              'metadata': {'name': 'settings-' + version}, 'immutable': True, 'data': {'mode': mode}})
    podspec = spec()
    podspec['containers'][0]['env'] = [{'name': 'MODE', 'valueFrom':
        {'configMapKeyRef': {'name': 'settings-v1', 'key': 'mode'}}}]
    deployment(namespace, 'web', podspec)
    first = deployment_pod(namespace, 'web')
    def mode(pod):
        return k('exec', '-n', namespace, pod['metadata']['name'], '--', 'python3', '-c',
                 'import os; print(os.environ["MODE"])').stdout.strip()
    assert mode(first) == 'blue'
    rejected = k('patch', 'configmap', 'settings-v1', '-n', namespace, '--type=merge',
                 '-p', json.dumps({'data': {'mode': 'changed'}}), expected=None)
    assert rejected.returncode != 0 and 'immutable' in rejected.stderr
    assert obj(namespace, 'configmap', 'settings-v1')['data']['mode'] == 'blue'
    def select(version):
        patch = [{'op': 'replace',
                  'path': '/spec/template/spec/containers/0/env/0/valueFrom/configMapKeyRef/name',
                  'value': 'settings-' + version}]
        k('patch', 'deployment', 'web', '-n', namespace, '--type=json', '-p', json.dumps(patch))
        return deployment_pod(namespace, 'web')
    second = select('v2')
    assert mode(second) == 'green' and second['metadata']['uid'] != first['metadata']['uid']
    third = select('v1')
    assert mode(third) == 'blue' and third['metadata']['uid'] != second['metadata']['uid']
    image_ids = [p['status']['containerStatuses'][0]['imageID'] for p in (first, second, third)]
    assert len(set(image_ids)) == 1
    assert len({p['metadata']['uid'] for p in (first, second, third)}) == 3
    print(json.dumps({'modes': ['blue', 'green', 'blue'], 'distinct_pod_uids': 3,
                      'same_image_id': image_ids[0], 'immutable_update_rejected': True,
                      'retained_config_rollback_verified': True}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')