结果验证与交付模块

摘要

在现代软件交付体系中，任务执行完成并不等于交付成功。一个可靠的交付平台必须能够回答三个关键问题：任务是否真正完成？、产生了哪些可交付产物？、本次变更如何被记录和追溯？本文将围绕 Harness 平台的“结果验证与交付模块”，深入剖析其判断任务完成、生成产物以及输出变更记录的核心机制，并结合完整的代码示例，帮助读者理解如何在自己的 CI/CD 流程中落地类似的验证与交付能力。

文章将从整体架构入手，逐步讲解完成判定、产物管理、变更记录、验证策略以及工程实践，最后给出可运行的示例代码与最佳实践建议。

引言

在传统的 CI/CD 流水线中，很多团队对“任务完成”的判断非常简单：只要脚本退出码为 0，就认为成功。然而在真实的生产环境中，这种判断往往过于粗糙：

• 单元测试通过了，但集成测试还没跑；
• 构建成功了，但镜像没有被推送到仓库；
• 部署完成了，但健康检查还没通过；
• 变更被应用了，但没有留下任何可审计的记录。

Harness 作为新一代软件交付平台，将“结果验证”和“交付”抽象为独立的模块，通过Step、Stage、Pipeline三层结构，配合Delegate、Connector、Artifact、Change Log等组件，形成了一个可观测、可追溯、可回滚的交付闭环。

本文将聚焦于这个闭环中最关键的一环：结果验证与交付模块。我们会回答以下问题：

1. Harness 如何判断一个任务真正完成？
2. 交付产物是如何生成、存储和引用的？
3. 变更记录是如何输出并用于审计和回滚的？
4. 如何用代码实现一个类似的验证与交付模块？

一、整体架构：从执行到交付的闭环

在深入细节之前，先看一下 Harness 结果验证与交付模块的整体结构。

┌──────────────────────────────────────────────────────────┐ │ Pipeline │ │ ┌────────────┐ ┌────────────┐ ┌────────────────┐ │ │ │ Stage 1 │ → │ Stage 2 │ → │ Stage 3 │ │ │ │ Build │ │ Verify │ │ Deploy │ │ │ └────────────┘ └────────────┘ └────────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ Artifact Verification Change Log │ │ (产物) (验证结果) (变更记录) │ └──────────────────────────────────────────────────────────┘

核心组件包括：

这个架构的核心思想是：执行、验证、交付三者解耦。执行负责“做事”，验证负责“确认”，交付负责“留痕”。接下来我们逐层拆解。

二、任务完成判定：从 Exit Code 到多维度验证

2.1 传统方式的局限

大多数 CI 工具判断任务完成只看退出码：

#!/bin/bashset-emvn clean packageecho"Build done"

只要mvn返回 0，就认为构建成功。但这种方式无法回答：

• 测试覆盖率是否达标？
• 产物是否被正确上传？
• 部署后服务是否健康？

2.2 Harness 的多维度完成判定

Harness 通过Step Outcome + Verification + Approval三重机制判断任务是否完成。

1）Step Outcome

每个 Step 执行完成后，会输出一个结构化的 Outcome：

step:name:BuildAndPushidentifier:BuildAndPushtype:Runspec:connectorRef:dockerhubimage:maven:3.9-eclipse-temurin-17command:|mvn clean package docker build -t myapp:${HARNESS_BUILD_ID} . docker push myapp:${HARNESS_BUILD_ID}outputs:-name:imageTagvalue:${HARNESS_BUILD_ID}

Harness 会解析outputs字段，作为后续 Stage 的输入。如果命令退出码非 0，Step 直接失败；如果退出码为 0，但 outputs 缺失，也会被标记为“部分成功”。

2）Verification Step

验证步骤会调用外部监控或日志系统，确认服务状态：

-step:name:VerifyServiceHealthidentifier:VerifyServiceHealthtype:Verifyspec:type:HealthCheckconnectorRef:k8sConnectorduration:5minterval:30sthresholds:-metric:http_5xx_ratecriteria:"< 1%"-metric:p99_latency_mscriteria:"< 500"

只有当所有阈值满足时，Stage 才会被判定为成功。

3）Approval Step

对于生产环境，Harness 支持人工审批：

-step:name:ManualApprovalidentifier:ManualApprovaltype:HarnessApprovalspec:approvalMessage:"请确认是否发布到生产环境"approvers:-user1@example.com-user2@example.com

2.3 用代码实现一个完成判定器

下面是一个简化版的 Python 实现，模拟 Harness 的多维度完成判定：

