Moby项目源码剖析 - 实战经验和最佳实践

graph TD
    Start[开始学习Moby] --> BasicConcepts[理解基础概念]
    BasicConcepts --> ClientAPI[掌握客户端API]
    ClientAPI --> CoreModules[深入核心模块]
    CoreModules --> NetworkStorage[网络和存储]
    NetworkStorage --> Advanced[高级特性]
    Advanced --> Contribution[贡献代码]
    
    BasicConcepts --> |包含| Containers[容器概念]
    BasicConcepts --> Images[镜像概念]
    BasicConcepts --> Networks[网络概念]
    
    ClientAPI --> |包含| SDKUsage[SDK使用]
    ClientAPI --> APIDesign[API设计]
    
    CoreModules --> |包含| Daemon[守护进程]
    CoreModules --> Runtime[运行时]
    
    Advanced --> |包含| Plugins[插件开发]
    Advanced --> Security[安全机制]
    Advanced --> Performance[性能优化]

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"text/tabwriter"
	"time"

	"github.com/moby/moby/client"
	"github.com/moby/moby/api/types/container"
	"github.com/moby/moby/api/types/image"
	"github.com/spf13/cobra"
)

// ContainerManager 容器管理器
type ContainerManager struct {
	client *client.Client
	ctx    context.Context
}

// NewContainerManager 创建容器管理器实例
func NewContainerManager() (*ContainerManager, error) {
	// 1. 创建Docker客户端
	cli, err := client.NewClientWithOpts(
		client.FromEnv,                    // 从环境变量读取配置
		client.WithAPIVersionNegotiation(), // 启用API版本协商
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create docker client: %v", err)
	}

	// 2. 验证连接
	ctx := context.Background()
	if _, err := cli.Ping(ctx); err != nil {
		return nil, fmt.Errorf("failed to connect to docker daemon: %v", err)
	}

	return &ContainerManager{
		client: cli,
		ctx:    ctx,
	}, nil
}

// ListContainers 列出容器，支持多种过滤选项
func (cm *ContainerManager) ListContainers(all bool, format string) error {
	// 1. 配置列表选项
	options := container.ListOptions{
		All: all,
	}

	// 2. 获取容器列表
	containers, err := cm.client.ContainerList(cm.ctx, options)
	if err != nil {
		return fmt.Errorf("failed to list containers: %v", err)
	}

	// 3. 格式化输出
	switch format {
	case "table":
		return cm.printContainerTable(containers)
	case "json":
		return cm.printContainerJSON(containers)
	default:
		return cm.printContainerTable(containers)
	}
}

// printContainerTable 以表格形式打印容器信息
func (cm *ContainerManager) printContainerTable(containers []types.Container) error {
	w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
	defer w.Flush()

	// 打印表头
	fmt.Fprintln(w, "CONTAINER ID\tIMAGE\tCOMMAND\tCREATED\tSTATUS\tPORTS\tNAMES")

	// 打印容器信息
	for _, c := range containers {
		// 格式化创建时间
		created := time.Unix(c.Created, 0).Format("2006-01-02 15:04:05")
		
		// 格式化容器名称（去掉前缀斜杠）
		names := ""
		if len(c.Names) > 0 {
			names = c.Names[0][1:] // 去掉前缀 '/'
		}

		// 格式化端口信息
		ports := cm.formatPorts(c.Ports)

		// 输出容器信息行
		fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
			c.ID[:12],           // 短ID
			c.Image,             // 镜像名
			c.Command,           // 命令
			created,             // 创建时间
			c.State+" "+c.Status, // 状态
			ports,               // 端口
			names,               // 名称
		)
	}

	return nil
}

// formatPorts 格式化端口映射信息
func (cm *ContainerManager) formatPorts(ports []types.Port) string {
	if len(ports) == 0 {
		return ""
	}

	var result []string
	for _, port := range ports {
		if port.PublicPort != 0 {
			result = append(result, fmt.Sprintf("%s:%d->%d/%s",
				port.IP, port.PublicPort, port.PrivatePort, port.Type))
		} else {
			result = append(result, fmt.Sprintf("%d/%s",
				port.PrivatePort, port.Type))
		}
	}

	return strings.Join(result, ", ")
}

