import "github.com/cloudwego/kitex/pkg/connpool"

// 推荐的长连接池配置
client.WithLongConnection(connpool.IdleConfig{
    MaxIdlePerAddress: 10,                    // 每个地址最大空闲连接数
    MaxIdleGlobal:     100,                   // 全局最大空闲连接数
    MaxIdleTimeout:    time.Minute * 3,       // 空闲超时时间
    MinIdlePerAddress: 2,                     // 每个地址最小空闲连接数
    MaxConnPerAddress: 50,                    // 每个地址最大连接数
})

// 连接池状态监控
func monitorConnPool(pool connpool.Pool) {
    if reporter, ok := pool.(connpool.Reporter); ok {
        go func() {
            ticker := time.NewTicker(time.Second * 30)
            defer ticker.Stop()
            
            for range ticker.C {
                stats := reporter.Reporter()
                log.Printf("ConnPool Stats: %+v", stats)
                
                // 发送到监控系统
                sendMetrics("connpool.active", stats.ActiveConnections)
                sendMetrics("connpool.idle", stats.IdleConnections)
                sendMetrics("connpool.total", stats.TotalConnections)
            }
        }()
    }
}

// 在生成代码时启用 Frugal
// kitex -use github.com/cloudwego/frugal hello.thrift

// 或者在客户端/服务端配置中启用
import "github.com/cloudwego/frugal"

// 客户端启用 Frugal
client.WithPayloadCodec(frugal.NewDefaultCodec())

// 服务端启用 Frugal  
server.WithPayloadCodec(frugal.NewDefaultCodec())

// 启用 RPCInfo 对象池
import "github.com/cloudwego/kitex/pkg/rpcinfo"

func init() {
    rpcinfo.EnablePool()
}

// 自定义对象池
type RequestPool struct {
    pool sync.Pool
}

func NewRequestPool() *RequestPool {
    return &RequestPool{
        pool: sync.Pool{
            New: func() interface{} {
                return &YourRequest{}
            },
        },
    }
}

func (p *RequestPool) Get() *YourRequest {
    return p.pool.Get().(*YourRequest)
}

func (p *RequestPool) Put(req *YourRequest) {
    req.Reset() // 重置对象状态
    p.pool.Put(req)
}

import "github.com/cloudwego/netpoll"

// 客户端网络优化
client.WithDialer(netpoll.NewDialer(
    netpoll.WithDialTimeout(time.Millisecond * 500),
    netpoll.WithKeepAlive(time.Minute * 5),
))

// 服务端网络优化
server.WithListener(netpoll.CreateListener("tcp", ":8888",
    netpoll.WithReusePort(true),
    netpoll.WithTCPNoDelay(true),
))

// 批量请求处理
type BatchProcessor struct {
    batchSize int
    timeout   time.Duration
    buffer    chan *Request
    client    YourServiceClient
}

func (p *BatchProcessor) Process(req *Request) error {
    select {
    case p.buffer <- req:
        return nil
    case <-time.After(p.timeout):
        return errors.New("batch buffer full")
    }
}

func (p *BatchProcessor) worker() {
    batch := make([]*Request, 0, p.batchSize)
    ticker := time.NewTicker(p.timeout)
    defer ticker.Stop()
    
    for {
        select {
        case req := <-p.buffer:
            batch = append(batch, req)
            if len(batch) >= p.batchSize {
                p.processBatch(batch)
                batch = batch[:0]
            }
            
        case <-ticker.C:
            if len(batch) > 0 {
                p.processBatch(batch)
                batch = batch[:0]
            }
        }
    }
}

func (p *BatchProcessor) processBatch(batch []*Request) {
    // 批量处理逻辑
    for _, req := range batch {
        go func(r *Request) {
            resp, err := p.client.Call(context.Background(), r)
            r.callback(resp, err)
        }(req)
    }
}

// 分层超时配置
const (
    // 连接超时：快速失败
    ConnectTimeout = time.Millisecond * 500
    
    // RPC 超时：业务处理时间 + 网络传输时间
    RPCTimeout = time.Second * 3
    
    // 读写超时：单次 IO 操作超时
    ReadWriteTimeout = time.Second * 2
)

client.WithConnectTimeout(ConnectTimeout)
client.WithRPCTimeout(RPCTimeout)  
client.WithReadWriteTimeout(ReadWriteTimeout)

// 基于历史延迟的动态超时
type DynamicTimeout struct {
    history []time.Duration
    mutex   sync.RWMutex
    maxSize int
}

