启动探针成功后，其他探针才开始执行

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-startup-' + uuid4().hex[:8]
k('create', 'namespace', namespace)
try:
    server = '''from http.server import BaseHTTPRequestHandler,HTTPServer
from pathlib import Path
import json
counts={'startup':0,'live':0,'ready':0}
class Handler(BaseHTTPRequestHandler):
 def do_GET(self):
  key=self.path.removeprefix('/'); code=200
  if key in counts:
   counts[key]+=1
   if key=='startup' and not Path('/tmp/start').exists(): code=503
  body=json.dumps(counts).encode(); self.send_response(code)
  self.send_header('Content-Length',str(len(body))); self.end_headers(); self.wfile.write(body)
 def log_message(self,*args): pass
HTTPServer(('0.0.0.0',8080),Handler).serve_forever()
'''
    podspec = spec(server)
    c = podspec['containers'][0]
    c['startupProbe'] = {'httpGet': {'path': '/startup', 'port': 8080}, 'periodSeconds': 1, 'failureThreshold': 60}
    for field, path in [('livenessProbe', '/live'), ('readinessProbe', '/ready')]:
        c[field] = {'httpGet': {'path': path, 'port': 8080}, 'periodSeconds': 1, 'failureThreshold': 2}
    apply(namespace, {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': 'slow'}, 'spec': podspec})
    wait_for(lambda: obj(namespace, 'pod', 'slow'), lambda p:
             any('running' in c['state'] for c in p.get('status', {}).get('containerStatuses', [])))
    read = 'from urllib.request import urlopen; print(urlopen("http://127.0.0.1:8080/stats",timeout=3).read().decode())'
    def counts():
        result = k('exec', '-n', namespace, 'slow', '--', 'python3', '-c', read, expected=None)
        return json.loads(result.stdout) if result.returncode == 0 else None
    before = wait_for(counts, lambda d: d and d['startup'] >= 3)
    assert before['live'] == before['ready'] == 0
    starting = obj(namespace, 'pod', 'slow')
    assert not pod_ready(starting)
    assert starting['status']['containerStatuses'][0]['started'] is False
    assert starting['status']['containerStatuses'][0]['restartCount'] == 0
    k('exec', '-n', namespace, 'slow', '--', 'python3', '-c',
      'from pathlib import Path; Path("/tmp/start").touch()')
    ready = wait_for(lambda: obj(namespace, 'pod', 'slow'), pod_ready)
    assert ready['status']['containerStatuses'][0]['started'] is True
    enabled = wait_for(counts, lambda d: d and d['live'] > 0 and d['ready'] > 0)
    time.sleep(3)
    after = counts()
    assert after['startup'] == enabled['startup']
    assert after['live'] > enabled['live'] and after['ready'] > enabled['ready']
    assert obj(namespace, 'pod', 'slow')['status']['containerStatuses'][0]['restartCount'] == 0
    print(json.dumps({'before_startup_success': before, 'after_startup_success': after,
                      'other_probes_suppressed_before_startup': True,
                      'startup_probe_stops_after_success': True, 'container_restarts': 0}))
finally:
    k('delete', 'namespace', namespace, '--wait=true', '--timeout=90s')