pip install requests pandas matplotlib seaborn collections

import time
import datetime
import logging
from collections import deque
import random

import requests
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def create_monitoring_state(
    service_name: str, url: str, window_size: int = 10
) -> dict:
    """
    Initializes the monitoring state dictionary for a given service.

    Parameters
    ----------
    service_name : str
        The name of the service to monitor.
    url : str
        The endpoint URL of the service to be monitored.
    window_size : int, optional
        The number of recent latency measurements to store in the sliding window,
        defaults to 10.

    Returns
    -------
    dict
        An initialized state dictionary for the service.
    """
    state = {
        "service_name": service_name,
        "url": url,
        "status": "UNKNOWN",  # e.g., 'UP', 'DOWN', 'DEGRADED'
        "last_check_time": None,
        "last_latency_ms": None,
        "latency_history": deque(maxlen=window_size),
        "uptime_downtime_events": [],  # Stores (timestamp, status) tuples
        "consecutive_failures": 0,
        "total_checks": 0,
        "successful_checks": 0,
    }
    logger.info(f"Monitoring state initialized for service: {service_name} with URL: {url}")
    return state

def check_service_status(
    state: dict, max_retries: int = 3, initial_backoff: float = 0.5
) -> dict:
    """
    Checks the status of a service by making an HTTP GET request with exponential backoff.

    Parameters
    ----------
    state : dict
        Current monitoring state dictionary.
    max_retries : int, optional
        Maximum number of retries for the HTTP request, defaults to 3.
    initial_backoff : float, optional
        Initial delay in seconds before the first retry, defaults to 0.5.

    Returns
    -------
    dict
        Updated monitoring state dictionary.
    """
    service_name = state["service_name"]
    url = state["url"]
    retries = 0
    current_backoff = initial_backoff
    success = False
    latency_ms = None

    state["total_checks"] += 1

    while retries <= max_retries:
        try:
            start_time = time.time()
            logger.debug(
                f"Attempting to reach {service_name} at {url} (Attempt {retries + 1}/{max_retries + 1})"
            )
            response = requests.get(url, timeout=5)  # 5-second timeout
            end_time = time.time()
            latency_ms = (end_time - start_time) * 1000

            if response.status_code == 200:
                logger.info(f"{service_name} is UP. Latency: {latency_ms:.2f}ms")
                state["status"] = "UP"
                state["last_latency_ms"] = latency_ms
                state["latency_history"].append(latency_ms)
                state["successful_checks"] += 1
                state["consecutive_failures"] = 0
                success = True
                break
            else:
                logger.warning(
                    f"{service_name} returned status code {response.status_code}. "
                    f"Retrying... (Attempt {retries + 1})"
                )
        except requests.exceptions.Timeout:
            logger.warning(
                f"{service_name} request timed out after 5 seconds. "
                f"Retrying... (Attempt {retries + 1})"
            )
        except requests.exceptions.ConnectionError:
            logger.warning(
                f"Connection error for {service_name}. "
                f"Retrying... (Attempt {retries + 1})"
            )
        except requests.exceptions.RequestException as e:
            logger.error(
                f"An unexpected request error occurred for {service_name}: {e}. "
                f"Retrying... (Attempt {retries + 1})"
            )

        retries += 1
        if retries <= max_retries:
            jitter = random.uniform(0.5, 1.5)  # Add random jitter
            sleep_time = current_backoff * jitter
            logger.debug(
                f"Waiting {sleep_time:.2f}s before next retry for {service_name}"
            )
            time.sleep(sleep_time)
            current_backoff *= 2  # Exponential increase

    state["last_check_time"] = datetime.datetime.now()

    if not success:
        state["status"] = "DOWN"
        state["last_latency_ms"] = None
        state["consecutive_failures"] += 1
        logger.error(
            f"{service_name} is DOWN after {max_retries + 1} attempts."
        )

    # Log uptime/downtime event if status changed
    current_status = state["status"]
    if not state["uptime_downtime_events"] or \
       state["uptime_downtime_events"][-1][1] != current_status:
        state["uptime_downtime_events"].append((state["last_check_time"], current_status))

    return state

def summarize_monitoring_metrics(state: dict) -> dict:
    """
    Calculates and returns key monitoring metrics from the service's state.

    Parameters
    ----------
    state : dict
        Current monitoring state dictionary.

    Returns
    -------
    dict
        A dictionary containing summarized metrics.
    """
    total_checks = state["total_checks"]
    successful_checks = state["successful_checks"]
    uptime_percentage = (successful_checks / total_checks * 100) if total_checks > 0 else 0.0
    average_latency_ms = (sum(state["latency_history"]) / len(state["latency_history"])) if state["latency_history"] else None

    summary = {
        "service_name": state["service_name"],
        "current_status": state["status"],
        "total_checks": total_checks,
        "successful_checks": successful_checks,
        "uptime_percentage": round(uptime_percentage, 2),
        "average_latency_ms": round(average_latency_ms, 2) if average_latency_ms is not None else None,
        "consecutive_failures": state["consecutive_failures"],
    }
    logger.info(f"Metrics summary for {state['service_name']}: {summary}")
    return summary

