Python pour les sysadmins : automatiser son infrastructure efficacement

En tant que sysadmin, l'automatisation reste votre meilleur allié. Bash excelle pour les tâches simples et les pipes de commandes. Python prend le relais sur les scripts complexes, la gestion des erreurs et l'intégration API. Ce guide couvre les bibliothèques essentielles et des cas d'usage concrets pour automatiser votre infrastructure.

1. Pourquoi Python plutôt que Bash ?, Comparaison et quand choisir quoi
2. Les bibliothèques essentielles, Paramiko, Fabric, psutil, requests, Jinja2
3. Scripts pratiques, Inventaire serveurs, monitoring, intégration API
4. Structurer ses scripts, Argparse, logging, gestion d'erreurs
5. Perspectives et évolutions

Les deux ont leur place. Bash brille pour les tâches ponctuelles et l'intégration native de commandes système. Python devient indispensable dès que vous avez besoin de logique métier, d'appels API, de structure et de maintenabilité.

Comparaison pratique

Cas Python : Inventaire serveurs, monitoring avec métriques, déploiement automatisé, ETL logs, intégration API. Cas Bash : One-liners, pipes simples, tests conditionnels rapides, scripts système légers.

Collectez les infos système sur plusieurs serveurs.

import paramiko
import csv
from pathlib import Path

def ssh_exec(hostname, username, key_path, command):
    """Exécute une commande sur un serveur via SSH"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname, username=username, key_filename=key_path, timeout=10)

    stdin, stdout, stderr = ssh.exec_command(command)
    result = stdout.read().decode().strip()
    ssh.close()
    return result

def inventory_servers(servers_file):
    """Collecte l'inventaire de tous les serveurs"""
    inventory = []

    with open(servers_file) as f:
        servers = [line.strip() for line in f if line.strip()]

    for hostname in servers:
        try:
            uptime = ssh_exec(hostname, 'admin', '~/.ssh/id_rsa', 'uptime')
            disk = ssh_exec(hostname, 'admin', '~/.ssh/id_rsa', 'df -h / | tail -1 | awk \'{print $5}\'')
            kernel = ssh_exec(hostname, 'admin', '~/.ssh/id_rsa', 'uname -r')

            inventory.append({
                'hostname': hostname,
                'uptime': uptime,
                'disk_usage': disk,
                'kernel': kernel
            })
        except Exception as e:
            print(f"Erreur {hostname}: {e}")

    # Sauvegarde en CSV
    with open('inventory.csv', 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=['hostname', 'uptime', 'disk_usage', 'kernel'])
        writer.writeheader()
        writer.writerows(inventory)

    print(f"Inventaire de {len(inventory)} serveurs sauvegardé")

if __name__ == '__main__':
    inventory_servers('servers.txt')

Générez des alertes basées sur les métriques de votre infrastructure.

import psutil
import json
from datetime import datetime

def check_system_health():
    """Vérifie la santé du système et retourne les métriques"""
    metrics = {
        'timestamp': datetime.now().isoformat(),
        'cpu_percent': psutil.cpu_percent(interval=1),
        'memory': {
            'percent': psutil.virtual_memory().percent,
            'used_gb': psutil.virtual_memory().used / (1024**3),
            'total_gb': psutil.virtual_memory().total / (1024**3)
        },
        'disk': {
            'percent': psutil.disk_usage('/').percent,
            'free_gb': psutil.disk_usage('/').free / (1024**3)
        },
        'process_count': len(psutil.pids()),
        'alerts': []
    }

    # Alertes
    if metrics['cpu_percent'] > 80:
        metrics['alerts'].append(f"CPU élevé: {metrics['cpu_percent']}%")
    if metrics['memory']['percent'] > 85:
        metrics['alerts'].append(f"Mémoire critique: {metrics['memory']['percent']}%")
    if metrics['disk']['percent'] > 90:
        metrics['alerts'].append(f"Disque presque plein: {metrics['disk']['percent']}%")

    return metrics

def send_webhook(metrics, webhook_url):
    """Envoie les métriques à un webhook (ex: monitoring SHPV)"""
    import requests
    try:
        requests.post(webhook_url, json=metrics, timeout=5)
    except requests.RequestException as e:
        print(f"Erreur webhook: {e}")

if __name__ == '__main__':
    health = check_system_health()
    print(json.dumps(health, indent=2))

    # Optionnel: envoyer à votre système de monitoring
    # send_webhook(health, 'https://monitoring.example.com/api/metrics')

Interagissez avec vos APIs d'infrastructure (load balancers, cloud providers, etc).

