Typha 架构与扩展性

type syncerPipeline struct {
    Type              syncproto.SyncerType
    Syncer            bapi.Syncer
    SyncerToValidator *calc.SyncerCallbacksDecoupler
    Validator         *calc.ValidationFilter
    ValidatorToCache  *calc.SyncerCallbacksDecoupler
    Cache             *snapcache.Cache
}

// typha/pkg/daemon/daemon.go
const (
    syncerTypeFelix      = "felix"      // Felix 需要的资源
    syncerTypeBGP        = "bgp"        // BGP/BIRD 需要的资源
    syncerTypeTunnelIPAlloc = "tunnel"  // 隧道 IP 分配
)

type Cache struct {
    // 当前数据快照
    kvs map[string]interface{}

    // 快照版本
    snapshotSequenceNumber uint64

    // 更新 breadcrumb（用于增量同步）
    breadcrumbs       []*Breadcrumb
    breadcrumbIndexes map[uint64]int

    // 客户端管理
    wakeUpCond *sync.Cond
}

type Breadcrumb struct {
    SequenceNumber uint64
    Timestamp      time.Time
    Updates        []syncproto.SerializedUpdate
}

func (c *Cache) RecordUpdate(update syncproto.SerializedUpdate) {
    c.lock.Lock()
    defer c.lock.Unlock()

    // 创建新的 breadcrumb
    c.snapshotSequenceNumber++
    bc := &Breadcrumb{
        SequenceNumber: c.snapshotSequenceNumber,
        Timestamp:      time.Now(),
        Updates:        []syncproto.SerializedUpdate{update},
    }

    // 添加到列表
    c.breadcrumbs = append(c.breadcrumbs, bc)
    c.breadcrumbIndexes[bc.SequenceNumber] = len(c.breadcrumbs) - 1

    // 清理过旧的 breadcrumb
    c.cleanOldBreadcrumbs()

    // 通知等待的客户端
    c.wakeUpCond.Broadcast()
}

service Sync {
    rpc Sync(stream SyncRequest) returns (stream SyncResponse);
}

message SyncRequest {
    oneof payload {
        SyncRequestStart start = 1;
        SyncRequestAck ack = 2;
    }
}

message SyncResponse {
    oneof payload {
        SyncResponseHeader header = 1;
        SyncResponseUpdate update = 2;
        SyncResponseFinished finished = 3;
        SyncResponsePing ping = 4;
    }
}

func (s *SyncServer) Sync(stream syncproto.Sync_SyncServer) error {
    // 1. 接收客户端请求
    req, err := stream.Recv()
    startReq := req.GetStart()

    // 2. 获取快照
    snapshot, seqNo := s.cache.GetSnapshot()

    // 3. 发送 Header
    stream.Send(&syncproto.SyncResponse{
        Payload: &syncproto.SyncResponse_Header{
            Header: &syncproto.SyncResponseHeader{
                SyncerType: startReq.SyncerType,
            },
        },
    })

    // 4. 发送快照数据
    for key, value := range snapshot {
        stream.Send(&syncproto.SyncResponse{
            Payload: &syncproto.SyncResponse_Update{
                Update: &syncproto.SerializedUpdate{
                    Key:   key,
                    Value: value,
                },
            },
        })
    }

    // 5. 发送快照完成标记
    stream.Send(&syncproto.SyncResponse{
        Payload: &syncproto.SyncResponse_Finished{
            Finished: &syncproto.SyncResponseFinished{},
        },
    })

    // 6. 持续发送增量更新
    for {
        updates := s.cache.WaitForUpdates(seqNo)
        for _, update := range updates {
            stream.Send(&syncproto.SyncResponse{
                Payload: &syncproto.SyncResponse_Update{
                    Update: &update,
                },
            })
        }
        seqNo = updates[len(updates)-1].SequenceNumber
    }
}

# Felix 配置
apiVersion: projectcalico.org/v3
kind: FelixConfiguration
metadata:
  name: default
spec:
  typhaAddr: typha-service:5473
  typhaK8sServiceName: calico-typha
  typhaK8sNamespace: kube-system

func connectToTypha(configParams *config.Config) (syncer bapi.Syncer, typhaClient *syncclient.SyncerClient) {
    // 发现 Typha 地址
    typhaAddr := discoverTyphaAddr(configParams)

    // 创建 Typha 客户端
    typhaClient = syncclient.New(
        typhaAddr,
        buildinfo.GitVersion,
        configParams.FelixHostname,
        "felix",  // syncer type
    )

    // 连接并开始同步
    err := typhaClient.Start(ctx)
    if err != nil {
        return nil, nil
    }

    return typhaClient, typhaClient
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: calico-typha
  namespace: kube-system
spec:
  replicas: 3
  selector:
    matchLabels:
      k8s-app: calico-typha
  template:
    metadata:
      labels:
        k8s-app: calico-typha
    spec:
      containers:
      - name: calico-typha
        image: calico/typha:v3.26.0
        ports:
        - containerPort: 5473
          name: calico-typha
        env:
        - name: TYPHA_LOGSEVERITYSCREEN
          value: "info"
        - name: TYPHA_DATASTORETYPE
          value: "kubernetes"
        livenessProbe:
          httpGet:
            path: /liveness
            port: 9098
        readinessProbe:
          httpGet:
            path: /readiness
            port: 9098
        resources:
          requests:
            cpu: 100m
            memory: 100Mi

func main() {
    // 1. 解析配置
    configParams := config.New()
    configParams.UpdateFrom(envVars)

    // 2. 创建 Daemon
    daemon := daemon.New()

    // 3. 启动
    daemon.Start(configParams)

    // 4. 等待信号
    daemon.WaitForShutdown()
}

func (d *Daemon) Start(configParams *config.Config) {
    // 1. 创建后端客户端
    backendClient := d.createBackendClient(configParams)

    // 2. 创建 Syncer Pipeline
    for _, syncerType := range []string{"felix", "bgp", "tunnel"} {
        pipeline := d.createSyncerPipeline(syncerType, backendClient)
        d.pipelines[syncerType] = pipeline
    }

    // 3. 启动 gRPC 服务
    d.server = syncserver.New(d.caches, configParams)
    d.server.Start()

    // 4. 启动所有 Syncer
    for _, pipeline := range d.pipelines {
        pipeline.Syncer.Start()
    }
}