// CreateContainer 创建容器的高级封装
func (cm *ContainerManager) CreateContainer(opts ContainerCreateOptions) (string, error) {
	// 1. 验证镜像是否存在，如果不存在则尝试拉取
	if err := cm.ensureImage(opts.Image); err != nil {
		return "", fmt.Errorf("failed to ensure image: %v", err)
	}

	// 2. 构建容器配置
	config := &container.Config{
		Image:        opts.Image,
		Cmd:          opts.Cmd,
		Env:          opts.Env,
		WorkingDir:   opts.WorkingDir,
		ExposedPorts: opts.ExposedPorts,
		Labels:       opts.Labels,
	}

	// 3. 构建主机配置
	hostConfig := &container.HostConfig{
		PortBindings:  opts.PortBindings,
		RestartPolicy: opts.RestartPolicy,
		AutoRemove:    opts.AutoRemove,
		Binds:         opts.Binds,
		Resources:     opts.Resources,
	}

	// 4. 创建容器
	resp, err := cm.client.ContainerCreate(
		cm.ctx,
		config,
		hostConfig,
		nil, // 网络配置
		nil, // 平台配置
		opts.Name,
	)
	if err != nil {
		return "", fmt.Errorf("failed to create container: %v", err)
	}

	// 5. 输出警告信息（如果有）
	for _, warning := range resp.Warnings {
		fmt.Printf("WARNING: %s\n", warning)
	}

	return resp.ID, nil
}

// ContainerCreateOptions 容器创建选项
type ContainerCreateOptions struct {
	Name          string
	Image         string
	Cmd           []string
	Env           []string
	WorkingDir    string
	ExposedPorts  nat.PortSet
	PortBindings  nat.PortMap
	RestartPolicy container.RestartPolicy
	AutoRemove    bool
	Binds         []string
	Resources     container.Resources
	Labels        map[string]string
}

// ensureImage 确保镜像存在，不存在则拉取
func (cm *ContainerManager) ensureImage(imageName string) error {
	// 1. 检查镜像是否已存在
	images, err := cm.client.ImageList(cm.ctx, image.ListOptions{})
	if err != nil {
		return err
	}

	// 2. 查找匹配的镜像
	for _, img := range images {
		for _, tag := range img.RepoTags {
			if tag == imageName {
				fmt.Printf("镜像 %s 已存在\n", imageName)
				return nil
			}
		}
	}

	// 3. 镜像不存在，开始拉取
	fmt.Printf("正在拉取镜像 %s...\n", imageName)
	reader, err := cm.client.ImagePull(cm.ctx, imageName, image.PullOptions{})
	if err != nil {
		return err
	}
	defer reader.Close()

	// 4. 显示拉取进度
	return cm.showPullProgress(reader)
}

// showPullProgress 显示镜像拉取进度
func (cm *ContainerManager) showPullProgress(reader io.Reader) error {
	decoder := json.NewDecoder(reader)
	
	for {
		var message struct {
			Status         string `json:"status"`
			Progress       string `json:"progress,omitempty"`
			ProgressDetail struct {
				Current int64 `json:"current"`
				Total   int64 `json:"total"`
			} `json:"progressDetail,omitempty"`
		}

		if err := decoder.Decode(&message); err != nil {
			if err == io.EOF {
				break
			}
			return err
		}

		// 显示进度信息
		if message.Progress != "" {
			fmt.Printf("\r%s: %s", message.Status, message.Progress)
		} else {
			fmt.Println(message.Status)
		}
	}

	fmt.Println("\n镜像拉取完成")
	return nil
}

// StartContainer 启动容器
func (cm *ContainerManager) StartContainer(containerID string) error {
	if err := cm.client.ContainerStart(cm.ctx, containerID, container.StartOptions{}); err != nil {
		return fmt.Errorf("failed to start container: %v", err)
	}
	
	fmt.Printf("容器 %s 已启动\n", containerID[:12])
	return nil
}

// StopContainer 停止容器
func (cm *ContainerManager) StopContainer(containerID string, timeout *int) error {
	options := container.StopOptions{}
	if timeout != nil {
		options.Timeout = timeout
	}

	if err := cm.client.ContainerStop(cm.ctx, containerID, options); err != nil {
		return fmt.Errorf("failed to stop container: %v", err)
	}
	
	fmt.Printf("容器 %s 已停止\n", containerID[:12])
	return nil
}

// RemoveContainer 删除容器
func (cm *ContainerManager) RemoveContainer(containerID string, force bool) error {
	options := container.RemoveOptions{
		Force: force,
	}

	if err := cm.client.ContainerRemove(cm.ctx, containerID, options); err != nil {
		return fmt.Errorf("failed to remove container: %v", err)
	}
	
	fmt.Printf("容器 %s 已删除\n", containerID[:12])
	return nil
}

// 命令行接口实现
func main() {
	var rootCmd = &cobra.Command{
		Use:   "container-mgr",
		Short: "一个简化的容器管理工具",
		Long:  "基于Moby客户端SDK开发的容器管理工具，用于学习Docker API",
	}

	// 初始化容器管理器
	mgr, err := NewContainerManager()
	if err != nil {
		log.Fatal("初始化失败:", err)
	}

	// 添加子命令
	rootCmd.AddCommand(
		createListCommand(mgr),
		createRunCommand(mgr),
		createStartCommand(mgr),
		createStopCommand(mgr),
		createRemoveCommand(mgr),
	)

	if err := rootCmd.Execute(); err != nil {
		log.Fatal(err)
	}
}

