叶哥Linux技术分享

主题

133-自动化工具开发 ​

课程目标 ​

本课程将详细介绍自动化工具开发的核心概念、技术栈和实践方法，帮助你掌握如何开发实用的自动化运维工具。通过本课程的学习，你将能够：

  • 理解自动化工具开发的基本原理和架构
  • 掌握Python、Go、Shell等语言的自动化工具开发技巧
  • 学习如何集成Ansible、Git等工具进行自动化开发
  • 开发实用的自动化运维工具，提高工作效率

1. 自动化工具开发概述 ​

1.1 自动化工具的定义和作用 ​

自动化工具是指通过编写代码或脚本，实现重复性任务的自动执行，从而提高工作效率和减少人为错误的工具。在运维领域，自动化工具的应用非常广泛，包括：

  • 服务器配置管理
  • 软件部署和更新
  • 系统监控和告警
  • 日志分析和处理
  • 备份和恢复操作
  • 网络配置和管理

1.2 自动化工具开发的技术栈 ​

技术用途适用场景
Python通用自动化脚本大多数自动化场景
Go高性能自动化工具需要并发处理的场景
Shell系统级自动化简单的系统操作
Ansible配置管理和编排多服务器配置管理
Git版本控制和代码管理代码管理和协作开发
JenkinsCI/CD流程自动化持续集成和部署

1.3 自动化工具开发的最佳实践 ​

  • 模块化设计：将工具分解为多个独立的模块，提高代码可维护性
  • 配置化管理：使用配置文件管理工具参数，提高灵活性
  • 错误处理：完善的错误处理机制，确保工具稳定运行
  • 日志记录：详细的日志记录，便于问题排查
  • 测试覆盖：编写测试用例，确保工具功能正常
  • 文档完善：提供详细的使用文档，方便他人使用

2. Python自动化工具开发 ​

2.1 Python自动化库介绍 ​

2.1.1 核心库 ​

  • os：系统操作相关功能
  • sys：Python解释器相关功能
  • subprocess：执行系统命令
  • shutil：高级文件操作
  • datetime：日期和时间处理
  • json：JSON数据处理
  • yaml：YAML配置文件处理

2.1.2 第三方库 ​

  • requests：HTTP请求处理
  • paramiko：SSH远程连接
  • fabric：远程执行命令和文件传输
  • ansible：配置管理和编排
  • pyyaml：YAML文件处理
  • argparse：命令行参数解析
  • logging：日志记录
  • colorama：终端彩色输出

2.2 Python自动化工具开发实例 ​

2.2.1 服务器状态检查工具 ​

python
#!/usr/bin/env python3
"""
服务器状态检查工具
"""

import argparse
import subprocess
import json
import logging
from datetime import datetime

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

def check_disk_usage():
    """检查磁盘使用情况"""
    try:
        output = subprocess.check_output(['df', '-h'], universal_newlines=True)
        lines = output.strip().split('\n')
        result = []
        for line in lines[1:]:
            parts = line.split()
            if len(parts) >= 5:
                result.append({
                    'filesystem': parts[0],
                    'size': parts[1],
                    'used': parts[2],
                    'avail': parts[3],
                    'use_percent': parts[4],
                    'mounted_on': parts[5]
                })
        return result
    except Exception as e:
        logger.error(f"检查磁盘使用情况失败: {e}")
        return []

def check_memory_usage():
    """检查内存使用情况"""
    try:
        output = subprocess.check_output(['free', '-h'], universal_newlines=True)
        lines = output.strip().split('\n')
        memory_line = lines[1]
        swap_line = lines[2]
        
        memory_parts = memory_line.split()
        swap_parts = swap_line.split()
        
        return {
            'memory': {
                'total': memory_parts[1],
                'used': memory_parts[2],
                'free': memory_parts[3],
                'shared': memory_parts[4],
                'buff_cache': memory_parts[5],
                'available': memory_parts[6]
            },
            'swap': {
                'total': swap_parts[1],
                'used': swap_parts[2],
                'free': swap_parts[3]
            }
        }
    except Exception as e:
        logger.error(f"检查内存使用情况失败: {e}")
        return {}

def check_cpu_usage():
    """检查CPU使用情况"""
    try:
        output = subprocess.check_output(['top', '-bn1'], universal_newlines=True)
        lines = output.strip().split('\n')
        cpu_line = None
        for line in lines:
            if line.startswith('%Cpu(s):'):
                cpu_line = line
                break
        
        if cpu_line:
            parts = cpu_line.split(':')[1].split(',')
            cpu_usage = {
                'us': parts[0].strip(),
                'sy': parts[1].strip(),
                'ni': parts[2].strip(),
                'id': parts[3].strip(),
                'wa': parts[4].strip(),
                'hi': parts[5].strip(),
                'si': parts[6].strip(),
                'st': parts[7].strip()
            }
            return cpu_usage
        return {}
    except Exception as e:
        logger.error(f"检查CPU使用情况失败: {e}")
        return {}

def check_processes():
    """检查进程情况"""
    try:
        output = subprocess.check_output(['ps', 'aux', '--sort=-%cpu'], universal_newlines=True)
        lines = output.strip().split('\n')
        processes = []
        for line in lines[1:11]:  # 取前10个CPU使用率最高的进程
            parts = line.split()
            if len(parts) >= 11:
                processes.append({
                    'user': parts[0],
                    'pid': parts[1],
                    'cpu_percent': parts[2],
                    'mem_percent': parts[3],
                    'vsz': parts[4],
                    'rss': parts[5],
                    'tty': parts[6],
                    'stat': parts[7],
                    'start': parts[8],
                    'time': parts[9],
                    'command': ' '.join(parts[10:])
                })
        return processes
    except Exception as e:
        logger.error(f"检查进程情况失败: {e}")
        return []

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='服务器状态检查工具')
    parser.add_argument('--output', '-o', choices=['text', 'json'], default='text',
                        help='输出格式 (默认: text)')
    args = parser.parse_args()
    
    logger.info('开始检查服务器状态...')
    
    # 收集状态信息
    status = {
        'timestamp': datetime.now().isoformat(),
        'disk': check_disk_usage(),
        'memory': check_memory_usage(),
        'cpu': check_cpu_usage(),
        'processes': check_processes()
    }
    
    # 输出结果
    if args.output == 'json':
        print(json.dumps(status, indent=2, ensure_ascii=False))
    else:
        print('=' * 80)
        print(f"服务器状态检查报告 - {status['timestamp']}")
        print('=' * 80)
        
        print('\n1. 磁盘使用情况:')
        print('-' * 60)
        print(f"{'文件系统':<20} {'总大小':<10} {'已用':<10} {'可用':<10} {'使用率':<10} {'挂载点':<10}")
        print('-' * 60)
        for disk in status['disk']:
            print(f"{disk['filesystem']:<20} {disk['size']:<10} {disk['used']:<10} {disk['avail']:<10} {disk['use_percent']:<10} {disk['mounted_on']:<10}")
        
        print('\n2. 内存使用情况:')
        print('-' * 60)
        memory = status['memory'].get('memory', {})
        swap = status['memory'].get('swap', {})
        print(f"内存总大小: {memory.get('total', 'N/A')}")
        print(f"已用内存: {memory.get('used', 'N/A')}")
        print(f"可用内存: {memory.get('available', 'N/A')}")
        print(f"交换分区总大小: {swap.get('total', 'N/A')}")
        print(f"已用交换分区: {swap.get('used', 'N/A')}")
        
        print('\n3. CPU使用情况:')
        print('-' * 60)
        cpu = status['cpu']
        print(f"用户空间: {cpu.get('us', 'N/A')}")
        print(f"系统空间: {cpu.get('sy', 'N/A')}")
        print(f"空闲: {cpu.get('id', 'N/A')}")
        print(f"等待IO: {cpu.get('wa', 'N/A')}")
        
        print('\n4. 进程情况 (CPU使用率前10):')
        print('-' * 60)
        print(f"{'用户':<10} {'PID':<10} {'CPU%':<10} {'MEM%':<10} {'命令':<40}")
        print('-' * 60)
        for proc in status['processes']:
            command = proc['command'][:40]
            print(f"{proc['user']:<10} {proc['pid']:<10} {proc['cpu_percent']:<10} {proc['mem_percent']:<10} {command:<40}")
        
        print('\n' + '=' * 80)
        print('检查完成')
        print('=' * 80)

