maintenance_guide

ContractAI Maintenance Guide

Complete guide to maintaining and operating ContractAI in production environments

Overview

This guide provides comprehensive instructions for maintaining ContractAI systems, including regular maintenance tasks, update procedures, health checks, and disaster recovery. It ensures the system remains secure, performant, and reliable.

Maintenance Schedule

1. Maintenance Types

graph TD
    A[Maintenance Types] --> B[Regular Maintenance]
    A --> C[Emergency Maintenance]
    A --> D[Planned Updates]

    B --> B1[Daily Checks]
    B --> B2[Weekly Tasks]
    B --> B3[Monthly Tasks]

    C --> C1[Incident Response]
    C --> C2[Security Patches]
    C --> C3[Critical Updates]

    D --> D1[Version Updates]
    D --> D2[Infrastructure Updates]
    D --> D3[Database Maintenance]

Loading

2. Maintenance Calendar

gantt
    title ContractAI Maintenance Schedule
    dateFormat  YYYY-MM-DD
    section Daily
    Health Checks    :daily, 2024-01-01, 2024-12-31
    Log Analysis     :daily, 2024-01-01, 2024-12-31
    Backup Verification :daily, 2024-01-01, 2024-12-31

    section Weekly
    Performance Review :weekly, 2024-01-01, 2024-12-31
    Security Scan    :weekly, 2024-01-01, 2024-12-31
    Database Optimization :weekly, 2024-01-01, 2024-12-31

    section Monthly
    System Updates   :monthly, 2024-01-01, 2024-12-31
    Capacity Planning :monthly, 2024-01-01, 2024-12-31
    Compliance Check :monthly, 2024-01-01, 2024-12-31

Loading

Regular Maintenance Tasks

1. Daily Tasks

Health Check Workflow

graph TD
    A[Start Health Check] --> B{Check System Status}
    B -->|Healthy| C[Log Status]
    B -->|Warning| D[Investigate]
    B -->|Critical| E[Alert Team]

    D --> F{Can Auto-recover?}
    F -->|Yes| G[Execute Recovery]
    F -->|No| E

    G --> H{Recovery Successful?}
    H -->|Yes| C
    H -->|No| E

    E --> I[Create Incident]
    I --> J[Execute Runbook]

Loading

Automated Health Checks

# health_check.py
from datetime import datetime
import psutil
import requests
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway

class HealthChecker:
    def __init__(self):
        self.registry = CollectorRegistry()
        self.metrics = {
            'cpu_usage': Gauge('cpu_usage_percent', 'CPU Usage', registry=self.registry),
            'memory_usage': Gauge('memory_usage_percent', 'Memory Usage', registry=self.registry),
            'disk_usage': Gauge('disk_usage_percent', 'Disk Usage', registry=self.registry),
            'api_latency': Gauge('api_latency_seconds', 'API Latency', registry=self.registry)
        }

    def check_system_health(self):
        # Check CPU
        self.metrics['cpu_usage'].set(psutil.cpu_percent())

        # Check Memory
        memory = psutil.virtual_memory()
        self.metrics['memory_usage'].set(memory.percent)

        # Check Disk
        disk = psutil.disk_usage('/')
        self.metrics['disk_usage'].set(disk.percent)

        # Check API
        try:
            start_time = datetime.now()
            response = requests.get('https://api.contractai.com/health')
            latency = (datetime.now() - start_time).total_seconds()
            self.metrics['api_latency'].set(latency)
        except Exception as e:
            logger.error(f"API health check failed: {e}")

        # Push metrics
        push_to_gateway('prometheus:9090', job='health_check', registry=self.registry)

2. Weekly Tasks

Performance Analysis

graph TD
    A[Start Performance Analysis] --> B[Collect Metrics]
    B --> C[Analyze Trends]
    C --> D{Performance OK?}
    D -->|Yes| E[Update Baseline]
    D -->|No| F[Identify Bottlenecks]
    F --> G[Generate Report]
    G --> H[Plan Optimization]
    H --> I[Schedule Updates]

Loading

Database Maintenance

-- Weekly Database Maintenance
-- Analyze table statistics
ANALYZE agents;
ANALYZE incidents;
ANALYZE knowledge_base;

-- Update statistics
UPDATE pg_stat_statements
SET calls = 0, total_time = 0, rows = 0, shared_blks_hit = 0,
    shared_blks_read = 0, shared_blks_written = 0,
    shared_blks_dirtied = 0, temp_blks_read = 0, temp_blks_written = 0;

-- Vacuum analyze
VACUUM ANALYZE;

