Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Monitoreo y Mantenimiento

🧭 Navegación:

👁️ Vigilancia continua: El supervisor del almacén

Imagina que has automatizado completamente tu almacén: los productos se mueven solos, los inventarios se actualizan automáticamente y los pedidos se procesan sin intervención humana. Pero, ¿qué sucede si una máquina falla o un proceso se detiene inesperadamente? Necesitas un supervisor que vigile constantemente que todo funcione correctamente.

El monitoreo y mantenimiento son ese supervisor para tus scripts automatizados, asegurando que sigan funcionando correctamente y alertándote cuando algo va mal.

🤔 ¿Por qué es crucial el monitoreo?

Incluso los mejores scripts pueden fallar por razones inesperadas:

  • Cambios externos: APIs que cambian, sitios web que se actualizan
  • Problemas de recursos: Memoria insuficiente, disco lleno
  • Errores inesperados: Casos extremos no contemplados
  • Problemas de red: Interrupciones de conexión
  • Actualizaciones del sistema: Cambios en dependencias o el entorno

Sin un buen sistema de monitoreo, podrías no darte cuenta de que tus scripts han dejado de funcionar hasta que sea demasiado tarde.

📊 Estrategias de monitoreo

Existen varias estrategias para monitorear tus scripts automatizados:

1. Logging detallado

El logging es la base de cualquier sistema de monitoreo:

import logging
from datetime import datetime

# Configurar logging con timestamp
log_file = f"script_{datetime.now().strftime('%Y%m%d')}.log"
logging.basicConfig(
    filename=log_file,
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

# Usar logging en puntos clave
logging.info("Script iniciado")
try:
    # Operación importante
    result = perform_critical_operation()
    logging.info(f"Operación completada con resultado: {result}")
except Exception as e:
    logging.error(f"Error en operación crítica: {str(e)}")
    # Posiblemente enviar notificación
finally:
    logging.info("Script finalizado")

2. Notificaciones de errores

Configura tu script para enviar alertas cuando ocurran problemas:

Notificaciones por correo electrónico:

import smtplib
from email.message import EmailMessage

def send_error_email(error_message):
    """Envía un correo electrónico de alerta."""
    msg = EmailMessage()
    msg.set_content(f"Error en script automatizado:\n\n{error_message}")
    
    msg['Subject'] = 'ALERTA: Error en script automatizado'
    msg['From'] = 'tu_correo@gmail.com'
    msg['To'] = 'destinatario@example.com'
    
    # Configurar servidor SMTP (ejemplo para Gmail)
    server = smtplib.SMTP_SSL('smtp.gmail.com', 465)
    server.login("tu_correo@gmail.com", "tu_contraseña_o_token")
    server.send_message(msg)
    server.quit()

# Usar en el manejo de excepciones
try:
    # Código que podría fallar
    result = risky_operation()
except Exception as e:
    error_msg = f"Error: {str(e)}"
    logging.error(error_msg)
    send_error_email(error_msg)

Nota de seguridad: Para servicios como Gmail, usa contraseñas de aplicación en lugar de tu contraseña principal.

Notificaciones por servicios de mensajería:

Puedes usar servicios como Telegram, Slack o Discord para recibir notificaciones:

# Ejemplo con Telegram (requiere python-telegram-bot)
from telegram import Bot

def send_telegram_alert(message):
    """Envía una alerta a través de Telegram."""
    bot_token = 'tu_token_de_bot'
    chat_id = 'tu_chat_id'
    
    bot = Bot(token=bot_token)
    bot.send_message(chat_id=chat_id, text=message)

# Usar en caso de error
try:
    # Código que podría fallar
    process_data()
except Exception as e:
    error_msg = f"⚠️ ALERTA: Script falló con error: {str(e)}"
    send_telegram_alert(error_msg)

3. Archivos de estado y heartbeats

Una técnica simple pero efectiva es crear “heartbeats” (latidos) que indiquen que tu script sigue funcionando:

def update_heartbeat():
    """Actualiza el archivo de heartbeat con la hora actual."""
    with open("heartbeat.txt", "w") as f:
        f.write(datetime.now().isoformat())

# Llamar periódicamente durante la ejecución
update_heartbeat()

Luego puedes tener otro script que verifique si el heartbeat está actualizado:

def check_heartbeat(max_age_minutes=60):
    """Verifica si el heartbeat está actualizado."""
    try:
        with open("heartbeat.txt", "r") as f:
            last_beat = datetime.fromisoformat(f.read().strip())
        
        age = datetime.now() - last_beat
        if age.total_seconds() > max_age_minutes * 60:
            send_alert(f"Script inactivo por {age.total_seconds() / 60:.1f} minutos")
            return False
        return True
    except Exception:
        send_alert("No se pudo leer el heartbeat")
        return False

4. Servicios de monitoreo

Para scripts críticos, considera usar servicios de monitoreo profesionales:

  • Uptime Robot: Monitoreo gratuito de sitios web y endpoints
  • Healthchecks.io: Servicio especializado en monitorear tareas cron
  • Sentry: Monitoreo de errores en aplicaciones
  • Prometheus + Grafana: Solución completa de monitoreo y visualización

Ejemplo con Healthchecks.io:

import requests

def ping_healthcheck(success=True):
    """Notifica a Healthchecks.io sobre el estado del script."""
    hc_url = "https://hc-ping.com/tu-uuid-de-healthcheck"
    
    if success:
        requests.get(hc_url)
    else:
        requests.get(f"{hc_url}/fail")

# Al inicio del script
try:
    # Código principal
    main_process()
    # Al finalizar con éxito
    ping_healthcheck(success=True)
except Exception as e:
    logging.error(f"Error: {str(e)}")
    ping_healthcheck(success=False)
    raise

🔧 Mantenimiento preventivo

Además del monitoreo, es importante realizar mantenimiento preventivo:

1. Rotación de logs

Evita que los archivos de log crezcan indefinidamente:

import logging
from logging.handlers import RotatingFileHandler

# Configurar rotación de logs (máximo 5 archivos de 1MB cada uno)
handler = RotatingFileHandler(
    "app.log",
    maxBytes=1_000_000,  # 1MB
    backupCount=5
)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[handler, logging.StreamHandler()]
)