// createListCommand 创建list命令
func createListCommand(mgr *ContainerManager) *cobra.Command {
	var all bool
	var format string

	cmd := &cobra.Command{
		Use:   "list",
		Short: "列出容器",
		RunE: func(cmd *cobra.Command, args []string) error {
			return mgr.ListContainers(all, format)
		},
	}

	cmd.Flags().BoolVarP(&all, "all", "a", false, "显示所有容器")
	cmd.Flags().StringVar(&format, "format", "table", "输出格式 (table|json)")

	return cmd
}

// createRunCommand 创建run命令
func createRunCommand(mgr *ContainerManager) *cobra.Command {
	var opts ContainerCreateOptions

	cmd := &cobra.Command{
		Use:   "run [OPTIONS] IMAGE [COMMAND] [ARG...]",
		Short: "创建并启动容器",
		Args:  cobra.MinimumNArgs(1),
		RunE: func(cmd *cobra.Command, args []string) error {
			opts.Image = args[0]
			if len(args) > 1 {
				opts.Cmd = args[1:]
			}

			// 创建容器
			containerID, err := mgr.CreateContainer(opts)
			if err != nil {
				return err
			}

			// 启动容器
			return mgr.StartContainer(containerID)
		},
	}

	// 添加常用选项
	cmd.Flags().StringVar(&opts.Name, "name", "", "容器名称")
	cmd.Flags().StringSliceVarP(&opts.Env, "env", "e", nil, "环境变量")
	cmd.Flags().StringVarP(&opts.WorkingDir, "workdir", "w", "", "工作目录")
	cmd.Flags().BoolVar(&opts.AutoRemove, "rm", false, "容器退出时自动删除")

	return cmd
}

graph TB
    subgraph "监控面板架构"
        WebUI[Web界面]
        API[REST API服务]
        Collector[数据收集器]
        Storage[时序数据库]
        Notifier[告警通知]
    end
    
    subgraph "Docker守护进程"
        DockerAPI[Docker API]
        EventsAPI[Events API]
        StatsAPI[Stats API]
    end
    
    WebUI --> API
    API --> Storage
    Collector --> DockerAPI
    Collector --> EventsAPI
    Collector --> StatsAPI
    Collector --> Storage
    Storage --> Notifier

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/moby/moby/client"
	"github.com/moby/moby/api/types"
	"github.com/moby/moby/api/types/events"
	"github.com/gorilla/mux"
	"github.com/gorilla/websocket"
)

// MonitoringService 监控服务
type MonitoringService struct {
	client         *client.Client
	ctx            context.Context
	statsCollector *StatsCollector
	eventCollector *EventCollector
	clients        map[*websocket.Conn]bool
	clientsMux     sync.RWMutex
	upgrader       websocket.Upgrader
}

// StatsCollector 统计信息收集器
type StatsCollector struct {
	client  *client.Client
	ctx     context.Context
	stats   map[string]*ContainerStats
	statsMux sync.RWMutex
}

// ContainerStats 容器统计信息
type ContainerStats struct {
	ContainerID   string    `json:"container_id"`
	Name          string    `json:"name"`
	CPUPercent    float64   `json:"cpu_percent"`
	MemoryUsage   uint64    `json:"memory_usage"`
	MemoryPercent float64   `json:"memory_percent"`
	NetworkRx     uint64    `json:"network_rx"`
	NetworkTx     uint64    `json:"network_tx"`
	BlockRead     uint64    `json:"block_read"`
	BlockWrite    uint64    `json:"block_write"`
	Timestamp     time.Time `json:"timestamp"`
}

// EventCollector 事件收集器
type EventCollector struct {
	client      *client.Client
	ctx         context.Context
	eventChan   chan events.Message
	subscribers []chan events.Message
	subMux      sync.RWMutex
}

// NewMonitoringService 创建监控服务
func NewMonitoringService() (*MonitoringService, error) {
	// 1. 创建Docker客户端
	cli, err := client.NewClientWithOpts(
		client.FromEnv,
		client.WithAPIVersionNegotiation(),
	)
	if err != nil {
		return nil, err
	}

	ctx := context.Background()

	// 2. 创建统计收集器
	statsCollector := &StatsCollector{
		client: cli,
		ctx:    ctx,
		stats:  make(map[string]*ContainerStats),
	}

	// 3. 创建事件收集器
	eventCollector := &EventCollector{
		client:    cli,
		ctx:       ctx,
		eventChan: make(chan events.Message, 100),
	}

	// 4. 创建监控服务
	service := &MonitoringService{
		client:         cli,
		ctx:            ctx,
		statsCollector: statsCollector,
		eventCollector: eventCollector,
		clients:        make(map[*websocket.Conn]bool),
		upgrader: websocket.Upgrader{
			CheckOrigin: func(r *http.Request) bool {
				return true // 允许跨域
			},
		},
	}

	return service, nil
}

