Informer 系统指南

Informer 系统指南

目录

1. 架构概述
2. 核心组件详解
3. 配置与部署
4. 使用指南
5. 监控与指标
6. 故障排除
7. API 参考
8. 最佳实践

架构概述

Informer 系统是一个用于监控 Kubernetes 资源变化并实时推送事件的高性能组件。它基于 Kubernetes 的 Informer 机制构建，提供了资源缓存、事件过滤、WebSocket 推送等功能。

系统架构图

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Kubernetes    │    │   Informer      │    │   WebSocket     │
│   API Server    │───▶│   Manager       │───▶│   Server        │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                              │                        │
                              ▼                        ▼
                       ┌─────────────────┐    ┌─────────────────┐
                       │   Cache         │    │   Client        │
                       │   Manager       │    │   Applications  │
                       └─────────────────┘    └─────────────────┘
                              │
                              ▼
                       ┌─────────────────┐
                       │   Metrics       │
                       │   Collector     │
                       └─────────────────┘

主要功能

• 资源监控: 监控 Kubernetes 资源（Pod、Service、Deployment 等）的变化
• 事件过滤: 根据应用标签和自定义规则过滤事件
• 实时推送: 通过 WebSocket 将事件实时推送给客户端
• 数据缓存: 缓存资源状态，提供高性能查询
• 指标收集: 收集系统运行指标，支持 Prometheus 监控

核心组件详解

InformerService

InformerService 是系统的核心组件，负责协调其他组件的工作。

type InformerService struct {
    // 上下文和配置
    ctx    context.Context
    config InformerServiceConfig
    
    // 核心组件
    clientManager    common.ClientManagerInterface
    podCacheManager  common.CacheManagerInterface
    kubeConfigManager common.KubeConfigManagerInterface
    websocketServer  WebSocketServerInterface
    
    // 事件处理
    eventHandlers    map[string]cache.ResourceEventHandler
    eventAdapter     *ResourceEventAdapter
    
    // 指标和监控
    metricsCollector *MetricsCollector
    
    // 同步控制
    mu               sync.RWMutex
    started          bool
}

主要方法

• Start(): 启动 InformerService 和所有依赖组件
• Stop(): 停止服务并清理资源
• RegisterHandler(resourceType string, handler cache.ResourceEventHandler): 注册资源事件处理器
• SetPodCacheManager(manager common.CacheManagerInterface): 设置 Pod 缓存管理器
• SetWebSocketServer(server WebSocketServerInterface): 设置 WebSocket 服务器

CacheManager

CacheManager 负责管理资源缓存，提供高性能的数据查询和更新。

type CacheManager struct {
    // 数据库连接
    db *sql.DB
    
    // 更新队列和批量处理
    updateQueue    chan CacheUpdateRequest
    batchSize      int
    batchTimeout   time.Duration
    
    // 指标
    metrics CacheMetrics
    
    // 运行状态
    mu     sync.RWMutex
    running bool
}

主要方法

• UpdatePod(applicationID string, podData map[string]interface{}): 更新 Pod 缓存
• GetPod(applicationID, podName, namespace string) (map[string]interface{}, error): 获取 Pod 信息
• GetApplicationStatus(applicationID string) (*ApplicationStatus, error): 获取应用状态
• Start(): 启动缓存管理器
• Stop(): 停止缓存管理器

WebSocketServer

WebSocketServer 负责与客户端建立 WebSocket 连接，并推送实时事件。

type Server struct {
    // 中心管理器
    hub *Hub
    
    // 配置
    config ServerConfig
    
    // 指标收集器
    metricsCollector *MetricsCollector
    
    // 运行状态
    mu     sync.RWMutex
    running bool
}

type Hub struct {
    // 客户端管理
    clients    map[*Client]bool
    register   chan *Client
    unregister chan *Client
    
    // 事件广播
    broadcast  chan []byte
    
    // 运行控制
    mu     sync.RWMutex
    closed bool
}

主要方法

• Start(ctx context.Context): 启动 WebSocket 服务器
• Stop(): 停止服务器
• HandleWebSocket(w http.ResponseWriter, r *http.Request): 处理 WebSocket 连接请求
• BroadcastEvent(eventType string, data interface{}): 广播事件给所有客户端

MetricsCollector

MetricsCollector 负责收集系统运行指标，支持 Prometheus 格式。