if __name__ == '__main__':
    main()

2.2.2 配置文件管理工具 ​

python
#!/usr/bin/env python3
"""
配置文件管理工具
"""

import argparse
import json
import yaml
import configparser
import os
import logging
from datetime import datetime

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

class ConfigManager:
    """配置文件管理器"""
    
    def __init__(self, config_file):
        """初始化配置文件管理器"""
        self.config_file = config_file
        self.file_extension = os.path.splitext(config_file)[1].lower()
        self.config_data = {}
        self._load_config()
    
    def _load_config(self):
        """加载配置文件"""
        try:
            if not os.path.exists(self.config_file):
                logger.warning(f"配置文件 {self.config_file} 不存在，将创建新文件")
                self._save_config()
                return
            
            with open(self.config_file, 'r', encoding='utf-8') as f:
                if self.file_extension == '.json':
                    self.config_data = json.load(f)
                elif self.file_extension in ['.yaml', '.yml']:
                    self.config_data = yaml.safe_load(f)
                elif self.file_extension == '.ini':
                    config = configparser.ConfigParser()
                    config.read(self.config_file, encoding='utf-8')
                    self.config_data = {}
                    for section in config.sections():
                        self.config_data[section] = {}
                        for key, value in config[section].items():
                            self.config_data[section][key] = value
                else:
                    logger.error(f"不支持的配置文件格式: {self.file_extension}")
            logger.info(f"成功加载配置文件: {self.config_file}")
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
    
    def _save_config(self):
        """保存配置文件"""
        try:
            # 确保目录存在
            os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
            
            with open(self.config_file, 'w', encoding='utf-8') as f:
                if self.file_extension == '.json':
                    json.dump(self.config_data, f, indent=2, ensure_ascii=False)
                elif self.file_extension in ['.yaml', '.yml']:
                    yaml.dump(self.config_data, f, default_flow_style=False, allow_unicode=True)
                elif self.file_extension == '.ini':
                    config = configparser.ConfigParser()
                    for section, values in self.config_data.items():
                        config[section] = values
                    config.write(f)
                else:
                    logger.error(f"不支持的配置文件格式: {self.file_extension}")
                    return False
            logger.info(f"成功保存配置文件: {self.config_file}")
            return True
        except Exception as e:
            logger.error(f"保存配置文件失败: {e}")
            return False
    
    def get(self, key, default=None):
        """获取配置值"""
        try:
            keys = key.split('.')
            value = self.config_data
            for k in keys:
                if k in value:
                    value = value[k]
                else:
                    return default
            return value
        except Exception as e:
            logger.error(f"获取配置失败: {e}")
            return default
    
    def set(self, key, value):
        """设置配置值"""
        try:
            keys = key.split('.')
            config = self.config_data
            for k in keys[:-1]:
                if k not in config:
                    config[k] = {}
                config = config[k]
            config[keys[-1]] = value
            return self._save_config()
        except Exception as e:
            logger.error(f"设置配置失败: {e}")
            return False
    
    def delete(self, key):
        """删除配置项"""
        try:
            keys = key.split('.')
            config = self.config_data
            for k in keys[:-1]:
                if k not in config:
                    return False
                config = config[k]
            if keys[-1] in config:
                del config[keys[-1]]
                return self._save_config()
            return False
        except Exception as e:
            logger.error(f"删除配置失败: {e}")
            return False
    
    def list(self):
        """列出所有配置"""
        return self.config_data
    
    def backup(self, backup_dir=None):
        """备份配置文件"""
        try:
            if backup_dir is None:
                backup_dir = os.path.dirname(self.config_file)
            
            backup_file = os.path.join(
                backup_dir,
                f"{os.path.basename(self.config_file)}.backup.{datetime.now().strftime('%Y%m%d%H%M%S')}"
            )
            
            import shutil
            shutil.copy2(self.config_file, backup_file)
            logger.info(f"成功备份配置文件到: {backup_file}")
            return backup_file
        except Exception as e:
            logger.error(f"备份配置文件失败: {e}")
            return None
    
    def restore(self, backup_file):
        """从备份恢复配置文件"""
        try:
            if not os.path.exists(backup_file):
                logger.error(f"备份文件不存在: {backup_file}")
                return False
            
            import shutil
            shutil.copy2(backup_file, self.config_file)
            self._load_config()
            logger.info(f"成功从备份恢复配置文件: {backup_file}")
            return True
        except Exception as e:
            logger.error(f"恢复配置文件失败: {e}")
            return False

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='配置文件管理工具')
    parser.add_argument('config_file', help='配置文件路径')
    parser.add_argument('--get', '-g', help='获取配置项')
    parser.add_argument('--set', '-s', nargs=2, metavar=('KEY', 'VALUE'), help='设置配置项')
    parser.add_argument('--delete', '-d', help='删除配置项')
    parser.add_argument('--list', '-l', action='store_true', help='列出所有配置')
    parser.add_argument('--backup', '-b', help='备份配置文件')
    parser.add_argument('--restore', '-r', help='从备份恢复配置文件')
    args = parser.parse_args()
    
    config_manager = ConfigManager(args.config_file)
    
    if args.get:
        value = config_manager.get(args.get)
        print(f"{args.get}: {value}")
    
    elif args.set:
        key, value = args.set
        # 尝试转换值类型
        try:
            if value.lower() == 'true':
                value = True
            elif value.lower() == 'false':
                value = False
            elif value.lower() == 'none':
                value = None
            else:
                # 尝试转换为数字
                if '.' in value:
                    value = float(value)
                else:
                    value = int(value)
        except:
            pass
        success = config_manager.set(key, value)
        if success:
            print(f"成功设置配置: {key} = {value}")
        else:
            print(f"设置配置失败: {key}")
    
    elif args.delete:
        success = config_manager.delete(args.delete)
        if success:
            print(f"成功删除配置: {args.delete}")
        else:
            print(f"删除配置失败: {args.delete}")
    
    elif args.list:
        config = config_manager.list()
        print(json.dumps(config, indent=2, ensure_ascii=False))
    
    elif args.backup:
        backup_file = config_manager.backup(args.backup)
        if backup_file:
            print(f"成功备份到: {backup_file}")
        else:
            print("备份失败")
    
    elif args.restore:
        success = config_manager.restore(args.restore)
        if success:
            print(f"成功从备份恢复: {args.restore}")
        else:
            print(f"恢复失败: {args.restore}")
    
    else:
        parser.print_help()

