从零开发 Virtual Kubelet Provider：深入理解节点注册与 Pod 生命周期

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer cancel()

    // 1. 创建 K8s 客户端
    kubeClient, err := nodeutil.ClientsetFromEnv(getEnv("KUBECONFIG", ""))
    if err != nil {
        log.Fatal(err)
    }

    // 2. 构建 NodeSpec（决定节点"样貌"）
    nodeSpec := corev1.Node{
        ObjectMeta: metav1.ObjectMeta{
            Name: nodeName,
            Labels: map[string]string{
                "kubernetes.io/role": "agent",
            },
        },
        Spec: corev1.NodeSpec{
            Taints: []corev1.Taint{{
                Key:    "virtual-kubelet.io/provider",
                Value:  "my-provider",
                Effect: corev1.TaintEffectNoSchedule,
            }},
        },
        Status: corev1.NodeStatus{
            Capacity: corev1.ResourceList{
                corev1.ResourceCPU:    resource.MustParse("8"),
                corev1.ResourceMemory: resource.MustParse("16Gi"),
                corev1.ResourcePods:   resource.MustParse("100"),
            },
            NodeInfo: corev1.NodeSystemInfo{
                OperatingSystem: "linux",
                Architecture:    "amd64",
            },
            Conditions: []corev1.NodeCondition{{
                Type:   corev1.NodeReady,
                Status: corev1.ConditionTrue,
            }},
        },
    }

    // 3. 启动虚拟节点
    n, err := nodeutil.NewNode(
        nodeName,
        // 工厂函数：接收 ProviderConfig，返回 Provider
        func(pc nodeutil.ProviderConfig) (nodeutil.Provider, node.NodeProvider, error) {
            return provider.NewMyProvider(cfg, pc)
        },
        nodeutil.WithNodeConfig(nodeutil.NodeConfig{
            NodeSpec:              nodeSpec,
            HTTPListenAddr:        ":10250",
            NumWorkers:            1,
        }),
        nodeutil.WithClient(kubeClient),
    )
    if err != nil {
        log.Fatal(err)
    }

    if err := n.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

// nodeutil.ProviderConfig 由 VK 框架构建并传入工厂函数
type ProviderConfig struct {
    Pods       corev1listers.PodLister       // Pod 缓存
    ConfigMaps corev1listers.ConfigMapLister // ConfigMap 缓存（用于解析 volume）
    Secrets    corev1listers.SecretLister    // Secret 缓存（用于解析 volume）
    Services   corev1listers.ServiceLister   // Service 缓存（用于 env vars）
    Node       *v1.Node                      // 节点对象的引用（可修改）
}

type Provider interface {
    // Pod 生命周期
    CreatePod(ctx context.Context, pod *corev1.Pod) error
    UpdatePod(ctx context.Context, pod *corev1.Pod) error
    DeletePod(ctx context.Context, pod *corev1.Pod) error
    GetPod(ctx context.Context, namespace, name string) (*corev1.Pod, error)
    GetPods(ctx context.Context) ([]*corev1.Pod, error)
    GetPodStatus(ctx context.Context, namespace, name string) (*corev1.PodStatus, error)

    // 容器交互
    GetContainerLogs(ctx context.Context, namespace, podName, containerName string, opts api.ContainerLogOpts) (io.ReadCloser, error)
    RunInContainer(ctx context.Context, namespace, podName, containerName string, cmd []string, attach api.AttachIO) error
    PortForward(ctx context.Context, namespace, pod string, port int32, stream io.ReadWriteCloser) error

    // 监控
    GetStatsSummary(context.Context) (*statsv1alpha1.Summary, error)
    GetMetricsResource(context.Context) ([]*dto.MetricFamily, error)
}

type ComposeProvider struct {
    configMapLister corev1listers.ConfigMapLister
    secretLister    corev1listers.SecretLister
    // ...
}