func (dt *DynamicTimeout) Record(duration time.Duration) {
    dt.mutex.Lock()
    defer dt.mutex.Unlock()
    
    dt.history = append(dt.history, duration)
    if len(dt.history) > dt.maxSize {
        dt.history = dt.history[1:]
    }
}

func (dt *DynamicTimeout) GetTimeout() time.Duration {
    dt.mutex.RLock()
    defer dt.mutex.RUnlock()
    
    if len(dt.history) == 0 {
        return time.Second * 3 // 默认超时
    }
    
    // 计算 P95 延迟
    sorted := make([]time.Duration, len(dt.history))
    copy(sorted, dt.history)
    sort.Slice(sorted, func(i, j int) bool {
        return sorted[i] < sorted[j]
    })
    
    p95Index := int(float64(len(sorted)) * 0.95)
    p95Latency := sorted[p95Index]
    
    // 超时时间 = P95 延迟 * 2 + 缓冲时间
    return p95Latency*2 + time.Millisecond*500
}

// 使用动态超时
dynamicTimeout := &DynamicTimeout{maxSize: 1000}

// 在中间件中记录延迟
func TimeoutMiddleware(dt *DynamicTimeout) endpoint.Middleware {
    return func(next endpoint.Endpoint) endpoint.Endpoint {
        return func(ctx context.Context, req, resp interface{}) error {
            start := time.Now()
            err := next(ctx, req, resp)
            duration := time.Since(start)
            
            if err == nil {
                dt.Record(duration)
            }
            
            return err
        }
    }
}

import "github.com/cloudwego/kitex/pkg/retry"

// 基于错误类型的重试策略
client.WithFailureRetry(retry.NewFailurePolicy(
    retry.WithMaxRetryTimes(3),
    retry.WithMaxDurationMS(10000),
    retry.WithInitialDelay(10),
    retry.WithMaxDelay(1000),
    retry.WithDelayPolicy(retry.BackOffDelayPolicy),
    
    // 自定义重试判断
    retry.WithRetryIfNeeded(func(err error, ri rpcinfo.RPCInfo) bool {
        // 网络错误重试
        if isNetworkError(err) {
            return true
        }
        
        // 服务端 5xx 错误重试
        if isServerError(err) {
            return true
        }
        
        // 业务错误不重试
        if isBizError(err) {
            return false
        }
        
        return false
    }),
))

func isNetworkError(err error) bool {
    if err == nil {
        return false
    }
    
    // 检查网络相关错误
    errStr := err.Error()
    return strings.Contains(errStr, "connection refused") ||
           strings.Contains(errStr, "timeout") ||
           strings.Contains(errStr, "connection reset")
}

// 备份请求配置
client.WithBackupRequest(retry.NewBackupPolicy(
    retry.WithRetryDelayMS(100),  // 100ms 后发起备份请求
    retry.WithStopPolicy(retry.StopPolicyType(1)), // 任一请求成功即停止
))

import "github.com/cloudwego/kitex/pkg/circuitbreak"

// 服务级熔断 + 实例级熔断
client.WithCircuitBreaker(circuitbreak.NewCBSuite(
    // 服务级熔断：保护整个服务
    circuitbreak.WithServiceCBConfig(circuitbreak.CBConfig{
        Enable:    true,
        ErrRate:   0.5,    // 错误率阈值 50%
        MinSample: 200,    // 最小采样数
        StatIntervalMS: 1000, // 统计间隔 1s
        StatSlidingWindowS: 10, // 滑动窗口 10s
    }),
    
    // 实例级熔断：保护单个实例
    circuitbreak.WithInstanceCBConfig(circuitbreak.CBConfig{
        Enable:    true,
        ErrRate:   0.3,    // 错误率阈值 30%
        MinSample: 100,    // 最小采样数
        StatIntervalMS: 1000,
        StatSlidingWindowS: 10,
    }),
))

// 基于延迟的熔断器
type LatencyCircuitBreaker struct {
    maxLatency    time.Duration
    windowSize    int
    latencies     []time.Duration
    state         circuitbreak.State
    mutex         sync.RWMutex
}

func (cb *LatencyCircuitBreaker) IsAllowed(ri rpcinfo.RPCInfo) bool {
    cb.mutex.RLock()
    defer cb.mutex.RUnlock()
    
    return cb.state == circuitbreak.Closed
}

