pip install requests tenacity pandas matplotlib seaborn

import logging
import time
import random
from collections import deque
from datetime import datetime, timedelta
from typing import Dict, Any, List, Tuple

import requests
from tenacity import retry, stop_after_attempt, wait_exponential
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

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

def create_monitor_state(
    service_url: str,
    monitor_interval_seconds: int,
    uptime_threshold_percent: float,
    rolling_window_size: int
) -> Dict[str, Any]:
    """
    Initializes the monitoring system's state dictionary.

    Parameters
    ----------
    service_url : str
        The URL of the service to monitor.
    monitor_interval_seconds : int
        The interval in seconds between monitoring checks.
    uptime_threshold_percent : float
        The minimum acceptable uptime percentage for a 'healthy' signal.
    rolling_window_size : int
        The number of recent check results to keep in memory for rolling calculations.

    Returns
    -------
    dict
        An initialized state dictionary for the monitoring system.
    """
    logger.info(f"Initializing monitor state for service: {service_url}")
    state = {
        "service_url": service_url,
        "monitor_interval_seconds": monitor_interval_seconds,
        "uptime_threshold_percent": uptime_threshold_percent,
        "rolling_window_size": rolling_window_size,
        "last_check_time": None,
        "check_history": deque(maxlen=rolling_window_size),  # Stores (timestamp, status_code, is_up)
        "signal_history": [],  # Stores (timestamp, signal_status, uptime_percent)
        "alert_history": [],  # Stores (timestamp, alert_message)
        "metrics": {
            "total_checks": 0,
            "successful_checks": 0,
            "failed_checks": 0,
            "total_alerts": 0
        }
    }
    logger.debug("Monitor state initialized successfully.")
    return state

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def check_service_status(state: Dict[str, Any], timeout: float = 10.0) -> Dict[str, Any]:
    """
    Checks the status of the service URL defined in the state.
    Includes retry logic with exponential backoff and random jitter.

    Parameters
    ----------
    state : dict
        Current state dictionary containing 'service_url'.
    timeout : float, optional
        The maximum time in seconds to wait for a response, defaults to 10.0.

    Returns
    -------
    dict
        Updated state with the latest check result.
    """
    service_url = state["service_url"]
    current_time = datetime.now()
    status_code = -1
    is_up = False
    logger.info(f"Attempting to check service status for {service_url}")

    try:
        # Add random jitter to prevent 'thundering herd' on retries
        time.sleep(random.uniform(0, 0.5))
        response = requests.get(service_url, timeout=timeout)
        status_code = response.status_code
        is_up = 200 <= status_code < 300
        logger.info(f"Service {service_url} responded with status {status_code}. Up: {is_up}")
    except requests.exceptions.Timeout:
        logger.warning(f"Service {service_url} timed out after {timeout} seconds.")
    except requests.exceptions.ConnectionError as e:
        logger.error(f"Connection error for {service_url}: {e}")
    except requests.exceptions.RequestException as e:
        logger.error(f"An unexpected request error occurred for {service_url}: {e}")
    finally:
        state["last_check_time"] = current_time
        state["check_history"].append((current_time, status_code, is_up))
        state["metrics"]["total_checks"] += 1
        if is_up:
            state["metrics"]["successful_checks"] += 1
        else:
            state["metrics"]["failed_checks"] += 1
        logger.debug(f"Check history updated. Current length: {len(state['check_history'])}")
    return state

