#!/usr/bin/env python3 """U33 公开入口：真实 loopback HTTP 与真实 Resilience4j 对象的编译测试，记录一次完整成功。 两个模块都不依赖外部服务。feign 模块启动进程内 HttpServer 并用 Feign/Apache HC5 发真实请求；resilience 模块用真实状态机、JDK HttpClient 与第二个 loopback 上游。 入口只运行本仓库测试，结束时回收自己创建的进程组。 """ import glob import hashlib import json import os import pathlib import platform import re import signal import subprocess import sys import time import tempfile import xml.etree.ElementTree as ET from datetime import datetime ROOT = pathlib.Path(__file__).resolve().parent EXPECTED = { 'study.remoteclient.feign.FeignClientLabTest': 13, 'study.remoteclient.feign.FeignConfigurationTest': 6, 'study.remoteclient.feign.HttpServiceClientLabTest': 1, 'study.remoteclient.resilience.BulkheadLabTest': 1, 'study.remoteclient.resilience.CircuitBreakerStateMachineTest': 1, 'study.remoteclient.resilience.DecorationOrderTest': 2, 'study.remoteclient.resilience.DeadlineCompositionHttpTest': 1, 'study.remoteclient.resilience.DurableRecoveryHttpTest': 2, 'study.remoteclient.resilience.ResiliencePolicyTest': 2, 'study.remoteclient.resilience.IdempotentRecoveryHttpTest': 3, 'study.remoteclient.resilience.SpringCloudCircuitBreakerLabTest': 1, 'study.remoteclient.resilience.TimeLimiterLabTest': 2, } MARKERS = [ 'U33 feignProxy=', 'U33 feignBinding=', 'U33 feignEncoder=', 'U33 feignContract=', 'U33 feignErrors=', 'U33 feignHiddenRetry=', 'U33 feignPostStatusRetry=', 'U33 feignRetryer=', 'U33 feignPool=', 'U33 feignHiddenRetrySuppressed=', 'U33 feignReadTimeout=', 'U33 feignRetryWrite=', 'U33 feignConnect=', 'U33 httpServiceClient=', 'U33 cbStateMachine=', 'U33 scbFallback=', 'U33 decorationOrder=', 'U33 bulkhead=', 'U33 timeLimiter=', 'U33 timeLimiterFuture=', 'U33 idempotentRecovery=', 'U33 idempotentStatus=', 'U33 feignOptionalDecoder=', 'U33 feignResponseEntityDecoder=', 'U33 feignCustomErrorDecoder=', 'U33 feignExplicitPool=', 'U33 feignConfiguredNoHiddenRetry=', 'U33 feignConfiguredTimeout=', 'U33 durableRecovery=', 'U33 durableConcurrent=', 'U33 deadlineComposition=', 'U33 timeWindow=', 'U33 retryPolicy=', ] DIGEST_INCLUDE = { 'spring-cloud-openfeign-core', 'feign-core', 'feign-hc5', 'httpclient5', 'spring-cloud-circuitbreaker-resilience4j', 'spring-cloud-circuitbreaker', 'resilience4j-circuitbreaker', 'resilience4j-retry', 'resilience4j-bulkhead', 'resilience4j-timelimiter', 'spring-boot', 'spring-web', 'h2', } def maven_binary(): found = os.environ.get('MAVEN_BIN') or subprocess.run( ['/usr/bin/which', 'mvn'], capture_output=True, text=True).stdout.strip() if found: return found candidates = sorted(glob.glob(os.path.expanduser( '~/.m2/wrapper/dists/apache-maven-3.9.9/*/bin/mvn'))) if candidates: return candidates[-1] raise SystemExit('需要 Maven 3.9.9：设置 MAVEN_BIN 或在 PATH 中提供 mvn') def interrupted(signum, _frame): raise SystemExit(128 + signum) signal.signal(signal.SIGTERM, interrupted) signal.signal(signal.SIGINT, interrupted) jdk = pathlib.Path(os.environ.get('JAVA_HOME') or subprocess.check_output( ['/usr/libexec/java_home', '-v', '21'], text=True).strip()) env = dict(os.environ, JAVA_HOME=str(jdk), PATH=str(jdk / 'bin') + os.pathsep + os.environ.get('PATH', '')) logs_root = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u33' logs_root.mkdir(parents=True, exist_ok=True) logs = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=logs_root)) started = datetime.now().astimezone().isoformat() results = [] mvn = maven_binary() def live_group_pids(pgid): # killpg(..., 0) also sees zombies; only non-zombie members require cleanup. snapshot = subprocess.run(['/bin/ps', '-axo', 'pid=,pgid=,stat='], capture_output=True, text=True, check=True, timeout=5) return [int(pid) for pid, group, state in (line.split() for line in snapshot.stdout.splitlines()) if int(group) == pgid and not state.startswith('Z')] def cleanup_group(process): # The leader can exit before its descendants, even on an ordinary exit=0. for signum, grace in ((signal.SIGTERM, 2), (signal.SIGKILL, 5)): process.poll() # Reap our child; orphan zombies belong to their reaper. remaining = live_group_pids(process.pid) if not remaining: return try: os.killpg(process.pid, signum) except ProcessLookupError: return deadline = time.monotonic() + grace while time.monotonic() < deadline: process.poll() if not live_group_pids(process.pid): return time.sleep(0.05) remaining = live_group_pids(process.pid) if remaining: raise RuntimeError('进程组仍有存活进程: ' + str(remaining)) def run(name, args, timeout=600, cwd=ROOT): log_path = logs / (name + '.log') process = None failure = None cleanup_error = None exit_code = 0 try: # File output cannot be held open as a PIPE by a surviving descendant. with log_path.open('w') as stream: try: process = subprocess.Popen(args, cwd=cwd, env=env, text=True, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True) return_code = process.wait(timeout=timeout) exit_code = return_code if return_code >= 0 else 128 - return_code except BaseException as exc: failure = exc if isinstance(exc, subprocess.TimeoutExpired): exit_code = 124 elif isinstance(exc, SystemExit) and isinstance(exc.code, int): exit_code = exc.code elif isinstance(exc, KeyboardInterrupt): exit_code = 130 else: exit_code = 1 finally: # A second signal must not abort bounded cleanup/log persistence. handlers = {sig: signal.signal(sig, signal.SIG_IGN) for sig in (signal.SIGINT, signal.SIGTERM)} try: if process is not None: cleanup_group(process) except Exception as exc: cleanup_error = str(exc) exit_code = exit_code or 1 finally: for sig, handler in handlers.items(): signal.signal(sig, handler) except OSError as exc: failure = exc exit_code = 1 output = log_path.read_text(errors='replace') if log_path.exists() else '' result = {'name': name, 'command': args, 'exitCode': process.returncode if process is not None else None, 'runnerExitCode': exit_code, 'timedOut': isinstance(failure, subprocess.TimeoutExpired), 'failure': str(failure) if failure is not None else None, 'cleanupError': cleanup_error, 'observed': [line for line in output.splitlines() if line.startswith('U33')]} results.append(result) (logs / (name + '.status.json')).write_text( json.dumps(result, ensure_ascii=False, indent=1) + '\n') if exit_code: print(name + ' exit=' + str(exit_code) + '; raw log: ' + str(log_path) + ('; cleanup: ' + cleanup_error if cleanup_error else '') + '\n' + output[-3500:], file=sys.stderr, flush=True) raise SystemExit(exit_code) print('PASS ' + name, flush=True) return output def validate_reports(reports): by_class = {} for report in reports: name = report['name'] if name in by_class: raise ValueError('重复 Surefire 报告: ' + name) for field in ('failures', 'errors', 'skipped'): if int(report[field]) != 0: raise ValueError(name + ' ' + field + '=' + report[field]) by_class[name] = int(report['tests']) if by_class != EXPECTED: raise ValueError('Surefire 计数不匹配: ' + str(by_class) + '; expected=' + str(EXPECTED)) test_count = sum(by_class.values()) if test_count != 35: raise ValueError('测试总数应为 35: ' + str(test_count)) return test_count version = run('jdk', [str(jdk / 'bin/java'), '-version']) assert re.search(r'version "21[.]', version), version maven_version = run('maven', [mvn, '-version']) reports = [] for module in ('feign', 'resilience'): run('test-' + module, [mvn, '-B', '-ntp', '-q', '-f', str(ROOT / module / 'pom.xml'), 'clean', 'test']) for path in sorted((ROOT / module / 'target/surefire-reports').glob('TEST-*.xml')): root = ET.parse(path).getroot() reports.append({key: root.attrib[key] for key in ['name', 'tests', 'failures', 'errors', 'skipped']}) (logs / 'reports.json').write_text(json.dumps(reports, ensure_ascii=False, indent=1) + '\n') try: test_count = validate_reports(reports) except (KeyError, ValueError) as exc: (logs / 'validation-failure.json').write_text( json.dumps({'failure': str(exc), 'exitCode': 1}, ensure_ascii=False) + '\n') raise SystemExit('Surefire 验证失败: ' + str(exc) + '; reports: ' + str(logs / 'reports.json')) def dependencies(module): tree = run('dependency-' + module, [mvn, '-B', '-ntp', '-q', '-f', str(ROOT / module / 'pom.xml'), 'dependency:list', '-DoutputFile=/dev/stdout']) versions = {} for line in tree.splitlines(): match = re.match(r'\s*[\w.\-]+:([\w.\-]+):(?:jar|pom):([\w.\-]+):', line) if match and match.group(1) in DIGEST_INCLUDE: versions[match.group(1)] = match.group(2) return versions dependencies_by_module = {module: dependencies(module) for module in ('feign', 'resilience')} observed = '\n'.join(line for result in results for line in result['observed']) for marker in MARKERS: assert marker in observed, '缺少观察行 ' + marker tracked = [ROOT / 'pom.xml', ROOT / 'run.py', ROOT / 'README.md'] tracked += sorted(ROOT.glob('*/pom.xml')) tracked += sorted(ROOT.glob('*/src/**/*.java')) digests = {} for path in tracked: if path.exists(): digests[str(path.relative_to(ROOT))] = hashlib.sha256(path.read_bytes()).hexdigest() evidence = { 'ticket': 'U33 #70', 'startedAt': started, 'finishedAt': datetime.now().astimezone().isoformat(), 'input': 'python3 labs/remote-client/run.py', 'exitCode': 0, 'environment': { 'jdk': version.strip(), 'maven': maven_version.strip(), 'os': platform.platform(), 'arch': platform.machine(), 'python': platform.python_version(), 'dependencies': dependencies_by_module, }, 'reports': reports, 'testCount': test_count, 'results': results, 'artifacts': digests, 'logs': str(logs), 'cleanup': { 'processes': '本入口创建的进程组已回收；HttpServer、测试 JVM 与 JDK HttpClient 均在用例内关闭', 'externalServices': '无外部服务；只使用 127.0.0.1 随机端口与进程内合成数据', }, 'limits': [ 'loopback HTTP 与受控上游；无真实注册中心、无 LB 集群、无生产网关', '未启用 TLS/mTLS、跨进程网络分区或容量压测', '未运行 OpenFeign 内建 CircuitBreaker 集成、Sentinel 组合；SCB默认线程池隔离舱经过调用但未专项验证容量/排队/拒绝', 'H2持久恢复仅本机优雅关闭重开与显式事务回滚；无OS进程强杀、断电、多实例租约与外部业务效果原子性证明', 'HC5连接池1连接/100ms获取等待已验证；keep-alive、生产容量/TLS/代理重发未运行', '时间窗口使用可控Clock验证秒桶淘汰，慢调用与退避用真实耗时；共享deadline只在本进程，不代表跨服务传播', '未运行 Micrometer/Prometheus 导出、Feign 客户端尝试指标与告警', 'R4j 默认窗口/默认 Retry 间隔等默认值只做源码核对，实验使用自定义小参数', ], } out = ROOT / 'evidence' / ('u33-' + datetime.now().strftime('%Y-%m-%d') + '.json') out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(evidence, ensure_ascii=False, indent=1) + '\n') print('WROTE ' + str(out.relative_to(ROOT)) + ' tests=' + str(test_count), flush=True)