func (cb *LatencyCircuitBreaker) OnRequestDone(ri rpcinfo.RPCInfo, err error) {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    // 记录延迟
    stats := ri.Stats()
    if stats != nil {
        latency := stats.GetEvent(stats.RPCFinish).Time().Sub(stats.GetEvent(stats.RPCStart).Time())
        cb.latencies = append(cb.latencies, latency)
        
        if len(cb.latencies) > cb.windowSize {
            cb.latencies = cb.latencies[1:]
        }
    }
    
    // 检查是否需要熔断
    if len(cb.latencies) >= cb.windowSize {
        avgLatency := cb.calculateAvgLatency()
        if avgLatency > cb.maxLatency {
            cb.state = circuitbreak.Open
        } else {
            cb.state = circuitbreak.Closed
        }
    }
}

func (cb *LatencyCircuitBreaker) calculateAvgLatency() time.Duration {
    var total time.Duration
    for _, latency := range cb.latencies {
        total += latency
    }
    return total / time.Duration(len(cb.latencies))
}

import "github.com/kitex-contrib/monitor-prometheus"

// 客户端监控
client.WithSuite(prometheus.NewClientSuite(
    prometheus.WithRegistry(prometheus.DefaultRegisterer),
    prometheus.WithDisableServer(false),
    prometheus.WithServerAddr(":9091"),
))

// 服务端监控
server.WithSuite(prometheus.NewServerSuite(
    prometheus.WithRegistry(prometheus.DefaultRegisterer),
    prometheus.WithDisableServer(false), 
    prometheus.WithServerAddr(":9092"),
))

import "github.com/prometheus/client_golang/prometheus"

var (
    // 请求计数器
    requestTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "kitex_requests_total",
            Help: "Total number of requests",
        },
        []string{"service", "method", "status"},
    )
    
    // 请求延迟直方图
    requestDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name: "kitex_request_duration_seconds",
            Help: "Request duration in seconds",
            Buckets: prometheus.DefBuckets,
        },
        []string{"service", "method"},
    )
    
    // 活跃连接数
    activeConnections = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "kitex_active_connections",
            Help: "Number of active connections",
        },
        []string{"service", "target"},
    )
)

func init() {
    prometheus.MustRegister(requestTotal)
    prometheus.MustRegister(requestDuration)
    prometheus.MustRegister(activeConnections)
}

// 监控中间件
func MetricsMiddleware(next endpoint.Endpoint) endpoint.Endpoint {
    return func(ctx context.Context, req, resp interface{}) error {
        start := time.Now()
        
        ri := rpcinfo.GetRPCInfo(ctx)
        service := ri.To().ServiceName()
        method := ri.To().Method()
        
        err := next(ctx, req, resp)
        
        duration := time.Since(start)
        status := "success"
        if err != nil {
            status = "error"
        }
        
        // 记录指标
        requestTotal.WithLabelValues(service, method, status).Inc()
        requestDuration.WithLabelValues(service, method).Observe(duration.Seconds())
        
        return err
    }
}

import "github.com/kitex-contrib/tracer-opentracing"

// 客户端追踪
client.WithSuite(opentracing.NewDefaultClientSuite())

// 服务端追踪
server.WithSuite(opentracing.NewDefaultServerSuite())

import "github.com/opentracing/opentracing-go"

// 自定义追踪中间件
func TracingMiddleware(tracer opentracing.Tracer) endpoint.Middleware {
    return func(next endpoint.Endpoint) endpoint.Endpoint {
        return func(ctx context.Context, req, resp interface{}) error {
            ri := rpcinfo.GetRPCInfo(ctx)
            
            // 创建 Span
            span := tracer.StartSpan(
                fmt.Sprintf("%s.%s", ri.To().ServiceName(), ri.To().Method()),
                opentracing.Tag{Key: "component", Value: "kitex"},
                opentracing.Tag{Key: "rpc.service", Value: ri.To().ServiceName()},
                opentracing.Tag{Key: "rpc.method", Value: ri.To().Method()},
            )
            defer span.Finish()
            
            // 将 Span 注入到上下文
            ctx = opentracing.ContextWithSpan(ctx, span)
            
            // 执行调用
            err := next(ctx, req, resp)
            
            // 记录错误信息
            if err != nil {
                span.SetTag("error", true)
                span.LogFields(
                    opentracing.String("error.message", err.Error()),
                )
            }
            
            return err
        }
    }
}

import "github.com/cloudwego/kitex/pkg/klog"

// 配置结构化日志
func init() {
    klog.SetLogger(&StructuredLogger{})
    klog.SetLevel(klog.LevelInfo)
}

type StructuredLogger struct{}

func (l *StructuredLogger) Trace(v ...interface{}) {
    l.log("TRACE", v...)
}