if __name__ == '__main__':
    main()

3. Go语言自动化工具开发 ​

3.1 Go语言自动化库介绍 ​

3.1.1 标准库 ​

  • os：系统操作相关功能
  • exec：执行系统命令
  • io/ioutil：文件操作
  • encoding/json：JSON数据处理
  • flag：命令行参数解析
  • log：日志记录
  • net/http：HTTP客户端和服务器
  • time：时间处理

3.1.2 第三方库 ​

  • spf13/viper：配置管理
  • spf13/cobra：命令行工具框架
  • golang.org/x/crypto：加密库
  • github.com/ssh/ssh：SSH客户端
  • github.com/prometheus/client_golang：Prometheus客户端

3.2 Go语言自动化工具开发实例 ​

3.2.1 服务器监控工具 ​

go
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"time"
)

// ServerStatus 服务器状态
type ServerStatus struct {
	Timestamp  string      `json:"timestamp"`
	CPU        CPUStatus   `json:"cpu"`
	Memory     MemoryStatus `json:"memory"`
	Disk       []DiskStatus `json:"disk"`
	Network    NetworkStatus `json:"network"`
	Processes  []ProcessStatus `json:"processes"`
}

// CPUStatus CPU状态
type CPUStatus struct {
	User      float64 `json:"user"`
	System    float64 `json:"system"`
	Idle      float64 `json:"idle"`
	IOWait    float64 `json:"iowait"`
	TotalUsed float64 `json:"total_used"`
}

// MemoryStatus 内存状态
type MemoryStatus struct {
	Total     uint64  `json:"total"`
	Used      uint64  `json:"used"`
	Free      uint64  `json:"free"`
	UsedPercent float64 `json:"used_percent"`
	SwapTotal uint64  `json:"swap_total"`
	SwapUsed  uint64  `json:"swap_used"`
	SwapFree  uint64  `json:"swap_free"`
}

// DiskStatus 磁盘状态
type DiskStatus struct {
	Filesystem string  `json:"filesystem"`
	Size       uint64  `json:"size"`
	Used       uint64  `json:"used"`
	Avail      uint64  `json:"avail"`
	UsePercent float64 `json:"use_percent"`
	MountedOn  string  `json:"mounted_on"`
}

// NetworkStatus 网络状态
type NetworkStatus struct {
	Interfaces []InterfaceStatus `json:"interfaces"`
}

// InterfaceStatus 网络接口状态
type InterfaceStatus struct {
	Name      string `json:"name"`
	IPAddress string `json:"ip_address"`
	MacAddress string `json:"mac_address"`
	RxBytes   uint64 `json:"rx_bytes"`
	TxBytes   uint64 `json:"tx_bytes"`
}

// ProcessStatus 进程状态
type ProcessStatus struct {
	PID        int    `json:"pid"`
	Command    string `json:"command"`
	User       string `json:"user"`
	CPUPercent float64 `json:"cpu_percent"`
	MemPercent float64 `json:"mem_percent"`
	RSS        uint64  `json:"rss"`
}

// executeCommand 执行系统命令
func executeCommand(command string, args ...string) (string, error) {
	cmd := exec.Command(command, args...)
	output, err := cmd.CombinedOutput()
	return string(output), err
}

// getCPUStatus 获取CPU状态
func getCPUStatus() (CPUStatus, error) {
	var cpu CPUStatus

	// 执行top命令获取CPU使用率
	output, err := executeCommand("top", "-bn1")
	if err != nil {
		return cpu, err
	}

	lines := strings.Split(output, "\n")
	for _, line := range lines {
		if strings.HasPrefix(line, "%Cpu(s):") {
			parts := strings.Fields(line)
			if len(parts) >= 10 {
				// 解析CPU使用率
				user, _ := strconv.ParseFloat(strings.TrimSuffix(parts[1], "%"), 64)
				system, _ := strconv.ParseFloat(strings.TrimSuffix(parts[3], "%"), 64)
				idle, _ := strconv.ParseFloat(strings.TrimSuffix(parts[7], "%"), 64)
				iowait, _ := strconv.ParseFloat(strings.TrimSuffix(parts[9], "%"), 64)

				cpu.User = user
				cpu.System = system
				cpu.Idle = idle
				cpu.IOWait = iowait
				cpu.TotalUsed = 100 - idle
			}
			break
		}
	}

	return cpu, nil
}

// getMemoryStatus 获取内存状态
func getMemoryStatus() (MemoryStatus, error) {
	var mem MemoryStatus

	// 执行free命令获取内存使用情况
	output, err := executeCommand("free", "-b")
	if err != nil {
		return mem, err
	}

	lines := strings.Split(output, "\n")
	if len(lines) >= 3 {
		// 解析内存使用情况
		memParts := strings.Fields(lines[1])
		if len(memParts) >= 4 {
			total, _ := strconv.ParseUint(memParts[1], 10, 64)
			used, _ := strconv.ParseUint(memParts[2], 10, 64)
			free, _ := strconv.ParseUint(memParts[3], 10, 64)

			mem.Total = total
			mem.Used = used
			mem.Free = free
			mem.UsedPercent = float64(used) / float64(total) * 100
		}

		// 解析交换分区使用情况
		swapParts := strings.Fields(lines[2])
		if len(swapParts) >= 4 {
			swapTotal, _ := strconv.ParseUint(swapParts[1], 10, 64)
			swapUsed, _ := strconv.ParseUint(swapParts[2], 10, 64)
			swapFree, _ := strconv.ParseUint(swapParts[3], 10, 64)

			mem.SwapTotal = swapTotal
			mem.SwapUsed = swapUsed
			mem.SwapFree = swapFree
		}
	}

	return mem, nil
}

// getDiskStatus 获取磁盘状态
func getDiskStatus() ([]DiskStatus, error) {
	var disks []DiskStatus

	// 执行df命令获取磁盘使用情况
	output, err := executeCommand("df", "-B1")
	if err != nil {
		return disks, err
	}

	lines := strings.Split(output, "\n")
	for i, line := range lines {
		if i == 0 {
			continue // 跳过表头
		}

		parts := strings.Fields(line)
		if len(parts) >= 6 {
			var disk DiskStatus
			disk.Filesystem = parts[0]
			size, _ := strconv.ParseUint(parts[1], 10, 64)
			used, _ := strconv.ParseUint(parts[2], 10, 64)
			avail, _ := strconv.ParseUint(parts[3], 10, 64)
			usePercent, _ := strconv.ParseFloat(strings.TrimSuffix(parts[4], "%"), 64)

			disk.Size = size
			disk.Used = used
			disk.Avail = avail
			disk.UsePercent = usePercent
			disk.MountedOn = parts[5]

			disks = append(disks, disk)
		}
	}

	return disks, nil
}

