CLARISSA Portal: Infrastructure Playground¶

Interactive playground for CLARISSA infrastructure management

This notebook provides hands-on access to infrastructure operations:

• 🏃 Runners - Monitor CI/CD runner status and performance
• 🔄 Sync - Check GitLab ↔ GitHub synchronization
• 📊 Metrics - Pipeline performance, job statistics
• 🚀 Deployment - Pages status, container registry

---

⚠️ Security Notice¶

This notebook uses project tokens for both GitLab and GitHub. Store tokens securely in Colab Secrets or environment variables.

In [ ]:

# Configuration
import os
import json
import requests
from datetime import datetime, timedelta
from typing import Dict, List, Any, Optional

# Try to get tokens from Colab secrets
try:
    from google.colab import userdata
    GITLAB_TOKEN = userdata.get('GITLAB_TOKEN')
    GITHUB_TOKEN = userdata.get('GITHUB_TOKEN')
    print("✅ Tokens loaded from Colab Secrets")
except:
    GITLAB_TOKEN = os.getenv('GITLAB_TOKEN', 'YOUR_GITLAB_TOKEN')
    GITHUB_TOKEN = os.getenv('GITHUB_TOKEN', 'YOUR_GITHUB_TOKEN')
    if 'YOUR_' in GITLAB_TOKEN:
        print("⚠️  No tokens found. Set environment variables or use Colab Secrets.")
    else:
        print("✅ Tokens loaded from environment")

# Project configuration
GITLAB_URL = "https://gitlab.com/api/v4"
GITHUB_URL = "https://api.github.com"
GITLAB_PROJECT_ID = 77260390
GITHUB_REPO = "wolfram-laube/clarissa"