func (l *StructuredLogger) Debug(v ...interface{}) {
    l.log("DEBUG", v...)
}

func (l *StructuredLogger) Info(v ...interface{}) {
    l.log("INFO", v...)
}

func (l *StructuredLogger) Notice(v ...interface{}) {
    l.log("NOTICE", v...)
}

func (l *StructuredLogger) Warn(v ...interface{}) {
    l.log("WARN", v...)
}

func (l *StructuredLogger) Error(v ...interface{}) {
    l.log("ERROR", v...)
}

func (l *StructuredLogger) Fatal(v ...interface{}) {
    l.log("FATAL", v...)
    os.Exit(1)
}

func (l *StructuredLogger) log(level string, v ...interface{}) {
    entry := map[string]interface{}{
        "timestamp": time.Now().UTC().Format(time.RFC3339),
        "level":     level,
        "message":   fmt.Sprint(v...),
    }
    
    // 添加调用信息
    if pc, file, line, ok := runtime.Caller(2); ok {
        entry["caller"] = fmt.Sprintf("%s:%d", filepath.Base(file), line)
        if fn := runtime.FuncForPC(pc); fn != nil {
            entry["function"] = fn.Name()
        }
    }
    
    jsonBytes, _ := json.Marshal(entry)
    fmt.Println(string(jsonBytes))
}

func RequestLoggingMiddleware(next endpoint.Endpoint) endpoint.Endpoint {
    return func(ctx context.Context, req, resp interface{}) error {
        start := time.Now()
        
        ri := rpcinfo.GetRPCInfo(ctx)
        
        // 生成请求 ID
        requestID := generateRequestID()
        ctx = context.WithValue(ctx, "request_id", requestID)
        
        // 记录请求开始
        klog.CtxInfof(ctx, "Request started: service=%s, method=%s, request_id=%s",
            ri.To().ServiceName(), ri.To().Method(), requestID)
        
        err := next(ctx, req, resp)
        
        duration := time.Since(start)
        
        // 记录请求结束
        if err != nil {
            klog.CtxErrorf(ctx, "Request failed: service=%s, method=%s, request_id=%s, duration=%v, error=%v",
                ri.To().ServiceName(), ri.To().Method(), requestID, duration, err)
        } else {
            klog.CtxInfof(ctx, "Request completed: service=%s, method=%s, request_id=%s, duration=%v",
                ri.To().ServiceName(), ri.To().Method(), requestID, duration)
        }
        
        return err
    }
}

func generateRequestID() string {
    return fmt.Sprintf("%d-%s", time.Now().UnixNano(), randomString(8))
}

# 多阶段构建
FROM golang:1.20-alpine AS builder

WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o main .

# 运行时镜像
FROM alpine:latest

# 安装 CA 证书
RUN apk --no-cache add ca-certificates tzdata

WORKDIR /root/

# 复制二进制文件
COPY --from=builder /app/main .

# 创建非 root 用户
RUN adduser -D -s /bin/sh appuser
USER appuser

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
    CMD ./main -health-check || exit 1

EXPOSE 8888

