第02章 Docker容器化¶

import docker
import time

class ContainerManager:
    """Docker容器管理类"""

    def __init__(self):
        self.client = docker.from_env()

    def create_container(self, image_name, container_name, **kwargs):
        """创建容器"""
        try:
            container = self.client.containers.run(
                image_name,
                name=container_name,
                detach=True,
                **kwargs
            )
            print(f"容器创建成功: {container_name}")
            return container
        except Exception as e:
            print(f"容器创建失败: {e}")
            return None

    def start_container(self, container_name):
        """启动容器"""
        try:
            container = self.client.containers.get(container_name)
            container.start()
            print(f"容器启动成功: {container_name}")
            return True
        except Exception as e:
            print(f"容器启动失败: {e}")
            return False

    def stop_container(self, container_name, timeout=10):
        """停止容器"""
        try:
            container = self.client.containers.get(container_name)
            container.stop(timeout=timeout)
            print(f"容器停止成功: {container_name}")
            return True
        except Exception as e:
            print(f"容器停止失败: {e}")
            return False

    def restart_container(self, container_name, timeout=10):
        """重启容器"""
        try:
            container = self.client.containers.get(container_name)
            container.restart(timeout=timeout)
            print(f"容器重启成功: {container_name}")
            return True
        except Exception as e:
            print(f"容器重启失败: {e}")
            return False

    def remove_container(self, container_name, force=False):
        """删除容器"""
        try:
            container = self.client.containers.get(container_name)
            container.remove(force=force)
            print(f"容器删除成功: {container_name}")
            return True
        except Exception as e:
            print(f"容器删除失败: {e}")
            return False

    def get_container_logs(self, container_name, tail=100):
        """获取容器日志"""
        try:
            container = self.client.containers.get(container_name)
            logs = container.logs(tail=tail)
            print(f"容器 {container_name} 日志:")
            print(logs.decode('utf-8'))
            return logs
        except Exception as e:
            print(f"获取日志失败: {e}")
            return None

    def get_container_stats(self, container_name):
        """获取容器统计信息"""
        try:
            container = self.client.containers.get(container_name)
            stats = container.stats(stream=False)

            print(f"容器 {container_name} 统计信息:")
            print(f"  CPU使用率: {self._calculate_cpu_percent(stats)}%")
            print(f"  内存使用: {stats['memory_stats']['usage'] / 1024 / 1024:.2f} MB")
            print(f"  网络接收: {stats['networks']['eth0']['rx_bytes'] / 1024:.2f} KB")
            print(f"  网络发送: {stats['networks']['eth0']['tx_bytes'] / 1024:.2f} KB")

            return stats
        except Exception as e:
            print(f"获取统计信息失败: {e}")
            return None

    def _calculate_cpu_percent(self, stats):
        """计算CPU使用率"""
        cpu_delta = stats['cpu_stats']['cpu_usage']['total_usage'] - \
                   stats['precpu_stats']['cpu_usage']['total_usage']
        system_delta = stats['cpu_stats']['system_cpu_usage'] - \
                      stats['precpu_stats']['system_cpu_usage']

        if system_delta > 0 and cpu_delta > 0:
            # cgroup v2 环境下 percpu_usage 可能为 None，需做兼容处理
            percpu_usage = stats['cpu_stats']['cpu_usage'].get('percpu_usage') or []
            num_cpus = len(percpu_usage) if percpu_usage else 1
            return (cpu_delta / system_delta) * num_cpus * 100.0
        return 0.0

    def list_containers(self, all=False):
        """列出容器"""
        containers = self.client.containers.list(all=all)

        print("容器列表:")
        print("-" * 80)
        print(f"{'容器ID':<15} {'名称':<20} {'状态':<15} {'镜像'}")
        print("-" * 80)

        for container in containers:
            print(f"{container.id[:12]:<15} {container.name:<20} {container.status:<15} {container.image.tags[0]}")  # 切片操作：[start:end:step]提取子序列

        return containers

    def exec_command(self, container_name, command):
        """在容器中执行命令"""
        try:
            container = self.client.containers.get(container_name)
            result = container.exec_run(command)
            print(f"命令执行结果:")
            print(result.output.decode('utf-8'))
            return result
        except Exception as e:
            print(f"命令执行失败: {e}")
            return None