-- Check for bloat
SELECT schemaname, tablename,
       pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as total_size,
       pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) as table_size,
       pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename) -
                     pg_relation_size(schemaname||'.'||tablename)) as bloat_size
FROM pg_tables
WHERE schemaname = 'public'
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC;

3. Monthly Tasks

System Updates

graph TD
    A[Start Update Process] --> B[Review Updates]
    B --> C[Test in Staging]
    C --> D{Tests Pass?}
    D -->|No| E[Fix Issues]
    E --> C
    D -->|Yes| F[Schedule Production]
    F --> G[Backup Systems]
    G --> H[Apply Updates]
    H --> I[Verify Systems]
    I --> J{All Systems OK?}
    J -->|No| K[Rollback]
    J -->|Yes| L[Update Documentation]

Loading

Capacity Planning

# capacity_planning.py
import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
from datetime import datetime, timedelta

class CapacityPlanner:
    def __init__(self):
        self.metrics_db = "prometheus"
        self.forecast_months = 3

    def analyze_growth(self):
        # Get historical data
        query = """
        SELECT
            timestamp,
            value
        FROM metrics
        WHERE metric_name IN ('cpu_usage', 'memory_usage', 'disk_usage')
        AND timestamp > NOW() - INTERVAL '6 months'
        """

        # Load data
        df = pd.read_sql(query, self.metrics_db)

        # Prepare for forecasting
        X = np.array(range(len(df))).reshape(-1, 1)
        y = df['value'].values

        # Fit model
        model = LinearRegression()
        model.fit(X, y)

        # Forecast
        future_dates = np.array(range(len(df), len(df) + self.forecast_months * 30)).reshape(-1, 1)
        forecast = model.predict(future_dates)

        return {
            'current_usage': df['value'].iloc[-1],
            'forecast': forecast.tolist(),
            'growth_rate': model.coef_[0],
            'recommendation': self._generate_recommendation(forecast[-1])
        }

    def _generate_recommendation(self, forecast):
        if forecast > 90:
            return "CRITICAL: Immediate capacity increase required"
        elif forecast > 75:
            return "WARNING: Plan capacity increase within 1 month"
        elif forecast > 60:
            return "INFO: Monitor growth, plan for future increase"
        else:
            return "OK: Current capacity sufficient"

Update Procedures

1. Version Updates

Update Workflow

graph TD
    A[Start Update] --> B[Create Update Branch]
    B --> C[Update Dependencies]
    C --> D[Run Tests]
    D --> E{Tests Pass?}
    E -->|No| F[Fix Issues]
    F --> D
    E -->|Yes| G[Update Documentation]
    G --> H[Create Release]
    H --> I[Deploy to Staging]
    I --> J{Staging OK?}
    J -->|No| K[Rollback]
    J -->|Yes| L[Deploy to Production]
    L --> M{Production OK?}
    M -->|No| K
    M -->|Yes| N[Monitor Systems]

Loading

Update Script

#!/bin/bash
# update.sh

# Configuration
APP_DIR="/opt/contractai"
BACKUP_DIR="/backups/contractai"
VERSION=$1

# Validate version
if [ -z "$VERSION" ]; then
    echo "Usage: $0 <version>"
    exit 1
fi

# Create backup
echo "Creating backup..."
backup_name="contractai_$(date +%Y%m%d_%H%M%S)"
tar -czf "$BACKUP_DIR/$backup_name.tar.gz" "$APP_DIR"

# Update application
echo "Updating to version $VERSION..."
cd "$APP_DIR"

# Pull new version
git fetch origin
git checkout "v$VERSION"

# Update dependencies
python3.12 -m pip install -r requirements.txt

# Run migrations
alembic upgrade head

# Restart services
systemctl restart contractai

# Verify update
echo "Verifying update..."
if curl -s http://localhost:8000/health | grep -q "healthy"; then
    echo "Update successful!"
else
    echo "Update failed, rolling back..."
    tar -xzf "$BACKUP_DIR/$backup_name.tar.gz" -C /
    systemctl restart contractai
    exit 1
fi

2. Security Updates

Security Update Process

graph TD
    A[Security Alert] --> B[Assess Impact]
    B --> C{Urgent?}
    C -->|Yes| D[Emergency Update]
    C -->|No| E[Schedule Update]

    D --> F[Create Hotfix]
    F --> G[Test Hotfix]
    G --> H{Tests Pass?}
    H -->|No| I[Fix Issues]
    I --> G
    H -->|Yes| J[Deploy Hotfix]

    E --> K[Plan Update]
    K --> L[Regular Update Process]

Loading

Security Update Script

# security_update.py
import subprocess
import logging
from datetime import datetime
import requests

