Add CI and deploy summaries

.gitea/workflows/deploy-controller.py

@@ -294,10 +294,47 @@ def follow(run_id, phase):
         time.sleep(3)
 
 
+def summary(run_id):
+    directory = run_directory(run_id)
+    request = json.loads((directory / 'request.json').read_text())
+    release = request['release']
+    plan_file = directory / 'plan.json'
+    lines = [
+        f'## Deploy `{release["sha"]}`',
+        '',
+        f'- Mode: `{request["mode"]}`',
+        f'- Refresh third-party images: `{request["refresh_images"]}`',
+    ]
+    if not plan_file.exists():
+        lines.extend(['', 'Plan was not created. Check the controller log.'])
+        print('\n'.join(lines))
+        return
+    plan = json.loads(plan_file.read_text())
+    lines.extend(['', '### Selected services'])
+    count = 0
+    for kind, services in plan['selected'].items():
+        for service in services:
+            lines.append(f'- `{kind}`: `{service}`')
+            count += 1
+    if not count:
+        lines.append('- None')
+    lines.extend(['', '### Selected Helm releases'])
+    lines.extend(f'- `{release}`' for release in plan.get('helm', []))
+    if not plan.get('helm'):
+        lines.append('- None')
+    lines.extend(['', '### Images pinned in the checked release'])
+    lines.extend(f'- `{image}@{digest}`' for image, digest in sorted(release['images'].items()))
+    lines.extend(['', '### Removed resources requiring manual review'])
+    lines.extend(f'- `{item}`' for item in plan.get('removed', []))
+    if not plan.get('removed'):
+        lines.append('- None')
+    print('\n'.join(lines))
+
+
 def main():
     os.umask(0o077)
     parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument('action', choices=('start', 'execute', 'recover', 'status', 'follow'))
+    parser.add_argument('action', choices=('start', 'execute', 'recover', 'status', 'follow', 'summary'))
     parser.add_argument('run_id')
     parser.add_argument('phase', nargs='?', choices=('apply', 'verify', 'smoke'))
     parser.add_argument('--retry', action='store_true', help='Retry failed recovery checks; never repeat apply')
@@ -315,6 +352,8 @@ def main():
         if (directory / 'plan.json').exists():
             plan = json.loads((directory / 'plan.json').read_text())
             print(json.dumps({k: plan[k] for k in ('sha', 'selected', 'helm', 'removed')}, indent=2))
+    elif args.action == 'summary':
+        summary(args.run_id)
     elif not follow(args.run_id, args.phase):
         sys.exit(1)