// Start 启动监控服务
func (ms *MonitoringService) Start() error {
	// 1. 启动统计信息收集
	go ms.statsCollector.Start()

	// 2. 启动事件收集
	go ms.eventCollector.Start()

	// 3. 启动事件分发
	go ms.eventDistributor()

	// 4. 启动HTTP服务器
	return ms.startHTTPServer()
}

// Start 启动统计信息收集
func (sc *StatsCollector) Start() {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			sc.collectStats()
		case <-sc.ctx.Done():
			return
		}
	}
}

// collectStats 收集统计信息
func (sc *StatsCollector) collectStats() {
	// 1. 获取运行中的容器列表
	containers, err := sc.client.ContainerList(sc.ctx, container.ListOptions{})
	if err != nil {
		log.Printf("Failed to list containers: %v", err)
		return
	}

	// 2. 并行收集每个容器的统计信息
	var wg sync.WaitGroup
	for _, c := range containers {
		wg.Add(1)
		go func(containerID string, name string) {
			defer wg.Done()
			
			// 获取容器统计信息
			stats, err := sc.client.ContainerStats(sc.ctx, containerID, false)
			if err != nil {
				log.Printf("Failed to get stats for container %s: %v", containerID, err)
				return
			}
			defer stats.Body.Close()

			// 解析统计信息
			var statsData types.Stats
			if err := json.NewDecoder(stats.Body).Decode(&statsData); err != nil {
				log.Printf("Failed to decode stats: %v", err)
				return
			}

			// 计算统计指标
			containerStats := sc.calculateStats(containerID, name, &statsData)

			// 更新统计信息缓存
			sc.statsMux.Lock()
			sc.stats[containerID] = containerStats
			sc.statsMux.Unlock()
		}(c.ID, c.Names[0])
	}

	wg.Wait()
}

// calculateStats 计算统计指标
func (sc *StatsCollector) calculateStats(containerID, name string, stats *types.Stats) *ContainerStats {
	// 1. 计算CPU使用率
	cpuPercent := calculateCPUPercent(stats)

	// 2. 计算内存使用情况
	memoryUsage := stats.MemoryStats.Usage
	memoryPercent := float64(memoryUsage) / float64(stats.MemoryStats.Limit) * 100

	// 3. 计算网络I/O
	var networkRx, networkTx uint64
	for _, network := range stats.Networks {
		networkRx += network.RxBytes
		networkTx += network.TxBytes
	}

	// 4. 计算磁盘I/O
	var blockRead, blockWrite uint64
	for _, blkio := range stats.BlkioStats.IoServiceBytesRecursive {
		if blkio.Op == "Read" {
			blockRead += blkio.Value
		} else if blkio.Op == "Write" {
			blockWrite += blkio.Value
		}
	}

	return &ContainerStats{
		ContainerID:   containerID,
		Name:          name,
		CPUPercent:    cpuPercent,
		MemoryUsage:   memoryUsage,
		MemoryPercent: memoryPercent,
		NetworkRx:     networkRx,
		NetworkTx:     networkTx,
		BlockRead:     blockRead,
		BlockWrite:    blockWrite,
		Timestamp:     time.Now(),
	}
}

// calculateCPUPercent 计算CPU使用百分比
func calculateCPUPercent(stats *types.Stats) float64 {
	// Docker统计信息中的CPU计算逻辑
	cpuDelta := float64(stats.CPUStats.CPUUsage.TotalUsage) - float64(stats.PreCPUStats.CPUUsage.TotalUsage)
	systemDelta := float64(stats.CPUStats.SystemUsage) - float64(stats.PreCPUStats.SystemUsage)

	if systemDelta > 0.0 && cpuDelta > 0.0 {
		return (cpuDelta / systemDelta) * float64(len(stats.CPUStats.CPUUsage.PercpuUsage)) * 100.0
	}
	return 0.0
}

// Start 启动事件收集
func (ec *EventCollector) Start() {
	// 监听Docker事件
	eventChan, errChan := ec.client.Events(ec.ctx, types.EventsOptions{})

	for {
		select {
		case event := <-eventChan:
			ec.eventChan <- event
		case err := <-errChan:
			log.Printf("Error receiving events: %v", err)
		case <-ec.ctx.Done():
			return
		}
	}
}

// Subscribe 订阅事件
func (ec *EventCollector) Subscribe() chan events.Message {
	ec.subMux.Lock()
	defer ec.subMux.Unlock()

	eventChan := make(chan events.Message, 10)
	ec.subscribers = append(ec.subscribers, eventChan)
	return eventChan
}