// getNetworkStatus 获取网络状态
func getNetworkStatus() (NetworkStatus, error) {
	var network NetworkStatus

	// 执行ip命令获取网络接口信息
	output, err := executeCommand("ip", "addr")
	if err != nil {
		return network, err
	}

	lines := strings.Split(output, "\n")
	var iface InterfaceStatus
	for _, line := range lines {
		if strings.HasPrefix(line, " ") {
			// 跳过空行
			if len(strings.TrimSpace(line)) == 0 {
				continue
			}

			// 解析IP地址
			if strings.Contains(line, "inet ") && !strings.Contains(line, "127.0.0.1") {
				parts := strings.Fields(line)
				if len(parts) >= 2 {
					ipAddr := strings.Split(parts[1], "/")[0]
					iface.IPAddress = ipAddr
				}
			}

			// 解析MAC地址
			if strings.Contains(line, "link/ether") {
				parts := strings.Fields(line)
				if len(parts) >= 2 {
					iface.MacAddress = parts[1]
				}
			}
		} else if strings.Contains(line, ": ") && !strings.Contains(line, "lo:") {
			// 新的网络接口
			if iface.Name != "" {
				network.Interfaces = append(network.Interfaces, iface)
			}

			// 解析接口名称
			parts := strings.Split(line, ":")
			if len(parts) >= 2 {
				iface = InterfaceStatus{}
				iface.Name = strings.TrimSpace(parts[1])
			}
		}
	}

	// 添加最后一个接口
	if iface.Name != "" {
		network.Interfaces = append(network.Interfaces, iface)
	}

	// 获取网络流量
	for i, iface := range network.Interfaces {
		output, _ := executeCommand("cat", "/proc/net/dev")
		lines := strings.Split(output, "\n")
		for _, line := range lines {
			if strings.Contains(line, iface.Name+":") {
				parts := strings.Fields(line)
				if len(parts) >= 10 {
					rxBytes, _ := strconv.ParseUint(parts[1], 10, 64)
					txBytes, _ := strconv.ParseUint(parts[9], 10, 64)
					network.Interfaces[i].RxBytes = rxBytes
					network.Interfaces[i].TxBytes = txBytes
				}
				break
			}
		}
	}

	return network, nil
}

// getProcessStatus 获取进程状态
func getProcessStatus() ([]ProcessStatus, error) {
	var processes []ProcessStatus

	// 执行ps命令获取进程信息
	output, err := executeCommand("ps", "aux", "--sort=-%cpu")
	if err != nil {
		return processes, err
	}

	lines := strings.Split(output, "\n")
	for i, line := range lines {
		if i == 0 || len(strings.TrimSpace(line)) == 0 {
			continue // 跳过表头和空行
		}

		parts := strings.Fields(line)
		if len(parts) >= 11 {
			var proc ProcessStatus
			pid, _ := strconv.Atoi(parts[1])
			cpuPercent, _ := strconv.ParseFloat(parts[2], 64)
			memPercent, _ := strconv.ParseFloat(parts[3], 64)
			rss, _ := strconv.ParseUint(parts[5], 10, 64)

			proc.PID = pid
			proc.User = parts[0]
			proc.CPUPercent = cpuPercent
			proc.MemPercent = memPercent
			proc.RSS = rss * 1024 // 转换为字节

			// 解析命令
			command := strings.Join(parts[10:], " ")
			proc.Command = command

			processes = append(processes, proc)

			// 只取前10个进程
			if len(processes) >= 10 {
				break
			}
		}
	}

	return processes, nil
}

// getServerStatus 获取服务器状态
func getServerStatus() (ServerStatus, error) {
	var status ServerStatus
	status.Timestamp = time.Now().Format(time.RFC3339)

	// 获取CPU状态
	cpu, err := getCPUStatus()
	if err != nil {
		log.Printf("获取CPU状态失败: %v", err)
	}
	status.CPU = cpu

	// 获取内存状态
	mem, err := getMemoryStatus()
	if err != nil {
		log.Printf("获取内存状态失败: %v", err)
	}
	status.Memory = mem

	// 获取磁盘状态
	disks, err := getDiskStatus()
	if err != nil {
		log.Printf("获取磁盘状态失败: %v", err)
	}
	status.Disk = disks

	// 获取网络状态
	network, err := getNetworkStatus()
	if err != nil {
		log.Printf("获取网络状态失败: %v", err)
	}
	status.Network = network

	// 获取进程状态
	processes, err := getProcessStatus()
	if err != nil {
		log.Printf("获取进程状态失败: %v", err)
	}
	status.Processes = processes

	return status, nil
}

func main() {
	// 解析命令行参数
	jsonOutput := flag.Bool("json", false, "以JSON格式输出")
	flag.Parse()

	// 获取服务器状态
	status, err := getServerStatus()
	if err != nil {
		log.Fatalf("获取服务器状态失败: %v", err)
	}

	// 输出结果
	if *jsonOutput {
		// 以JSON格式输出
		jsonData, err := json.MarshalIndent(status, "", "  ")
		if err != nil {
			log.Fatalf("JSON编码失败: %v", err)
		}
		fmt.Println(string(jsonData))
	} else {
		// 以文本格式输出
		fmt.Println("=========================================")
		fmt.Printf("服务器状态报告 - %s\n", status.Timestamp)
		fmt.Println("=========================================")

		// 输出CPU状态
		fmt.Println("\n1. CPU状态:")
		fmt.Println("-----------------------------------------")
		fmt.Printf("用户空间: %.2f%%\n", status.CPU.User)
		fmt.Printf("系统空间: %.2f%%\n", status.CPU.System)
		fmt.Printf("空闲: %.2f%%\n", status.CPU.Idle)
		fmt.Printf("IO等待: %.2f%%\n", status.CPU.IOWait)
		fmt.Printf("总使用率: %.2f%%\n", status.CPU.TotalUsed)

		// 输出内存状态
		fmt.Println("\n2. 内存状态:")
		fmt.Println("-----------------------------------------")
		fmt.Printf("总内存: %.2f GB\n", float64(status.Memory.Total)/1024/1024/1024)
		fmt.Printf("已用内存: %.2f GB\n", float64(status.Memory.Used)/1024/1024/1024)
		fmt.Printf("可用内存: %.2f GB\n", float64(status.Memory.Free)/1024/1024/1024)
		fmt.Printf("内存使用率: %.2f%%\n", status.Memory.UsedPercent)
		fmt.Printf("交换分区总大小: %.2f GB\n", float64(status.Memory.SwapTotal)/1024/1024/1024)
		fmt.Printf("已用交换分区: %.2f GB\n", float64(status.Memory.SwapUsed)/1024/1024/1024)

		// 输出磁盘状态
		fmt.Println("\n3. 磁盘状态:")
		fmt.Println("-----------------------------------------")
		fmt.Printf("%s\t%s\t%s\t%s\t%s\t%s\n", "文件系统", "总大小", "已用", "可用", "使用率", "挂载点")
		fmt.Println("-----------------------------------------")
		for _, disk := range status.Disk {
			fmt.Printf("%s\t%.2f GB\t%.2f GB\t%.2f GB\t%.2f%%\t%s\n",
				disk.Filesystem,
				float64(disk.Size)/1024/1024/1024,
				float64(disk.Used)/1024/1024/1024,
				float64(disk.Avail)/1024/1024/1024,
				disk.UsePercent,
				disk.MountedOn)
		}

		// 输出网络状态
		fmt.Println("\n4. 网络状态:")
		fmt.Println("-----------------------------------------")
		for _, iface := range status.Network.Interfaces {
			fmt.Printf("接口: %s\n", iface.Name)
			fmt.Printf("  IP地址: %s\n", iface.IPAddress)
			fmt.Printf("  MAC地址: %s\n", iface.MacAddress)
			fmt.Printf("  接收字节: %.2f MB\n", float64(iface.RxBytes)/1024/1024)
			fmt.Printf("  发送字节: %.2f MB\n", float64(iface.TxBytes)/1024/1024)
			fmt.Println()
		}

		// 输出进程状态
		fmt.Println("5. 进程状态 (CPU使用率前10):")
		fmt.Println("-----------------------------------------")
		fmt.Printf("%s\t%s\t%s\t%s\t%s\n", "PID", "CPU%", "MEM%", "用户", "命令")
		fmt.Println("-----------------------------------------")
		for _, proc := range status.Processes {
			cmd := proc.Command
			if len(cmd) > 50 {
				cmd = cmd[:50] + "..."
			}
			fmt.Printf("%d\t%.2f\t%.2f\t%s\t%s\n",
				proc.PID, proc.CPUPercent, proc.MemPercent, proc.User, cmd)
		}

		fmt.Println("\n=========================================")
		fmt.Println("报告结束")
		fmt.Println("=========================================")
	}
}