# 使用示例
if __name__ == '__main__':
    manager = ContainerManager()

    # 创建容器
    container = manager.create_container(
        'nginx:latest',
        'my-nginx',
        ports={'80/tcp': 8080},
        environment={'ENV': 'production'}
    )

    time.sleep(2)

    # 列出容器
    manager.list_containers()

    # 获取日志
    manager.get_container_logs('my-nginx')

    # 获取统计信息
    manager.get_container_stats('my-nginx')

    # 停止容器
    manager.stop_container('my-nginx')

    # 删除容器
    manager.remove_container('my-nginx')

import docker
import time

class NetworkManager:
    """Docker网络管理类"""

    def __init__(self):
        self.client = docker.from_env()
        self.networks = {}  # 存储已创建的网络对象

    def create_network(self, network_name, subnet=None, gateway=None):
        """创建网络"""
        try:
            ipam_config = None
            if subnet and gateway:
                ipam_config = docker.types.IPAMConfig(
                    pool_configs=[
                        docker.types.IPAMPool(
                            subnet=subnet,
                            gateway=gateway
                        )
                    ]
                )

            network = self.client.networks.create(
                network_name,
                driver='bridge',
                ipam=ipam_config
            )
            self.networks[network_name] = network
            print(f"网络创建成功: {network_name}")
            return network
        except Exception as e:
            print(f"网络创建失败: {e}")
            return None

    def deploy_microservices(self):
        """部署微服务架构"""

        # 1. 创建网络
        frontend_network = self.create_network('frontend-network', '172.20.0.0/16', '172.20.0.1')
        backend_network = self.create_network('backend-network', '172.21.0.0/16', '172.21.0.1')

        time.sleep(1)

        # 2. 部署数据库
        # ⚠️ 安全警告：以下代码示例包含硬编码密码，仅用于演示目的。
        # 在生产环境中，请务必使用环境变量或密钥管理服务来存储敏感信息。
        db = self.client.containers.run(
            'mysql:8.4',
            name='mysql-db',
            environment={
                'MYSQL_ROOT_PASSWORD': 'rootpassword',
                'MYSQL_DATABASE': 'appdb',
                'MYSQL_USER': 'appuser',
                'MYSQL_PASSWORD': 'apppassword'
            },
            network='backend-network',
            detach=True
        )
        print(f"数据库容器启动: {db.name}")

        # 3. 部署Redis缓存
        redis = self.client.containers.run(
            'redis:alpine',
            name='redis-cache',
            network='backend-network',
            detach=True
        )
        print(f"Redis容器启动: {redis.name}")

        # 4. 部署后端API
        backend = self.client.containers.run(
            'python:3.12-slim',
            name='backend-api',
            command='python -m http.server 8000',
            environment={
                'DB_HOST': 'mysql-db',
                'REDIS_HOST': 'redis-cache',
                'DB_PASSWORD': 'apppassword'
            },
            ports={'8000/tcp': 8000},
            network='backend-network',
            detach=True
        )
        # containers.run() 只支持 network 参数（单个网络），
        # 多网络需在启动后通过 connect() 追加
        self.networks['frontend-network'].connect(backend)
        print(f"后端API容器启动: {backend.name}")

        # 5. 部署前端
        frontend = self.client.containers.run(
            'nginx:alpine',
            name='frontend-web',
            ports={'80/tcp': 80},
            network='frontend-network',
            detach=True
        )
        print(f"前端容器启动: {frontend.name}")

        # 6. 等待容器启动
        time.sleep(5)

        # 7. 测试网络连通性
        print("\n测试网络连通性:")

        # 后端连接数据库
        result = backend.exec_run('ping -c 2 mysql-db')
        print(f"后端 -> 数据库: {'成功' if result.exit_code == 0 else '失败'}")

        # 后端连接Redis
        result = backend.exec_run('ping -c 2 redis-cache')
        print(f"后端 -> Redis: {'成功' if result.exit_code == 0 else '失败'}")

        # 前端连接后端
        result = frontend.exec_run('ping -c 2 backend-api')
        print(f"前端 -> 后端: {'成功' if result.exit_code == 0 else '失败'}")

        return {
            'db': db,
            'redis': redis,
            'backend': backend,
            'frontend': frontend
        }

# 使用示例
if __name__ == '__main__':
    manager = NetworkManager()
    services = manager.deploy_microservices()

    print("\n微服务部署完成！")
    print("访问地址: http://localhost")

import docker
import time

