Monitoring and Observability Guide¶

import asyncio
from arcp import ARCPClient

async def get_metrics_with_arcp():
    # 1. Initialize ARCP client
    client = ARCPClient("http://localhost:8001")

    try:
        # 2. Login as admin (use your actual admin credentials)
        await client.login_admin("ARCP", "ARCP")  # Replace with your admin credentials

        # 3. Access metrics endpoint using the client
        metrics_data = await client.get_system_metrics()
        print(f"Metrics retrieved: {len(metrics_data)} bytes")
        return metrics_data

    finally:
        # 4. Clean up
        await client.close()

# Usage
asyncio.run(get_metrics_with_arcp())

"""
Simple ARCP Metrics Example

This example shows the basic usage of ARCP's built-in metrics system 
and how to add custom metrics to your agents.
"""

from prometheus_client import Counter, Histogram
import time
import asyncio

from arcp import ARCPClient
from arcp.services.metrics import metrics_service

# Add your own custom metrics for specific agent logic
CUSTOM_AGENT_REQUESTS = Counter(
    'my_agent_custom_requests_total', 
    'Custom agent requests', 
    ['operation']
)
CUSTOM_PROCESSING_TIME = Histogram(
    'my_agent_processing_duration_seconds', 
    'Custom processing time'
)

class MyAgent:
    def __init__(self):
        self.arcp_client = ARCPClient("http://localhost:8001")

    async def process_request(self):
        """Dummy processing method"""
        await asyncio.sleep(0.1)  # Simulate some processing
        return "processed"

    async def handle_request(self, operation: str):
        start_time = time.time()

        try:
            # Your request handling logic
            result = await self.process_request()

            # Record custom metrics (in addition to built-in ARCP metrics)
            CUSTOM_AGENT_REQUESTS.labels(operation=operation).inc()
            CUSTOM_PROCESSING_TIME.observe(time.time() - start_time)

            return result
        except Exception as e:
            CUSTOM_AGENT_REQUESTS.labels(operation=operation).inc()
            CUSTOM_PROCESSING_TIME.observe(time.time() - start_time)
            raise

async def main():
    """Main function to demonstrate metrics usage"""

    # 1. Access built-in ARCP metrics using the correct method
    print("1. Built-in ARCP Metrics:")
    prometheus_metrics, content_type = metrics_service.get_prometheus_metrics()

    print(f"   Content-Type: {content_type}")
    print(f"   Metrics size: {len(prometheus_metrics)} bytes")

    # Show sample of metrics
    metrics_text = prometheus_metrics.decode()
    arcp_lines = [line for line in metrics_text.split('\n') 
                  if 'arcp_' in line and not line.startswith('#')]
    print(f"   Built-in ARCP metrics found: {len(arcp_lines)}")
    print("   Sample metrics:")
    for line in arcp_lines[:3]:
        if line.strip():
            print(f"     {line}")
    print()

    # 2. Show current system resource utilization
    print("2. System Resource Utilization:")
    resource_util = await metrics_service.get_resource_utilization()
    for resource, value in resource_util.items():
        print(f"   {resource.capitalize()}: {value}%")
    print()

    # 3. Test custom agent functionality
    print("3. Testing Custom Agent Metrics:")
    agent = MyAgent()

    # Perform some operations
    operations = ["data_process", "user_query", "system_check"]

    for op in operations:
        print(f"   Processing {op}...")
        result = await agent.handle_request(op)
        print(f"   Result: {result}")

    print()

    # 4. Show metrics service status
    print("4. Metrics Service Status:")
    status = metrics_service.get_status()
    for key, value in status.items():
        if key in ['prometheus_available', 'psutil_available']:
            print(f"   {key}: {value}")

if __name__ == "__main__":
    asyncio.run(main())

# Example trace for agent registration
from opentelemetry import trace

from arcp import client

tracer = trace.get_tracer(__name__)

async def register_agent(agent_data):
    with tracer.start_as_current_span("register_agent") as span:
        span.set_attribute("agent_id", agent_data["agent_id"])
        span.set_attribute("agent_type", agent_data["agent_type"])

        # This will create child spans for Redis operations
        result = await client.ARCPClient.register_agent(agent_data)

        span.set_attribute("registration_success", True)
        return result

from opentelemetry import trace
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.redis import RedisInstrumentor

# Instrument your agent
HTTPXClientInstrumentor().instrument()
RedisInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

class MyAgent:
    async def process_request(self, data):
        with tracer.start_as_current_span("process_request") as span:
            span.set_attribute("request_size", len(data))

            # Your processing logic
            result = await self.analyze_data(data)

            span.set_attribute("result_size", len(result))
            return result

# metrics_proxy.py
import asyncio
from flask import Flask, Response
from arcp import ARCPClient
import os

app = Flask(__name__)