4. Shell自动化工具开发 ​

4.1 Shell脚本基础回顾 ​

  • 变量定义：VAR=value
  • 条件判断：if [ condition ]; then ... fi
  • 循环结构：for i in ...; do ... done、while [ condition ]; do ... done
  • 函数定义：function_name() { ... }
  • 命令替换：$(command) 或 `command`
  • 管道操作：command1 | command2

4.2 Shell自动化工具开发实例 ​

4.2.1 系统备份脚本 ​

bash
#!/bin/bash

# 系统备份脚本
# 功能：备份系统重要文件和配置

# 配置变量
BACKUP_DIR="/backup"
DATE=$(date +"%Y%m%d%H%M%S")
BACKUP_NAME="system-backup-$DATE"
BACKUP_FILE="$BACKUP_DIR/$BACKUP_NAME.tar.gz"
EXCLUDE_FILE="$BACKUP_DIR/exclude.txt"
LOG_FILE="$BACKUP_DIR/backup.log"

# 备份目录列表
BACKUP_PATHS=(  "/etc"  "/home"  "/var/www"  "/var/lib/mysql"  "/var/spool/cron"  "/root" )

# 排除文件列表
EXCLUDE_PATHS=(  "/home/*/.cache"  "/home/*/.local/share"  "/var/www/*/node_modules"  "/var/lib/mysql/*log*"  "/var/lib/mysql/*binlog*" )

# 日志函数
log() {
    echo "[$(date +"%Y-%m-%d %H:%M:%S")] $1" >> "$LOG_FILE"
    echo "[$(date +"%Y-%m-%d %H:%M:%S")] $1"
}

# 检查备份目录
check_backup_dir() {
    if [ ! -d "$BACKUP_DIR" ]; then
        log "创建备份目录: $BACKUP_DIR"
        mkdir -p "$BACKUP_DIR"
        if [ $? -ne 0 ]; then
            log "错误: 创建备份目录失败"
            exit 1
        fi
    fi
}

# 创建排除文件
create_exclude_file() {
    log "创建排除文件: $EXCLUDE_FILE"
    > "$EXCLUDE_FILE"
    for path in "${EXCLUDE_PATHS[@]}"; do
        echo "$path" >> "$EXCLUDE_FILE"
    done
}

# 检查磁盘空间
check_disk_space() {
    log "检查磁盘空间"
    FREE_SPACE=$(df -h "$BACKUP_DIR" | tail -1 | awk '{print $4}' | sed 's/G//')
    REQUIRED_SPACE=50  # 假设需要50G空间
    
    if (( $(echo "$FREE_SPACE < $REQUIRED_SPACE" | bc -l) )); then
        log "警告: 磁盘空间不足，当前可用: ${FREE_SPACE}G，建议: ${REQUIRED_SPACE}G"
    else
        log "磁盘空间充足: ${FREE_SPACE}G"
    fi
}

# 执行备份
perform_backup() {
    log "开始执行备份..."
    
    # 构建tar命令
    TAR_CMD="tar -czf "$BACKUP_FILE" --exclude-from="$EXCLUDE_FILE""
    
    for path in "${BACKUP_PATHS[@]}"; do
        if [ -d "$path" ] || [ -f "$path" ]; then
            TAR_CMD="$TAR_CMD "$path""
        else
            log "警告: 路径不存在，跳过: $path"
        fi
    done
    
    log "执行备份命令: $TAR_CMD"
    eval $TAR_CMD
    
    if [ $? -eq 0 ]; then
        log "备份成功: $BACKUP_FILE"
        log "备份大小: $(du -h "$BACKUP_FILE" | awk '{print $1}')"
        return 0
    else
        log "错误: 备份失败"
        return 1
    fi
}

# 备份验证
verify_backup() {
    log "验证备份文件..."
    if [ -f "$BACKUP_FILE" ]; then
        tar -tzf "$BACKUP_FILE" > /dev/null 2>&1
        if [ $? -eq 0 ]; then
            log "备份文件验证成功"
            return 0
        else
            log "错误: 备份文件验证失败"
            return 1
        fi
    else
        log "错误: 备份文件不存在"
        return 1
    fi
}

# 清理旧备份
cleanup_old_backups() {
    log "清理旧备份文件..."
    # 保留最近7天的备份
    find "$BACKUP_DIR" -name "system-backup-*.tar.gz" -mtime +7 -delete
    if [ $? -eq 0 ]; then
        log "清理旧备份成功"
    else
        log "错误: 清理旧备份失败"
    fi
}

# 发送通知
send_notification() {
    local status=$1
    if [ "$status" -eq 0 ]; then
        log "备份任务完成，状态: 成功"
        # 这里可以添加邮件通知或其他通知方式
    else
        log "备份任务完成，状态: 失败"
        # 这里可以添加错误通知方式
    fi
}

# 主函数
main() {
    log "========================================"
    log "开始系统备份任务"
    log "========================================"
    
    check_backup_dir
    create_exclude_file
    check_disk_space
    
    perform_backup
    BACKUP_STATUS=$?
    
    if [ $BACKUP_STATUS -eq 0 ]; then
        verify_backup
        VERIFY_STATUS=$?
        if [ $VERIFY_STATUS -eq 0 ]; then
            cleanup_old_backups
        fi
    fi
    
    send_notification $BACKUP_STATUS
    
    log "========================================"
    log "系统备份任务结束"
    log "========================================"
}

# 执行主函数
main

4.2.2 服务监控脚本 ​

bash
#!/bin/bash

# 服务监控脚本
# 功能：监控系统服务状态，异常时发送告警

# 配置变量
CONFIG_FILE="/etc/service-monitor.conf"
LOG_FILE="/var/log/service-monitor.log"
ALERT_FILE="/var/log/service-alert.log"

# 服务列表
SERVICES=(  "nginx"  "mysql"  "redis"  "sshd"  "docker" )