// eventDistributor 事件分发器
func (ms *MonitoringService) eventDistributor() {
	for {
		select {
		case event := <-ms.eventCollector.eventChan:
			// 分发事件给所有订阅者
			ms.eventCollector.subMux.RLock()
			for _, subscriber := range ms.eventCollector.subscribers {
				select {
				case subscriber <- event:
				default:
					// 非阻塞发送，避免慢订阅者影响其他订阅者
				}
			}
			ms.eventCollector.subMux.RUnlock()

			// 通过WebSocket分发给前端客户端
			ms.broadcastEvent(event)
		case <-ms.ctx.Done():
			return
		}
	}
}

// broadcastEvent 广播事件到WebSocket客户端
func (ms *MonitoringService) broadcastEvent(event events.Message) {
	ms.clientsMux.RLock()
	defer ms.clientsMux.RUnlock()

	message, _ := json.Marshal(map[string]interface{}{
		"type":  "event",
		"event": event,
	})

	for conn := range ms.clients {
		if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
			log.Printf("Failed to write to WebSocket: %v", err)
			conn.Close()
			delete(ms.clients, conn)
		}
	}
}

// startHTTPServer 启动HTTP服务器
func (ms *MonitoringService) startHTTPServer() error {
	router := mux.NewRouter()

	// API路由
	api := router.PathPrefix("/api/v1").Subrouter()
	api.HandleFunc("/stats", ms.handleStats).Methods("GET")
	api.HandleFunc("/containers", ms.handleContainers).Methods("GET")

	// WebSocket路由
	router.HandleFunc("/ws", ms.handleWebSocket)

	// 静态文件服务
	router.PathPrefix("/").Handler(http.FileServer(http.Dir("./web/")))

	log.Println("监控服务启动在端口 :8080")
	return http.ListenAndServe(":8080", router)
}

// handleStats 处理统计信息API
func (ms *MonitoringService) handleStats(w http.ResponseWriter, r *http.Request) {
	ms.statsCollector.statsMux.RLock()
	defer ms.statsCollector.statsMux.RUnlock()

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(ms.statsCollector.stats)
}

// handleContainers 处理容器列表API
func (ms *MonitoringService) handleContainers(w http.ResponseWriter, r *http.Request) {
	containers, err := ms.client.ContainerList(ms.ctx, container.ListOptions{All: true})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(containers)
}

// handleWebSocket 处理WebSocket连接
func (ms *MonitoringService) handleWebSocket(w http.ResponseWriter, r *http.Request) {
	conn, err := ms.upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Printf("Failed to upgrade to WebSocket: %v", err)
		return
	}

	// 注册客户端
	ms.clientsMux.Lock()
	ms.clients[conn] = true
	ms.clientsMux.Unlock()

	// 发送初始数据
	ms.sendInitialData(conn)

	// 处理客户端消息（心跳等）
	for {
		_, _, err := conn.ReadMessage()
		if err != nil {
			log.Printf("WebSocket read error: %v", err)
			break
		}
	}

	// 清理连接
	ms.clientsMux.Lock()
	delete(ms.clients, conn)
	ms.clientsMux.Unlock()
	conn.Close()
}

// sendInitialData 发送初始数据
func (ms *MonitoringService) sendInitialData(conn *websocket.Conn) {
	// 发送当前统计信息
	ms.statsCollector.statsMux.RLock()
	stats := ms.statsCollector.stats
	ms.statsCollector.statsMux.RUnlock()

	message, _ := json.Marshal(map[string]interface{}{
		"type":  "initial_stats",
		"stats": stats,
	})

	conn.WriteMessage(websocket.TextMessage, message)
}

func main() {
	service, err := NewMonitoringService()
	if err != nil {
		log.Fatal("Failed to create monitoring service:", err)
	}

	log.Fatal(service.Start())
}

sequenceDiagram
    participant Docker as Docker Daemon
    participant Plugin as Network Plugin
    participant Driver as Network Driver
    participant System as Host Network

    Note over Docker,System: 网络创建流程
    Docker->>Plugin: CreateNetwork Request
    Plugin->>Driver: Initialize Network Driver
    Driver->>System: Configure Host Network
    System-->>Driver: Network Created
    Driver-->>Plugin: Network Info
    Plugin-->>Docker: CreateNetwork Response

    Note over Docker,System: 端点创建流程  
    Docker->>Plugin: CreateEndpoint Request
    Plugin->>Driver: Create Network Endpoint
    Driver->>System: Configure Interface
    System-->>Driver: Interface Ready
    Driver-->>Plugin: Endpoint Info
    Plugin-->>Docker: CreateEndpoint Response

    Note over Docker,System: 容器加入网络
    Docker->>Plugin: Join Request
    Plugin->>Driver: Attach Container
    Driver->>System: Setup Container Network
    System-->>Driver: Container Connected
    Driver-->>Plugin: Join Info
    Plugin-->>Docker: Join Response

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"

	"github.com/docker/go-plugins-helpers/network"
	"github.com/vishvananda/netlink"
	"github.com/vishvananda/netns"
)