type MetricsCollector struct {
    // Prometheus 指标
    informerStarts     prometheus.Counter
    websocketConnections prometheus.Counter
    eventsProcessed    prometheus.Counter
    cacheOperations    prometheus.Counter
    errorCount         prometheus.Counter
    
    // 持续时间直方图
    eventProcessingDuration prometheus.Histogram
    cacheOperationDuration  prometheus.Histogram
    
    // 当前状态指标
    activeConnections   prometheus.Gauge
    cacheSize          prometheus.Gauge
}

主要方法

• RecordInformerStart(clusterID string): 记录 Informer 启动
• RecordWebSocketConnection(clusterID string): 记录 WebSocket 连接
• RecordEventProcessed(eventType string): 记录事件处理
• RecordCacheOperation(operation string): 记录缓存操作
• RecordError(errorType string): 记录错误

配置与部署

基本配置

type InformerServiceConfig struct {
    // 重新同步间隔
    ResyncInterval time.Duration
    
    // 工作线程数
    Workers int
    
    // 是否启用指标
    EnableMetrics bool
    
    // 命名空间过滤器
    NamespaceSelector string
    
    // 标签选择器
    LabelSelector string
}

默认配置

func DefaultInformerServiceConfig() InformerServiceConfig {
    return InformerServiceConfig{
        ResyncInterval:   10 * time.Minute,
        Workers:          5,
        EnableMetrics:    true,
        NamespaceSelector: "",
        LabelSelector:    "",
    }
}

部署配置

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: informer-service
spec:
  replicas: 2
  selector:
    matchLabels:
      app: informer-service
  template:
    metadata:
      labels:
        app: informer-service
    spec:
      containers:
      - name: informer-service
        image: your-registry/informer-service:latest
        ports:
        - containerPort: 8080
        env:
        - name: RESYNC_INTERVAL
          value: "10m"
        - name: WORKERS
          value: "5"
        - name: ENABLE_METRICS
          value: "true"
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"

服务配置

# service.yaml
apiVersion: v1
kind: Service
metadata:
  name: informer-service
spec:
  selector:
    app: informer-service
  ports:
  - name: http
    port: 80
    targetPort: 8080
  - name: websocket
    port: 8081
    targetPort: 8081
  type: ClusterIP

使用指南

初始化 InformerService

package main

import (
    "context"
    "database/sql"
    "log"
    
    _ "github.com/mattn/go-sqlite3"
    "cloud-deployment-api/model/informer"
)

func main() {
    // 创建数据库连接
    db, err := sql.Open("sqlite3", "informer.db")
    if err != nil {
        log.Fatalf("打开数据库失败: %v", err)
    }
    defer db.Close()
    
    // 创建组件
    cacheManager := informer.NewCacheManager(db)
    websocketServer := informer.NewWebSocketServer()
    
    // 创建 InformerService
    config := informer.DefaultInformerServiceConfig()
    informerService := informer.NewInformerService(config)
    
    // 设置依赖关系
    informerService.SetPodCacheManager(cacheManager)
    informerService.SetWebSocketServer(websocketServer)
    
    // 启动组件
    ctx := context.Background()
    
    cacheManager.Start()
    defer cacheManager.Stop()
    
    err = websocketServer.Start(ctx)
    if err != nil {
        log.Fatalf("启动WebSocket服务器失败: %v", err)
    }
    defer websocketServer.Stop()
    
    err = informerService.Start()
    if err != nil {
        log.Fatalf("启动InformerService失败: %v", err)
    }
    defer informerService.Stop()
    
    // 服务运行中...
    select {}
}

注册自定义事件处理器

// 创建自定义事件处理器
type CustomEventHandler struct {
    // 自定义字段
}

func (h *CustomEventHandler) OnAdd(obj interface{}) {
    // 处理资源添加事件
    log.Printf("资源添加: %v", obj)
}

func (h *CustomEventHandler) OnUpdate(oldObj, newObj interface{}) {
    // 处理资源更新事件
    log.Printf("资源更新: %v -> %v", oldObj, newObj)
}

func (h *CustomEventHandler) OnDelete(obj interface{}) {
    // 处理资源删除事件
    log.Printf("资源删除: %v", obj)
}

// 注册事件处理器
customHandler := &CustomEventHandler{}
informerService.RegisterHandler("pods", customHandler)

客户端连接示例