class VolumeManager:
    """Docker卷管理类"""

    def __init__(self):
        self.client = docker.from_env()

    def create_volume(self, volume_name, driver='local', **kwargs):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        """创建卷"""
        try:
            volume = self.client.volumes.create(
                name=volume_name,
                driver=driver,
                driver_opts=kwargs
            )
            print(f"卷创建成功: {volume_name}")
            return volume
        except Exception as e:
            print(f"卷创建失败: {e}")
            return None

    def deploy_persistent_database(self):
        """部署持久化数据库"""

        # 1. 创建数据卷
        mysql_data = self.create_volume('mysql-data')
        mysql_logs = self.create_volume('mysql-logs')

        # 2. 创建配置卷
        mysql_config = self.create_volume('mysql-config')

        # 3. 部署MySQL
        # ⚠️ 安全警告：以下代码示例包含硬编码密码，仅用于演示目的。
        # 在生产环境中，请务必使用环境变量或密钥管理服务来存储敏感信息。
        mysql = self.client.containers.run(
            'mysql:8.4',
            name='mysql-server',
            environment={
                'MYSQL_ROOT_PASSWORD': 'rootpassword',
                'MYSQL_DATABASE': 'appdb',
                'MYSQL_USER': 'appuser',
                'MYSQL_PASSWORD': 'apppassword'
            },
            volumes={
                'mysql-data': {'bind': '/var/lib/mysql', 'mode': 'rw'},
                'mysql-logs': {'bind': '/var/log/mysql', 'mode': 'rw'},
                'mysql-config': {'bind': '/etc/mysql/conf.d', 'mode': 'ro'}
            },
            ports={'3306/tcp': 3306},
            detach=True
        )
        print(f"MySQL容器启动: {mysql.name}")

        # 4. 等待MySQL启动
        time.sleep(10)

        # 5. 测试数据库连接
        test_container = self.client.containers.run(
            'mysql:8.4',
            command='mysql -hmysql-server -uappuser -papppassword -e "SHOW DATABASES;"',
            network='host',
            remove=True
        )

        print("数据库连接测试:")
        print(test_container.logs().decode('utf-8'))

        # 6. 创建测试数据
        create_table = """
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(100),
            email VARCHAR(100),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
        """

        insert_data = """
        INSERT INTO users (name, email) VALUES
        ('Alice', 'alice@example.com'),
        ('Bob', 'bob@example.com'),
        ('Charlie', 'charlie@example.com');
        """

        # 执行SQL
        self.client.containers.run(
            'mysql:8.4',
            command=f'mysql -hmysql-server -uappuser -papppassword appdb -e "{create_table}"',
            network='host',
            remove=True
        )

        self.client.containers.run(
            'mysql:8.4',
            command=f'mysql -hmysql-server -uappuser -papppassword appdb -e "{insert_data}"',
            network='host',
            remove=True
        )

        # 7. 查询数据
        query_result = self.client.containers.run(
            'mysql:8.4',
            command='mysql -hmysql-server -uappuser -papppassword appdb -e "SELECT * FROM users;"',
            network='host',
            remove=True
        )

        print("\n查询结果:")
        print(query_result.logs().decode('utf-8'))

        # 8. 备份数据
        backup_container = self.client.containers.run(
            'mysql:8.4',
            command='mysqldump -hmysql-server -uroot -prootpassword appdb > /backup/appdb_backup.sql',
            volumes={
                '/tmp/backup': {'bind': '/backup', 'mode': 'rw'}
            },
            network='host',
            detach=True
        )

        print("\n数据备份完成")

        return mysql

    def backup_volume(self, volume_name, backup_path):
        """备份卷"""
        try:
            # 创建临时容器挂载卷
            backup_container = self.client.containers.run(
                'alpine',
                command=f'tar czf /backup/{volume_name}.tar.gz /data',
                volumes={
                    volume_name: {'bind': '/data', 'mode': 'ro'},
                    backup_path: {'bind': '/backup', 'mode': 'rw'}
                },
                remove=True
            )

            print(f"卷 {volume_name} 备份成功: {backup_path}/{volume_name}.tar.gz")
            return True
        except Exception as e:
            print(f"卷备份失败: {e}")
            return False

    def restore_volume(self, volume_name, backup_path):
        """恢复卷"""
        try:
            # 创建临时容器恢复卷
            restore_container = self.client.containers.run(
                'alpine',
                command=f'tar xzf /backup/{volume_name}.tar.gz -C /data',
                volumes={
                    volume_name: {'bind': '/data', 'mode': 'rw'},
                    backup_path: {'bind': '/backup', 'mode': 'ro'}
                },
                remove=True
            )

            print(f"卷 {volume_name} 恢复成功")
            return True
        except Exception as e:
            print(f"卷恢复失败: {e}")
            return False

