Hands-on Lab

#Lab 01: Python Automation Script

Build a practical Python automation script for server health monitoring.

#🎯 Objectives

  • Create a reusable health check script
  • Handle errors gracefully
  • Generate reports in multiple formats

#📋 Prerequisites

  • Python 3.8+ installed
  • Basic Python knowledge

#🔬 Lab Steps

#Step 1: Project Setup

bash
1mkdir health-checker
2cd health-checker
3python -m venv venv
4source venv/bin/activate  # Windows: venv\Scripts\activate
5pip install requests pyyaml

#Step 2: Create the Script

Create health_checker.py:

python
1#!/usr/bin/env python3
2"""
3Server Health Checker
4A DevOps automation script for monitoring server health.
5"""
6
7import socket
8import json
9import yaml
10import argparse
11from datetime import datetime
12from concurrent.futures import ThreadPoolExecutor
13
14
15def check_port(host: str, port: int, timeout: int = 5) -> dict:
16    """Check if a port is open on a host."""
17    try:
18        start = datetime.now()
19        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
20        sock.settimeout(timeout)
21        result = sock.connect_ex((host, port))
22        sock.close()
23        latency = (datetime.now() - start).total_seconds() * 1000
24        
25        return {
26            "host": host,
27            "port": port,
28            "status": "healthy" if result == 0 else "unhealthy",
29            "latency_ms": round(latency, 2),
30            "checked_at": datetime.now().isoformat()
31        }
32    except Exception as e:
33        return {
34            "host": host,
35            "port": port,
36            "status": "error",
37            "error": str(e),
38            "checked_at": datetime.now().isoformat()
39        }
40
41
42def load_servers(config_file: str) -> list:
43    """Load server list from YAML config."""
44    with open(config_file, 'r') as f:
45        config = yaml.safe_load(f)
46    return config.get('servers', [])
47
48
49def check_all_servers(servers: list, max_workers: int = 10) -> list:
50    """Check all servers in parallel."""
51    results = []
52    
53    with ThreadPoolExecutor(max_workers=max_workers) as executor:
54        futures = []
55        for server in servers:
56            future = executor.submit(
57                check_port,
58                server['host'],
59                server.get('port', 80)
60            )
61            futures.append((server.get('name', server['host']), future))
62        
63        for name, future in futures:
64            result = future.result()
65            result['name'] = name
66            results.append(result)
67    
68    return results
69
70
71def print_report(results: list):
72    """Print formatted report to console."""
73    print("\n" + "=" * 60)
74    print("SERVER HEALTH REPORT")
75    print("=" * 60)
76    
77    for result in results:
78        status_icon = "✅" if result['status'] == "healthy" else "❌"
79        print(f"{status_icon} {result['name']}: {result['status']}")
80        if result['status'] == "healthy":
81            print(f"   Latency: {result['latency_ms']}ms")
82    
83    healthy = sum(1 for r in results if r['status'] == "healthy")
84    print("\n" + "-" * 60)
85    print(f"Summary: {healthy}/{len(results)} servers healthy")
86    print("=" * 60 + "\n")
87
88
89def save_report(results: list, output_file: str):
90    """Save report to JSON file."""
91    report = {
92        "generated_at": datetime.now().isoformat(),
93        "total_servers": len(results),
94        "healthy_count": sum(1 for r in results if r['status'] == "healthy"),
95        "results": results
96    }
97    
98    with open(output_file, 'w') as f:
99        json.dump(report, f, indent=2)
100    
101    print(f"Report saved to {output_file}")
102
103
104def main():
105    parser = argparse.ArgumentParser(description='Server Health Checker')
106    parser.add_argument('-c', '--config', default='servers.yaml',
107                        help='Config file path')
108    parser.add_argument('-o', '--output', help='Output JSON file')
109    args = parser.parse_args()
110    
111    servers = load_servers(args.config)
112    results = check_all_servers(servers)
113    print_report(results)
114    
115    if args.output:
116        save_report(results, args.output)
117
118
119if __name__ == "__main__":
120    main()

#Step 3: Create Config File

Create servers.yaml:

yaml
1servers:
2  - name: Google
3    host: google.com
4    port: 443
5  - name: GitHub
6    host: github.com
7    port: 443
8  - name: Local Web
9    host: localhost
10    port: 8080

#Step 4: Run the Script

bash
1# Basic usage
2python health_checker.py
3
4# With custom config
5python health_checker.py -c servers.yaml
6
7# Save JSON report
8python health_checker.py -o report.json

#✅ Success Criteria

  • Script runs without errors
  • Parallel checks complete quickly
  • JSON report is valid
  • Handles connection failures gracefully

#🎓 What You Learned

  • Python socket programming
  • Concurrent execution with ThreadPoolExecutor
  • YAML and JSON handling
  • CLI argument parsing