// WebSocket 客户端连接示例
const ws = new WebSocket('ws://localhost:8081/ws');

ws.onopen = function(event) {
    console.log('WebSocket连接已建立');
};

ws.onmessage = function(event) {
    const message = JSON.parse(event.data);
    console.log('收到消息:', message);
    
    switch(message.type) {
        case 'pod_added':
            handlePodAdded(message.data);
            break;
        case 'pod_updated':
            handlePodUpdated(message.data);
            break;
        case 'pod_deleted':
            handlePodDeleted(message.data);
            break;
        default:
            console.log('未知消息类型:', message.type);
    }
};

ws.onclose = function(event) {
    console.log('WebSocket连接已关闭');
};

ws.onerror = function(error) {
    console.error('WebSocket错误:', error);
};

function handlePodAdded(podData) {
    // 处理Pod添加事件
    console.log('Pod已添加:', podData);
}

function handlePodUpdated(podData) {
    // 处理Pod更新事件
    console.log('Pod已更新:', podData);
}

function handlePodDeleted(podData) {
    // 处理Pod删除事件
    console.log('Pod已删除:', podData);
}

监控与指标

Prometheus 指标

Informer 系统提供了丰富的 Prometheus 指标，用于监控系统运行状态：

指标配置

// 启用指标收集
config := informer.InformerServiceConfig{
    EnableMetrics: true,
}

// 获取指标收集器
metricsCollector := informerService.GetMetricsCollector()

// 注册指标到 Prometheus
prometheus.MustRegister(metricsCollector.GetPrometheusCollectors()...)

// 启动指标 HTTP 服务器
http.Handle("/metrics", promhttp.Handler())
log.Fatal(http.ListenAndServe(":9090", nil))

Grafana 仪表板

可以使用以下 Grafana 仪表板模板监控 Informer 系统：

{
  "dashboard": {
    "title": "Informer 系统监控",
    "panels": [
      {
        "title": "事件处理速率",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(events_processed_total[5m])",
            "legendFormat": "事件处理速率"
          }
        ]
      },
      {
        "title": "活跃连接数",
        "type": "singlestat",
        "targets": [
          {
            "expr": "active_connections",
            "legendFormat": "活跃连接数"
          }
        ]
      },
      {
        "title": "缓存大小",
        "type": "graph",
        "targets": [
          {
            "expr": "cache_size",
            "legendFormat": "缓存大小"
          }
        ]
      },
      {
        "title": "错误率",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(errors_total[5m])",
            "legendFormat": "错误率"
          }
        ]
      }
    ]
  }
}

故障排除

常见问题

1. Informer 无法启动

症状: InformerService 启动失败，报错 "无法连接到 Kubernetes API"

可能原因:

• Kubernetes 配置文件不存在或无效
• 网络连接问题
• 权限不足

解决方案:

// 检查 Kubernetes 配置
kubeconfig := os.Getenv("KUBECONFIG")
if kubeconfig == "" {
    kubeconfig = filepath.Join(os.Getenv("HOME"), ".kube", "config")
}

// 验证配置文件
if _, err := os.Stat(kubeconfig); os.IsNotExist(err) {
    log.Fatalf("Kubernetes 配置文件不存在: %s", kubeconfig)
}

// 使用配置文件创建客户端
config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
if err != nil {
    log.Fatalf("构建 Kubernetes 配置失败: %v", err)
}

2. WebSocket 连接频繁断开

症状: 客户端 WebSocket 连接频繁断开

可能原因:

• 网络不稳定
• 服务器负载过高
• 客户端心跳超时

解决方案:

// 增加读写超时时间
conn.SetReadDeadline(time.Now().Add(60 * time.Second))
conn.SetWriteDeadline(time.Now().Add(10 * time.Second))

// 实现心跳机制
func (c *Client) sendPing() {
    ticker := time.NewTicker(54 * time.Second)
    defer func() {
        ticker.Stop()
    }()
    
    for {
        select {
        case <-ticker.C:
            if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
        case <-c.close:
            return
        }
    }
}

3. 缓存更新缓慢

症状: 资源状态更新不及时

可能原因:

• 数据库性能问题
• 批量处理参数设置不当
• 更新队列阻塞

解决方案:

// 调整批量处理参数
cacheManager := informer.NewCacheManager(db)
cacheManager.SetBatchSize(100)        // 增加批量大小
cacheManager.SetBatchTimeout(100 * time.Millisecond)  // 减少批量超时

