ELK 日志收集 + 分布式追踪高级配置 + 全链路安全加固

2025-01-04T14:21:02+08:00 | 6分钟阅读 | 更新于 2025-01-04T14:21:02+08:00

@ MoonlightMaze

一、ELK 全链路日志收集（Filebeat+Elasticsearch+Kibana）

字节跳动内部日志系统核心架构：应用输出结构化 JSON 日志 → Filebeat 采集 → Kafka 缓冲 → Logstash 过滤 → Elasticsearch 存储 → Kibana 可视化，支持 PB 级日志秒级检索

1. 完整 ELK Stack 部署

# k8s/elk/namespace.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: elk
---
# k8s/elk/elasticsearch.yaml
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: elasticsearch
  namespace: elk
spec:
  replicas: 3
  selector:
    matchLabels:
      app: elasticsearch
  template:
    metadata:
      labels:
        app: elasticsearch
    spec:
      containers:
      - name: elasticsearch
        image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0
        ports:
        - containerPort: 9200
        - containerPort: 9300
        env:
        - name: discovery.type
          value: zen
        - name: ES_JAVA_OPTS
          value: "-Xms2g -Xmx2g"
        - name: ELASTIC_PASSWORD
          value: "elastic123"
        volumeMounts:
        - name: es-data
          mountPath: /usr/share/elasticsearch/data
  volumeClaimTemplates:
  - metadata:
      name: es-data
    spec:
      accessModes: ["ReadWriteOnce"]
      resources:
        requests:
          storage: 100Gi
---
apiVersion: v1
kind: Service
metadata:
  name: elasticsearch
  namespace: elk
spec:
  selector:
    app: elasticsearch
  ports:
  - port: 9200
    targetPort: 9200
---
# k8s/elk/kibana.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kibana
  namespace: elk
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kibana
  template:
    metadata:
      labels:
        app: kibana
    spec:
      containers:
      - name: kibana
        image: docker.elastic.co/kibana/kibana:8.13.0
        ports:
        - containerPort: 5601
        env:
        - name: ELASTICSEARCH_HOSTS
          value: "http://elasticsearch:9200"
        - name: ELASTICSEARCH_USERNAME
          value: "elastic"
        - name: ELASTICSEARCH_PASSWORD
          value: "elastic123"
---
apiVersion: v1
kind: Service
metadata:
  name: kibana
  namespace: elk
spec:
  type: LoadBalancer
  selector:
    app: kibana
  ports:
  - port: 5601
    targetPort: 5601
---
# k8s/elk/filebeat.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: filebeat
  namespace: elk
spec:
  selector:
    matchLabels:
      app: filebeat
  template:
    metadata:
      labels:
        app: filebeat
    spec:
      containers:
      - name: filebeat
        image: docker.elastic.co/beats/filebeat:8.13.0
        volumeMounts:
        - name: filebeat-config
          mountPath: /usr/share/filebeat/filebeat.yml
          subPath: filebeat.yml
        - name: varlog
          mountPath: /var/log
        - name: varlibdockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
      volumes:
      - name: filebeat-config
        configMap:
          name: filebeat-config
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: filebeat-config
  namespace: elk