class SecurityUpdater:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.vulnerability_db = "https://vulndb.contractai.com"

    def check_security_updates(self):
        # Check for security updates
        updates = self._get_security_updates()

        for update in updates:
            if update['severity'] == 'critical':
                self._handle_critical_update(update)
            else:
                self._schedule_update(update)

    def _handle_critical_update(self, update):
        self.logger.warning(f"Critical security update: {update['id']}")

        # Create hotfix branch
        subprocess.run([
            'git', 'checkout', '-b',
            f'hotfix/security-{update["id"]}'
        ])

        # Apply update
        try:
            subprocess.run([
                'pip', 'install', '--upgrade',
                f"{update['package']}=={update['version']}"
            ], check=True)

            # Run tests
            if self._run_tests():
                self._deploy_hotfix()
            else:
                self.logger.error("Security update tests failed")
                self._rollback_update()
        except Exception as e:
            self.logger.error(f"Security update failed: {e}")
            self._rollback_update()

    def _schedule_update(self, update):
        # Schedule non-critical updates
        scheduled_time = self._calculate_update_window()
        self.logger.info(
            f"Scheduling update {update['id']} for {scheduled_time}"
        )

Health Checks

1. System Health

Health Check Components

graph TD
    A[System Health] --> B[Application Health]
    A --> C[Database Health]
    A --> D[Cache Health]
    A --> E[Network Health]

    B --> B1[API Status]
    B --> B2[Worker Status]
    B --> B3[Queue Status]

    C --> C1[Connection Pool]
    C --> C2[Query Performance]
    C --> C3[Replication Lag]

    D --> D1[Redis Status]
    D --> D2[Cache Hit Rate]
    D --> D3[Memory Usage]

    E --> E1[Load Balancer]
    E --> E2[DNS Resolution]
    E --> E3[SSL Certificates]

Loading

Health Check Implementation

# health_checker.py
from dataclasses import dataclass
from typing import Dict, List
import psycopg2
import redis
import requests
from prometheus_client import CollectorRegistry, Gauge

@dataclass
class HealthStatus:
    component: str
    status: str
    details: Dict
    timestamp: str

class SystemHealthChecker:
    def __init__(self):
        self.registry = CollectorRegistry()
        self.metrics = {
            'health_status': Gauge(
                'component_health_status',
                'Health status of system components',
                ['component'],
                registry=self.registry
            )
        }

    def check_all_components(self) -> List[HealthStatus]:
        statuses = []

        # Check application
        statuses.append(self._check_application())

        # Check database
        statuses.append(self._check_database())

        # Check cache
        statuses.append(self._check_cache())

        # Check network
        statuses.append(self._check_network())

        return statuses

    def _check_application(self) -> HealthStatus:
        try:
            response = requests.get('https://api.contractai.com/health')
            status = 'healthy' if response.status_code == 200 else 'unhealthy'
            self.metrics['health_status'].labels('application').set(
                1 if status == 'healthy' else 0
            )
            return HealthStatus(
                component='application',
                status=status,
                details={'response_time': response.elapsed.total_seconds()},
                timestamp=datetime.now().isoformat()
            )
        except Exception as e:
            self.metrics['health_status'].labels('application').set(0)
            return HealthStatus(
                component='application',
                status='unhealthy',
                details={'error': str(e)},
                timestamp=datetime.now().isoformat()
            )

2. Performance Health

Performance Monitoring

graph TD
    A[Performance Monitoring] --> B[Collect Metrics]
    B --> C[Analyze Patterns]
    C --> D[Set Baselines]
    D --> E[Detect Anomalies]
    E --> F{Anomaly Detected?}
    F -->|Yes| G[Investigate]
    F -->|No| H[Update Baselines]
    G --> I[Identify Cause]
    I --> J[Plan Action]
    J --> K[Execute Fix]
    K --> L[Verify Improvement]

Loading

Performance Analysis

# performance_analyzer.py
import numpy as np
from scipy import stats
from prometheus_client import start_http_server, Gauge
import pandas as pd