CMD ["./main"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kitex-service
  labels:
    app: kitex-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: kitex-service
  template:
    metadata:
      labels:
        app: kitex-service
    spec:
      containers:
      - name: kitex-service
        image: your-registry/kitex-service:latest
        ports:
        - containerPort: 8888
        - containerPort: 9090  # metrics
        env:
        - name: SERVICE_NAME
          value: "kitex-service"
        - name: LOG_LEVEL
          value: "info"
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 9090
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 9090
          initialDelaySeconds: 5
          periodSeconds: 5
        volumeMounts:
        - name: config
          mountPath: /etc/config
      volumes:
      - name: config
        configMap:
          name: kitex-service-config

---
apiVersion: v1
kind: Service
metadata:
  name: kitex-service
  labels:
    app: kitex-service
spec:
  selector:
    app: kitex-service
  ports:
  - name: rpc
    port: 8888
    targetPort: 8888
  - name: metrics
    port: 9090
    targetPort: 9090
  type: ClusterIP

import "github.com/fsnotify/fsnotify"

type ConfigManager struct {
    configPath string
    config     *Config
    callbacks  []func(*Config)
    mutex      sync.RWMutex
}

func NewConfigManager(configPath string) *ConfigManager {
    cm := &ConfigManager{
        configPath: configPath,
        callbacks:  make([]func(*Config), 0),
    }
    
    // 初始加载配置
    cm.loadConfig()
    
    // 启动配置监听
    go cm.watchConfig()
    
    return cm
}

func (cm *ConfigManager) loadConfig() {
    data, err := ioutil.ReadFile(cm.configPath)
    if err != nil {
        log.Printf("Failed to read config file: %v", err)
        return
    }
    
    var config Config
    if err := yaml.Unmarshal(data, &config); err != nil {
        log.Printf("Failed to parse config: %v", err)
        return
    }
    
    cm.mutex.Lock()
    cm.config = &config
    cm.mutex.Unlock()
    
    // 通知配置更新
    for _, callback := range cm.callbacks {
        callback(&config)
    }
}

func (cm *ConfigManager) watchConfig() {
    watcher, err := fsnotify.NewWatcher()
    if err != nil {
        log.Printf("Failed to create file watcher: %v", err)
        return
    }
    defer watcher.Close()
    
    err = watcher.Add(cm.configPath)
    if err != nil {
        log.Printf("Failed to watch config file: %v", err)
        return
    }
    
    for {
        select {
        case event := <-watcher.Events:
            if event.Op&fsnotify.Write == fsnotify.Write {
                log.Println("Config file modified, reloading...")
                time.Sleep(time.Millisecond * 100) // 防止重复触发
                cm.loadConfig()
            }
        case err := <-watcher.Errors:
            log.Printf("Config watcher error: %v", err)
        }
    }
}

func (cm *ConfigManager) OnConfigChange(callback func(*Config)) {
    cm.callbacks = append(cm.callbacks, callback)
}

func (cm *ConfigManager) GetConfig() *Config {
    cm.mutex.RLock()
    defer cm.mutex.RUnlock()
    return cm.config
}

import (
    "os"
    "os/signal"
    "syscall"
    "time"
)

type GracefulServer struct {
    server     server.Server
    httpServer *http.Server
    shutdown   chan struct{}
    done       chan struct{}
}

func NewGracefulServer(s server.Server) *GracefulServer {
    return &GracefulServer{
        server:   s,
        shutdown: make(chan struct{}),
        done:     make(chan struct{}),
    }
}

func (gs *GracefulServer) Run() error {
    // 启动 HTTP 服务器（用于健康检查和指标）
    gs.httpServer = &http.Server{
        Addr: ":9090",
        Handler: gs.createHTTPHandler(),
    }
    
    go func() {
        if err := gs.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Printf("HTTP server error: %v", err)
        }
    }()
    
    // 启动 RPC 服务器（非阻塞）
    go func() {
        defer close(gs.done)
        
        if err := gs.server.Run(); err != nil {
            log.Printf("RPC server error: %v", err)
        }
    }()
    
    // 等待关闭信号
    gs.waitForShutdown()
    
    // 执行优雅关闭
    return gs.gracefulShutdown()
}

func (gs *GracefulServer) waitForShutdown() {
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    
    select {
    case <-quit:
        log.Println("Received shutdown signal")
        close(gs.shutdown)
    case <-gs.done:
        log.Println("Server stopped")
    }
}

func (gs *GracefulServer) gracefulShutdown() error {
    log.Println("Starting graceful shutdown...")
    
    // 设置关闭超时
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
    defer cancel()
    
    // 关闭 HTTP 服务器
    if gs.httpServer != nil {
        if err := gs.httpServer.Shutdown(ctx); err != nil {
            log.Printf("HTTP server shutdown error: %v", err)
        }
    }
    
    // 停止接受新连接
    log.Println("Stopping RPC server...")
    
    // 等待现有请求完成
    done := make(chan error, 1)
    go func() {
        done <- gs.server.Stop()
    }()
    
    select {
    case err := <-done:
        if err != nil {
            log.Printf("RPC server stop error: %v", err)
            return err
        }
        log.Println("RPC server stopped gracefully")
        return nil
    case <-ctx.Done():
        log.Println("Shutdown timeout, forcing exit")
        return ctx.Err()
    }
}

func (gs *GracefulServer) createHTTPHandler() http.Handler {
    mux := http.NewServeMux()
    
    // 健康检查
    mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        w.WriteHeader(http.StatusOK)
        w.Write([]byte("OK"))
    })
    
    // 就绪检查
    mux.HandleFunc("/ready", func(w http.ResponseWriter, r *http.Request) {
        select {
        case <-gs.shutdown:
            w.WriteHeader(http.StatusServiceUnavailable)
            w.Write([]byte("Shutting down"))
        default:
            w.WriteHeader(http.StatusOK)
            w.Write([]byte("Ready"))
        }
    })
    
    // Prometheus 指标
    mux.Handle("/metrics", promhttp.Handler())
    
    return mux
}