func (p *ComposeProvider) CreatePod(ctx context.Context, pod *corev1.Pod) error {
    podDir := p.podDir(pod)

    // 1. 转换 Pod Spec 为 docker-compose.yml
    composeBytes, err := ConvertToCompose(ctx, pod)

    // 2. 解析 Volume：从 Listers 读取 ConfigMap/Secret 数据
    extraFiles, err := p.resolveVolumes(ctx, pod)
    if err != nil {
        return err
    }

    // 3. 上传文件 + 启动容器
    p.syncClient.Upload(ctx, podDir, composeBytes, extraFiles)
    p.sshClient.Exec(ctx, fmt.Sprintf("cd %s && docker compose up -d", podDir))

    // 4. 缓存 Pod 状态为 Running
    cached := pod.DeepCopy()
    cached.Status = corev1.PodStatus{
        Phase: corev1.PodRunning,
        ContainerStatuses: buildContainerStatuses(pod.Spec.Containers),
    }
    p.pods.Store(key(pod), cached)

    // 5. 推送状态更新
    if p.notifyFunc != nil {
        p.notifyFunc(cached)
    }
    return nil
}

func (p *ComposeProvider) resolveVolumes(ctx context.Context, pod *corev1.Pod) ([]ExtraFile, error) {
    var files []ExtraFile

    for _, vol := range pod.Spec.Volumes {
        switch {
        case vol.ConfigMap != nil:
            // 通过 Lister 读取（带缓存，不直接调 API）
            cm, err := p.configMapLister.ConfigMaps(pod.Namespace).Get(vol.ConfigMap.Name)
            if err != nil {
                return nil, fmt.Errorf("fetch configmap %s/%s: %w", pod.Namespace, vol.ConfigMap.Name, err)
            }
            for key, data := range cm.Data {
                if !includeVolumeKey(vol.ConfigMap.Items, key) {
                    continue // 尊重 optional items 过滤
                }
                files = append(files, ExtraFile{
                    Path:    filepath.Join("volumes", vol.Name, key),
                    Content: []byte(data),
                })
            }

        case vol.Secret != nil:
            secret, err := p.secretLister.Secrets(pod.Namespace).Get(vol.Secret.SecretName)
            if err != nil {
                return nil, fmt.Errorf("fetch secret %s/%s: %w", pod.Namespace, vol.Secret.SecretName, err)
            }
            for key, data := range secret.Data {
                if !includeVolumeKey(vol.Secret.Items, key) {
                    continue
                }
                files = append(files, ExtraFile{
                    Path:    filepath.Join("volumes", vol.Name, key),
                    Content: data,
                })
            }

        case vol.EmptyDir != nil:
            // EmptyDir 只需要创建一个空目录
            files = append(files, ExtraFile{
                Path:    filepath.Join("volumes", vol.Name, ".gitkeep"),
                Content: []byte{},
            })
        }
    }
    return files, nil
}

// includeVolumeKey 检查 key 是否在 Items 白名单中
func includeVolumeKey(items []corev1.KeyToPath, key string) bool {
    if len(items) == 0 {
        return true // 没有指定 items，所有 key 都包含
    }
    for _, item := range items {
        if item.Key == key {
            return true
        }
    }
    return false
}

func (p *ComposeProvider) GetPodStatus(ctx context.Context, namespace, name string) (*corev1.PodStatus, error) {
    dir := p.podDir(namespace, name)
    out, err := p.sshClient.Exec(ctx, fmt.Sprintf("cd %s && docker compose ps --format json 2>&1", dir))
    if err != nil {
        // 容器不存在 → 返回 nil, nil
        return &corev1.PodStatus{Phase: corev1.PodFailed}, nil
    }

    // 解析后端状态，映射为 PodStatus
    status := parseContainerStates(out)
    return status, nil
}

func (p *ComposeProvider) DeletePod(ctx context.Context, pod *corev1.Pod) error {
    dir := p.podDir(pod)
    // docker compose down -v 会清理容器和网络，然后删除整个目录
    p.sshClient.Exec(ctx, fmt.Sprintf("cd %s && docker compose down -v 2>&1; rm -rf %s", dir, dir))

    p.pods.Delete(key(pod))
    return nil
}

services:
  nginx:
    image: nginx:latest
    volumes:
      - "./volumes/config:/etc/nginx/conf.d:ro"
      - "./volumes/secret:/etc/secret:ro"
      - "./volumes/cache:/tmp/cache"