# 告警阈值
CPU_THRESHOLD=80
MEMORY_THRESHOLD=80
DISK_THRESHOLD=90

# 监控间隔（秒）
MONITOR_INTERVAL=60

# 日志函数
log() {
    echo "[$(date +"%Y-%m-%d %H:%M:%S")] $1" >> "$LOG_FILE"
    echo "[$(date +"%Y-%m-%d %H:%M:%S")] $1"
}

# 告警函数
alert() {
    local message="$1"
    echo "[$(date +"%Y-%m-%d %H:%M:%S")] ALERT: $message" >> "$ALERT_FILE"
    echo "[$(date +"%Y-%m-%d %H:%M:%S")] ALERT: $message"
    
    # 这里可以添加邮件通知、短信通知等
    # 示例：echo "$message" | mail -s "Service Monitor Alert" admin@example.com
}

# 加载配置文件
load_config() {
    if [ -f "$CONFIG_FILE" ]; then
        log "加载配置文件: $CONFIG_FILE"
        source "$CONFIG_FILE"
    else
        log "配置文件不存在，使用默认配置"
    fi
}

# 检查服务状态
check_service() {
    local service="$1"
    log "检查服务状态: $service"
    
    if systemctl is-active --quiet "$service"; then
        log "服务 $service 运行正常"
        return 0
    else
        local status=$(systemctl status "$service" | grep -E "Active:" | cut -d ':' -f2- | sed 's/^[ \t]*//')
        alert "服务 $service 异常: $status"
        return 1
    fi
}

# 检查CPU使用率
check_cpu() {
    log "检查CPU使用率"
    local cpu_usage=$(top -bn1 | grep "%Cpu(s):" | sed "s/.*, *\([0-9.]*\)%* id.*/\1/" | awk '{print 100 - $1}')
    local cpu_usage_int=$(printf "%.0f" "$cpu_usage")
    
    log "当前CPU使用率: ${cpu_usage_int}%"
    
    if [ "$cpu_usage_int" -gt "$CPU_THRESHOLD" ]; then
        alert "CPU使用率过高: ${cpu_usage_int}% (阈值: ${CPU_THRESHOLD}%)"
        return 1
    else
        return 0
    fi
}

# 检查内存使用率
check_memory() {
    log "检查内存使用率"
    local memory_usage=$(free | grep Mem | awk '{print $3/$2 * 100.0}')
    local memory_usage_int=$(printf "%.0f" "$memory_usage")
    
    log "当前内存使用率: ${memory_usage_int}%"
    
    if [ "$memory_usage_int" -gt "$MEMORY_THRESHOLD" ]; then
        alert "内存使用率过高: ${memory_usage_int}% (阈值: ${MEMORY_THRESHOLD}%)"
        return 1
    else
        return 0
    fi
}

# 检查磁盘使用率
check_disk() {
    log "检查磁盘使用率"
    local disk_usage=$(df -h | grep -v tmpfs | grep -v devtmpfs | awk '{print $5}' | sed 's/%//')
    
    for usage in $disk_usage; do
        if [ "$usage" -gt "$DISK_THRESHOLD" ]; then
            local mount_point=$(df -h | grep -v tmpfs | grep -v devtmpfs | grep "$usage%" | awk '{print $6}')
            alert "磁盘使用率过高: $usage% (阈值: ${DISK_THRESHOLD}%)，挂载点: $mount_point"
            return 1
        fi
    done
    
    log "磁盘使用率正常"
    return 0
}

# 检查系统负载
check_load() {
    log "检查系统负载"
    local load_avg=$(uptime | awk '{print $10 $11 $12}' | sed 's/,//g')
    local cpu_count=$(nproc)
    
    log "当前系统负载: $load_avg (CPU核心数: $cpu_count)"
    
    # 检查15分钟负载
    local load_15=$(echo $load_avg | awk '{print $3}')
    if (( $(echo "$load_15 > $cpu_count" | bc -l) )); then
        alert "系统负载过高: $load_15 (CPU核心数: $cpu_count)"
        return 1
    else
        return 0
    fi
}

# 检查网络连接
check_network() {
    log "检查网络连接"
    local ping_result=$(ping -c 3 google.com > /dev/null 2>&1)
    
    if [ $? -eq 0 ]; then
        log "网络连接正常"
        return 0
    else
        alert "网络连接异常"
        return 1
    fi
}

# 主监控函数
monitor() {
    log "========================================"
    log "开始监控任务"
    log "========================================"
    
    # 检查所有服务
    for service in "${SERVICES[@]}"; do
        check_service "$service"
    done
    
    # 检查系统状态
    check_cpu
    check_memory
    check_disk
    check_load
    check_network
    
    log "========================================"
    log "监控任务结束"
    log "========================================"
}

# 主函数
main() {
    # 创建日志目录
    mkdir -p $(dirname "$LOG_FILE")
    mkdir -p $(dirname "$ALERT_FILE")
    
    # 加载配置
    load_config
    
    # 执行监控
    monitor
    
    # 如果是作为守护进程运行，可以添加循环
    # while true; do
    #     monitor
    #     sleep $MONITOR_INTERVAL
    # done
}

# 执行主函数
main

5. Ansible自动化集成 ​

5.1 Ansible简介和安装 ​

Ansible是一个开源的配置管理和自动化工具，使用Python开发，基于SSH协议进行通信，无需在目标服务器上安装客户端。

安装Ansible：

bash
# 在CentOS/RHEL上安装
sudo yum install epel-release
sudo yum install ansible

# 在Ubuntu/Debian上安装
sudo apt update
sudo apt install ansible

# 使用pip安装
pip install ansible

5.2 Ansible Playbook开发 ​

基本Playbook结构：

yaml
---
- name: 安装和配置Nginx
  hosts: web_servers
  become: yes
  tasks:
    - name: 安装Nginx
      yum:
        name: nginx
        state: present

    - name: 启动Nginx服务
      service:
        name: nginx
        state: started
        enabled: yes

    - name: 复制Nginx配置文件
      copy:
        src: files/nginx.conf
        dest: /etc/nginx/nginx.conf
      notify:
        - 重启Nginx

  handlers:
    - name: 重启Nginx
      service:
        name: nginx
        state: restarted

5.3 Ansible模块开发 ​

自定义Ansible模块示例：

python
#!/usr/bin/env python3
"""
自定义Ansible模块：检查文件权限
"""

import os
import stat
import json