importsubprocessfromdataclassesimportdataclassfromtypingimportList,Optional@dataclassclassStepResult:"""单个 Step 的执行结果"""name:strexit_code:intoutputs:dicterror:Optional[str]=None@dataclassclassVerificationResult:"""验证结果"""name:strpassed:booldetails:strclassCompletionJudge:"""任务完成判定器"""def__init__(self):self.steps:List[StepResult]=[]self.verifications:List[VerificationResult]=[]defrun_step(self,name:str,command:str)->StepResult:"""执行一个 Step 并记录结果"""print(f"[Step] 执行:{name}")proc=subprocess.run(command,shell=True,capture_output=True,text=True)# 解析 stdout 中形如 KEY=VALUE 的输出outputs={}forlineinproc.stdout.splitlines():if"="inlineandnotline.startswith(" "):k,v=line.split("=",1)outputs[k.strip()]=v.strip()result=StepResult(name=name,exit_code=proc.returncode,outputs=outputs,error=proc.stderrifproc.returncode!=0elseNone,)self.steps.append(result)returnresultdefverify(self,name:str,condition:bool,details:str)->VerificationResult:"""执行一个验证步骤"""result=VerificationResult(name=name,passed=condition,details=details)self.verifications.append(result)print(f"[Verify]{name}:{'PASS'ifconditionelse'FAIL'}-{details}")returnresultdefis_complete(self)->bool:"""综合判定任务是否完成"""# 1. 所有 Step 必须退出码为 0steps_ok=all(s.exit_code==0forsinself.steps)# 2. 所有验证必须通过verifications_ok=all(v.passedforvinself.verifications)# 3. 至少有一个 Step 产生了 outputs（说明有产物）has_outputs=any(s.outputsforsinself.steps)returnsteps_okandverifications_okandhas_outputsdefreport(self)->dict:"""生成完成判定报告"""return{"complete":self.is_complete(),"steps":[{"name":s.name,"exit_code":s.exit_code,"outputs":s.outputs}forsinself.steps],"verifications":[{"name":v.name,"passed":v.passed,"details":v.details}forvinself.verifications],}# ============ 使用示例 ============if__name__=="__main__":judge=CompletionJudge()# Step 1: 构建judge.run_step("Build","echo BUILD_ID=12345 && echo 'build success'")# Step 2: 验证judge.verify("UnitTest",condition=True,details="覆盖率 85%")judge.verify("HealthCheck",condition=True,details="5xx 比例 0.2%")# 输出报告importjsonprint(json.dumps(judge.report(),indent=2,ensure_ascii=False))

运行结果会输出一个结构化的 JSON，清晰地展示每个 Step 和 Verification 的状态，最终给出complete: true/false的判定。

三、产物生成：Artifact 的生命周期管理

3.1 什么是产物

在 Harness 中，Artifact（产物）是构建阶段输出的、可被后续阶段引用的对象，常见类型包括：

• Docker 镜像
• JAR / WAR 包
• Helm Chart
• 二进制文件
• 配置文件

产物有两个关键属性：唯一标识和可寻址。前者用于区分不同版本，后者用于在部署阶段拉取。

3.2 产物的生成与推送

在 Harness 中，产物通常由 Build Stage 生成并推送到制品库：

-step:name:BuildAndPushDockeridentifier:BuildAndPushDockertype:BuildAndPushDockerRegistryspec:connectorRef:dockerHubConnectorrepo:myorg/myapptags:-<+pipeline.sequenceId>-latestdockerfile:Dockerfilecontext:.

Harness 会自动将镜像标签作为 Artifact 的标识，并在后续 Stage 中通过表达式引用：

-step:name:DeployToK8sidentifier:DeployToK8stype:K8sRollingDeployspec:image:myorg/myapp:<+pipeline.stages.Build.spec.execution.steps.BuildAndPushDocker.output.imageTag>

3.3 用代码实现一个 Artifact 管理器

下面是一个简化版的 Artifact 管理器，支持生成、存储和引用产物：