data:
  filebeat.yml: |
    filebeat.inputs:
    - type: container
      paths:
        - /var/log/containers/*.log
      processors:
        - add_kubernetes_metadata:
            host: ${NODE_NAME}
            matchers:
            - logs_path:
                logs_path: "/var/log/containers/"

    output.elasticsearch:
      hosts: ["elasticsearch:9200"]
      username: "elastic"
      password: "elastic123"
      index: "mall-logs-%{+yyyy.MM.dd}"

    setup.ilm.enabled: false
    setup.template.name: "mall-logs"
    setup.template.pattern: "mall-logs-*"

2. 应用日志增强（集成 TraceID）

修改pkg/logger/logger.go，自动在日志中注入链路追踪 ID：

package logger

import (
    "context"
    "os"
    "go.opentelemetry.io/otel/trace"
    "go.uber.org/zap"
    "go.uber.org/zap/zapcore"
)

var Logger *zap.Logger

// Ctx 从上下文获取trace_id和span_id，返回带上下文的logger
func Ctx(ctx context.Context) *zap.Logger {
    span := trace.SpanFromContext(ctx)
    if span.SpanContext().IsValid() {
        return Logger.With(
            zap.String("trace_id", span.SpanContext().TraceID().String()),
            zap.String("span_id", span.SpanContext().SpanID().String()),
        )
    }
    return Logger
}

// 所有日志方法都支持上下文参数
func InfoCtx(ctx context.Context, msg string, fields ...zap.Field) {
    Ctx(ctx).Info(msg, fields...)
}

func ErrorCtx(ctx context.Context, msg string, fields ...zap.Field) {
    Ctx(ctx).Error(msg, fields...)
}

func FatalCtx(ctx context.Context, msg string, fields ...zap.Field) {
    Ctx(ctx).Fatal(msg, fields...)
}

3. 业务代码中使用增强日志

// 示例：internal/userservice/service/user_service.go
func (s *UserServiceImpl) GetUserInfo(ctx context.Context, req *user.GetUserInfoRequest) (*user.GetUserInfoResponse, error) {
    // 自动包含trace_id和span_id
    logger.InfoCtx(ctx, "GetUserInfo request", zap.Uint64("user_id", req.UserId))

    userModel, err := s.userDAO.GetByID(ctx, req.UserId)
    if err != nil {
        logger.ErrorCtx(ctx, "GetUserInfo failed", zap.Error(err))
        return nil, err
    }

    return &user.GetUserInfoResponse{
        UserId:   userModel.ID,
        Username: userModel.Username,
        Email:    userModel.Email,
        Phone:    userModel.Phone,
    }, nil
}

4. Kibana 日志查询最佳实践

按 TraceID 查询全链路日志：trace_id:"xxxxxx"
按服务查询：kubernetes.labels.app:"userservice"
按错误级别查询：level:"ERROR"
按时间范围查询：@timestamp:[now-1h TO now]

二、OpenTelemetry 分布式追踪高级配置

1. 自定义追踪属性与事件

// 示例：在订单创建接口添加自定义追踪信息
func (s *OrderServiceImpl) CreateOrder(ctx context.Context, req *order.CreateOrderRequest) (*order.CreateOrderResponse, error) {
    userID := ctx.Value("user_id").(uint64)
    
    // 获取当前span
    span := trace.SpanFromContext(ctx)
    
    // 添加自定义属性
    span.SetAttributes(
        attribute.Int64("user_id", int64(userID)),
        attribute.Int("item_count", len(req.Items)),
        attribute.Float64("total_amount", totalAmount),
    )
    
    // 添加事件
    span.AddEvent("开始创建订单")
    
    // ... 业务逻辑
    
    span.AddEvent("订单创建成功", trace.WithAttributes(
        attribute.Int64("order_id", int64(orderModel.ID)),
    ))
    
    return &order.CreateOrderResponse{OrderId: int64(orderModel.ID)}, nil
}

2. 智能采样策略（高并发必备）

// pkg/otel/otel.go
func InitProvider(serviceName, endpoint string) provider.OtelProvider {
    // 混合采样器：
    // 1. 所有错误请求100%采样
    // 2. 正常请求按10%概率采样
    sampler := sdktrace.NewParentBasedSampler(
        sdktrace.NewTraceIDRatioBased(0.1),
        sdktrace.WithRemoteParentSampled(sdktrace.AlwaysSample()),
        sdktrace.WithRemoteParentNotSampled(sdktrace.NeverSample()),
    )

    return provider.NewOpenTelemetryProvider(
        provider.WithServiceName(serviceName),
        provider.WithExportEndpoint(endpoint),
        provider.WithInsecure(),
        provider.WithSampler(sampler), // 应用采样策略
    )
}

3. Baggage 跨服务上下文透传

用于在整个调用链中传递用户 ID、请求 ID 等全局信息：

// 网关层设置Baggage
func JWTAuth() app.HandlerFunc {
    return func(ctx context.Context, c *app.RequestContext) {
        // ... 验证token
        
        // 将用户ID添加到Baggage，自动透传到所有下游服务
        baggageMember, _ := baggage.NewMember("user_id", strconv.FormatUint(claims.UserID, 10))
        baggage, _ := baggage.New(baggageMember)
        ctx = baggage.ContextWithBaggage(ctx, baggage)
        
        c.Next(ctx)
    }
}

// 下游服务获取Baggage
func (s *OrderServiceImpl) CreateOrder(ctx context.Context, req *order.CreateOrderRequest) (*order.CreateOrderResponse, error) {
    // 从Baggage获取用户ID
    b := baggage.FromContext(ctx)
    userIDStr := b.Member("user_id").Value()
    userID, _ := strconv.ParseUint(userIDStr, 10, 64)
    
    // ... 业务逻辑
}

4. 日志与链路追踪关联

在 Kibana 中点击日志中的trace_id，可以直接跳转到 Jaeger 查看完整链路，实现日志 - 链路一键跳转。

三、全链路安全加固

1. API 网关安全防护

1.1 HTTPS 强制加密

# k8s/istio/gateway.yaml
apiVersion: networking.istio.io/v1alpha3
kind: Gateway
metadata:
  name: mall-gateway
  namespace: cloudwego-mall
spec:
  selector:
    istio: ingressgateway
  servers:
  - port:
      number: 80
      name: http
      protocol: HTTP
    hosts:
    - "mall.example.com"
    tls:
      httpsRedirect: true # 强制HTTP跳转到HTTPS
  - port:
      number: 443
      name: https
      protocol: HTTPS
    hosts:
    - "mall.example.com"
    tls:
      mode: SIMPLE
      credentialName: mall-tls # TLS证书

1.2 CORS 跨域安全配置

// internal/api-gateway/middleware/cors.go
package middleware

import (
    "context"
    "github.com/cloudwego/hertz/pkg/app"
)

func CORS() app.HandlerFunc {
    return func(ctx context.Context, c *app.RequestContext) {
        c.Header("Access-Control-Allow-Origin", "https://mall.example.com")
        c.Header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
        c.Header("Access-Control-Allow-Headers", "Content-Type, Authorization")
        c.Header("Access-Control-Max-Age", "86400")
        c.Header("X-Content-Type-Options", "nosniff")
        c.Header("X-Frame-Options", "DENY")
        c.Header("X-XSS-Protection", "1; mode=block")

        if c.Request.Method() == "OPTIONS" {
            c.AbortWithStatus(204)
            return
        }

        c.Next(ctx)
    }
}

1.3 接口速率限制

// internal/api-gateway/middleware/ratelimit.go
package middleware

import (
    "context"
    "time"
    "github.com/cloudwego/hertz/pkg/app"
    "github.com/redis/go-redis/v9"
    "github.com/yourname/cloudwego-mall/pkg/redis"
    "github.com/yourname/cloudwego-mall/pkg/errors"
    "github.com/yourname/cloudwego-mall/internal/api-gateway/model"
)

func RateLimit() app.HandlerFunc {
    return func(ctx context.Context, c *app.RequestContext) {
        // 按IP限流，每分钟最多100次请求
        ip := c.ClientIP()
        key := "ratelimit:" + ip

        count, err := redis.Client.Incr(ctx, key).Result()
        if err != nil {
            model.Error(c, errors.New(errors.CodeInternalError, "限流服务异常"))
            c.Abort()
            return
        }

        if count == 1 {
            redis.Client.Expire(ctx, key, 1*time.Minute)
        }

        if count > 100 {
            model.Error(c, errors.New(errors.CodeRateLimitExceeded, "请求过于频繁，请稍后再试"))
            c.Abort()
            return
        }

        c.Next(ctx)
    }
}

2. 微服务间 mTLS 双向认证

使用 Istio 实现服务间自动双向 TLS 加密，防止中间人攻击

# k8s/istio/peer-authentication.yaml
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: cloudwego-mall
spec:
  mtls:
    mode: STRICT # 强制所有服务间通信使用mTLS

3. 敏感信息加密与配置安全

3.1 使用 Vault 管理敏感信息

不要在配置文件中硬编码数据库密码、API 密钥等敏感信息，使用 HashiCorp Vault 统一管理：

// pkg/vault/vault.go
package vault

import (
    "github.com/hashicorp/vault/api"
    "github.com/yourname/cloudwego-mall/pkg/config"
)

var Client *api.Client

func Init() error {
    cfg := api.DefaultConfig()
    cfg.Address = config.GlobalConfig.Vault.Address

    var err error
    Client, err = api.NewClient(cfg)
    if err != nil {
        return err
    }

    Client.SetToken(config.GlobalConfig.Vault.Token)
    return nil
}

// GetSecret 获取敏感信息
func GetSecret(path string) (map[string]interface{}, error) {
    secret, err := Client.Logical().Read(path)
    if err != nil {
        return nil, err
    }
    return secret.Data, nil
}

3.2 数据库密码加密示例

// pkg/mysql/mysql.go
func Init() error {
    // 从Vault获取数据库密码
    secret, err := vault.GetSecret("database/mysql")
    if err != nil {
        return err
    }
    password := secret["password"].(string)

    dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=True&loc=Local",
        config.GlobalConfig.Mysql.Username,
        password,
        config.GlobalConfig.Mysql.Address,
        config.GlobalConfig.Mysql.Database,
    )

    // ... 剩余代码
}

4. 容器与 Kubernetes 安全加固

4.1 非 root 用户运行容器

# Dockerfile 示例
FROM golang:1.22-alpine AS builder
WORKDIR /app
COPY . .
RUN go build -o userservice cmd/userservice/main.go

FROM alpine:3.19
RUN addgroup -S appgroup && adduser -S appuser -G appgroup
USER appuser # 使用非root用户
WORKDIR /app
COPY --from=builder /app/userservice .
EXPOSE 8081
CMD ["./userservice"]

4.2 Kubernetes SecurityContext

# k8s/userservice.yaml 新增
spec:
  securityContext:
    runAsNonRoot: true
    runAsUser: 1000
    runAsGroup: 1000
    fsGroup: 1000
  containers:
  - name: userservice
    image: your-docker-registry/userservice:latest
    securityContext:
      allowPrivilegeEscalation: false
      readOnlyRootFilesystem: true # 只读根文件系统
      capabilities:
        drop:
        - ALL # 删除所有Linux能力
    volumeMounts:
    - name: tmp
      mountPath: /tmp
  volumes:
  - name: tmp
    emptyDir: {}

4.3 网络策略限制 Pod 间通信

# k8s/security/network-policy.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: default-deny-all
  namespace: cloudwego-mall
spec:
  podSelector: {}
  policyTypes:
  - Ingress
  - Egress
---
# 允许API网关访问所有服务
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-api-gateway
  namespace: cloudwego-mall
spec:
  podSelector: {}
  ingress:
  - from:
    - podSelector:
        matchLabels:
          app: api-gateway
  policyTypes:
  - Ingress
---
# 允许订单服务访问商品和支付服务
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-order-service
  namespace: cloudwego-mall
spec:
  podSelector:
    matchLabels:
      app: orderservice
  egress:
  - to:
    - podSelector:
        matchLabels:
          app: productservice
    - podSelector:
        matchLabels:
          app: paymentservice
  policyTypes:
  - Egress

四、项目最终完整生产级能力全景

能力维度	已实现特性
业务能力	用户认证、购物车、商品管理、订单管理、支付集成
架构能力	微服务架构、API 网关、服务注册发现、配置中心
高并发能力	Redis 多级缓存、库存预扣减、异步消息队列、连接池优化
一致性能力	Seata 分布式事务、乐观锁防超卖、最终一致性补偿
可观测性	Zap 结构化日志、ELK 日志收集、OpenTelemetry 全链路追踪、Prometheus 监控、Grafana 可视化
高可用能力	Sentinel 熔断限流、Istio 流量治理、金丝雀灰度发布、K8s 多副本部署、健康检查
安全能力	HTTPS 加密、JWT 认证、mTLS 双向认证、CORS 防护、XSS/CSRF 防护、速率限制、敏感信息加密、容器安全、网络隔离
部署能力	Docker 容器化、K8s 编排、完整部署脚本、CI/CD 集成

五、面试加分亮点总结

微服务架构设计：服务拆分原则、API 网关设计、服务间通信方式
高并发处理：缓存设计、库存防超卖、异步解耦、性能优化
分布式系统：分布式事务、分布式锁、服务注册发现、链路追踪
可观测性：日志、监控、告警、追踪的最佳实践
生产级安全：HTTPS、mTLS、敏感信息保护、容器安全
工程化能力：项目结构、代码规范、CI/CD、K8s 部署

上一页 Kitex面试指导

下一页灰度发布 + Prometheus 监控告警 + Vegeta 压测