def main():
    """主函数"""
    # 模块参数
    module = {
        "argument_spec": {
            "path": {
                "required": True,
                "type": "str"
            },
            "mode": {
                "required": False,
                "type": "str"
            }
        },
        "supports_check_mode": True
    }
    
    # 导入Ansible模块库
    try:
        from ansible.module_utils.basic import AnsibleModule
        ansible_module = AnsibleModule(**module)
    except ImportError:
        # 用于测试
        class MockModule:
            def __init__(self, **kwargs):
                self.params = {
                    "path": "/etc/passwd",
                    "mode": None
                }
            def exit_json(self, **kwargs):
                print(json.dumps(kwargs))
            def fail_json(self, **kwargs):
                print(json.dumps(kwargs))
        ansible_module = MockModule(**module)
    
    path = ansible_module.params["path"]
    mode = ansible_module.params["mode"]
    
    # 检查文件是否存在
    if not os.path.exists(path):
        ansible_module.fail_json(msg=f"文件不存在: {path}")
    
    # 获取文件权限
    file_stat = os.stat(path)
    current_mode = stat.filemode(file_stat.st_mode)
    octal_mode = oct(file_stat.st_mode & 0o777)
    
    # 准备返回结果
    result = {
        "changed": False,
        "path": path,
        "current_mode": current_mode,
        "octal_mode": octal_mode,
        "owner": file_stat.st_uid,
        "group": file_stat.st_gid,
        "size": file_stat.st_size,
        "mtime": file_stat.st_mtime
    }
    
    # 如果指定了权限模式，则修改权限
    if mode:
        try:
            # 转换模式为八进制
            if not mode.startswith('0'):
                mode = '0' + mode
            new_mode = int(mode, 8)
            
            # 检查是否需要修改
            if (file_stat.st_mode & 0o777) != new_mode:
                if ansible_module.check_mode:
                    result["changed"] = True
                else:
                    os.chmod(path, new_mode)
                    result["changed"] = True
                    result["new_mode"] = stat.filemode(new_mode)
                    result["new_octal_mode"] = oct(new_mode)
        except Exception as e:
            ansible_module.fail_json(msg=f"修改权限失败: {str(e)}")
    
    ansible_module.exit_json(**result)

if __name__ == '__main__':
    main()

6. 自动化工具集成和部署 ​

6.1 工具集成方案 ​

  • 版本控制：使用Git管理工具代码
  • 依赖管理：使用pip、go mod等管理依赖
  • 配置管理：使用环境变量或配置文件管理配置
  • 日志管理：统一日志格式，便于分析
  • 监控告警：集成监控系统，及时发现问题

6.2 工具部署最佳实践 ​

  1. 标准化部署路径：

    • 工具脚本：/usr/local/bin/
    • 配置文件：/etc/tool-name/
    • 日志文件：/var/log/tool-name/

  2. 权限管理：

    • 执行权限：chmod +x tool.sh
    • 所有者：chown root:root tool.sh
    • 敏感配置：chmod 600 config.conf

  3. 自动化部署：

    • 使用Ansible部署工具到多服务器
    • 使用CI/CD流水线自动构建和部署
    • 使用容器化技术封装工具环境

  4. 文档和示例：

    • 提供详细的使用文档
    • 提供示例配置和使用场景
    • 提供故障排查指南

7. 自动化工具开发实战案例 ​

7.1 实例：服务器初始化工具 ​

功能需求：

  • 自动安装常用软件包
  • 配置系统参数
  • 设置防火墙规则
  • 创建用户和权限
  • 配置SSH服务
  • 安装监控代理

实现方案：

python
#!/usr/bin/env python3
"""
服务器初始化工具
"""

import argparse
import subprocess
import os
import json
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='/var/log/server-init.log'
)
logger = logging.getLogger(__name__)

class ServerInitializer:
    """服务器初始化器"""
    
    def __init__(self, config_file=None):
        """初始化服务器初始化器"""
        self.config = {
            "packages": [
                "vim", "git", "wget", "curl", "htop", "iftop",
                "iotop", "net-tools", "bash-completion", "chrony"
            ],
            "firewall_rules": [
                "ssh", "http", "https"
            ],
            "users": [],
            "ssh_config": {
                "Port": 22,
                "PermitRootLogin": "no",
                "PasswordAuthentication": "yes",
                "PubkeyAuthentication": "yes"
            },
            "sysctl": {
                "net.core.somaxconn": 65535,
                "net.ipv4.tcp_max_syn_backlog": 65535,
                "net.ipv4.tcp_fin_timeout": 30,
                "net.ipv4.tcp_keepalive_time": 1200
            }
        }
        
        if config_file:
            self._load_config(config_file)
    
    def _load_config(self, config_file):
        """加载配置文件"""
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
                self.config.update(config)
            logger.info(f"成功加载配置文件: {config_file}")
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
    
    def _run_command(self, command, sudo=False):
        """执行命令"""
        try:
            if sudo and os.geteuid() != 0:
                command = f"sudo {command}"
            
            logger.info(f"执行命令: {command}")
            result = subprocess.run(
                command, shell=True, capture_output=True, text=True
            )
            
            if result.returncode == 0:
                logger.info(f"命令执行成功: {command}")
                return True, result.stdout
            else:
                logger.error(f"命令执行失败: {command}\n错误信息: {result.stderr}")
                return False, result.stderr
        except Exception as e:
            logger.error(f"执行命令异常: {command}\n异常信息: {str(e)}")
            return False, str(e)
    
    def update_system(self):
        """更新系统"""
        logger.info("开始更新系统")
        
        # 检测系统类型
        if os.path.exists('/etc/redhat-release'):
            # RHEL/CentOS/Rocky
            success, _ = self._run_command('yum update -y', sudo=True)
        elif os.path.exists('/etc/debian_version'):
            # Debian/Ubuntu
            success, _ = self._run_command('apt update -y', sudo=True)
            if success:
                success, _ = self._run_command('apt upgrade -y', sudo=True)
        else:
            logger.error("不支持的系统类型")
            return False
        
        logger.info("系统更新完成")
        return success
    
    def install_packages(self):
        """安装软件包"""
        logger.info("开始安装软件包")
        
        packages = self.config.get('packages', [])
        if not packages:
            logger.info("没有需要安装的软件包")
            return True
        
        # 检测系统类型
        if os.path.exists('/etc/redhat-release'):
            # RHEL/CentOS/Rocky
            pkg_list = ' '.join(packages)
            success, _ = self._run_command(f'yum install -y {pkg_list}', sudo=True)
        elif os.path.exists('/etc/debian_version'):
            # Debian/Ubuntu
            pkg_list = ' '.join(packages)
            success, _ = self._run_command(f'apt install -y {pkg_list}', sudo=True)
        else:
            logger.error("不支持的系统类型")
            return False
        
        logger.info("软件包安装完成")
        return success
    
    def configure_firewall(self):
        """配置防火墙"""
        logger.info("开始配置防火墙")
        
        # 检测防火墙类型
        success, _ = self._run_command('which firewalld', sudo=True)
        if success:
            # 使用firewalld
            for service in self.config.get('firewall_rules', []):
                self._run_command(
                    f'firewall-cmd --permanent --add-service={service}', 
                    sudo=True
                )
            self._run_command('firewall-cmd --reload', sudo=True)
        else:
            # 尝试使用ufw
            success, _ = self._run_command('which ufw', sudo=True)
            if success:
                for service in self.config.get('firewall_rules', []):
                    self._run_command(
                        f'ufw allow {service}', 
                        sudo=True
                    )
                self._run_command('ufw --force enable', sudo=True)
            else:
                logger.warning("未找到可用的防火墙工具")
        
        logger.info("防火墙配置完成")
        return True
    
    def configure_ssh(self):
        """配置SSH服务"""
        logger.info("开始配置SSH服务")
        
        # 备份SSH配置文件
        self._run_command('cp /etc/ssh/sshd_config /etc/ssh/sshd_config.bak', sudo=True)
        
        # 读取当前配置
        ssh_config = {}
        try:
            with open('/etc/ssh/sshd_config', 'r') as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith('#'):
                        continue
                    if '=' in line:
                        key, value = line.split('=', 1)
                        ssh_config[key.strip()] = value.strip()
                    elif ' ' in line:
                        parts = line.split(' ', 1)
                        if len(parts) == 2:
                            ssh_config[parts[0]] = parts[1]
        except Exception as e:
            logger.error(f"读取SSH配置文件失败: {e}")
        
        # 更新配置
        for key, value in self.config.get('ssh_config', {}).items():
            ssh_config[key] = value
        
        # 写回配置文件
        try:
            with open('/etc/ssh/sshd_config', 'w') as f:
                f.write('# SSH服务配置\n')
                for key, value in ssh_config.items():
                    f.write(f'{key} {value}\n')
            
            # 重启SSH服务
            self._run_command('systemctl restart sshd', sudo=True)
            logger.info("SSH服务配置完成")
            return True
        except Exception as e:
            logger.error(f"配置SSH服务失败: {e}")
            return False
    
    def configure_sysctl(self):
        """配置系统参数"""
        logger.info("开始配置系统参数")
        
        # 备份sysctl配置
        self._run_command('cp /etc/sysctl.conf /etc/sysctl.conf.bak', sudo=True)
        
        # 读取当前配置
        sysctl_config = {}
        try:
            with open('/etc/sysctl.conf', 'r') as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith('#'):
                        continue
                    if '=' in line:
                        key, value = line.split('=', 1)
                        sysctl_config[key.strip()] = value.strip()
        except Exception as e:
            logger.error(f"读取sysctl配置文件失败: {e}")
        
        # 更新配置
        for key, value in self.config.get('sysctl', {}).items():
            sysctl_config[key] = value
        
        # 写回配置文件
        try:
            with open('/etc/sysctl.conf', 'w') as f:
                f.write('# 系统参数配置\n')
                for key, value in sysctl_config.items():
                    f.write(f'{key} = {value}\n')
            
            # 应用配置
            self._run_command('sysctl -p', sudo=True)
            logger.info("系统参数配置完成")
            return True
        except Exception as e:
            logger.error(f"配置系统参数失败: {e}")
            return False
    
    def create_users(self):
        """创建用户"""
        logger.info("开始创建用户")
        
        for user in self.config.get('users', []):
            username = user.get('name')
            password = user.get('password')
            sudo = user.get('sudo', False)
            
            if not username:
                continue
            
            # 检查用户是否存在
            success, _ = self._run_command(f'id {username}')
            if not success:
                # 创建用户
                if password:
                    self._run_command(
                        f"useradd -m -p $(openssl passwd -1 {password}) {username}", 
                        sudo=True
                    )
                else:
                    self._run_command(f'useradd -m {username}', sudo=True)
                
                # 添加到sudo组
                if sudo:
                    self._run_command(f'usermod -aG sudo {username}', sudo=True)
                
                logger.info(f"创建用户成功: {username}")
            else:
                logger.info(f"用户已存在: {username}")
        
        logger.info("用户创建完成")
        return True
    
    def run(self):
        """执行初始化"""
        logger.info("========================================")
        logger.info("开始服务器初始化")
        logger.info("========================================")
        
        # 执行各个初始化步骤
        self.update_system()
        self.install_packages()
        self.configure_firewall()
        self.configure_ssh()
        self.configure_sysctl()
        self.create_users()
        
        logger.info("========================================")
        logger.info("服务器初始化完成")
        logger.info("========================================")
        return True

