利用Python进行微服务架构的监控与日志分析

import requests
​
def check_service_health(url):
    try:
        response = requests.get(url)
        if response.status_code == 200:
            print(f"{url} is healthy")
        else:
            print(f"{url} is unhealthy")
    except requests.exceptions.RequestException as e:
        print(f"Failed to connect to {url}: {e}")
​
if __name__ == "__main__":
    services = ["http://service1.example.com", "http://service2.example.com"]
    for service in services:
        check_service_health(service)

import re
​
def analyze_logs(log_file):
    error_count = 0
    warning_count = 0
    with open(log_file, 'r') as file:
        for line in file:
            if re.search(r'ERROR', line):
                error_count += 1
            elif re.search(r'WARNING', line):
                warning_count += 1
    print(f"Errors: {error_count}, Warnings: {warning_count}")
​
if __name__ == "__main__":
    log_file = "service.log"
    analyze_logs(log_file)

import requests
import re
​
def check_service_health(url):
    try:
        response = requests.get(url)
        if response.status_code == 200:
            print(f"{url} is healthy")
        else:
            print(f"{url} is unhealthy")
            # 记录异常到日志
            with open("error.log", "a") as log_file:
                log_file.write(f"{url} is unhealthy\n")
    except requests.exceptions.RequestException as e:
        print(f"Failed to connect to {url}: {e}")
        # 记录异常到日志
        with open("error.log", "a") as log_file:
            log_file.write(f"Failed to connect to {url}: {e}\n")
​
def analyze_logs(log_file):
    error_count = 0
    warning_count = 0
    with open(log_file, 'r') as file:
        for line in file:
            if re.search(r'ERROR', line):
                error_count += 1
            elif re.search(r'WARNING', line):
                warning_count += 1
    print(f"Errors: {error_count}, Warnings: {warning_count}")
​
if __name__ == "__main__":
    services = ["http://service1.example.com", "http://service2.example.com"]
    for service in services:
        check_service_health(service)
    
    log_file = "service.log"
    analyze_logs(log_file)

import matplotlib.pyplot as plt
​
def plot_service_health(healthy_count, unhealthy_count):
    labels = ['Healthy', 'Unhealthy']
    sizes = [healthy_count, unhealthy_count]
    colors = ['green', 'red']
    plt.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%', startangle=140)
    plt.axis('equal')
    plt.title('Service Health Status')
    plt.show()
​
if __name__ == "__main__":
    healthy_count = 8
    unhealthy_count = 2
    plot_service_health(healthy_count, unhealthy_count)

import smtplib
from email.mime.text import MIMEText
​
def send_alert_email():
    sender_email = "your_email@example.com"
    receiver_email = "recipient_email@example.com"
    subject = "Service Alert: Microservice Down"
    body = "One of the microservices is down. Please take action immediately."
    
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = sender_email
    msg['To'] = receiver_email
​
    smtp_server = "smtp.example.com"
    smtp_port = 587
    smtp_username = "your_smtp_username"
    smtp_password = "your_smtp_password"
​
    try:
        server = smtplib.SMTP(smtp_server, smtp_port)
        server.starttls()
        server.login(smtp_username, smtp_password)
        server.sendmail(sender_email, receiver_email, msg.as_string())
        print("Alert email sent successfully")
    except Exception as e:
        print(f"Failed to send alert email: {e}")
    finally:
        server.quit()
​
if __name__ == "__main__":
    send_alert_email()

from apscheduler.schedulers.background import BackgroundScheduler
import time
​
def backup_database():
    # Placeholder function for database backup
    print("Backing up database...")
    time.sleep(5)
    print("Database backup complete.")
​
def cleanup_logs():
    # Placeholder function for log cleanup
    print("Cleaning up logs...")
    time.sleep(3)
    print("Log cleanup complete.")
​
if __name__ == "__main__":
    scheduler = BackgroundScheduler()
    scheduler.add_job(backup_database, 'interval', hours=24)
    scheduler.add_job(cleanup_logs, 'cron', day_of_week='sun', hour=0)
    scheduler.start()
    print("Scheduler started. Press Ctrl+C to exit.")
    try:
        while True:
            time.sleep(2)
    except KeyboardInterrupt:
        print("Exiting...")
        scheduler.shutdown()

from cryptography.fernet import Fernet
​
def generate_key():
    return Fernet.generate_key()
​
def encrypt_data(key, data):
    cipher = Fernet(key)
    return cipher.encrypt(data.encode()).decode()
​
def decrypt_data(key, encrypted_data):
    cipher = Fernet(key)
    return cipher.decrypt(encrypted_data.encode()).decode()
​
if __name__ == "__main__":
    key = generate_key()
    sensitive_data = "This is sensitive data"
    encrypted_data = encrypt_data(key, sensitive_data)
    print("Encrypted data:", encrypted_data)
    decrypted_data = decrypt_data(key, encrypted_data)
    print("Decrypted data:", decrypted_data)

from flask import Flask, render_template, request, redirect, url_for
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required
​
app = Flask(__name__)
app.secret_key = 'your_secret_key'
login_manager = LoginManager()
login_manager.init_app(app)
​
class User(UserMixin):
    def __init__(self, username, password):
        self.id = username
        self.password = password
​
users = {
    'admin': User('admin', 'admin123'),
    'user': User('user', 'user123')
}
​
@login_manager.user_loader
def load_user(user_id):
    return users.get(user_id)
​
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        user = users.get(username)
        if user and user.password == password:
            login_user(user)
            return redirect(url_for('dashboard'))
    return render_template('login.html')
​
@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('login'))
​
@app.route('/dashboard')
@login_required
def dashboard():
    return render_template('dashboard.html')
​
if __name__ == '__main__':
    app.run(debug=True)