ARCP_BASE_URL = os.getenv('ARCP_BASE_URL', 'http://localhost:8001')
ADMIN_USERNAME = os.getenv('ADMIN_USERNAME', 'ARCP')
ADMIN_PASSWORD = os.getenv('ADMIN_PASSWORD', 'ARCP')

@app.route('/metrics')
def proxy_metrics():
    async def get_metrics():
        client = ARCPClient(ARCP_BASE_URL)
        try:
            await client.login_admin(ADMIN_USERNAME, ADMIN_PASSWORD)
            return await client.get_system_metrics()
        finally:
            await client.close()

    try:
        metrics_data = asyncio.run(get_metrics())
        return Response(metrics_data, mimetype='text/plain')
    except Exception as e:
        return Response(f'Metrics fetch failed: {e}', status=503)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

#!/usr/bin/env python3
"""
Health check script for ARCP monitoring.
Validates all monitoring endpoints and provides detailed status.
"""
import sys
import requests

def check_endpoint(url: str, name: str = "") -> bool:
    """Check if an endpoint is accessible."""
    try:
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            print(f"{name or url}: OK ({response.status_code})")
            return True
        else:
            print(f"{name or url}: Failed ({response.status_code})")
            return False
    except requests.exceptions.RequestException as e:
        print(f"{name or url}: Connection failed - {e}")
        return False