// 优化数据库性能
db.SetMaxOpenConns(25)
db.SetMaxIdleConns(25)
db.SetConnMaxLifetime(5 * time.Minute)

4. 内存使用过高

症状: 服务内存使用持续增长

可能原因:

• 缓存数据过多
• 内存泄漏
• 垃圾回收不及时

解决方案:

// 定期清理过期数据
func (cm *CacheManager) startCleanupRoutine() {
    ticker := time.NewTicker(10 * time.Minute)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            cm.cleanupExpiredData()
        case <-cm.ctx.Done():
            return
        }
    }
}

// 限制缓存大小
func (cm *CacheManager) enforceCacheSizeLimit() {
    if cm.cacheSize > cm.maxCacheSize {
        cm.evictLRU()
    }
}

日志分析

启用详细日志

// 设置日志级别
log.SetLevel(log.DebugLevel)

// 启用组件日志
informerService.SetLogger(log.New(os.Stdout, "[InformerService] ", log.LstdFlags))
cacheManager.SetLogger(log.New(os.Stdout, "[CacheManager] ", log.LstdFlags))
websocketServer.SetLogger(log.New(os.Stdout, "[WebSocketServer] ", log.LstdFlags))

关键日志信息

• [InformerService] Starting informer for resource: pods
• [CacheManager] Cache update batch processed: 50 items
• [WebSocketServer] Client connected: 127.0.0.1:54321
• [MetricsCollector] Event processed: pod_added

调试工具

健康检查端点

// 添加健康检查端点
http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
    status := map[string]interface{}{
        "status": "healthy",
        "components": map[string]string{
            "informer_service": "running",
            "cache_manager":    "running",
            "websocket_server": "running",
        },
        "metrics": map[string]interface{}{
            "active_connections": websocketServer.ActiveConnections(),
            "cache_size":        cacheManager.CacheSize(),
            "events_processed":  metricsCollector.EventsProcessed(),
        },
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(status)
})

调试 API

// 添加调试 API
http.HandleFunc("/debug/cache", func(w http.ResponseWriter, r *http.Request) {
    applicationID := r.URL.Query().Get("app")
    if applicationID == "" {
        http.Error(w, "缺少 app 参数", http.StatusBadRequest)
        return
    }
    
    status, err := cacheManager.GetApplicationStatus(applicationID)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(status)
})

API 参考

InformerService API

方法

NewInformerService(config InformerServiceConfig) *InformerService

创建新的 InformerService 实例。

参数:

• config: InformerService 配置

返回值:

• *InformerService: InformerService 实例

Start() error

启动 InformerService 和所有依赖组件。

返回值:

• error: 错误信息

Stop()

停止 InformerService 并清理资源。

RegisterHandler(resourceType string, handler cache.ResourceEventHandler)

注册资源事件处理器。

参数:

• resourceType: 资源类型（如 "pods", "services"）
• handler: 事件处理器

SetPodCacheManager(manager common.CacheManagerInterface)

设置 Pod 缓存管理器。

参数:

• manager: 缓存管理器接口

SetWebSocketServer(server WebSocketServerInterface)

设置 WebSocket 服务器。

参数:

• server: WebSocket 服务器接口

GetMetricsCollector() *MetricsCollector

获取指标收集器。

返回值:

• *MetricsCollector: 指标收集器实例

CacheManager API

方法

NewCacheManager(db *sql.DB) *CacheManager

创建新的 CacheManager 实例。

参数:

• db: 数据库连接

返回值:

• *CacheManager: CacheManager 实例

Start()

启动缓存管理器。

Stop()

停止缓存管理器。

UpdatePod(applicationID string, podData map[string]interface{}) error

更新 Pod 缓存。

参数:

• applicationID: 应用 ID
• podData: Pod 数据

返回值:

• error: 错误信息

GetPod(applicationID, podName, namespace string) (map[string]interface{}, error)

获取 Pod 信息。

参数:

• applicationID: 应用 ID
• podName: Pod 名称
• namespace: 命名空间

返回值:

• map[string]interface{}: Pod 数据
• error: 错误信息

GetApplicationStatus(applicationID string) (*ApplicationStatus, error)

获取应用状态。

参数:

• applicationID: 应用 ID

返回值:

• *ApplicationStatus: 应用状态
• error: 错误信息

