To enforce a CI gate that blocks merges when a change raises a service's critical-incident risk above a configurable threshold, we can aggregate risk contributions from available artifacts (unit/integration failures, canary negative feedback, estimated SLO impact), compute a composite risk score, compare to threshold, and report back to code review tooling (e.g., GitHub Checks / GitLab Pipeline / Bitbucket).python
# Pseudocode: CI risk-check step (Python-like)
CONFIG = {
"risk_threshold": 0.6, # configurable 0.0-1.0
"weights": { # relative weights for inputs
"unit_fail": 0.2,
"integration_fail": 0.3,
"canary_negative": 0.25,
"slo_impact": 0.25
},
"min_artifacts_required": 1
}
def fetch_artifacts(pr_id):
# Integrations: test runner API, canary feedback store, SLO estimator
unit = get_unit_test_results(pr_id) # {"failures": int, "total": int}
integration = get_integration_results(pr_id) # similar shape
canary = get_canary_feedback(pr_id) # {"negative_count": int, "total": int}
slo = get_slo_estimate(pr_id) # {"expected_error_rate_increase": float}
return {"unit": unit, "integration": integration, "canary": canary, "slo": slo}
def score_from_tests(test):
if test is None: return None
if test["total"] == 0: return 0.0
fail_rate = test["failures"] / test["total"]
# map fail_rate -> risk 0..1 (non-linear mapping possible)
return min(1.0, fail_rate)
def score_from_canary(canary):
if canary is None or canary["total"]==0: return None
neg_rate = canary["negative_count"] / canary["total"]
return min(1.0, neg_rate)
def score_from_slo(slo):
if slo is None: return None
# example: expected_error_rate_increase is fractional (0.0..1.0)
return min(1.0, slo.get("expected_error_rate_increase", 0.0))
def aggregate_scores(scores, config):
weighted_sum = 0.0
weight_total = 0.0
for key, raw in scores.items():
if raw is None: continue
w = config["weights"].get(key, 0.0)
weighted_sum += raw * w
weight_total += w
if weight_total == 0: return 0.0
return weighted_sum / weight_total # normalized 0..1
def run_risk_check(pr_id):
artifacts = fetch_artifacts(pr_id)
mapped = {
"unit_fail": score_from_tests(artifacts["unit"]),
"integration_fail": score_from_tests(artifacts["integration"]),
"canary_negative": score_from_canary(artifacts["canary"]),
"slo_impact": score_from_slo(artifacts["slo"])
}
provided = sum(1 for v in mapped.values() if v is not None)
if provided < CONFIG["min_artifacts_required"]:
report_status(pr_id, "neutral", "Insufficient artifacts for risk assessment")
return True # allow merge or follow org policy
risk = aggregate_scores(mapped, CONFIG)
if risk > CONFIG["risk_threshold"]:
report_status(pr_id, "failure",
f"Risk {risk:.2f} exceeds threshold {CONFIG['risk_threshold']:.2f}")
block_merge(pr_id)
return False
else:
report_status(pr_id, "success", f"Risk OK ({risk:.2f})")
allow_merge(pr_id)
return True