2. Limpieza de archivos temporales

Implementa rutinas para limpiar archivos temporales:

def cleanup_temp_files(directory, max_age_days=7):
    """Elimina archivos temporales más antiguos que max_age_days."""
    now = datetime.now()
    count = 0
    
    for filename in os.listdir(directory):
        if not filename.endswith('.tmp'):
            continue
            
        filepath = os.path.join(directory, filename)
        file_time = datetime.fromtimestamp(os.path.getmtime(filepath))
        age_days = (now - file_time).days
        
        if age_days > max_age_days:
            os.remove(filepath)
            count += 1
    
    logging.info(f"Limpieza completada: {count} archivos eliminados")

3. Verificación de dependencias

Verifica periódicamente que las dependencias estén actualizadas:

import pkg_resources
import requests

def check_dependencies():
    """Verifica si hay actualizaciones disponibles para las dependencias."""
    outdated = []
    
    for dist in pkg_resources.working_set:
        package_name = dist.project_name
        current_version = dist.version
        
        try:
            # Consultar la última versión en PyPI
            response = requests.get(f"https://pypi.org/pypi/{package_name}/json")
            if response.status_code == 200:
                latest_version = response.json()["info"]["version"]
                
                if latest_version != current_version:
                    outdated.append(f"{package_name}: {current_version} -> {latest_version}")
        except Exception:
            pass
    
    if outdated:
        message = "Dependencias desactualizadas:\n" + "\n".join(outdated)
        logging.warning(message)
        # Posiblemente enviar notificación

4. Pruebas periódicas

Implementa pruebas automáticas que verifiquen que todo sigue funcionando:

def run_self_tests():
    """Ejecuta pruebas básicas para verificar la funcionalidad."""
    tests_passed = True
    
    # Prueba 1: Verificar conexión a la base de datos
    try:
        db_connection = connect_to_database()
        logging.info("Test de conexión a BD: OK")
    except Exception as e:
        logging.error(f"Test de conexión a BD: FALLÓ - {str(e)}")
        tests_passed = False
    
    # Prueba 2: Verificar acceso a API externa
    try:
        api_response = call_external_api()
        if api_response.status_code == 200:
            logging.info("Test de API externa: OK")
        else:
            logging.error(f"Test de API externa: FALLÓ - Código {api_response.status_code}")
            tests_passed = False
    except Exception as e:
        logging.error(f"Test de API externa: FALLÓ - {str(e)}")
        tests_passed = False
    
    return tests_passed