def main():
    base_url = "http://localhost:8001"

    # Test essential ARCP endpoints (note: /metrics requires admin auth)
    endpoints = [
        (f"{base_url}/health", "Health Endpoint"),
        (f"{base_url}/", "Root Endpoint"),  
        (f"{base_url}/docs", "API Documentation"),
    ]

    results = []
    print("ARCP Health Check")
    print("=" * 30)

    for url, name in endpoints:
        results.append(check_endpoint(url, name))

    # Summary
    passed = sum(results)
    total = len(results)
    success_rate = (passed / total) * 100

    print(f"\nSummary: {passed}/{total} ({success_rate:.1f}%)")

    if success_rate >= 80:
        print("Monitoring system is healthy!")
        sys.exit(0)
    else:
        print("Monitoring system has issues")
        sys.exit(1)

    if success_rate >= 80:
        print("Monitoring system is healthy!")
        sys.exit(0)
    else:
        print("Monitoring system has issues")
        sys.exit(1)

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Metrics collection script for ARCP monitoring.
Collects, parses, and analyzes ARCP metrics for validation and monitoring.
"""
import os
import json
import asyncio
from datetime import datetime
from arcp import ARCPClient

async def fetch_and_analyze_metrics_arcp(base_url: str):
    """Fetch metrics using ARCP client library and provide comprehensive analysis."""
    client = ARCPClient(base_url)

    try:
        # First, authenticate as admin
        print("1. Authenticating as admin...")
        await client.login_admin("ARCP", "ARCP")  # Use your actual admin credentials
        print("Admin authentication successful")

        # Fetch raw metrics with authentication
        print("2. Fetching metrics...")
        raw_metrics = await client.get_system_metrics()
        print(f"Metrics fetched successfully ({len(raw_metrics)} bytes)")

        # Create timestamped output directory
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_dir = f"metrics_collection_{timestamp}"
        os.makedirs(output_dir, exist_ok=True)

        # Save raw metrics
        with open(f"{output_dir}/raw_metrics.txt", 'w') as f:
            f.write(raw_metrics)

        # Analyze metrics
        print("3. Analyzing metrics...")
        lines = raw_metrics.split('\n')
        arcp_metrics = [line for line in lines if 'arcp_' in line and not line.startswith('#')]
        standard_metrics = [line for line in lines if any(prefix in line for prefix in ['process_', 'python_']) and not line.startswith('#')]

        # Create summary
        summary = {
            'timestamp': datetime.now().isoformat(),
            'total_metrics_lines': len([l for l in lines if l.strip() and not l.startswith('#')]),
            'arcp_specific_metrics': len(arcp_metrics),
            'standard_metrics': len(standard_metrics),
            'collection_successful': True
        }

        with open(f"{output_dir}/summary.json", 'w') as f:
            json.dump(summary, f, indent=2)

        print(f"Analysis complete:")
        print(f"  - Total metrics lines: {summary['total_metrics_lines']}")
        print(f"  - ARCP-specific metrics: {summary['arcp_specific_metrics']}")
        print(f"  - Standard metrics: {summary['standard_metrics']}")
        print(f"  - Output directory: {output_dir}")

        return True

    except Exception as e:
        print(f"Error collecting metrics: {e}")
        return False
    finally:
        await client.close()

def main():
    base_url = "http://localhost:8001"

    print("ARCP Metrics Collection")
    print("=" * 40)

    try:
        print("Using ARCP client library...")
        success = asyncio.run(fetch_and_analyze_metrics_arcp(base_url))
    except Exception as e:
        print(f"ARCP client failed ({e})")

    return 0 if success else 1

if __name__ == "__main__":
    exit(main())

#!/usr/bin/env python3
"""
Comprehensive ARCP monitoring validation script.
Tests all monitoring components and provides detailed reporting.
"""
import sys
from urllib import request

class ARCPMonitoringValidator:
    def __init__(self, base_url="http://localhost:8001"):
        self.base_url = base_url
        self.results = []

    def validate_endpoints(self):
        """Test all monitoring endpoints."""
        endpoints = ["/health", "/metrics", "/dashboard", "/docs"]
        for endpoint in endpoints:
            try:
                url = f"{self.base_url}{endpoint}"
                with request.urlopen(url, timeout=10) as response:
                    success = response.status == 200
                    self.results.append(("Endpoint " + endpoint, success))
            except Exception:
                self.results.append(("Endpoint " + endpoint, False))

    def validate_metrics_content(self):
        """Validate metrics contain expected ARCP metrics."""
        try:
            url = f"{self.base_url}/metrics"
            with request.urlopen(url, timeout=10) as response:
                content = response.read().decode()

            expected = [
                "arcp_agent_registrations_total",
                "arcp_active_agents",
                "arcp_requests_total", 
                "arcp_system_cpu_utilization_percent"
            ]

            for metric in expected:
                found = metric in content
                self.results.append(("Metric " + metric, found))

        except Exception:
            self.results.append(("Metrics validation", False))

    def run_validation(self):
        """Run all validations and report results."""
        print("ARCP Monitoring Validation")
        print("=" * 50)

        self.validate_endpoints()
        self.validate_metrics_content()

        passed = sum(1 for _, success in self.results if success)
        total = len(self.results)

        print(f"\nResults: {passed}/{total} ({passed/total*100:.1f}%)")

        for test, success in self.results:
            print(f"{success} {test}")

        return 0 if passed >= total * 0.8 else 1

if __name__ == "__main__":
    validator = ARCPMonitoringValidator()
    sys.exit(validator.run_validation())

import asyncio
from arcp import ARCPClient

class ARCPMonitor:
    def __init__(self, base_url: str, admin_username: str = "ARCP", admin_password: str = "ARCP"):
        self.client = ARCPClient(base_url)
        self.admin_username = admin_username
        self.admin_password = admin_password
        self._authenticated = False

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.close()

    async def ensure_authenticated(self):
        """Ensure we're authenticated for admin operations"""
        if not self._authenticated:
            await self.client.login_admin(self.admin_username, self.admin_password)
            self._authenticated = True

    async def get_health(self):
        """Get ARCP health status (public endpoint)"""
        return await self.client.get_system_info()

    async def get_stats(self):
        """Get ARCP statistics (admin required)"""
        await self.ensure_authenticated()
        return await self.client.get_system_stats()

    async def get_metrics(self):
        """Get ARCP metrics (admin required)"""
        await self.ensure_authenticated()
        return await self.client.get_system_metrics()

    async def monitor(self, interval: int = 30, duration: int = None):
        """Continuous monitoring with ARCP client"""
        await self.ensure_authenticated()

        await self.client.monitor_system(
            interval=interval, 
            duration=duration,
            callback=self.metrics_callback
        )

    def metrics_callback(self, snapshot):
        """Callback for processing metrics snapshots"""
        print(f"\n[{snapshot.timestamp}] System Metrics:")

        # Extract key metrics
        cpu_util = snapshot.resource_utilization.get('cpu', 'N/A')
        memory_util = snapshot.resource_utilization.get('memory', 'N/A')

        print(f"  CPU: {cpu_util}%")
        print(f"  Memory: {memory_util}%")
        print(f"  Active Agents: {snapshot.agent_stats.get('total_agents', 0)}")

# Usage - Modern ARCP client approach
async def main():
    async with ARCPMonitor("http://localhost:8001") as monitor:
        # Get one-time status
        health = await monitor.get_health()
        print(f"System Status: {health.get('status', 'Unknown')}")

        # Start continuous monitoring
        await monitor.monitor(interval=30, duration=300)  # Monitor for 5 minutes

asyncio.run(main())

from prometheus_client import Counter, Histogram, Gauge

# Use appropriate metric types
REQUEST_COUNT = Counter('requests_total', 'Total requests', ['method', 'endpoint'])
REQUEST_DURATION = Histogram('request_duration_seconds', 'Request duration', buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0])
ACTIVE_CONNECTIONS = Gauge('active_connections', 'Active connections')

# Batch metric updates
class MetricsBatcher:
    def __init__(self, batch_size=100):
        self.batch_size = batch_size
        self.batch = []

    def add_metric(self, metric, value, labels=None):
        self.batch.append((metric, value, labels))

        if len(self.batch) >= self.batch_size:
            self.flush()

    def flush(self):
        for metric, value, labels in self.batch:
            if labels:
                metric.labels(**labels).set(value)
            else:
                metric.set(value)
        self.batch.clear()