def main():
    """主函数"""
    parser = argparse.ArgumentParser(description='服务器初始化工具')
    parser.add_argument('--config', '-c', help='配置文件路径')
    args = parser.parse_args()
    
    initializer = ServerInitializer(args.config)
    initializer.run()

if __name__ == '__main__':
    main()

7. 自动化工具开发最佳实践 ​

7.1 代码质量保证 ​

  • 代码风格：遵循PEP 8（Python）、Go Code Review Comments（Go）等代码风格规范
  • 代码审查：使用代码审查工具和流程，确保代码质量
  • 单元测试：编写单元测试，提高代码可靠性
  • 集成测试：编写集成测试，确保工具在真实环境中正常工作
  • 代码覆盖率：使用代码覆盖率工具，确保测试覆盖足够的代码

7.2 工具性能优化 ​

  • 资源管理：合理管理内存、CPU等资源，避免资源泄漏
  • 并发处理：使用并发技术，提高工具执行效率
  • 缓存机制：使用缓存，减少重复计算和IO操作
  • 算法优化：选择合适的算法，提高工具性能
  • 日志级别：合理设置日志级别，减少日志输出对性能的影响

7.3 工具安全性 ​

  • 输入验证：对用户输入进行验证，防止注入攻击
  • 权限管理：合理设置工具权限，避免权限滥用
  • 敏感信息处理：避免在代码中硬编码敏感信息，使用环境变量或配置文件
  • 加密传输：使用加密传输，保护数据安全
  • 安全审计：定期进行安全审计，发现和修复安全问题

7.4 工具可维护性 ​

  • 模块化设计：将工具分解为多个独立的模块，提高代码可维护性
  • 文档完善：提供详细的文档，包括使用说明、API文档等
  • 版本控制：使用版本控制工具，管理代码变更
  • 依赖管理：使用依赖管理工具，管理工具依赖
  • 错误处理：完善的错误处理机制，提高工具稳定性

8. 课程总结 ​

本课程详细介绍了自动化工具开发的核心概念、技术栈和实践方法，包括：

  • 自动化工具开发概述：了解自动化工具的定义、作用和技术栈
  • Python自动化工具开发：掌握Python自动化库和工具开发技巧
  • Go语言自动化工具开发：学习Go语言在自动化工具开发中的应用
  • Shell自动化工具开发：使用Shell脚本开发系统级自动化工具
  • Ansible自动化集成：集成Ansible进行配置管理和编排
  • 自动化工具集成和部署：学习工具集成方案和部署最佳实践
  • 自动化工具开发实战案例：通过实战案例，掌握自动化工具开发技巧
  • 自动化工具开发最佳实践：学习代码质量保证、性能优化、安全性和可维护性

通过本课程的学习，你已经掌握了自动化工具开发的核心技能，能够开发实用的自动化运维工具，提高工作效率。在实际工作中，你可以根据具体需求，选择合适的技术栈和方法，开发出更加高效、可靠的自动化工具。

后续学习建议 ​

  1. 深入学习特定领域的自动化工具开发：根据自己的工作需求，深入学习特定领域的自动化工具开发，如监控自动化、部署自动化等
  2. 学习DevOps相关技术：学习CI/CD、容器化等DevOps相关技术，将自动化工具与DevOps流程集成
  3. 参与开源项目：参与开源自动化工具项目，积累实战经验
  4. 持续关注技术发展：关注自动化领域的技术发展，学习新技术和新方法
  5. 实践项目：通过实际项目，巩固所学知识，提高自动化工具开发能力

自动化工具开发是运维开发工程师的核心技能之一，掌握这一技能将大大提高你的工作效率和竞争力。希望本课程对你有所帮助，祝你在自动化工具开发的道路上越走越远！

评论区

专业的Linux技术学习平台，从入门到精通的完整学习路径

Copyright © 2026 叶哥Linux技术分享 - linuxeasy.site