WebSocketServer API

方法

NewWebSocketServer() *Server

创建新的 WebSocket 服务器实例。

返回值:

• *Server: WebSocket 服务器实例

Start(ctx context.Context) error

启动 WebSocket 服务器。

参数:

• ctx: 上下文

返回值:

• error: 错误信息

Stop()

停止 WebSocket 服务器。

HandleWebSocket(w http.ResponseWriter, r *http.Request)

处理 WebSocket 连接请求。

参数:

• w: HTTP 响应写入器
• r: HTTP 请求

BroadcastEvent(eventType string, data interface{})

广播事件给所有客户端。

参数:

• eventType: 事件类型
• data: 事件数据

MetricsCollector API

方法

NewMetricsCollector() *MetricsCollector

创建新的指标收集器实例。

返回值:

• *MetricsCollector: 指标收集器实例

RecordInformerStart(clusterID string)

记录 Informer 启动。

参数:

• clusterID: 集群 ID

RecordWebSocketConnection(clusterID string)

记录 WebSocket 连接。

参数:

• clusterID: 集群 ID

RecordEventProcessed(eventType string)

记录事件处理。

参数:

• eventType: 事件类型

RecordCacheOperation(operation string)

记录缓存操作。

参数:

• operation: 操作类型

RecordError(errorType string)

记录错误。

参数:

• errorType: 错误类型

GetPrometheusCollectors() []prometheus.Collector

获取 Prometheus 指标收集器。

返回值:

• []prometheus.Collector: Prometheus 指标收集器列表

最佳实践

性能优化

1. 合理设置工作线程数

   // 根据 CPU 核心数设置工作线程数
   numWorkers := runtime.NumCPU()
   config := informer.InformerServiceConfig{
       Workers: numWorkers,
   }
   

2. 批量处理缓存更新

   // 设置适当的批量大小和超时
   cacheManager.SetBatchSize(100)
   cacheManager.SetBatchTimeout(100 * time.Millisecond)
   

3. 使用连接池

   // 配置数据库连接池
   db.SetMaxOpenConns(25)
   db.SetMaxIdleConns(25)
   db.SetConnMaxLifetime(5 * time.Minute)
   

4. 限制缓存大小

   // 设置最大缓存大小
   cacheManager.SetMaxCacheSize(10000)
   

可靠性保障

1. 实现优雅关闭

   // 处理系统信号
   sigCh := make(chan os.Signal, 1)
   signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
   
   go func() {
       <-sigCh
       log.Println("收到关闭信号，开始优雅关闭...")
       informerService.Stop()
       cacheManager.Stop()
       websocketServer.Stop()
       os.Exit(0)
   }()
   

2. 添加重试机制

   // 实现指数退避重试
   func retryWithBackoff(fn func() error, maxRetries int) error {
       var err error
       for i := 0; i < maxRetries; i++ {
           err = fn()
           if err == nil {
               return nil
           }
           
           backoff := time.Duration(math.Pow(2, float64(i))) * time.Second
           time.Sleep(backoff)
       }
       return fmt.Errorf("重试 %d 次后仍然失败: %v", maxRetries, err)
   }
   

3. 实现断路器模式

   // 使用断路器防止级联故障
   type CircuitBreaker struct {
       maxFailures int
       resetTimeout time.Duration
       failures     int
       lastFailTime time.Time
       mu           sync.Mutex
   }
   
   func (cb *CircuitBreaker) Call(fn func() error) error {
       cb.mu.Lock()
       defer cb.mu.Unlock()
       
       if cb.failures >= cb.maxFailures {
           if time.Since(cb.lastFailTime) < cb.resetTimeout {
               return errors.New("断路器打开")
           }
           // 重置断路器
           cb.failures = 0
       }
       
       err := fn()
       if err != nil {
           cb.failures++
           cb.lastFailTime = time.Now()
           return err
       }
       
       // 成功时重置失败计数
       cb.failures = 0
       return nil
   }
   

安全考虑

1. 认证和授权

   // 实现 WebSocket 认证
   func (s *Server) authenticateToken(r *http.Request) (bool, error) {
       token := r.URL.Query().Get("token")
       if token == "" {
           return false, errors.New("缺少认证令牌")
       }
       
       // 验证令牌
       return validateToken(token), nil
   }
   