def simulate_monitoring(
    service_states: list[dict], iterations: int, check_interval_seconds: float = 5.0
) -> list[dict]:
    """
    Simulates monitoring of multiple services over a specified number of iterations.

    Parameters
    ----------
    service_states : list[dict]
        A list of initialized service state dictionaries.
    iterations : int
        The number of monitoring cycles to perform.
    check_interval_seconds : float, optional
        The delay between monitoring cycles in seconds, defaults to 5.0.

    Returns
    -------
    list[dict]
        The updated list of service state dictionaries after the simulation.
    """
    logger.info(f"Starting monitoring simulation for {len(service_states)} services over {iterations} iterations.")
    for i in range(iterations):
        logger.info(f"--- Monitoring Cycle {i + 1}/{iterations} ---")
        for state in service_states:
            state = check_service_status(state)
            # Ensure latency history is captured over time for plotting
            if state['last_latency_ms'] is not None:
                # Append current latency and check time for detailed history
                if 'full_latency_history' not in state:
                    state['full_latency_history'] = []
                state['full_latency_history'].append({'timestamp': state['last_check_time'], 'latency_ms': state['last_latency_ms']})
            else:
                # If down, still record the timestamp with a null latency
                if 'full_latency_history' not in state:
                    state['full_latency_history'] = []
                state['full_latency_history'].append({'timestamp': state['last_check_time'], 'latency_ms': None})

        if i < iterations - 1:
            # Add random jitter to the check interval
            jitter = random.uniform(0.8, 1.2)
            sleep_time = check_interval_seconds * jitter
            logger.debug(f"Sleeping for {sleep_time:.2f}s before next cycle.")
            time.sleep(sleep_time)

    logger.info("Monitoring simulation completed.")
    return service_states

services_to_monitor_configs = [
    {"name": "Dummy Local Service", "url": "https://www.google.com"}, # Simulating a local, always-up dummy service
    {"name": "Binance API", "url": "https://api.binance.com/api/v3/ping"},
    {"name": "Fake Down Service", "url": "http://this-url-does-not-exist-12345.com"} # This will simulate a perpetually down service
]

# Initialize monitoring states for each service
service_states = []
for config in services_to_monitor_configs:
    state = create_monitoring_state(config["name"], config["url"])
    service_states.append(state)

# Run the monitoring simulation
# Setting a shorter interval and fewer iterations for demonstration purposes
simulated_service_states = simulate_monitoring(service_states, iterations=10, check_interval_seconds=2.0)

logger.info("Simulation completed. Summarizing final states:")
for state in simulated_service_states:
    summary = summarize_monitoring_metrics(state)
    logger.info(f"  Service: {summary['service_name']}, Status: {summary['current_status']}, "
                f"Uptime: {summary['uptime_percentage']}%, Avg Latency: {summary['average_latency_ms']}ms")

def plot_monitoring_results(service_states: list[dict]) -> None:
    """
    Plots latency history and service status over time for all monitored services.

    Parameters
    ----------
    service_states : list[dict]
        A list of finalized service state dictionaries from the simulation.

    Returns
    -------
    None
        Displays plots directly.
    """
    num_services = len(service_states)
    plt.figure(figsize=(15, 5 * num_services))

    for i, state in enumerate(service_states):
        service_name = state["service_name"]
        full_history = state.get('full_latency_history', [])

        if not full_history:
            logger.warning(f"No full latency history data found for {service_name}. Skipping plot.")
            continue

        df_history = pd.DataFrame(full_history)
        df_history['timestamp'] = pd.to_datetime(df_history['timestamp'])
        df_history = df_history.set_index('timestamp')

        plt.subplot(num_services, 1, i + 1)
        sns.lineplot(x=df_history.index, y='latency_ms', data=df_history, marker='o', label=f'{service_name} Latency')

        # Mark downtime periods
        downtime_periods = df_history[df_history['latency_ms'].isnull()]
        if not downtime_periods.empty:
            for start, end in zip(downtime_periods.index[:-1], downtime_periods.index[1:]):
                plt.axvspan(start, end, color='red', alpha=0.3, label='_nolegend_')

            # For single downtime point or the very last one
            if len(downtime_periods) == 1:
                plt.axvline(x=downtime_periods.index[0], color='red', linestyle='--', alpha=0.5, label='Downtime')
            else:
                plt.axvline(x=downtime_periods.index[-1], color='red', linestyle='--', alpha=0.5, label='Downtime')

        plt.title(f'Latency and Uptime for {service_name}')
        plt.ylabel('Latency (ms)')
        plt.xlabel('Time')
        plt.grid(True)
        plt.legend()
        plt.tight_layout()

    plt.show()

plot_monitoring_results(simulated_service_states)

summary_data = []
for state in simulated_service_states:
    summary_data.append(summarize_monitoring_metrics(state))

df_summary = pd.DataFrame(summary_data)
print(df_summary.to_markdown(index=False))