# 使用示例
if __name__ == '__main__':
    manager = VolumeManager()

    # 部署持久化数据库
    mysql = manager.deploy_persistent_database()

    # 备份卷
    manager.backup_volume('mysql-data', '/tmp/backup')

    print("\n数据持久化部署完成！")

from flask import Flask, jsonify, request
from flask_cors import CORS
import redis
import pymysql
import logging
from datetime import datetime
import os

app = Flask(__name__)
CORS(app)

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 数据库连接
def get_db_connection():
    """获取数据库连接"""
    try:  # try/except捕获异常
        connection = pymysql.connect(
            host=os.getenv('DB_HOST', 'db'),
            user=os.getenv('DB_USER', 'appuser'),
            password=os.getenv('DB_PASSWORD', 'apppassword'),
            database=os.getenv('DB_NAME', 'appdb'),
            cursorclass=pymysql.cursors.DictCursor
        )
        return connection
    except Exception as e:
        logger.error(f"数据库连接失败: {e}")
        return None

# Redis连接
def get_redis_connection():
    """获取Redis连接"""
    try:
        r = redis.Redis(
            host=os.getenv('REDIS_HOST', 'redis'),
            port=int(os.getenv('REDIS_PORT', 6379)),
            password=os.getenv('REDIS_PASSWORD', 'redispassword'),
            decode_responses=True
        )
        return r
    except Exception as e:
        logger.error(f"Redis连接失败: {e}")
        return None

@app.route('/')
def index():
    """首页"""
    return jsonify({
        'message': 'Welcome to Flask App',
        'timestamp': datetime.now().isoformat(),
        'status': 'running'
    })

@app.route('/health')
def health():
    """健康检查"""
    # 检查数据库连接
    db_status = 'ok'
    try:
        conn = get_db_connection()
        if conn:
            conn.close()
        else:
            db_status = 'error'
    except:
        db_status = 'error'

    # 检查Redis连接
    redis_status = 'ok'
    try:
        r = get_redis_connection()
        if r:
            r.ping()
        else:
            redis_status = 'error'
    except:
        redis_status = 'error'

    return jsonify({
        'status': 'healthy',
        'database': db_status,
        'redis': redis_status,
        'timestamp': datetime.now().isoformat()
    })

@app.route('/api/users', methods=['GET', 'POST'])
def users():
    """用户API"""
    if request.method == 'GET':
        # 尝试从缓存获取
        r = get_redis_connection()
        cache_key = 'users:list'

        if r:
            cached = r.get(cache_key)
            if cached:
                logger.info("从缓存获取用户列表")
                return jsonify({'source': 'cache', 'data': cached})

        # 从数据库获取
        conn = get_db_connection()
        if not conn:
            return jsonify({'error': 'Database connection failed'}), 500

        try:
            with conn.cursor() as cursor:
                cursor.execute('SELECT * FROM users')
                users = cursor.fetchall()

            # 缓存结果
            if r:
                r.setex(cache_key, 300, str(users))  # 缓存5分钟

            return jsonify({'source': 'database', 'data': users})
        finally:
            conn.close()

    elif request.method == 'POST':
        data = request.json
        conn = get_db_connection()
        if not conn:
            return jsonify({'error': 'Database connection failed'}), 500

        try:
            with conn.cursor() as cursor:
                sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
                cursor.execute(sql, (data['name'], data['email']))
            conn.commit()

            # 清除缓存
            r = get_redis_connection()
            if r:
                r.delete('users:list')

            return jsonify({'message': 'User created successfully'}), 201
        except Exception as e:
            conn.rollback()
            return jsonify({'error': str(e)}), 500
        finally:
            conn.close()

@app.route('/api/stats')
def stats():
    """统计信息"""
    r = get_redis_connection()
    if not r:
        return jsonify({'error': 'Redis connection failed'}), 500

    # 增加访问计数
    r.incr('stats:visits')

    # 获取统计信息
    visits = r.get('stats:visits')

    return jsonify({
        'visits': visits,
        'timestamp': datetime.now().isoformat()
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)