2. 速率限制

   // 实现连接速率限制
   type RateLimiter struct {
       requests map[string]*rate.Limiter
       mu       sync.RWMutex
       r        rate.Limit
       b        int
   }
   
   func (rl *RateLimiter) Allow(ip string) bool {
       rl.mu.Lock()
       defer rl.mu.Unlock()
       
       if _, exists := rl.requests[ip]; !exists {
           rl.requests[ip] = rate.NewLimiter(rl.r, rl.b)
       }
       
       return rl.requests[ip].Allow()
   }
   

3. 输入验证

   // 验证输入数据
   func validatePodData(podData map[string]interface{}) error {
       if podData == nil {
           return errors.New("Pod 数据不能为空")
       }
       
       metadata, ok := podData["metadata"].(map[string]interface{})
       if !ok {
           return errors.New("无效的 metadata 字段")
       }
       
       name, ok := metadata["name"].(string)
       if !ok || name == "" {
           return errors.New("Pod 名称不能为空")
       }
       
       return nil
   }
   

监控和告警

1. 设置关键指标告警

   # Prometheus 告警规则
   groups:
   - name: informer_alerts
     rules:
     - alert: InformerDown
       expr: informer_starts_total == 0
       for: 5m
       labels:
         severity: critical
       annotations:
         summary: "Informer 服务停止运行"
         description: "Informer 服务在过去 5 分钟内没有启动记录"
   

2. 实现自定义健康检查

   // 实现详细健康检查
   func (s *InformerService) HealthCheck() map[string]interface{} {
       health := map[string]interface{}{
           "status": "healthy",
           "components": make(map[string]interface{}),
       }
       
       components := health["components"].(map[string]interface{})
       
       // 检查缓存管理器
       if s.podCacheManager != nil && s.podCacheManager.IsRunning() {
           components["cache_manager"] = map[string]interface{}{
               "status": "healthy",
               "cache_size": s.podCacheManager.CacheSize(),
           }
       } else {
           components["cache_manager"] = map[string]interface{}{
               "status": "unhealthy",
           }
           health["status"] = "unhealthy"
       }
       
       // 检查 WebSocket 服务器
       if s.websocketServer != nil && s.websocketServer.IsRunning() {
           components["websocket_server"] = map[string]interface{}{
               "status": "healthy",
               "active_connections": s.websocketServer.ActiveConnections(),
           }
       } else {
           components["websocket_server"] = map[string]interface{}{
               "status": "unhealthy",
           }
           health["status"] = "unhealthy"
       }
       
       return health
   }
   

3. 实现分布式追踪

   // 使用 OpenTelemetry 实现分布式追踪
   import (
       "go.opentelemetry.io/otel"
       "go.opentelemetry.io/otel/trace"
   )
   
   func (s *InformerService) processEventWithTracing(eventType string, data interface{}) error {
       tracer := otel.Tracer("informer-service")
       ctx, span := tracer.Start(context.Background(), "processEvent")
       defer span.End()
       
       span.SetAttributes(
           attribute.String("event.type", eventType),
       )
       
       // 处理事件
       err := s.processEvent(ctx, eventType, data)
       if err != nil {
           span.SetAttributes(
               attribute.String("error.message", err.Error()),
           )
           span.SetStatus(codes.Error, "事件处理失败")
       }
       
       return err
   }
   

版本历史

v1.0.0 (2023-10-01)

• 初始版本发布
• 基本的 Informer 功能
• WebSocket 事件推送
• 缓存管理

v1.1.0 (2023-11-15)

• 添加 Prometheus 指标
• 改进错误处理
• 性能优化

v1.2.0 (2024-01-20)

• 支持多集群
• 添加应用过滤器
• 改进缓存机制

v1.3.0 (2024-03-10)

• 添加分布式追踪
• 改进 WebSocket 连接管理
• 增强监控和告警

贡献指南

欢迎贡献代码！请遵循以下步骤：

1. Fork 项目
2. 创建功能分支 (git checkout -b feature/amazing-feature)
3. 提交更改 (git commit -m 'Add some amazing feature')
4. 推送到分支 (git push origin feature/amazing-feature)
5. 创建 Pull Request

许可证

本项目采用 MIT 许可证。详情请参阅 LICENSE 文件。

联系方式

如有问题或建议，请通过以下方式联系：

• 邮箱: support@example.com
• GitHub Issues: [项目 Issues 页面]
• 文档: [项目文档站点]