class PerformanceAnalyzer:
    def __init__(self):
        self.metrics = {
            'response_time': Gauge(
                'api_response_time_seconds',
                'API response time',
                ['endpoint']
            ),
            'error_rate': Gauge(
                'api_error_rate',
                'API error rate',
                ['endpoint']
            ),
            'throughput': Gauge(
                'api_requests_per_second',
                'API throughput',
                ['endpoint']
            )
        }

    def analyze_performance(self, metric_data: pd.DataFrame):
        # Calculate baseline
        baseline = self._calculate_baseline(metric_data)

        # Detect anomalies
        anomalies = self._detect_anomalies(metric_data, baseline)

        # Generate report
        report = {
            'baseline': baseline,
            'anomalies': anomalies,
            'recommendations': self._generate_recommendations(anomalies)
        }

        return report

    def _detect_anomalies(self, data: pd.DataFrame, baseline: dict) -> list:
        anomalies = []

        for metric in baseline:
            z_scores = stats.zscore(data[metric])
            threshold = 3  # 3 standard deviations

            for i, z_score in enumerate(z_scores):
                if abs(z_score) > threshold:
                    anomalies.append({
                        'metric': metric,
                        'timestamp': data.index[i],
                        'value': data[metric].iloc[i],
                        'z_score': z_score,
                        'baseline': baseline[metric]
                    })

        return anomalies

Disaster Recovery

1. Recovery Procedures

Recovery Workflow

graph TD
    A[Disaster Detected] --> B[Assess Impact]
    B --> C{System Critical?}
    C -->|Yes| D[Emergency Recovery]
    C -->|No| E[Planned Recovery]

    D --> F[Activate DR Site]
    F --> G[Restore Data]
    G --> H[Verify Systems]
    H --> I{Recovery Successful?}
    I -->|No| J[Escalate]
    I -->|Yes| K[Monitor Systems]

    E --> L[Schedule Recovery]
    L --> M[Execute Recovery Plan]
    M --> N[Verify Systems]
    N --> O{Recovery Successful?}
    O -->|No| P[Rollback]
    O -->|Yes| Q[Update Documentation]

Loading

Recovery Script

# disaster_recovery.py
import boto3
import subprocess
from datetime import datetime
import logging

class DisasterRecovery:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.s3 = boto3.client('s3')
        self.dr_bucket = 'contractai-dr'

    def execute_recovery(self, disaster_type: str):
        self.logger.info(f"Starting recovery for {disaster_type}")

        try:
            # Stop services
            self._stop_services()

            # Restore data
            self._restore_data()

            # Verify systems
            if self._verify_systems():
                self._start_services()
                self.logger.info("Recovery completed successfully")
            else:
                self.logger.error("System verification failed")
                self._rollback_recovery()
        except Exception as e:
            self.logger.error(f"Recovery failed: {e}")
            self._rollback_recovery()

    def _restore_data(self):
        # Get latest backup
        backup = self._get_latest_backup()

        # Restore database
        self._restore_database(backup['database'])

        # Restore configurations
        self._restore_configurations(backup['config'])

        # Restore application data
        self._restore_application_data(backup['application'])

    def _verify_systems(self) -> bool:
        checks = [
            self._verify_database(),
            self._verify_application(),
            self._verify_network(),
            self._verify_security()
        ]
        return all(checks)

2. Backup Verification

Backup Verification Process

graph TD
    A[Start Verification] --> B[Select Backup]
    B --> C[Restore to Test]
    C --> D[Run Tests]
    D --> E{Tests Pass?}
    E -->|No| F[Mark Backup Invalid]
    E -->|Yes| G[Verify Data]
    G --> H{Data Valid?}
    H -->|No| F
    H -->|Yes| I[Update Status]
    I --> J[Cleanup Test]

Loading

Backup Verification Script

# backup_verifier.py
import boto3
import subprocess
from datetime import datetime, timedelta
import psycopg2
import json

class BackupVerifier:
    def __init__(self):
        self.s3 = boto3.client('s3')
        self.backup_bucket = 'contractai-backups'

    def verify_backups(self, days: int = 7):
        """Verify backups for the last N days"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)

        backups = self._list_backups(start_date, end_date)

        for backup in backups:
            try:
                # Create test environment
                test_env = self._create_test_environment()

                # Restore backup
                self._restore_backup(backup, test_env)

                # Run verification
                if self._verify_backup(test_env):
                    self._update_backup_status(backup, 'verified')
                else:
                    self._update_backup_status(backup, 'invalid')

                # Cleanup
                self._cleanup_test_environment(test_env)
            except Exception as e:
                self.logger.error(f"Backup verification failed: {e}")
                self._update_backup_status(backup, 'verification_failed')

    def _verify_backup(self, test_env: dict) -> bool:
        checks = [
            self._verify_database_integrity(test_env),
            self._verify_application_data(test_env),
            self._verify_configurations(test_env)
        ]
        return all(checks)

Additional Resources

1. Maintenance Tools

• Monitoring Dashboard
• Log Analysis
• Performance Metrics
• Security Scanner

2. Support Resources

• Maintenance Portal
• Knowledge Base
• Support Team
• Emergency Contact

---

Need help with maintenance? Contact our operations team at ops@contractai.com or visit our Maintenance Portal