import requests
import logging
from typing import Optional, Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class APIClient:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}

    def get(self, endpoint: str) -> Optional[Dict[str, Any]]:
        """GET request avec gestion d'erreur"""
        try:
            response = requests.get(
                f"{self.base_url}/{endpoint}",
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            logger.error(f"Erreur GET {endpoint}: {e}")
            return None

    def post(self, endpoint: str, data: Dict) -> Optional[Dict[str, Any]]:
        """POST request"""
        try:
            response = requests.post(
                f"{self.base_url}/{endpoint}",
                json=data,
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            logger.error(f"Erreur POST {endpoint}: {e}")
            return None

# Usage
if __name__ == '__main__':
    client = APIClient('https://api.cloud.example.com', 'your_token_here')

    # Récupérer instances
    instances = client.get('compute/instances')
    if instances:
        logger.info(f"Instances trouvées: {len(instances.get('data', []))}")

    # Créer une ressource
    new_instance = {
        'name': 'web-server-01',
        'size': 'small',
        'region': 'fr-toulouse'
    }
    result = client.post('compute/instances', new_instance)

Utiliser argparse pour un CLI propre

import argparse
import sys

def main():
    parser = argparse.ArgumentParser(
        description='Outil d\'administration système',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog='Exemples:\n  %(prog)s inventory --format csv\n  %(prog)s monitor --alert-cpu 80'
    )

    subparsers = parser.add_subparsers(dest='command', help='Commande')

    # Subcommande inventory
    inv_parser = subparsers.add_parser('inventory', help='Collecter inventaire')
    inv_parser.add_argument('--servers', required=True, help='Fichier serveurs')
    inv_parser.add_argument('--format', choices=['csv', 'json'], default='csv')

    # Subcommande monitor
    mon_parser = subparsers.add_parser('monitor', help='Monitoring temps réel')
    mon_parser.add_argument('--alert-cpu', type=int, default=80, help='Seuil CPU %')
    mon_parser.add_argument('--interval', type=int, default=5, help='Intervalle secondes')

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    if args.command == 'inventory':
        print(f"Inventaire depuis {args.servers} en format {args.format}")
    elif args.command == 'monitor':
        print(f"Monitoring avec seuil CPU à {args.alert_cpu}%")

if __name__ == '__main__':
    main()

Logging structuré

import logging
import logging.handlers

def setup_logging(logfile='app.log', level=logging.INFO):
    """Configure les logs avec rotation"""
    logger = logging.getLogger('sysadmin')
    logger.setLevel(level)

    # Fichier avec rotation
    file_handler = logging.handlers.RotatingFileHandler(
        logfile, maxBytes=10*1024*1024, backupCount=5
    )

    # Format structuré
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    return logger

# Usage
logger = setup_logging('sysadmin.log')
logger.info('Script démarré')
logger.warning('Attention: mémoire élevée')
logger.error('Erreur de connexion SSH')

Voici les mêmes cas en Bash (souvent plus complexes) :

# Inventaire basique (Bash)
while IFS= read -r host; do
    uptime=$(ssh admin@$host uptime 2>/dev/null)
    disk=$(ssh admin@$host 'df -h / | tail -1 | awk "{print $5}"' 2>/dev/null)
    echo "$host,$uptime,$disk" >> inventory.csv
done < servers.txt

# Monitoring (Bash)
while true; do
    cpu=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
    if (( $(echo "$cpu > 80" | bc -l) )); then
        echo "ALERTE: CPU à $cpu%"
    fi
    sleep 5
done

# API avec curl (Bash)
token="your_token"
curl -H "Authorization: Bearer $token" \
     -H "Content-Type: application/json" \
     https://api.cloud.example.com/compute/instances

Vous voyez : gestion d'erreurs minimale, parsing fragile, moins de portabilité.

Pour approfondir votre automatisation :

• Orchestration : Ansible pour le déploiement multi-serveurs
• Bash avancé : Bash scripting pour les cas légers
• Infrastructure as Code : Terraform, CloudFormation pour l'IaC

• Documentation Python officielle
• Fabric: SSH deployment in Python
• psutil: cross-platform system monitoring

Python est devenu le langage de facto pour l'automatisation d'infrastructure. Ses bibliothèques (paramiko, fabric, psutil, requests) servent à écrire des scripts maintenables et portables d'un OS à l'autre.

Chez SHPV, l'automatisation décide de la tenue de votre infrastructure. Que vous orchestriez des serveurs sur nos infrastructures d'hébergement, supervisiez des instances cloud ou branchiez vos APIs monitoring, Python donne la structure attendue sans sacrifier les temps de réponse.

Commencez par les bibliothèques essentielles, structurez avec argparse et logging, puis évoluez vers des outils comme Fabric pour l'orchestration multi-serveurs.