importhashlibimportjsonimportosimportshutilfromdataclassesimportdataclass,field,asdictfromdatetimeimportdatetimefromtypingimportDict,List,Optional@dataclassclassArtifact:"""产物元数据"""name:strversion:strpath:strchecksum:strsize:intcreated_at:strmetadata:Dict[str,str]=field(default_factory=dict)@propertydefidentifier(self)->str:"""唯一标识：name:version"""returnf"{self.name}:{self.version}"classArtifactManager:"""产物管理器：负责生成、存储、引用产物"""def__init__(self,storage_dir:str="./artifacts"):self.storage_dir=storage_dir os.makedirs(storage_dir,exist_ok=True)self.index_file=os.path.join(storage_dir,"index.json")self.index:Dict[str,dict]=self._load_index()def_load_index(self)->Dict[str,dict]:ifos.path.exists(self.index_file):withopen(self.index_file,"r",encoding="utf-8")asf:returnjson.load(f)return{}def_save_index(self):withopen(self.index_file,"w",encoding="utf-8")asf:json.dump(self.index,f,indent=2,ensure_ascii=False)@staticmethoddef_checksum(file_path:str)->str:"""计算文件 SHA256"""h=hashlib.sha256()withopen(file_path,"rb")asf:forchunkiniter(lambda:f.read(8192),b""):h.update(chunk)returnh.hexdigest()defpublish(self,name:str,version:str,source_path:str,metadata:Optional[Dict[str,str]]=None,)->Artifact:"""发布一个产物"""ifnotos.path.exists(source_path):raiseFileNotFoundError(f"源文件不存在:{source_path}")# 1. 计算校验和checksum=self._checksum(source_path)size=os.path.getsize(source_path)# 2. 拷贝到存储目录target_dir=os.path.join(self.storage_dir,name,version)os.makedirs(target_dir,exist_ok=True)target_path=os.path.join(target_dir,os.path.basename(source_path))shutil.copy2(source_path,target_path)# 3. 构建元数据artifact=Artifact(name=name,version=version,path=target_path,checksum=checksum,size=size,created_at=datetime.utcnow().isoformat(),metadata=metadataor{},)# 4. 更新索引self.index[artifact.identifier]=asdict(artifact)self._save_index()print(f"[Publish]{artifact.identifier}->{target_path}")returnartifactdefresolve(self,identifier:str)->Artifact:"""根据标识引用产物"""ifidentifiernotinself.index:raiseKeyError(f"产物不存在:{identifier}")returnArtifact(**self.index[identifier])deflist(self,name:Optional[str]=None)->List[Artifact]:"""列出产物"""result=[]forident,datainself.index.items():ifnameisNoneordata["name"]==name:result.append(Artifact(**data))returnresult# ============ 使用示例 ============if__name__=="__main__":# 准备一个测试文件os.makedirs("./build",exist_ok=True)withopen("./build/app.jar","w")asf:f.write("fake jar content")mgr=ArtifactManager("./artifacts")# 发布产物artifact=mgr.publish(name="myapp",version="1.0.0",source_path="./build/app.jar",metadata={"git_sha":"abc123","branch":"main"},)# 引用产物resolved=mgr.resolve("myapp:1.0.0")print(f"引用产物:{resolved.identifier}, 校验和:{resolved.checksum[:12]}...")# 列出所有版本forainmgr.list("myapp"):print(f" -{a.identifier}({a.size}bytes)")

这个管理器实现了产物的完整生命周期：发布 → 存储 → 引用 → 列出。在实际生产环境中，可以把storage_dir换成 S3、OSS 或 Harbor 等制品库。

四、变更记录：从 Diff 到可追溯的审计日志

4.1 为什么需要变更记录

变更记录（Change Log）是交付模块中最容易被忽视、但最关键的部分。它的作用有三：

1. 审计：谁在什么时候改了什么？
2. 回滚：出问题时能快速定位到上一个稳定版本；
3. 关联：把代码提交、构建产物、部署实例串成一条链。

在 Harness 中，变更记录通常包含以下字段：

4.2 生成变更记录

Harness 会在 Pipeline 执行过程中自动收集变更信息，并输出到 Change Log。一个典型的输出如下：

{"changeId":"chg-20240115-001","type":"DEPLOYMENT","commitSha":"a1b2c3d4e5f6","author":"alice@example.com","timestamp":"2024-01-15T10:23:45Z","artifact":"myorg/myapp:12345","environment":"production","diff":{"previous":"myorg/myapp:12344","current":"myorg/myapp:12345","changes":["feat: add /health endpoint","fix: correct timeout config"]}}

4.3 用代码实现变更记录生成器

下面是一个 Python 实现，从 Git 仓库和产物元数据中生成结构化的变更记录：

importjsonimportsubprocessimportuuidfromdataclassesimportdataclass,asdict,fieldfromdatetimeimportdatetimefromtypingimportList,Optional@dataclassclassChangeEntry:"""单条变更记录"""change_id:strtype:strcommit_sha:strauthor:strtimestamp:strartifact:str