// CustomNetworkDriver 自定义网络驱动
type CustomNetworkDriver struct {
	networks  map[string]*NetworkInfo
	endpoints map[string]*EndpointInfo
}

// NetworkInfo 网络信息
type NetworkInfo struct {
	ID      string                 `json:"id"`
	Name    string                 `json:"name"`
	Gateway string                 `json:"gateway"`
	Subnet  string                 `json:"subnet"`
	Bridge  string                 `json:"bridge"`
	Options map[string]interface{} `json:"options"`
}

// EndpointInfo 端点信息
type EndpointInfo struct {
	ID        string `json:"id"`
	NetworkID string `json:"network_id"`
	Interface string `json:"interface"`
	IPAddress string `json:"ip_address"`
	MacAddress string `json:"mac_address"`
}

// NewCustomNetworkDriver 创建自定义网络驱动
func NewCustomNetworkDriver() *CustomNetworkDriver {
	return &CustomNetworkDriver{
		networks:  make(map[string]*NetworkInfo),
		endpoints: make(map[string]*EndpointInfo),
	}
}

// GetCapabilities 获取驱动能力
func (d *CustomNetworkDriver) GetCapabilities() (*network.CapabilitiesResponse, error) {
	log.Printf("收到GetCapabilities请求")
	
	return &network.CapabilitiesResponse{
		Scope:             "local",        // 作用域：local/global
		ConnectivityScope: "local",        // 连接性作用域
	}, nil
}

// AllocateNetwork 分配网络资源
func (d *CustomNetworkDriver) AllocateNetwork(req *network.AllocateNetworkRequest) (*network.AllocateNetworkResponse, error) {
	log.Printf("收到AllocateNetwork请求: %+v", req)
	
	// 对于local作用域的驱动，通常不需要特殊的网络分配逻辑
	return &network.AllocateNetworkResponse{}, nil
}

// FreeNetwork 释放网络资源
func (d *CustomNetworkDriver) FreeNetwork(req *network.FreeNetworkRequest) error {
	log.Printf("收到FreeNetwork请求: %+v", req)
	
	// 清理网络资源
	delete(d.networks, req.NetworkID)
	return nil
}

// CreateNetwork 创建网络
func (d *CustomNetworkDriver) CreateNetwork(req *network.CreateNetworkRequest) error {
	log.Printf("收到CreateNetwork请求: %+v", req)
	
	// 1. 解析网络配置
	netInfo := &NetworkInfo{
		ID:      req.NetworkID,
		Options: req.Options,
	}
	
	// 2. 从IPv4Data中获取子网和网关信息
	if len(req.IPv4Data) > 0 {
		ipv4Data := req.IPv4Data[0]
		netInfo.Subnet = ipv4Data.Pool
		netInfo.Gateway = ipv4Data.Gateway
	}
	
	// 3. 生成桥接设备名称
	bridgeName := fmt.Sprintf("br-%s", req.NetworkID[:12])
	netInfo.Bridge = bridgeName
	
	// 4. 创建Linux桥接设备
	if err := d.createBridge(bridgeName, netInfo.Gateway, netInfo.Subnet); err != nil {
		return fmt.Errorf("创建网桥失败: %v", err)
	}
	
	// 5. 存储网络信息
	d.networks[req.NetworkID] = netInfo
	
	log.Printf("网络创建成功: %s (桥接: %s)", req.NetworkID, bridgeName)
	return nil
}

// createBridge 创建Linux桥接设备
func (d *CustomNetworkDriver) createBridge(name, gateway, subnet string) error {
	// 1. 创建桥接设备
	bridge := &netlink.Bridge{
		LinkAttrs: netlink.LinkAttrs{
			Name: name,
		},
	}
	
	// 2. 检查桥接是否已存在
	if _, err := netlink.LinkByName(name); err == nil {
		log.Printf("桥接 %s 已存在", name)
		return nil
	}
	
	// 3. 添加桥接设备
	if err := netlink.LinkAdd(bridge); err != nil {
		return fmt.Errorf("添加桥接设备失败: %v", err)
	}
	
	// 4. 启用桥接设备
	if err := netlink.LinkSetUp(bridge); err != nil {
		return fmt.Errorf("启用桥接设备失败: %v", err)
	}
	
	// 5. 配置网关地址
	if gateway != "" && subnet != "" {
		gatewayAddr, err := netlink.ParseAddr(fmt.Sprintf("%s/%s", gateway, subnet))
		if err != nil {
			return fmt.Errorf("解析网关地址失败: %v", err)
		}
		
		if err := netlink.AddrAdd(bridge, gatewayAddr); err != nil {
			return fmt.Errorf("添加网关地址失败: %v", err)
		}
	}
	
	log.Printf("桥接 %s 创建成功", name)
	return nil
}

