#!/usr/bin/env python3 """U37 公开入口：Sentinel 1.8.9 纯 JVM 规则语义、数据源、注解与 SCA starter 实验。 不依赖 Dashboard、Nacos、Redis 或任何外部服务：全部 12 个测试运行在单个 Surefire fork 内，通过 Sentinel 1.8.9 的固定源码语义（slot 链、滑动窗口、 熔断状态机、参数流控、System/Authority 规则、FileRefreshableDataSource、 SentinelResourceAspect、spring-cloud-starter-alibaba-sentinel 2025.1.0.0 在 Boot 4.0.9 上的最小上下文）观察行为。入口只运行本仓库测试，结束时回收 自己创建的进程组并清理 Sentinel 日志目录。 """ import glob import hashlib import json import os import pathlib import platform import re import shutil import signal import subprocess import sys import tempfile import time import xml.etree.ElementTree as ET from datetime import datetime ROOT = pathlib.Path(__file__).resolve().parent EVIDENCE = ROOT / 'evidence/u37-2026-10-28.json' EXPECTED_CLASSES = { 'study.sentinel.RuleBehaviorTest': [ 'entryAndContextPairing', 'slotChainOrderAndSilentOverflow', 'slidingWindowAndTracerAccounting', 'flowQpsFastRejectAndPrioritized', 'threadIsolationWithoutQueueing', 'warmUpRampAndThrottlingSpacing', 'degradeStrategiesAndHalfOpenProbe', 'paramFlowItemThresholdsAndBurst', 'systemRuleInOnlyAndAuthorityList', ], 'study.sentinel.DataSourceAndAnnotationTest': [ 'fileDataSourceHotReload', 'sentinelResourceAnnotation', ], 'study.sentinel.StarterIntegrationTest': [ 'starterDependencyResolution', ], } EXPECTED_TEST_COUNT = 12 MARKERS = [ 'U37 entryContext=PASS', 'U37 slotChain=PASS', 'U37 slidingWindow=PASS', 'U37 flowQps=PASS', 'U37 flowThread=PASS', 'U37 warmupThrottle=PASS', 'U37 degrade=PASS', 'U37 paramFlow=PASS', 'U37 systemAuthority=PASS', 'U37 fileDatasource=PASS', 'U37 annotation=PASS', ] STARTER_PREFIX = 'U37 starterBoot4=' STARTER_VALUES = {'PASS', 'INCOMPATIBLE-DEDUCED'} EXPECTED_VERSIONS = { 'com.alibaba.csp:sentinel-core': '1.8.9', 'com.alibaba.csp:sentinel-parameter-flow-control': '1.8.9', 'com.alibaba.csp:sentinel-datasource-extension': '1.8.9', 'com.alibaba.csp:sentinel-annotation-aspectj': '1.8.9', 'org.aspectj:aspectjweaver': '1.9.25.1', 'com.alibaba.cloud:spring-cloud-starter-alibaba-sentinel': '2025.1.0.0', 'com.alibaba.cloud:spring-cloud-circuitbreaker-sentinel': '2025.1.0.0', 'org.springframework.boot:spring-boot': '4.0.9', } # 1.8.6 是 ~/.m2 里最容易被误用的旧版本；所有 sentinel 工件必须精确为 1.8.9。 SENTINEL_GROUP = 'com.alibaba.csp:sentinel-' SENTINEL_VERSION = '1.8.9' def now(): return datetime.now().astimezone().isoformat() def write_json(path, value): """Replace the evidence atomically; a failed run must not leave old success.""" path.parent.mkdir(parents=True, exist_ok=True) temporary = None try: with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent, prefix=path.name + '.', delete=False) as stream: temporary = pathlib.Path(stream.name) json.dump(value, stream, ensure_ascii=False, indent=2) stream.write('\n') temporary.replace(path) finally: if temporary is not None: temporary.unlink(missing_ok=True) def file_sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def artifact_digests(): paths = [ROOT / name for name in ('pom.xml', 'run.py', 'README.md')] paths += sorted(path for path in (ROOT / 'src').rglob('*') if path.is_file()) return {str(path.relative_to(ROOT)): file_sha(path) for path in paths} def maven_binary(): supplied = os.environ.get('MAVEN_BIN') if supplied: return shutil.which(supplied) or supplied found = shutil.which('mvn') 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 RuntimeError('需要 Maven 3.9.x：设置 MAVEN_BIN 或在 PATH 中提供 mvn') def interrupted(signum, _frame): raise SystemExit(128 + signum) def live_group_pids(pgid): # killpg(..., 0) also sees zombies; they cannot hold ports or do more work. snapshot = subprocess.run(['/bin/ps', '-axo', 'pid=,pgid=,stat='], capture_output=True, text=True, check=True, timeout=5) members = [] for line in snapshot.stdout.splitlines(): pid, group, state = line.split() if int(group) == pgid and not state.startswith('Z'): members.append(int(pid)) return members def cleanup_group(process): # The leader can exit before its descendants, including on exit=0. for signum, grace in ((signal.SIGTERM, 2), (signal.SIGKILL, 5)): process.poll() if not live_group_pids(process.pid): 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, logs, env, results, timeout=600): log_path = logs / (name + '.log') started = now() process = None failure = None cleanup_error = None exit_code = 0 try: # A surviving descendant cannot keep a PIPE reader waiting indefinitely. with log_path.open('w', encoding='utf-8') as stream: try: process = subprocess.Popen(args, cwd=ROOT, env=env, text=True, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True) code = process.wait(timeout=timeout) exit_code = code if code >= 0 else 128 - code except BaseException as exc: failure = exc exit_code = failure_code(exc) finally: # A second signal cannot interrupt bounded cleanup/status writing. 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(encoding='utf-8', errors='replace') if log_path.exists() else '' result = { 'name': name, 'command': args, 'cwd': str(ROOT), 'startedAt': started, 'finishedAt': now(), '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, 'log': str(log_path), 'logSha256': file_sha(log_path) if log_path.exists() else None, 'observed': [line for line in output.splitlines() if line.startswith('U37 ')], } results.append(result) write_json(logs / (name + '.status.json'), result) 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('DONE ' + name + ' exit=0; log: ' + str(log_path), flush=True) return output def failure_code(exc): if isinstance(exc, subprocess.TimeoutExpired): return 124 if isinstance(exc, SystemExit) and isinstance(exc.code, int) and exc.code: return exc.code if isinstance(exc, KeyboardInterrupt): return 130 return 1 def validate_reports(logs, not_before_ns): report_dir = ROOT / 'target/surefire-reports' paths = sorted(report_dir.glob('TEST-*.xml')) expected_files = sorted('TEST-' + name + '.xml' for name in EXPECTED_CLASSES) if [path.name for path in paths] != expected_files: raise ValueError('Surefire 报告缺失或多出报告: ' + str([path.name for path in paths]) + '; expected=' + str(expected_files)) reports = [] for path in paths: if path.stat().st_mtime_ns < not_before_ns: raise ValueError('Surefire 报告不是本次测试生成: ' + str(path)) shutil.copyfile(path, logs / path.name) suite = ET.parse(path).getroot() class_name = suite.get('name') if suite.tag != 'testsuite' or class_name not in EXPECTED_CLASSES: raise ValueError('Surefire testsuite 名称不匹配: ' + str(class_name)) report = {key: suite.attrib[key] for key in ('name', 'tests', 'failures', 'errors', 'skipped')} for field in ('failures', 'errors', 'skipped'): if int(report[field]) != 0: raise ValueError(class_name + ' ' + field + '=' + report[field]) cases = suite.findall('testcase') names = [case.attrib['name'] for case in cases] expected = EXPECTED_CLASSES[class_name] if (len(expected) != int(report['tests']) or len(cases) != len(expected) or len(set(names)) != len(names) or set(names) != set(expected)): raise ValueError(class_name + ' Surefire 测试名/数量不匹配: ' + str(names) + '; expected=' + str(sorted(expected))) for case in cases: if case.get('classname') != class_name or any( case.find(tag) is not None for tag in ('failure', 'error', 'skipped')): raise ValueError('Surefire 用例失败、跳过或类名不匹配: ' + case.attrib['name']) report['testNames'] = names report['sha256'] = file_sha(path) reports.append(report) write_json(logs / 'reports.json', reports) return reports def validate_observations(observed): for marker in MARKERS: matches = [line for line in observed if line == marker] if len(matches) != 1: raise ValueError('观察行必须出现一次: ' + marker + '; actual=' + str(len(matches))) starter = [line for line in observed if line.startswith(STARTER_PREFIX)] if len(starter) != 1: raise ValueError('starterBoot4 观察行必须唯一: ' + str(starter)) value = starter[0][len(STARTER_PREFIX):] if value not in STARTER_VALUES: raise ValueError('starterBoot4 取值非法: ' + value) expected_lines = len(MARKERS) + 1 if len(observed) != expected_lines: raise ValueError('存在未登记或重复的 U37 观察行: ' + str(observed)) return value def read_dependencies(path): versions = {} for line in path.read_text(encoding='utf-8').splitlines(): fields = line.strip().split(' ', 1)[0].split(':') if len(fields) not in (5, 6) or fields[2] not in ('jar', 'pom', 'test-jar'): continue if fields[-1] not in ('compile', 'provided', 'runtime', 'test', 'system'): continue coordinate = ':'.join(fields[:2]) version = fields[-2] if coordinate in versions and versions[coordinate] != version: raise ValueError('依赖版本冲突: ' + coordinate) versions[coordinate] = version for coordinate, expected in EXPECTED_VERSIONS.items(): if versions.get(coordinate) != expected: raise ValueError('依赖版本不匹配: ' + coordinate + '=' + str(versions.get(coordinate)) + '; expected=' + expected) sentinel_versions = {coordinate: version for coordinate, version in versions.items() if coordinate.startswith(SENTINEL_GROUP)} wrong = {coordinate: version for coordinate, version in sentinel_versions.items() if version != SENTINEL_VERSION} if wrong: raise ValueError('Sentinel 工件必须全部为 ' + SENTINEL_VERSION + '（严禁 1.8.6）: ' + str(wrong)) return versions def validate_pom(): pom = ET.parse(ROOT / 'pom.xml').getroot() ns = {'m': 'http://maven.apache.org/POM/4.0.0'} expected = { 'm:parent/m:version': '4.0.9', 'm:properties/m:java.version': '21', 'm:properties/m:spring-cloud.version': '2025.1.4', 'm:properties/m:spring-cloud-alibaba.version': '2025.1.0.0', 'm:properties/m:sentinel.version': '1.8.9', 'm:properties/m:aspectjweaver.version': '1.9.25.1', } for query, value in expected.items(): if pom.findtext(query, namespaces=ns) != value: raise ValueError('POM 版本不匹配: ' + query + '; expected=' + value) arg_line = ''.join(pom.itertext()) if '-Dcsp.sentinel.log.dir=' not in arg_line or '${project.build.directory}/sentinel-logs' not in arg_line: raise ValueError('POM surefire argLine 必须把 Sentinel 日志重定向到 target/sentinel-logs') return {'spring-boot': '4.0.9', 'spring-cloud': '2025.1.4', 'spring-cloud-alibaba': '2025.1.0.0', 'sentinel': '1.8.9', 'aspectjweaver': '1.9.25.1', 'java': '21'} def redirect_and_clean_sentinel_logs(logs, test_started_ns): """Verify the Surefire argLine kept Sentinel logs out of ~/logs and keep a copy.""" home_csp = pathlib.Path.home() / 'logs/csp' offenders = [] if home_csp.is_dir(): offenders = [str(path) for path in home_csp.iterdir() if path.is_file() and path.stat().st_mtime_ns >= test_started_ns] if offenders: raise ValueError('Sentinel 日志泄漏到 ~/logs/csp（argLine 未生效）: ' + str(offenders)) target_logs = ROOT / 'target/sentinel-logs' copied = [] if target_logs.is_dir(): destination = logs / 'sentinel-logs' destination.mkdir(exist_ok=True) for path in sorted(target_logs.iterdir()): if path.is_file(): shutil.copyfile(path, destination / path.name) copied.append(path.name) shutil.rmtree(target_logs) if not copied: raise ValueError('target/sentinel-logs 为空：测试运行未产生 Sentinel 日志文件') return {'redirectedTo': 'target/sentinel-logs', 'copiedTo': str(logs / 'sentinel-logs'), 'files': copied, 'homeDirWrites': 0} def main(): signal.signal(signal.SIGTERM, interrupted) signal.signal(signal.SIGINT, interrupted) logs = None evidence = { 'ticket': 'U37 #74', 'status': 'running', 'startedAt': now(), 'finishedAt': None, 'input': 'python3 labs/sentinel/run.py', 'exitCode': None, 'environment': {'os': platform.platform(), 'arch': platform.machine(), 'python': platform.python_version()}, 'expectedTestCount': EXPECTED_TEST_COUNT, 'expectedTests': {name: sorted(tests) for name, tests in EXPECTED_CLASSES.items()}, 'reports': [], 'testCount': 0, 'observed': [], 'results': [], 'starterBoot4': None, 'sentinelLogs': None, 'artifacts': {}, 'logs': None, 'cleanup': {'processGroups': '尚未完成', 'resources': '测试 fixture 自行删除临时规则文件；runner 验证并清理 target/sentinel-logs'}, 'limits': [ '纯 JVM 单模块实验：无 Sentinel Dashboard、Nacos/Apollo/ZooKeeper 数据源、token server 与集群流控', '所有规则经内存 API（*RuleManager.loadRules）装载；文件数据源为本地临时文件热更新，不是生产配置中心', 'Boot 4.0.9 + Sentinel 1.8.9/SCA 2025.1.0.0 超出官方支持矩阵；starterBoot4 只覆盖最小无 Web 上下文启动', '不验证 Web 容器拦截器（SCA WebMvc/WebFlux adapter）、Feign 适配与网关（SCG 4.x adapter 与基线 5.0.4 错配）', '熔断/流控是单 JVM 本地统计；集群模式、Dashboard 规则推送与生产容量不在证明范围', ], } exit_code = 1 try: # Invalidate success even if cache setup fails or this run is killed later. write_json(EVIDENCE, evidence) logs_root = pathlib.Path.home() / '.cache/software-engineering-study/java-depth-impl/u37' logs_root.mkdir(parents=True, exist_ok=True) logs = pathlib.Path(tempfile.mkdtemp(prefix='run-', dir=logs_root)) evidence['logs'] = str(logs) print('LOGS ' + str(logs), flush=True) if len(sys.argv) != 1: raise ValueError('本入口不接受参数；使用 python3 labs/sentinel/run.py') if len(MARKERS) != EXPECTED_TEST_COUNT - 1 or sum( len(tests) for tests in EXPECTED_CLASSES.values()) != EXPECTED_TEST_COUNT: raise ValueError('U37 测试/观察行清单尚未完整登记，必须有 ' + str(EXPECTED_TEST_COUNT) + ' 项') evidence['artifacts'] = artifact_digests() evidence['environment']['declaredVersions'] = validate_pom() env = dict(os.environ) if not env.get('JAVA_HOME') and sys.platform == 'darwin': java_home = run('java-home', ['/usr/libexec/java_home', '-v', '21'], logs, env, evidence['results'], timeout=15).strip() env['JAVA_HOME'] = java_home if env.get('JAVA_HOME'): env['PATH'] = str(pathlib.Path(env['JAVA_HOME']) / 'bin') + os.pathsep + env.get('PATH', '') java = shutil.which('java', path=env.get('PATH')) if not java: raise RuntimeError('需要 JDK 21：设置 JAVA_HOME 或在 PATH 中提供 java') version = run('jdk', [java, '-version'], logs, env, evidence['results'], timeout=15) evidence['environment']['jdk'] = version.strip() if not re.search(r'version "21[.\"]', version): raise ValueError('需要 JDK 21；实际版本见 jdk.log') mvn = maven_binary() maven = run('maven', [mvn, '-version'], logs, env, evidence['results'], timeout=30) evidence['environment']['maven'] = maven.strip() if not re.search(r'Apache Maven 3[.]9[.]', maven): raise ValueError('需要 Maven 3.9.x；实际版本见 maven.log') base = [mvn, '-B', '-ntp', '-q', '-f', str(ROOT / 'pom.xml')] test_started_ns = time.time_ns() run('test', base + ['clean', 'test', '-DskipTests=false', '-Dmaven.test.skip=false', '-Dmaven.clean.skip=false', '-Dmaven.test.failure.ignore=false', '-DfailIfNoTests=true'], logs, env, evidence['results']) evidence['reports'] = validate_reports(logs, test_started_ns) evidence['observed'] = evidence['results'][-1]['observed'] evidence['starterBoot4'] = validate_observations(evidence['observed']) evidence['sentinelLogs'] = redirect_and_clean_sentinel_logs(logs, test_started_ns) dependencies = logs / 'dependencies.txt' run('dependencies', base + ['dependency:list', '-DincludeScope=test', '-DappendOutput=false', '-DoutputFile=' + str(dependencies)], logs, env, evidence['results']) evidence['environment']['dependencies'] = read_dependencies(dependencies) evidence['dependencyReport'] = {'path': str(dependencies), 'sha256': file_sha(dependencies)} if artifact_digests() != evidence['artifacts']: raise ValueError('运行期间实验文件发生变化；请在修改结束后重新运行') evidence['testCount'] = sum(int(report['tests']) for report in evidence['reports']) if evidence['testCount'] != EXPECTED_TEST_COUNT: raise ValueError('测试总数不为 ' + str(EXPECTED_TEST_COUNT) + ': ' + str(evidence['testCount'])) evidence['status'] = 'passed' exit_code = 0 except BaseException as exc: exit_code = failure_code(exc) evidence['status'] = 'failed' evidence['failure'] = {'type': type(exc).__name__, 'message': str(exc)} print('FAILED exit=' + str(exit_code) + ': ' + str(exc) + '; logs: ' + str(logs), file=sys.stderr, flush=True) finally: handlers = {sig: signal.signal(sig, signal.SIG_IGN) for sig in (signal.SIGINT, signal.SIGTERM)} try: evidence['exitCode'] = exit_code evidence['finishedAt'] = now() evidence['observed'] = [line for result in evidence['results'] if result['name'] == 'test' for line in result['observed']] if not evidence['results']: evidence['cleanup']['processGroups'] = '未启动命令进程' elif any(result['cleanupError'] is not None for result in evidence['results']): evidence['cleanup']['processGroups'] = '进程组清理失败，详见 results.cleanupError' else: evidence['cleanup']['processGroups'] = '本次启动的全部进程组已定向回收' if logs is not None: write_json(logs / 'run.status.json', evidence) write_json(EVIDENCE, evidence) finally: for sig, handler in handlers.items(): signal.signal(sig, handler) if exit_code == 0: print('PASS U37 tests=' + str(evidence['testCount']) + ' starterBoot4=' + str(evidence['starterBoot4']) + '; evidence: ' + str(EVIDENCE), flush=True) return exit_code if __name__ == '__main__': sys.exit(main())