print(f"
📅 Session: {datetime.now().isoformat()}")

# Configuration import os import json import requests from datetime import datetime, timedelta from typing import Dict, List, Any, Optional # Try to get tokens from Colab secrets try: from google.colab import userdata GITLAB_TOKEN = userdata.get('GITLAB_TOKEN') GITHUB_TOKEN = userdata.get('GITHUB_TOKEN') print("✅ Tokens loaded from Colab Secrets") except: GITLAB_TOKEN = os.getenv('GITLAB_TOKEN', 'YOUR_GITLAB_TOKEN') GITHUB_TOKEN = os.getenv('GITHUB_TOKEN', 'YOUR_GITHUB_TOKEN') if 'YOUR_' in GITLAB_TOKEN: print("⚠️ No tokens found. Set environment variables or use Colab Secrets.") else: print("✅ Tokens loaded from environment") # Project configuration GITLAB_URL = "https://gitlab.com/api/v4" GITHUB_URL = "https://api.github.com" GITLAB_PROJECT_ID = 77260390 GITHUB_REPO = "wolfram-laube/clarissa" print(f" 📅 Session: {datetime.now().isoformat()}")

In [ ]:

# Infrastructure Manager
class InfrastructureManager:
    """Manager for CLARISSA infrastructure operations."""
    
    def __init__(self, gitlab_token: str, github_token: str, 
                 gitlab_project_id: int, github_repo: str):
        self.gitlab_headers = {'PRIVATE-TOKEN': gitlab_token}
        self.github_headers = {'Authorization': f'token {github_token}'}
        self.gitlab_project_id = gitlab_project_id
        self.github_repo = github_repo
    
    def _gitlab_request(self, endpoint: str, method: str = 'GET', data: Dict = None) -> Any:
        url = f"{GITLAB_URL}/{endpoint}"
        try:
            response = requests.request(method, url, headers=self.gitlab_headers, json=data, timeout=30)
            response.raise_for_status()
            return response.json() if response.text else {'status': 'success'}
        except Exception as e:
            return {'error': str(e)}
    
    def _github_request(self, endpoint: str, method: str = 'GET', data: Dict = None) -> Any:
        url = f"{GITHUB_URL}/{endpoint}"
        try:
            response = requests.request(method, url, headers=self.github_headers, json=data, timeout=30)
            response.raise_for_status()
            return response.json() if response.text else {'status': 'success'}
        except Exception as e:
            return {'error': str(e)}
    
    # ============ RUNNERS ============
    def get_runners(self) -> List:
        """Get all project runners."""
        return self._gitlab_request(f'projects/{self.gitlab_project_id}/runners')
    
    def get_runner_details(self, runner_id: int) -> Dict:
        """Get details for a specific runner."""
        return self._gitlab_request(f'runners/{runner_id}')
    
    def get_runner_jobs(self, runner_id: int, per_page: int = 10) -> List:
        """Get recent jobs for a runner."""
        return self._gitlab_request(f'runners/{runner_id}/jobs?per_page={per_page}')
    
    # ============ SYNC STATUS ============
    def get_mirror_status(self) -> Dict:
        """Get GitLab push mirror status."""
        mirrors = self._gitlab_request(f'projects/{self.gitlab_project_id}/remote_mirrors')
        if isinstance(mirrors, list) and len(mirrors) > 0:
            return mirrors[0]
        return {'status': 'no_mirror_configured'}
    
    def trigger_mirror_sync(self) -> Dict:
        """Manually trigger mirror sync."""
        mirrors = self._gitlab_request(f'projects/{self.gitlab_project_id}/remote_mirrors')
        if isinstance(mirrors, list) and len(mirrors) > 0:
            mirror_id = mirrors[0]['id']
            return self._gitlab_request(
                f'projects/{self.gitlab_project_id}/remote_mirrors/{mirror_id}/sync',
                method='POST'
            )
        return {'error': 'No mirror configured'}
    
    def compare_repos(self) -> Dict:
        """Compare GitLab and GitHub repositories."""
        gitlab_commits = self._gitlab_request(
            f'projects/{self.gitlab_project_id}/repository/commits?per_page=5'
        )
        github_commits = self._github_request(
            f'repos/{self.github_repo}/commits?per_page=5'
        )
        
        result = {
            'gitlab': [],
            'github': [],
            'in_sync': False
        }
        
        if isinstance(gitlab_commits, list):
            result['gitlab'] = [{'sha': c['id'][:8], 'title': c['title']} for c in gitlab_commits]
        
        if isinstance(github_commits, list):
            result['github'] = [{'sha': c['sha'][:8], 'title': c['commit']['message'].split('
')[0]} for c in github_commits]
        
        if result['gitlab'] and result['github']:
            result['in_sync'] = result['gitlab'][0]['sha'] == result['github'][0]['sha']
        
        return result
    
    # ============ PIPELINE METRICS ============
    def get_pipeline_stats(self, days: int = 7) -> Dict:
        """Get pipeline statistics for the last N days."""
        since = (datetime.now() - timedelta(days=days)).isoformat()
        pipelines = self._gitlab_request(
            f'projects/{self.gitlab_project_id}/pipelines?per_page=100&updated_after={since}'
        )
        
        if not isinstance(pipelines, list):
            return {'error': 'Could not fetch pipelines'}
        
        stats = {
            'total': len(pipelines),
            'success': 0,
            'failed': 0,
            'running': 0,
            'pending': 0,
            'canceled': 0,
            'other': 0,
            'success_rate': 0.0
        }
        
        for p in pipelines:
            status = p.get('status', 'other')
            if status in stats:
                stats[status] += 1
            else:
                stats['other'] += 1
        
        completed = stats['success'] + stats['failed']
        if completed > 0:
            stats['success_rate'] = round(stats['success'] / completed * 100, 1)
        
        return stats
    
    def get_job_performance(self, per_page: int = 50) -> Dict:
        """Analyze job performance across runners."""
        jobs = self._gitlab_request(
            f'projects/{self.gitlab_project_id}/jobs?per_page={per_page}'
        )
        
        if not isinstance(jobs, list):
            return {'error': 'Could not fetch jobs'}
        
        runner_stats = {}
        for job in jobs:
            if job.get('runner'):
                runner_name = job['runner'].get('description', 'Unknown')
                if runner_name not in runner_stats:
                    runner_stats[runner_name] = {'jobs': 0, 'total_duration': 0, 'success': 0}
                
                runner_stats[runner_name]['jobs'] += 1
                if job.get('duration'):
                    runner_stats[runner_name]['total_duration'] += job['duration']
                if job.get('status') == 'success':
                    runner_stats[runner_name]['success'] += 1
        
        # Calculate averages
        for runner in runner_stats:
            stats = runner_stats[runner]
            stats['avg_duration'] = round(stats['total_duration'] / stats['jobs'], 1) if stats['jobs'] > 0 else 0
            stats['success_rate'] = round(stats['success'] / stats['jobs'] * 100, 1) if stats['jobs'] > 0 else 0
        
        return runner_stats
    
    # ============ DEPLOYMENT ============
    def get_pages_status(self) -> Dict:
        """Get GitLab Pages deployment status."""
        deployments = self._gitlab_request(
            f'projects/{self.gitlab_project_id}/deployments?per_page=5&environment=production'
        )
        
        if isinstance(deployments, list) and len(deployments) > 0:
            latest = deployments[0]
            return {
                'status': latest.get('status'),
                'created_at': latest.get('created_at'),
                'environment': latest.get('environment', {}).get('name'),
                'ref': latest.get('ref')
            }
        
        # Check pages settings
        project = self._gitlab_request(f'projects/{self.gitlab_project_id}')
        if isinstance(project, dict):
            return {
                'pages_url': project.get('pages_access_level'),
                'visibility': project.get('visibility')
            }
        
        return {'status': 'unknown'}
    
    def get_container_registry(self) -> Dict:
        """Get container registry info."""
        repos = self._gitlab_request(
            f'projects/{self.gitlab_project_id}/registry/repositories'
        )
        
        if isinstance(repos, list):
            return {
                'repositories': len(repos),
                'details': [{'id': r['id'], 'path': r['path']} for r in repos[:5]]
            }
        return {'status': 'no_registry_or_error', 'response': repos}
    
    # ============ GITHUB ACTIONS ============
    def get_github_actions_status(self) -> Dict:
        """Get GitHub Actions workflow runs."""
        runs = self._github_request(
            f'repos/{self.github_repo}/actions/runs?per_page=10'
        )
        
        if isinstance(runs, dict) and 'workflow_runs' in runs:
            return {
                'total_count': runs.get('total_count'),
                'recent_runs': [
                    {
                        'name': r.get('name'),
                        'status': r.get('status'),
                        'conclusion': r.get('conclusion'),
                        'created_at': r.get('created_at')
                    }
                    for r in runs['workflow_runs'][:5]
                ]
            }
        return runs
    
    # ============ HEALTH CHECK ============
    def health_check(self) -> Dict:
        """Comprehensive infrastructure health check."""
        health = {
            'timestamp': datetime.now().isoformat(),
            'gitlab_api': '❓',
            'github_api': '❓',
            'mirror_sync': '❓',
            'runners': '❓',
            'pipelines': '❓'
        }
        
        # GitLab API
        project = self._gitlab_request(f'projects/{self.gitlab_project_id}')
        health['gitlab_api'] = '✅' if 'name' in project else '❌'
        
        # GitHub API
        repo = self._github_request(f'repos/{self.github_repo}')
        health['github_api'] = '✅' if 'name' in repo else '❌'
        
        # Mirror sync
        sync = self.compare_repos()
        health['mirror_sync'] = '✅' if sync.get('in_sync') else '⚠️ Out of sync'
        
        # Runners
        runners = self.get_runners()
        if isinstance(runners, list):
            online = sum(1 for r in runners if r.get('online'))
            health['runners'] = f'✅ {online}/{len(runners)} online'
        else:
            health['runners'] = '❌'
        
        # Pipelines
        stats = self.get_pipeline_stats(days=1)
        if 'total' in stats:
            health['pipelines'] = f"✅ {stats['success_rate']}% success rate (24h)"
        else:
            health['pipelines'] = '❌'
        
        return health

# Initialize manager
infra = InfrastructureManager(GITLAB_TOKEN, GITHUB_TOKEN, GITLAB_PROJECT_ID, GITHUB_REPO)
print("✅ InfrastructureManager initialized")

# Infrastructure Manager class InfrastructureManager: """Manager for CLARISSA infrastructure operations.""" def __init__(self, gitlab_token: str, github_token: str, gitlab_project_id: int, github_repo: str): self.gitlab_headers = {'PRIVATE-TOKEN': gitlab_token} self.github_headers = {'Authorization': f'token {github_token}'} self.gitlab_project_id = gitlab_project_id self.github_repo = github_repo def _gitlab_request(self, endpoint: str, method: str = 'GET', data: Dict = None) -> Any: url = f"{GITLAB_URL}/{endpoint}" try: response = requests.request(method, url, headers=self.gitlab_headers, json=data, timeout=30) response.raise_for_status() return response.json() if response.text else {'status': 'success'} except Exception as e: return {'error': str(e)} def _github_request(self, endpoint: str, method: str = 'GET', data: Dict = None) -> Any: url = f"{GITHUB_URL}/{endpoint}" try: response = requests.request(method, url, headers=self.github_headers, json=data, timeout=30) response.raise_for_status() return response.json() if response.text else {'status': 'success'} except Exception as e: return {'error': str(e)} # ============ RUNNERS ============ def get_runners(self) -> List: """Get all project runners.""" return self._gitlab_request(f'projects/{self.gitlab_project_id}/runners') def get_runner_details(self, runner_id: int) -> Dict: """Get details for a specific runner.""" return self._gitlab_request(f'runners/{runner_id}') def get_runner_jobs(self, runner_id: int, per_page: int = 10) -> List: """Get recent jobs for a runner.""" return self._gitlab_request(f'runners/{runner_id}/jobs?per_page={per_page}') # ============ SYNC STATUS ============ def get_mirror_status(self) -> Dict: """Get GitLab push mirror status.""" mirrors = self._gitlab_request(f'projects/{self.gitlab_project_id}/remote_mirrors') if isinstance(mirrors, list) and len(mirrors) > 0: return mirrors[0] return {'status': 'no_mirror_configured'} def trigger_mirror_sync(self) -> Dict: """Manually trigger mirror sync.""" mirrors = self._gitlab_request(f'projects/{self.gitlab_project_id}/remote_mirrors') if isinstance(mirrors, list) and len(mirrors) > 0: mirror_id = mirrors[0]['id'] return self._gitlab_request( f'projects/{self.gitlab_project_id}/remote_mirrors/{mirror_id}/sync', method='POST' ) return {'error': 'No mirror configured'} def compare_repos(self) -> Dict: """Compare GitLab and GitHub repositories.""" gitlab_commits = self._gitlab_request( f'projects/{self.gitlab_project_id}/repository/commits?per_page=5' ) github_commits = self._github_request( f'repos/{self.github_repo}/commits?per_page=5' ) result = { 'gitlab': [], 'github': [], 'in_sync': False } if isinstance(gitlab_commits, list): result['gitlab'] = [{'sha': c['id'][:8], 'title': c['title']} for c in gitlab_commits] if isinstance(github_commits, list): result['github'] = [{'sha': c['sha'][:8], 'title': c['commit']['message'].split(' ')[0]} for c in github_commits] if result['gitlab'] and result['github']: result['in_sync'] = result['gitlab'][0]['sha'] == result['github'][0]['sha'] return result # ============ PIPELINE METRICS ============ def get_pipeline_stats(self, days: int = 7) -> Dict: """Get pipeline statistics for the last N days.""" since = (datetime.now() - timedelta(days=days)).isoformat() pipelines = self._gitlab_request( f'projects/{self.gitlab_project_id}/pipelines?per_page=100&updated_after={since}' ) if not isinstance(pipelines, list): return {'error': 'Could not fetch pipelines'} stats = { 'total': len(pipelines), 'success': 0, 'failed': 0, 'running': 0, 'pending': 0, 'canceled': 0, 'other': 0, 'success_rate': 0.0 } for p in pipelines: status = p.get('status', 'other') if status in stats: stats[status] += 1 else: stats['other'] += 1 completed = stats['success'] + stats['failed'] if completed > 0: stats['success_rate'] = round(stats['success'] / completed * 100, 1) return stats def get_job_performance(self, per_page: int = 50) -> Dict: """Analyze job performance across runners.""" jobs = self._gitlab_request( f'projects/{self.gitlab_project_id}/jobs?per_page={per_page}' ) if not isinstance(jobs, list): return {'error': 'Could not fetch jobs'} runner_stats = {} for job in jobs: if job.get('runner'): runner_name = job['runner'].get('description', 'Unknown') if runner_name not in runner_stats: runner_stats[runner_name] = {'jobs': 0, 'total_duration': 0, 'success': 0} runner_stats[runner_name]['jobs'] += 1 if job.get('duration'): runner_stats[runner_name]['total_duration'] += job['duration'] if job.get('status') == 'success': runner_stats[runner_name]['success'] += 1 # Calculate averages for runner in runner_stats: stats = runner_stats[runner] stats['avg_duration'] = round(stats['total_duration'] / stats['jobs'], 1) if stats['jobs'] > 0 else 0 stats['success_rate'] = round(stats['success'] / stats['jobs'] * 100, 1) if stats['jobs'] > 0 else 0 return runner_stats # ============ DEPLOYMENT ============ def get_pages_status(self) -> Dict: """Get GitLab Pages deployment status.""" deployments = self._gitlab_request( f'projects/{self.gitlab_project_id}/deployments?per_page=5&environment=production' ) if isinstance(deployments, list) and len(deployments) > 0: latest = deployments[0] return { 'status': latest.get('status'), 'created_at': latest.get('created_at'), 'environment': latest.get('environment', {}).get('name'), 'ref': latest.get('ref') } # Check pages settings project = self._gitlab_request(f'projects/{self.gitlab_project_id}') if isinstance(project, dict): return { 'pages_url': project.get('pages_access_level'), 'visibility': project.get('visibility') } return {'status': 'unknown'} def get_container_registry(self) -> Dict: """Get container registry info.""" repos = self._gitlab_request( f'projects/{self.gitlab_project_id}/registry/repositories' ) if isinstance(repos, list): return { 'repositories': len(repos), 'details': [{'id': r['id'], 'path': r['path']} for r in repos[:5]] } return {'status': 'no_registry_or_error', 'response': repos} # ============ GITHUB ACTIONS ============ def get_github_actions_status(self) -> Dict: """Get GitHub Actions workflow runs.""" runs = self._github_request( f'repos/{self.github_repo}/actions/runs?per_page=10' ) if isinstance(runs, dict) and 'workflow_runs' in runs: return { 'total_count': runs.get('total_count'), 'recent_runs': [ { 'name': r.get('name'), 'status': r.get('status'), 'conclusion': r.get('conclusion'), 'created_at': r.get('created_at') } for r in runs['workflow_runs'][:5] ] } return runs # ============ HEALTH CHECK ============ def health_check(self) -> Dict: """Comprehensive infrastructure health check.""" health = { 'timestamp': datetime.now().isoformat(), 'gitlab_api': '❓', 'github_api': '❓', 'mirror_sync': '❓', 'runners': '❓', 'pipelines': '❓' } # GitLab API project = self._gitlab_request(f'projects/{self.gitlab_project_id}') health['gitlab_api'] = '✅' if 'name' in project else '❌' # GitHub API repo = self._github_request(f'repos/{self.github_repo}') health['github_api'] = '✅' if 'name' in repo else '❌' # Mirror sync sync = self.compare_repos() health['mirror_sync'] = '✅' if sync.get('in_sync') else '⚠️ Out of sync' # Runners runners = self.get_runners() if isinstance(runners, list): online = sum(1 for r in runners if r.get('online')) health['runners'] = f'✅ {online}/{len(runners)} online' else: health['runners'] = '❌' # Pipelines stats = self.get_pipeline_stats(days=1) if 'total' in stats: health['pipelines'] = f"✅ {stats['success_rate']}% success rate (24h)" else: health['pipelines'] = '❌' return health # Initialize manager infra = InfrastructureManager(GITLAB_TOKEN, GITHUB_TOKEN, GITLAB_PROJECT_ID, GITHUB_REPO) print("✅ InfrastructureManager initialized")

---

🏥 Health Check¶

In [ ]:

# Run health check
health = infra.health_check()
print("🏥 Infrastructure Health Check")
print("=" * 40)
for key, value in health.items():
    print(f"  {key}: {value}")

# Run health check health = infra.health_check() print("🏥 Infrastructure Health Check") print("=" * 40) for key, value in health.items(): print(f" {key}: {value}")

---

🏃 Runners Status¶

In [ ]:

# Get all runners
runners = infra.get_runners()
print("🏃 CI/CD Runners")
print("=" * 60)

if isinstance(runners, list):
    for r in runners:
        status = '🟢' if r.get('online') else '🔴'
        print(f"
{status} {r['description']}")
        print(f"   ID: {r['id']}")
        print(f"   Tags: {', '.join(r.get('tag_list', []))}")
        print(f"   Active: {r.get('active')}")
else:
    print(f"Error: {runners}")

# Get all runners runners = infra.get_runners() print("🏃 CI/CD Runners") print("=" * 60) if isinstance(runners, list): for r in runners: status = '🟢' if r.get('online') else '🔴' print(f" {status} {r['description']}") print(f" ID: {r['id']}") print(f" Tags: {', '.join(r.get('tag_list', []))}") print(f" Active: {r.get('active')}") else: print(f"Error: {runners}")

In [ ]:

# Runner performance analysis
performance = infra.get_job_performance(per_page=100)
print("📊 Runner Performance (last 100 jobs)")
print("=" * 60)

if isinstance(performance, dict) and 'error' not in performance:
    for runner, stats in sorted(performance.items(), key=lambda x: x[1]['jobs'], reverse=True):
        print(f"
🏃 {runner}")
        print(f"   Jobs: {stats['jobs']}")
        print(f"   Avg Duration: {stats['avg_duration']}s")
        print(f"   Success Rate: {stats['success_rate']}%")
else:
    print(f"Error: {performance}")

# Runner performance analysis performance = infra.get_job_performance(per_page=100) print("📊 Runner Performance (last 100 jobs)") print("=" * 60) if isinstance(performance, dict) and 'error' not in performance: for runner, stats in sorted(performance.items(), key=lambda x: x[1]['jobs'], reverse=True): print(f" 🏃 {runner}") print(f" Jobs: {stats['jobs']}") print(f" Avg Duration: {stats['avg_duration']}s") print(f" Success Rate: {stats['success_rate']}%") else: print(f"Error: {performance}")

---

🔄 Repository Sync Status¶

In [ ]:

# Compare GitLab and GitHub
sync_status = infra.compare_repos()

sync_icon = '✅' if sync_status.get('in_sync') else '⚠️'
print(f"{sync_icon} Repository Sync Status")
print("=" * 60)

print("
🦊 GitLab (latest commits):")
for c in sync_status.get('gitlab', []):
    print(f"   {c['sha']} - {c['title'][:50]}")

print("
🐙 GitHub (latest commits):")
for c in sync_status.get('github', []):
    print(f"   {c['sha']} - {c['title'][:50]}")

print(f"
{'✅ Repos are in sync!' if sync_status.get('in_sync') else '⚠️ Repos are NOT in sync!'}")

# Compare GitLab and GitHub sync_status = infra.compare_repos() sync_icon = '✅' if sync_status.get('in_sync') else '⚠️' print(f"{sync_icon} Repository Sync Status") print("=" * 60) print(" 🦊 GitLab (latest commits):") for c in sync_status.get('gitlab', []): print(f" {c['sha']} - {c['title'][:50]}") print(" 🐙 GitHub (latest commits):") for c in sync_status.get('github', []): print(f" {c['sha']} - {c['title'][:50]}") print(f" {'✅ Repos are in sync!' if sync_status.get('in_sync') else '⚠️ Repos are NOT in sync!'}")

In [ ]:

# Mirror configuration
mirror = infra.get_mirror_status()
print("🔄 Mirror Configuration")
print("=" * 60)
if 'url' in mirror:
    print(f"   URL: {mirror.get('url')}")
    print(f"   Enabled: {mirror.get('enabled')}")
    print(f"   Last Update: {mirror.get('last_update_at')}")
    print(f"   Last Error: {mirror.get('last_error') or 'None'}")
else:
    print(f"   {mirror}")

# Mirror configuration mirror = infra.get_mirror_status() print("🔄 Mirror Configuration") print("=" * 60) if 'url' in mirror: print(f" URL: {mirror.get('url')}") print(f" Enabled: {mirror.get('enabled')}") print(f" Last Update: {mirror.get('last_update_at')}") print(f" Last Error: {mirror.get('last_error') or 'None'}") else: print(f" {mirror}")

In [ ]:

# Trigger manual sync
# Uncomment to trigger:

# result = infra.trigger_mirror_sync()
# print(f"🔄 Sync triggered: {result}")

print("⚠️ Manual sync trigger is commented out. Uncomment to test.")

# Trigger manual sync # Uncomment to trigger: # result = infra.trigger_mirror_sync() # print(f"🔄 Sync triggered: {result}") print("⚠️ Manual sync trigger is commented out. Uncomment to test.")

---

📊 Pipeline Metrics¶

In [ ]:

# Pipeline statistics
stats = infra.get_pipeline_stats(days=7)
print("📊 Pipeline Statistics (Last 7 Days)")
print("=" * 40)

if 'total' in stats:
    print(f"   Total Pipelines: {stats['total']}")
    print(f"   ✅ Success: {stats['success']}")
    print(f"   ❌ Failed: {stats['failed']}")
    print(f"   🔄 Running: {stats['running']}")
    print(f"   ⏳ Pending: {stats['pending']}")
    print(f"   🚫 Canceled: {stats['canceled']}")
    print(f"
   📈 Success Rate: {stats['success_rate']}%")
else:
    print(f"Error: {stats}")

# Pipeline statistics stats = infra.get_pipeline_stats(days=7) print("📊 Pipeline Statistics (Last 7 Days)") print("=" * 40) if 'total' in stats: print(f" Total Pipelines: {stats['total']}") print(f" ✅ Success: {stats['success']}") print(f" ❌ Failed: {stats['failed']}") print(f" 🔄 Running: {stats['running']}") print(f" ⏳ Pending: {stats['pending']}") print(f" 🚫 Canceled: {stats['canceled']}") print(f" 📈 Success Rate: {stats['success_rate']}%") else: print(f"Error: {stats}")

In [ ]:

# Visualize pipeline stats
try:
    import matplotlib.pyplot as plt
    
    if 'total' in stats and stats['total'] > 0:
        labels = ['Success', 'Failed', 'Running', 'Pending', 'Canceled']
        sizes = [stats['success'], stats['failed'], stats['running'], stats['pending'], stats['canceled']]
        colors = ['#28a745', '#dc3545', '#007bff', '#ffc107', '#6c757d']
        
        # Filter out zeros
        filtered = [(l, s, c) for l, s, c in zip(labels, sizes, colors) if s > 0]
        if filtered:
            labels, sizes, colors = zip(*filtered)
            
            fig, ax = plt.subplots(figsize=(8, 6))
            ax.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%', startangle=90)
            ax.set_title(f'Pipeline Status Distribution (Last 7 Days, n={stats["total"]})')
            plt.show()
except ImportError:
    print("⚠️ matplotlib not available")

# Visualize pipeline stats try: import matplotlib.pyplot as plt if 'total' in stats and stats['total'] > 0: labels = ['Success', 'Failed', 'Running', 'Pending', 'Canceled'] sizes = [stats['success'], stats['failed'], stats['running'], stats['pending'], stats['canceled']] colors = ['#28a745', '#dc3545', '#007bff', '#ffc107', '#6c757d'] # Filter out zeros filtered = [(l, s, c) for l, s, c in zip(labels, sizes, colors) if s > 0] if filtered: labels, sizes, colors = zip(*filtered) fig, ax = plt.subplots(figsize=(8, 6)) ax.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%', startangle=90) ax.set_title(f'Pipeline Status Distribution (Last 7 Days, n={stats["total"]})') plt.show() except ImportError: print("⚠️ matplotlib not available")

---

🐙 GitHub Actions Status¶

In [ ]:

# GitHub Actions runs
actions = infra.get_github_actions_status()
print("🐙 GitHub Actions Status")
print("=" * 60)

if 'recent_runs' in actions:
    print(f"   Total workflow runs: {actions.get('total_count')}")
    print("
   Recent runs:")
    for run in actions['recent_runs']:
        status_icon = {'completed': '✅', 'in_progress': '🔄', 'queued': '⏳'}.get(run['status'], '❓')
        conclusion_icon = {'success': '✅', 'failure': '❌', 'cancelled': '🚫'}.get(run.get('conclusion'), '')
        print(f"   {status_icon}{conclusion_icon} {run['name']} ({run['created_at'][:10]})")
else:
    print(f"   {actions}")

# GitHub Actions runs actions = infra.get_github_actions_status() print("🐙 GitHub Actions Status") print("=" * 60) if 'recent_runs' in actions: print(f" Total workflow runs: {actions.get('total_count')}") print(" Recent runs:") for run in actions['recent_runs']: status_icon = {'completed': '✅', 'in_progress': '🔄', 'queued': '⏳'}.get(run['status'], '❓') conclusion_icon = {'success': '✅', 'failure': '❌', 'cancelled': '🚫'}.get(run.get('conclusion'), '') print(f" {status_icon}{conclusion_icon} {run['name']} ({run['created_at'][:10]})") else: print(f" {actions}")

---

🚀 Deployment Status¶

In [ ]:

# Pages status
pages = infra.get_pages_status()
print("📚 GitLab Pages Status")
print("=" * 40)
print(json.dumps(pages, indent=2))

print("
🔗 Portal URL: https://clarissa-40cc50.gitlab.io/")

# Pages status pages = infra.get_pages_status() print("📚 GitLab Pages Status") print("=" * 40) print(json.dumps(pages, indent=2)) print(" 🔗 Portal URL: https://clarissa-40cc50.gitlab.io/")

In [ ]:

# Container registry
registry = infra.get_container_registry()
print("🐳 Container Registry")
print("=" * 40)
print(json.dumps(registry, indent=2))

# Container registry registry = infra.get_container_registry() print("🐳 Container Registry") print("=" * 40) print(json.dumps(registry, indent=2))

---

🎮 Interactive Playground¶

In [ ]:

# 🎮 YOUR PLAYGROUND
# Try your own infrastructure operations here!

# Examples:
# infra.get_runner_details(12345)
# infra.get_pipeline_stats(days=30)
# infra.trigger_mirror_sync()

print("🎮 Edit this cell to try your own operations!")

# 🎮 YOUR PLAYGROUND # Try your own infrastructure operations here! # Examples: # infra.get_runner_details(12345) # infra.get_pipeline_stats(days=30) # infra.trigger_mirror_sync() print("🎮 Edit this cell to try your own operations!")

In [ ]:

# 📋 Quick Reference
print("""
╔══════════════════════════════════════════════════════════════════╗
║                Infrastructure Quick Reference                    ║
╠══════════════════════════════════════════════════════════════════╣
║ RUNNERS                                                          ║
║   infra.get_runners()                  → List all runners        ║
║   infra.get_runner_details(id)         → Runner details          ║
║   infra.get_job_performance()          → Performance stats       ║
╠══════════════════════════════════════════════════════════════════╣
║ SYNC                                                             ║
║   infra.compare_repos()                → Compare GL/GH           ║
║   infra.get_mirror_status()            → Mirror config           ║
║   infra.trigger_mirror_sync()          → Force sync              ║
╠══════════════════════════════════════════════════════════════════╣
║ PIPELINES                                                        ║
║   infra.get_pipeline_stats(days)       → Statistics              ║
║   infra.get_github_actions_status()    → GH Actions runs         ║
╠══════════════════════════════════════════════════════════════════╣
║ DEPLOYMENT                                                       ║
║   infra.get_pages_status()             → Pages deployment        ║
║   infra.get_container_registry()       → Docker registry         ║
║   infra.health_check()                 → Full health check       ║
╚══════════════════════════════════════════════════════════════════╝
""")

# 📋 Quick Reference print(""" ╔══════════════════════════════════════════════════════════════════╗ ║ Infrastructure Quick Reference ║ ╠══════════════════════════════════════════════════════════════════╣ ║ RUNNERS ║ ║ infra.get_runners() → List all runners ║ ║ infra.get_runner_details(id) → Runner details ║ ║ infra.get_job_performance() → Performance stats ║ ╠══════════════════════════════════════════════════════════════════╣ ║ SYNC ║ ║ infra.compare_repos() → Compare GL/GH ║ ║ infra.get_mirror_status() → Mirror config ║ ║ infra.trigger_mirror_sync() → Force sync ║ ╠══════════════════════════════════════════════════════════════════╣ ║ PIPELINES ║ ║ infra.get_pipeline_stats(days) → Statistics ║ ║ infra.get_github_actions_status() → GH Actions runs ║ ╠══════════════════════════════════════════════════════════════════╣ ║ DEPLOYMENT ║ ║ infra.get_pages_status() → Pages deployment ║ ║ infra.get_container_registry() → Docker registry ║ ║ infra.health_check() → Full health check ║ ╚══════════════════════════════════════════════════════════════════╝ """)

---

📚 Resources¶

• 📖 GitLab CI/CD Documentation
• 🏃 GitLab Runners
• 🐙 GitHub Actions
• 📝 ADR-027: Repository Sync

---

CLARISSA Portal - Infrastructure Playground