def evaluate_signal(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Evaluates a signal based on the recent service check history.

    Parameters
    ----------
    state : dict
        Current state dictionary with 'check_history' and 'uptime_threshold_percent'.

    Returns
    -------
    dict
        Updated state with the latest signal status.
    """
    current_time = datetime.now()
    uptime_checks = [res[2] for res in state["check_history"]]

    if not uptime_checks:
        logger.warning("No check history available to evaluate signal.")
        uptime_percent = 0.0
    else:
        uptime_percent = (sum(uptime_checks) / len(uptime_checks)) * 100

    signal_status = "healthy" if uptime_percent >= state["uptime_threshold_percent"] else "unhealthy"

    state["signal_history"].append((current_time, signal_status, uptime_percent))
    logger.info(f"Signal evaluated: Uptime {uptime_percent:.2f}% (Threshold: {state['uptime_threshold_percent']:.2f}%). Status: {signal_status}")
    return state

def send_alert(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Simulates sending an alert if the latest signal is 'unhealthy'.
    Prevents alert storms by checking previous signal states.

    Parameters
    ----------
    state : dict
        Current state dictionary with 'signal_history' and 'alert_history'.

    Returns
    -------
    dict
        Updated state, potentially with a new alert.
    """
    if not state["signal_history"]:
        logger.debug("No signal history to evaluate for alerting.")
        return state

    latest_signal_time, latest_signal_status, _ = state["signal_history"][-1]

    # Check if the previous signal was also 'unhealthy' to avoid repeated alerts
    previous_signal_status = "healthy" # Assume healthy if no previous signal
    if len(state["signal_history"]) > 1:
        _, previous_signal_status, _ = state["signal_history"][-2]

    if latest_signal_status == "unhealthy" and previous_signal_status == "healthy":
        alert_message = f"ALERT! Service {state['service_url']} is UNHEALTHY! Uptime below {state['uptime_threshold_percent']}%. " \
                        f"Current uptime: {state['signal_history'][-1][2]:.2f}%"
        state["alert_history"].append((latest_signal_time, alert_message))
        state["metrics"]["total_alerts"] += 1
        logger.critical(alert_message)
    elif latest_signal_status == "healthy" and previous_signal_status == "unhealthy":
        recovery_message = f"RECOVERY! Service {state['service_url']} is now HEALTHY. " \
                           f"Current uptime: {state['signal_history'][-1][2]:.2f}%"
        state["alert_history"].append((latest_signal_time, recovery_message))
        logger.info(recovery_message)
    else:
        logger.debug(f"No new alert condition for current signal status: {latest_signal_status}")

    return state

def summarize_metrics(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Summarizes the collected monitoring metrics.

    Parameters
    ----------
    state : dict
        Current state dictionary with 'metrics'.

    Returns
    -------
    dict
        A dictionary containing summarized metrics.
    """
    metrics = state["metrics"]
    total_checks = metrics["total_checks"]
    successful_checks = metrics["successful_checks"]

    overall_uptime = (successful_checks / total_checks) * 100 if total_checks > 0 else 0.0

    summary = {
        "total_checks": total_checks,
        "successful_checks": successful_checks,
        "failed_checks": metrics["failed_checks"],
        "overall_uptime_percent": overall_uptime,
        "total_alerts_triggered": metrics["total_alerts"],
        "last_check_time": state["last_check_time"]
    }
    logger.info("Monitoring metrics summarized.")
    logger.debug(f"Summary: {summary}")
    return summary

def run_monitoring_cycle(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Executes a single monitoring cycle: checks status, evaluates signal, sends alert.

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

    Returns
    -------
    dict
        Updated state after one complete monitoring cycle.
    """
    logger.info("Starting a new monitoring cycle...")
    state = check_service_status(state)
    state = evaluate_signal(state)
    state = send_alert(state)
    logger.info("Monitoring cycle completed.")
    return state

# Define simulation parameters
TARGET_SERVICE_URL = "https://www.google.com" # A more reliable test URL
FAILING_SERVICE_URL = "https://httpbin.org/status/500" # Simulate a failure
SIMULATED_MONITOR_INTERVAL = 1 # seconds (for quick simulation)
SIMULATION_DURATION_MINUTES = 1 # Changed from 1 to 2 minutes
UPTIME_THRESHOLD = 90.0 # percent
ROLLING_WINDOW = 5 # number of checks

# Initialize state
monitor_state = create_monitor_state(
    service_url=TARGET_SERVICE_URL,
    monitor_interval_seconds=SIMULATED_MONITOR_INTERVAL,
    uptime_threshold_percent=UPTIME_THRESHOLD,
    rolling_window_size=ROLLING_WINDOW
)

print(f"\n--- Starting Monitoring Simulation for {SIMULATION_DURATION_MINUTES} minutes ---")

start_time = datetime.now()
cycle_count = 0

while (datetime.now() - start_time).total_seconds() < (SIMULATION_DURATION_MINUTES * 60):
    cycle_count += 1
    logger.info(f"Simulation Cycle: {cycle_count}")

    # Simulate service going down and coming back up
    if cycle_count == 5 or cycle_count == 15 or cycle_count == 25: # At certain cycles, simulate failure
        monitor_state['service_url'] = FAILING_SERVICE_URL
        logger.warning(f"SIMULATION: Service URL changed to {FAILING_SERVICE_URL} to simulate failure.")
    elif cycle_count == 10 or cycle_count == 20 or cycle_count == 30: # After some cycles, revert to healthy
        monitor_state['service_url'] = TARGET_SERVICE_URL
        logger.info(f"SIMULATION: Service URL reverted to {TARGET_SERVICE_URL} to simulate recovery.")

    monitor_state = run_monitoring_cycle(monitor_state)
    time.sleep(monitor_state["monitor_interval_seconds"] + random.uniform(0, 0.2)) # Add jitter to interval

print(f"\n--- Simulation Finished after {cycle_count} cycles ---")

final_summary = summarize_metrics(monitor_state)
display(pd.DataFrame([final_summary]))

# Prepare data for plotting
check_df = pd.DataFrame(monitor_state["check_history"], columns=["timestamp", "status_code", "is_up"])
check_df["is_up_int"] = check_df["is_up"].astype(int) # Convert boolean to int for plotting

signal_df = pd.DataFrame(monitor_state["signal_history"], columns=["timestamp", "signal_status", "uptime_percent"])
alert_df = pd.DataFrame(monitor_state["alert_history"], columns=["timestamp", "alert_message"])

# Plotting
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(15, 10), sharex=True)

# Subplot 1: Service Check Status
sns.scatterplot(x='timestamp', y='is_up_int', data=check_df, hue='is_up', style='is_up', s=100, ax=ax1, palette={True: 'green', False: 'red'})
ax1.set_title('Service Check Status Over Time')
ax1.set_ylabel('Service Up (1=True, 0=False)')
ax1.set_yticks([0, 1])
ax1.axhline(y=1, color='green', linestyle='--', alpha=0.5, label='Service Expected Up')
ax1.grid(True, linestyle='--', alpha=0.6)

# Subplot 2: Uptime Percentage and Signal Status
sns.lineplot(x='timestamp', y='uptime_percent', data=signal_df, ax=ax2, marker='o', label='Rolling Uptime %')
ax2.axhline(y=UPTIME_THRESHOLD, color='orange', linestyle='-', label=f'Uptime Threshold ({UPTIME_THRESHOLD}%)')

# Annotate alerts on the signal plot
for _, row in alert_df.iterrows():
    alert_type = 'ALERT' if 'UNHEALTHY' in row['alert_message'] else 'RECOVERY'
    color = 'red' if alert_type == 'ALERT' else 'blue'
    marker = 'v' if alert_type == 'ALERT' else '^'
    ax2.axvline(x=row['timestamp'], color=color, linestyle=':', alpha=0.7)
    ax2.text(row['timestamp'], ax2.get_ylim()[1] * 0.95, alert_type, rotation=90, va='top', ha='center', color=color, fontsize=9)

ax2.set_title('Rolling Uptime Percentage and Signal Evaluation')
ax2.set_xlabel('Time')
ax2.set_ylabel('Uptime Percentage')
ax2.set_ylim(-5, 105) # Give some padding
ax2.legend()
ax2.grid(True, linestyle='--', alpha=0.6)

plt.tight_layout()
plt.show()