// DeleteNetwork 删除网络
func (d *CustomNetworkDriver) DeleteNetwork(req *network.DeleteNetworkRequest) error {
	log.Printf("收到DeleteNetwork请求: %+v", req)
	
	// 1. 获取网络信息
	netInfo, exists := d.networks[req.NetworkID]
	if !exists {
		log.Printf("网络 %s 不存在", req.NetworkID)
		return nil
	}
	
	// 2. 删除桥接设备
	if err := d.deleteBridge(netInfo.Bridge); err != nil {
		log.Printf("删除桥接失败: %v", err)
		// 不返回错误，继续清理
	}
	
	// 3. 清理网络信息
	delete(d.networks, req.NetworkID)
	
	log.Printf("网络删除成功: %s", req.NetworkID)
	return nil
}

// deleteBridge 删除Linux桥接设备
func (d *CustomNetworkDriver) deleteBridge(name string) error {
	// 1. 获取桥接设备
	bridge, err := netlink.LinkByName(name)
	if err != nil {
		// 桥接不存在
		return nil
	}
	
	// 2. 关闭桥接设备
	if err := netlink.LinkSetDown(bridge); err != nil {
		log.Printf("关闭桥接设备失败: %v", err)
	}
	
	// 3. 删除桥接设备
	if err := netlink.LinkDel(bridge); err != nil {
		return fmt.Errorf("删除桥接设备失败: %v", err)
	}
	
	log.Printf("桥接 %s 删除成功", name)
	return nil
}

// CreateEndpoint 创建端点
func (d *CustomNetworkDriver) CreateEndpoint(req *network.CreateEndpointRequest) (*network.CreateEndpointResponse, error) {
	log.Printf("收到CreateEndpoint请求: %+v", req)
	
	// 1. 获取网络信息
	netInfo, exists := d.networks[req.NetworkID]
	if !exists {
		return nil, fmt.Errorf("网络 %s 不存在", req.NetworkID)
	}
	
	// 2. 解析接口信息
	var ipAddress, macAddress string
	if req.Interface != nil {
		ipAddress = req.Interface.Address
		macAddress = req.Interface.MacAddress
	}
	
	// 3. 创建veth对
	vethName := fmt.Sprintf("veth%s", req.EndpointID[:8])
	peerName := fmt.Sprintf("peer%s", req.EndpointID[:8])
	
	if err := d.createVethPair(vethName, peerName); err != nil {
		return nil, fmt.Errorf("创建veth对失败: %v", err)
	}
	
	// 4. 将veth的一端连接到桥接
	if err := d.attachToBridge(vethName, netInfo.Bridge); err != nil {
		return nil, fmt.Errorf("连接到桥接失败: %v", err)
	}
	
	// 5. 存储端点信息
	endpointInfo := &EndpointInfo{
		ID:         req.EndpointID,
		NetworkID:  req.NetworkID,
		Interface:  peerName,
		IPAddress:  ipAddress,
		MacAddress: macAddress,
	}
	d.endpoints[req.EndpointID] = endpointInfo
	
	// 6. 返回接口信息
	response := &network.CreateEndpointResponse{}
	if req.Interface == nil {
		// Docker没有提供接口配置，我们需要返回配置信息
		response.Interface = &network.EndpointInterface{
			Address: ipAddress,
		}
	}
	
	log.Printf("端点创建成功: %s", req.EndpointID)
	return response, nil
}

// createVethPair 创建veth对
func (d *CustomNetworkDriver) createVethPair(vethName, peerName string) error {
	// 1. 创建veth链接
	veth := &netlink.Veth{
		LinkAttrs: netlink.LinkAttrs{
			Name: vethName,
		},
		PeerName: peerName,
	}
	
	// 2. 添加veth对
	if err := netlink.LinkAdd(veth); err != nil {
		return fmt.Errorf("添加veth对失败: %v", err)
	}
	
	// 3. 启用veth设备
	if err := netlink.LinkSetUp(veth); err != nil {
		return fmt.Errorf("启用veth设备失败: %v", err)
	}
	
	// 4. 启用peer设备
	peer, err := netlink.LinkByName(peerName)
	if err != nil {
		return fmt.Errorf("获取peer设备失败: %v", err)
	}
	
	if err := netlink.LinkSetUp(peer); err != nil {
		return fmt.Errorf("启用peer设备失败: %v", err)
	}
	
	log.Printf("veth对创建成功: %s <-> %s", vethName, peerName)
	return nil
}