📈 Monitoreo de rendimiento

Para scripts que procesan grandes cantidades de datos o se ejecutan frecuentemente, es útil monitorear el rendimiento:

1. Tiempo de ejecución

import time

def measure_execution_time(func):
    """Decorador para medir el tiempo de ejecución de una función."""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        execution_time = end_time - start_time
        logging.info(f"Función {func.__name__} ejecutada en {execution_time:.2f} segundos")
        return result
    return wrapper

@measure_execution_time
def process_large_dataset():
    # Procesamiento que podría ser lento
    pass

2. Uso de recursos

import psutil

def log_resource_usage():
    """Registra el uso actual de recursos."""
    process = psutil.Process()
    
    # Uso de memoria
    memory_info = process.memory_info()
    memory_mb = memory_info.rss / (1024 * 1024)
    
    # Uso de CPU
    cpu_percent = process.cpu_percent(interval=1)
    
    # Uso de disco
    disk = psutil.disk_usage('/')
    disk_percent = disk.percent
    
    logging.info(f"Recursos: Memoria={memory_mb:.1f}MB, CPU={cpu_percent}%, Disco={disk_percent}%")

🔄 Actualizaciones y mejoras

Mantén tus scripts actualizados y mejorándolos continuamente:

1. Control de versiones

Usa un sistema de control de versiones como Git para rastrear cambios:

# Inicializar repositorio
git init

# Añadir archivos
git add .

# Crear commit inicial
git commit -m "Versión inicial del script de automatización"

# Para cada actualización
git add script.py
git commit -m "Mejora: añadida verificación de errores en la función X"

2. Documentación de cambios

Mantén un archivo CHANGELOG.md para documentar cambios importantes:

# Changelog

## [1.2.0] - 2023-07-15
### Añadido
- Notificaciones por Telegram cuando ocurren errores
- Rotación automática de logs

### Corregido
- Error al procesar archivos con nombres especiales

## [1.1.0] - 2023-06-20
### Añadido
- Soporte para nuevos formatos de archivo
- Mejor manejo de errores

### Cambiado
- Optimizado el procesamiento de archivos grandes

3. Actualizaciones graduales

Cuando implementes cambios importantes:

  1. Haz una copia de seguridad del script actual
  2. Prueba exhaustivamente los cambios en un entorno de desarrollo
  3. Implementa gradualmente en producción
  4. Monitorea de cerca después de la actualización
  5. Ten un plan de rollback en caso de problemas

🔍 Comprueba tu comprensión

  1. ¿Por qué es importante implementar logging en scripts automatizados?
  2. ¿Qué ventajas ofrece un servicio como Healthchecks.io sobre un sistema de heartbeat casero?
  3. ¿Cómo podrías detectar si un script está consumiendo demasiada memoria?
  4. ¿Por qué es importante la rotación de logs?
  5. ¿Qué estrategias puedes usar para ser notificado inmediatamente si un script falla?

🛠️ Ejercicio práctico

Mejora uno de los scripts de automatización que creamos en el capítulo anterior añadiendo:

  1. Logging detallado con rotación de archivos
  2. Un sistema de notificación por correo electrónico o Telegram
  3. Medición y registro del tiempo de ejecución
  4. Verificación de recursos disponibles antes de ejecutar operaciones intensivas
  5. Un archivo de heartbeat que se actualice periódicamente

📝 Resumen

En esta sección, has aprendido:

  • La importancia del monitoreo para scripts automatizados
  • Diferentes estrategias de monitoreo: logging, notificaciones, heartbeats y servicios profesionales
  • Técnicas de mantenimiento preventivo para evitar problemas
  • Cómo monitorear el rendimiento de tus scripts
  • Buenas prácticas para actualizar y mejorar tus scripts a lo largo del tiempo

Con estos conocimientos, puedes asegurarte de que tus scripts automatizados no solo funcionen correctamente al principio, sino que sigan siendo confiables y eficientes a largo plazo.

En la siguiente sección, haremos un resumen del capítulo y veremos cómo todos estos conceptos se integran para crear sistemas de automatización robustos y profesionales.