// attachToBridge 将设备连接到桥接
func (d *CustomNetworkDriver) attachToBridge(deviceName, bridgeName string) error {
	// 1. 获取设备
	device, err := netlink.LinkByName(deviceName)
	if err != nil {
		return fmt.Errorf("获取设备失败: %v", err)
	}
	
	// 2. 获取桥接
	bridge, err := netlink.LinkByName(bridgeName)
	if err != nil {
		return fmt.Errorf("获取桥接失败: %v", err)
	}
	
	// 3. 连接设备到桥接
	if err := netlink.LinkSetMaster(device, bridge); err != nil {
		return fmt.Errorf("连接到桥接失败: %v", err)
	}
	
	log.Printf("设备 %s 已连接到桥接 %s", deviceName, bridgeName)
	return nil
}

// DeleteEndpoint 删除端点
func (d *CustomNetworkDriver) DeleteEndpoint(req *network.DeleteEndpointRequest) error {
	log.Printf("收到DeleteEndpoint请求: %+v", req)
	
	// 1. 获取端点信息
	endpointInfo, exists := d.endpoints[req.EndpointID]
	if !exists {
		log.Printf("端点 %s 不存在", req.EndpointID)
		return nil
	}
	
	// 2. 删除veth设备（会自动删除整个veth对）
	vethName := fmt.Sprintf("veth%s", req.EndpointID[:8])
	if err := d.deleteVeth(vethName); err != nil {
		log.Printf("删除veth失败: %v", err)
	}
	
	// 3. 清理端点信息
	delete(d.endpoints, req.EndpointID)
	
	log.Printf("端点删除成功: %s", req.EndpointID)
	return nil
}

// deleteVeth 删除veth设备
func (d *CustomNetworkDriver) deleteVeth(vethName string) error {
	// 获取veth设备
	veth, err := netlink.LinkByName(vethName)
	if err != nil {
		// 设备不存在
		return nil
	}
	
	// 删除veth设备
	if err := netlink.LinkDel(veth); err != nil {
		return fmt.Errorf("删除veth设备失败: %v", err)
	}
	
	log.Printf("veth %s 删除成功", vethName)
	return nil
}

// Join 容器加入网络
func (d *CustomNetworkDriver) Join(req *network.JoinRequest) (*network.JoinResponse, error) {
	log.Printf("收到Join请求: %+v", req)
	
	// 1. 获取端点信息
	endpointInfo, exists := d.endpoints[req.EndpointID]
	if !exists {
		return nil, fmt.Errorf("端点 %s 不存在", req.EndpointID)
	}
	
	// 2. 返回加入信息
	response := &network.JoinResponse{
		InterfaceName: network.InterfaceName{
			SrcName:   endpointInfo.Interface,
			DstPrefix: "eth",
		},
	}
	
	log.Printf("容器加入网络成功: 端点 %s", req.EndpointID)
	return response, nil
}

// Leave 容器离开网络
func (d *CustomNetworkDriver) Leave(req *network.LeaveRequest) error {
	log.Printf("收到Leave请求: %+v", req)
	
	// 对于这个简单的实现，离开网络时不需要特殊处理
	// 实际的清理工作在DeleteEndpoint中完成
	
	log.Printf("容器离开网络成功: 端点 %s", req.EndpointID)
	return nil
}

// DiscoverNew 发现新资源
func (d *CustomNetworkDriver) DiscoverNew(req *network.DiscoveryNotification) error {
	log.Printf("收到DiscoverNew请求: %+v", req)
	return nil
}

// DiscoverDelete 删除发现的资源
func (d *CustomNetworkDriver) DiscoverDelete(req *network.DiscoveryNotification) error {
	log.Printf("收到DiscoverDelete请求: %+v", req)
	return nil
}

// ProgramExternalConnectivity 配置外部连接性
func (d *CustomNetworkDriver) ProgramExternalConnectivity(req *network.ProgramExternalConnectivityRequest) error {
	log.Printf("收到ProgramExternalConnectivity请求: %+v", req)
	return nil
}

// RevokeExternalConnectivity 撤销外部连接性
func (d *CustomNetworkDriver) RevokeExternalConnectivity(req *network.RevokeExternalConnectivityRequest) error {
	log.Printf("收到RevokeExternalConnectivity请求: %+v", req)
	return nil
}

// main 主函数
func main() {
	// 创建网络驱动
	driver := NewCustomNetworkDriver()
	
	// 创建网络处理器
	handler := network.NewHandler(driver)
	
	// 启动HTTP服务器
	log.Printf("自定义网络插件启动在 /run/docker/plugins/custom-network.sock")
	
	// 确保插件目录存在
	pluginDir := "/run/docker/plugins"
	if err := os.MkdirAll(pluginDir, 0755); err != nil {
		log.Fatal("创建插件目录失败:", err)
	}
	
	// 监听Unix socket
	socketPath := filepath.Join(pluginDir, "custom-network.sock")
	if err := handler.ServeUnix(socketPath, 0); err != nil {
		log.Fatal("启动网络插件失败:", err)
	}
}