Kubernetes 扩展机制

支持的版本: Kubernetes 1.31, 1.32, 1.33 最后更新: February 21, 2026

概述

Kubernetes 提供了多种扩展机制，用于扩展和自定义其基础功能。本文档探讨 Kubernetes 的主要扩展机制，并说明实际使用场景和实现方法。

Custom Resource Definitions (CRD)

Custom Resource Definitions (CRDs) 是一种机制，允许你扩展 Kubernetes API 来定义自定义资源。

CRD 基本概念

使用 CRDs 可带来以下好处：

声明式 API: 你可以利用 Kubernetes 的声明式 API 模型。
kubectl 集成: 自定义资源可以像原生 Kubernetes 资源一样进行管理。
版本管理: 资源 schema 可以通过 API 版本管理逐步演进。
校验: 可以通过 OpenAPI v3 schema 执行资源校验。

CRD 创建示例

yaml

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: webapps.example.com
spec:
  group: example.com
  names:
    kind: WebApp
    listKind: WebAppList
    plural: webapps
    singular: webapp
    shortNames:
      - wa
  scope: Namespaced
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                replicas:
                  type: integer
                  minimum: 1
                image:
                  type: string
                port:
                  type: integer
              required: ["image"]
            status:
              type: object
              properties:
                availableReplicas:
                  type: integer
                conditions:
                  type: array
                  items:
                    type: object
                    properties:
                      type:
                        type: string
                      status:
                        type: string
                      lastTransitionTime:
                        type: string
      additionalPrinterColumns:
        - name: Replicas
          type: integer
          jsonPath: .spec.replicas
        - name: Image
          type: string
          jsonPath: .spec.image
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp
      subresources:
        status: {}
        scale:
          specReplicasPath: .spec.replicas
          statusReplicasPath: .status.availableReplicas

自定义资源实例创建

yaml

apiVersion: example.com/v1
kind: WebApp
metadata:
  name: my-webapp
spec:
  replicas: 3
  image: nginx:1.21
  port: 80

Custom Controllers

仅有自定义资源无法实现实际行为。Custom controllers 会监视自定义资源的状态，并执行操作以达到期望状态。

Controller 模式

Kubernetes controllers 遵循以下模式：

观察: 观察资源的当前状态。
分析: 分析当前状态与期望状态之间的差异。
执行: 采取行动以达到期望状态。

Controller 实现方法

1. 使用 client-go

package main

import (
    "context"
    "fmt"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/homedir"
    "path/filepath"
)

func main() {
    // Load kubeconfig
    kubeconfig := filepath.Join(homedir.HomeDir(), ".kube", "config")
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        panic(err)
    }

    // Create Kubernetes client
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err)
    }

    // Get pod list
    pods, err := clientset.CoreV1().Pods("default").List(context.TODO(), metav1.ListOptions{})
    if err != nil {
        panic(err)
    }

    fmt.Printf("There are %d pods in the default namespace\n", len(pods.Items))

    // Watch pods
    watch, err := clientset.CoreV1().Pods("default").Watch(context.TODO(), metav1.ListOptions{})
    if err != nil {
        panic(err)
    }

    // Handle events
    for event := range watch.ResultChan() {
        fmt.Printf("Event: %s\n", event.Type)
    }
}

2. 使用 controller-runtime

package main

import (
    "context"

    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"

    examplev1 "example.com/api/v1"
)

// WebAppReconciler reconciles a WebApp object
type WebAppReconciler struct {
    client.Client
    Scheme *runtime.Scheme
}

func (r *WebAppReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)

    // Get WebApp instance
    var webapp examplev1.WebApp
    if err := r.Get(ctx, req.NamespacedName, &webapp); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // Create or update Deployment
    deployment := &appsv1.Deployment{}
    err := r.Get(ctx, client.ObjectKey{Namespace: webapp.Namespace, Name: webapp.Name}, deployment)
    if client.IgnoreNotFound(err) != nil {
        return ctrl.Result{}, err
    }

    if err != nil {
        // Create Deployment if it doesn't exist
        deployment = &appsv1.Deployment{
            ObjectMeta: metav1.ObjectMeta{
                Name:      webapp.Name,
                Namespace: webapp.Namespace,
            },
        }

        if err := ctrl.SetControllerReference(&webapp, deployment, r.Scheme); err != nil {
            return ctrl.Result{}, err
        }

        // Set Deployment spec
        replicas := int32(webapp.Spec.Replicas)
        deployment.Spec.Replicas = &replicas
        deployment.Spec.Selector = &metav1.LabelSelector{
            MatchLabels: map[string]string{"app": webapp.Name},
        }
        deployment.Spec.Template.ObjectMeta.Labels = map[string]string{"app": webapp.Name}
        deployment.Spec.Template.Spec.Containers = []corev1.Container{
            {
                Name:  "webapp",
                Image: webapp.Spec.Image,
                Ports: []corev1.ContainerPort{
                    {
                        ContainerPort: int32(webapp.Spec.Port),
                    },
                },
            },
        }

        if err := r.Create(ctx, deployment); err != nil {
            log.Error(err, "Failed to create Deployment")
            return ctrl.Result{}, err
        }

        log.Info("Created Deployment", "Deployment.Namespace", deployment.Namespace, "Deployment.Name", deployment.Name)
    } else {
        // Update Deployment if it exists
        replicas := int32(webapp.Spec.Replicas)
        deployment.Spec.Replicas = &replicas
        deployment.Spec.Template.Spec.Containers[0].Image = webapp.Spec.Image

        if err := r.Update(ctx, deployment); err != nil {
            log.Error(err, "Failed to update Deployment")
            return ctrl.Result{}, err
        }

        log.Info("Updated Deployment", "Deployment.Namespace", deployment.Namespace, "Deployment.Name", deployment.Name)
    }

    // Update status
    webapp.Status.AvailableReplicas = int(deployment.Status.AvailableReplicas)
    if err := r.Status().Update(ctx, &webapp); err != nil {
        log.Error(err, "Failed to update WebApp status")
        return ctrl.Result{}, err
    }

    return ctrl.Result{}, nil
}

func (r *WebAppReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&examplev1.WebApp{}).
        Owns(&appsv1.Deployment{}).
        Complete(r)
}

func main() {
    scheme := runtime.NewScheme()
    _ = examplev1.AddToScheme(scheme)
    _ = appsv1.AddToScheme(scheme)
    _ = corev1.AddToScheme(scheme)

    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme: scheme,
    })
    if err != nil {
        panic(err)
    }

    if err := (&WebAppReconciler{
        Client: mgr.GetClient(),
        Scheme: mgr.GetScheme(),
    }).SetupWithManager(mgr); err != nil {
        panic(err)
    }

    if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
        panic(err)
    }
}

Operator 模式

Operator 是一种将 CRDs 与 controllers 结合起来的模式，用于自动化特定应用的运维知识。

Operators 的关键特性：

领域知识自动化: 将应用领域知识以代码形式实现。
声明式管理: 用户声明期望状态，Operator 执行操作以达到该状态。
自愈: 检测故障状况并自动恢复。
升级管理: 安全地处理应用升级。

Operator 示例：

Prometheus Operator: 管理 Prometheus 监控栈。
Elasticsearch Operator: 管理 Elasticsearch 集群。
PostgreSQL Operator: 管理 PostgreSQL 数据库。

Operator SDK

Operator SDK 是一种简化 Operator 开发的工具。

Operator 创建：

bash

# Install Operator SDK
curl -LO https://github.com/operator-framework/operator-sdk/releases/download/v1.25.0/operator-sdk_linux_amd64
chmod +x operator-sdk_linux_amd64
sudo mv operator-sdk_linux_amd64 /usr/local/bin/operator-sdk

# Create Operator project
operator-sdk init --domain example.com --repo github.com/example/webapp-operator

# Create API
operator-sdk create api --group apps --version v1 --kind WebApp --resource --controller

# Generate CRD
make manifests

# Build and deploy Operator
make docker-build docker-push
make deploy

API Server 扩展

API server 扩展提供了扩展 Kubernetes API server 功能的方式。

1. Aggregation Layer

Aggregation layer 是一种机制，允许将额外的 APIs 注册到 Kubernetes API server。

关键特性：

API 扩展: 可以向现有 API server 添加新的 APIs。
集群内执行: 扩展 API servers 在集群内运行。
认证委托: 主 API server 处理认证，并委托给扩展 API server。

APIService 示例：

yaml

apiVersion: apiregistration.k8s.io/v1
kind: APIService
metadata:
  name: v1.metrics.k8s.io
spec:
  service:
    name: metrics-server
    namespace: kube-system
  group: metrics.k8s.io
  version: v1
  insecureSkipTLSVerify: true
  groupPriorityMinimum: 100
  versionPriority: 100

扩展 API Server 实现：

package main

import (
    "fmt"
    "net/http"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/runtime/schema"
    "k8s.io/apimachinery/pkg/runtime/serializer"
    "k8s.io/apiserver/pkg/registry/rest"
    genericapiserver "k8s.io/apiserver/pkg/server"
    genericoptions "k8s.io/apiserver/pkg/server/options"
)

var (
    scheme = runtime.NewScheme()
    codecs = serializer.NewCodecFactory(scheme)
)

func main() {
    // Create server options
    serverOptions := genericoptions.NewRecommendedOptions("/tmp/apiserver.etcd", codecs.LegacyCodec())

    // Configure server
    config := genericapiserver.NewRecommendedConfig(codecs)
    if err := serverOptions.ApplyTo(config); err != nil {
        panic(err)
    }

    // Create API server
    server, err := config.Complete().New("example-apiserver", genericapiserver.NewEmptyDelegate())
    if err != nil {
        panic(err)
    }

    // Install API group
    apiGroupInfo := genericapiserver.NewDefaultAPIGroupInfo("example.com", scheme, metav1.ParameterCodec, codecs)
    server.InstallAPIGroup(&apiGroupInfo)

    // Run server
    server.PrepareRun().Run(make(chan struct{}))
}

2. Webhooks

Webhooks 是一种机制，当特定事件发生时，Kubernetes API server 会调用外部服务来执行额外处理。

Admission Webhooks

Admission webhooks 可以在 API 请求持久化到存储之前对其进行校验或修改。

主要类型：

MutatingAdmissionWebhook: 可以修改请求。
ValidatingAdmissionWebhook: 仅校验请求，不进行修改。

Webhook 配置示例：

yaml

apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  name: example-webhook
webhooks:
- name: example.webhook.com
  clientConfig:
    url: https://example.webhook.com/mutate
    caBundle: <BASE64_ENCODED_CA_CERT>
  rules:
  - apiGroups: [""]
    apiVersions: ["v1"]
    resources: ["pods"]
    operations: ["CREATE", "UPDATE"]
    scope: "Namespaced"
  admissionReviewVersions: ["v1"]
  sideEffects: None
  timeoutSeconds: 5

Webhook Server 实现：

package main

import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net/http"

    admissionv1 "k8s.io/api/admission/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/runtime/serializer"
)

var (
    runtimeScheme = runtime.NewScheme()
    codecs        = serializer.NewCodecFactory(runtimeScheme)
    deserializer  = codecs.UniversalDeserializer()
)

func handleMutate(w http.ResponseWriter, r *http.Request) {
    // Read request body
    body, err := ioutil.ReadAll(r.Body)
    if err != nil {
        http.Error(w, fmt.Sprintf("Failed to read body: %v", err), http.StatusBadRequest)
        return
    }

    // Convert to AdmissionReview object
    var admissionReview admissionv1.AdmissionReview
    if _, _, err := deserializer.Decode(body, nil, &admissionReview); err != nil {
        http.Error(w, fmt.Sprintf("Failed to decode body: %v", err), http.StatusBadRequest)
        return
    }

    // Extract pod object
    var pod corev1.Pod
    if err := json.Unmarshal(admissionReview.Request.Object.Raw, &pod); err != nil {
        http.Error(w, fmt.Sprintf("Failed to unmarshal pod: %v", err), http.StatusBadRequest)
        return
    }

    // Create patch
    patch := []map[string]interface{}{
        {
            "op":    "add",
            "path":  "/metadata/labels/example.com~1injected",
            "value": "true",
        },
    }

    patchBytes, err := json.Marshal(patch)
    if err != nil {
        http.Error(w, fmt.Sprintf("Failed to marshal patch: %v", err), http.StatusInternalServerError)
        return
    }

    // Create response
    admissionResponse := admissionv1.AdmissionResponse{
        UID:     admissionReview.Request.UID,
        Allowed: true,
        Patch:   patchBytes,
        PatchType: func() *admissionv1.PatchType {
            pt := admissionv1.PatchTypeJSONPatch
            return &pt
        }(),
    }

    admissionReview.Response = &admissionResponse

    // Send response
    resp, err := json.Marshal(admissionReview)
    if err != nil {
        http.Error(w, fmt.Sprintf("Failed to marshal response: %v", err), http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "application/json")
    w.Write(resp)
}

func main() {
    http.HandleFunc("/mutate", handleMutate)
    http.ListenAndServeTLS(":8443", "tls.crt", "tls.key", nil)
}

Scheduler 扩展

Kubernetes scheduler 决定将 pods 放置在哪个 node 上。Scheduler 扩展允许你自定义这个决策过程。

1. Scheduler Framework

Scheduler framework 提供了一种扩展机制，可在调度流水线的不同阶段添加 plugins。

关键扩展点：

Filter: 过滤掉 pod 无法运行的 nodes。
Score: 为合适的 nodes 分配分数。
Bind: 将 pod 绑定到 node。
Reserve/Unreserve: 保留或释放 node 资源。
Permit: 允许、拒绝或延迟 pod 调度。

Scheduler 配置示例：

yaml

apiVersion: kubescheduler.config.k8s.io/v1
kind: KubeSchedulerConfiguration
profiles:
- schedulerName: custom-scheduler
  plugins:
    filter:
      enabled:
      - name: NodeResourcesFit
      - name: NodeName
      - name: CustomFilter
    score:
      enabled:
      - name: NodeResourcesBalancedAllocation
        weight: 1
      - name: CustomScore
        weight: 5
  pluginConfig:
  - name: CustomFilter
    args:
      foo: bar

Scheduler Plugin 实现：

package main

import (
    "context"

    v1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/kubernetes/pkg/scheduler/framework"
)

// CustomPlugin is a scheduler framework plugin.
type CustomPlugin struct {
    handle framework.Handle
}

var _ framework.FilterPlugin = &CustomPlugin{}
var _ framework.ScorePlugin = &CustomPlugin{}

// Name returns the name of the plugin.
func (p *CustomPlugin) Name() string {
    return "CustomPlugin"
}

// Filter filters nodes where the pod can run.
func (p *CustomPlugin) Filter(ctx context.Context, state *framework.CycleState, pod *v1.Pod, node *framework.NodeInfo) *framework.Status {
    // Implement filtering logic
    return framework.NewStatus(framework.Success, "")
}

// Score assigns scores to nodes.
func (p *CustomPlugin) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    // Implement score calculation logic
    return 100, framework.NewStatus(framework.Success, "")
}

// ScoreExtensions provides score normalization methods.
func (p *CustomPlugin) ScoreExtensions() framework.ScoreExtensions {
    return p
}

// NormalizeScore normalizes scores.
func (p *CustomPlugin) NormalizeScore(ctx context.Context, state *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
    // Implement score normalization logic
    return framework.NewStatus(framework.Success, "")
}

// New creates a new instance of the plugin.
func New(configuration runtime.Object, f framework.Handle) (framework.Plugin, error) {
    return &CustomPlugin{handle: f}, nil
}

2. Scheduler Extender

Scheduler extender 是一个外部进程，可以通过 HTTP webhooks 影响调度决策。

Scheduler 配置示例：

yaml

apiVersion: kubescheduler.config.k8s.io/v1
kind: KubeSchedulerConfiguration
extenders:
- urlPrefix: "http://extender.example.com"
  filterVerb: "filter"
  prioritizeVerb: "prioritize"
  weight: 5
  bindVerb: "bind"
  enableHTTPS: false

Extender Server 实现：

package main

import (
    "encoding/json"
    "net/http"

    v1 "k8s.io/api/core/v1"
    extender "k8s.io/kube-scheduler/extender/v1"
)

func filter(w http.ResponseWriter, r *http.Request) {
    var extenderArgs extender.ExtenderArgs
    var extenderFilterResult extender.ExtenderFilterResult

    // Parse request
    if err := json.NewDecoder(r.Body).Decode(&extenderArgs); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    // Implement filtering logic
    filteredNodes := make([]v1.Node, 0, len(extenderArgs.Nodes.Items))
    failedNodes := make(map[string]string)

    for _, node := range extenderArgs.Nodes.Items {
        // Node filtering logic
        if /* check if node is suitable */ true {
            filteredNodes = append(filteredNodes, node)
        } else {
            failedNodes[node.Name] = "Node is not suitable"
        }
    }

    // Create result
    extenderFilterResult = extender.ExtenderFilterResult{
        Nodes: &v1.NodeList{
            Items: filteredNodes,
        },
        FailedNodes: failedNodes,
        Error:       "",
    }

    // Send response
    if err := json.NewEncoder(w).Encode(extenderFilterResult); err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
}

func prioritize(w http.ResponseWriter, r *http.Request) {
    var extenderArgs extender.ExtenderArgs
    var hostPriorityList extender.HostPriorityList

    // Parse request
    if err := json.NewDecoder(r.Body).Decode(&extenderArgs); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    // Implement priority logic
    hostPriorityList = make(extender.HostPriorityList, 0, len(extenderArgs.Nodes.Items))

    for _, node := range extenderArgs.Nodes.Items {
        // Node score calculation logic
        score := int64(0)
        hostPriorityList = append(hostPriorityList, extender.HostPriority{
            Host:  node.Name,
            Score: score,
        })
    }

    // Send response
    if err := json.NewEncoder(w).Encode(hostPriorityList); err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
}

func main() {
    http.HandleFunc("/filter", filter)
    http.HandleFunc("/prioritize", prioritize)
    http.ListenAndServe(":8888", nil)
}

Network Plugins

Kubernetes 通过 Container Network Interface (CNI) 支持 network plugins。

CNI (Container Network Interface)

CNI 定义了 container runtimes 与 network plugins 之间的标准接口。

主要 CNI Plugins：

Calico: 提供基于 BGP 的网络和 network policies。
Cilium: 提供基于 eBPF 的网络、安全和可观测性。
Flannel: 提供简单的 overlay networking。
Weave Net: 提供多主机 container networking。

CNI 配置示例：

json

{
  "cniVersion": "0.4.0",
  "name": "mynet",
  "type": "bridge",
  "bridge": "cni0",
  "isGateway": true,
  "ipMasq": true,
  "ipam": {
    "type": "host-local",
    "subnet": "10.244.0.0/16",
    "routes": [
      { "dst": "0.0.0.0/0" }
    ]
  }
}

CNI Plugin 实现：

package main

import (
    "encoding/json"
    "net"

    "github.com/containernetworking/cni/pkg/skel"
    "github.com/containernetworking/cni/pkg/types"
    current "github.com/containernetworking/cni/pkg/types/100"
    "github.com/containernetworking/cni/pkg/version"
)

func cmdAdd(args *skel.CmdArgs) error {
    // Parse configuration
    conf := &types.NetConf{}
    if err := json.Unmarshal(args.StdinData, conf); err != nil {
        return err
    }

    // Implement network setup logic
    // ...

    // Return result
    result := &current.Result{
        CNIVersion: conf.CNIVersion,
        IPs: []*current.IPConfig{
            {
                Address: net.IPNet{
                    IP:   net.ParseIP("10.244.0.2"),
                    Mask: net.CIDRMask(24, 32),
                },
                Gateway: net.ParseIP("10.244.0.1"),
            },
        },
    }

    return types.PrintResult(result, conf.CNIVersion)
}

func cmdDel(args *skel.CmdArgs) error {
    // Implement network teardown logic
    // ...

    return nil
}

func cmdCheck(args *skel.CmdArgs) error {
    // Implement network status check logic
    // ...

    return nil
}

func main() {
    skel.PluginMain(cmdAdd, cmdCheck, cmdDel, version.All, "My CNI Plugin v0.1.0")
}

Storage Plugins

Kubernetes 通过 Container Storage Interface (CSI) 支持 storage plugins。

CSI (Container Storage Interface)

CSI 定义了 container orchestration systems 与 storage providers 之间的标准接口。

主要 CSI Plugins：

AWS EBS CSI Driver: 提供 Amazon EBS volumes。
GCE PD CSI Driver: 提供 Google Compute Engine persistent disks。
Azure Disk CSI Driver: 提供 Azure disks。
Ceph CSI: 提供 Ceph RBD 和 CephFS。

CSI Driver 部署示例：

yaml

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: csi-sc
provisioner: example.csi.driver
parameters:
  type: ssd
  fsType: ext4
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: csi-pvc
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 5Gi
  storageClassName: csi-sc

CSI Driver 实现：

CSI drivers 必须实现三个主要服务：

Identity Service: Driver 标识和能力发现
Controller Service: Volume 供给和管理
Node Service: 在 nodes 上挂载和卸载 volumes

package main

import (
    "context"
    "net"
    "os"
    "os/signal"
    "syscall"

    "github.com/container-storage-interface/spec/lib/go/csi"
    "google.golang.org/grpc"
)

type driver struct {
    csi.UnimplementedIdentityServer
    csi.UnimplementedControllerServer
    csi.UnimplementedNodeServer
}

// Identity Service
func (d *driver) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
    return &csi.GetPluginInfoResponse{
        Name:          "example.csi.driver",
        VendorVersion: "v0.1.0",
    }, nil
}

// Controller Service
func (d *driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
    // Implement volume creation logic
    // ...

    return &csi.CreateVolumeResponse{
        Volume: &csi.Volume{
            VolumeId:      "vol-123",
            CapacityBytes: req.GetCapacityRange().GetRequiredBytes(),
            VolumeContext: req.GetParameters(),
        },
    }, nil
}

// Node Service
func (d *driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
    // Implement volume mount logic
    // ...

    return &csi.NodePublishVolumeResponse{}, nil
}

func main() {
    // Set up gRPC server
    server := grpc.NewServer()

    // Create CSI driver instance
    d := &driver{}

    // Register CSI services
    csi.RegisterIdentityServer(server, d)
    csi.RegisterControllerServer(server, d)
    csi.RegisterNodeServer(server, d)

    // Create socket listener
    listener, err := net.Listen("unix", "/csi/csi.sock")
    if err != nil {
        panic(err)
    }

    // Start server
    go server.Serve(listener)

    // Handle termination signals
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    <-sigCh

    server.GracefulStop()
}

结论

Kubernetes 扩展机制提供了强大的方式，可针对不同使用场景和需求自定义 Kubernetes。你可以使用 CRDs 和 custom controllers 定义新的 APIs，使用 admission webhooks 校验或修改 API 请求，使用 scheduler extensions 自定义 pod 放置决策，并通过 CNI 和 CSI 集成网络与存储解决方案。

通过利用这些扩展机制，你可以根据组织的特定需求定制 Kubernetes，自动化复杂的应用管理，并最大化云原生生态系统的收益。

端到端可观测性实验

ArgoCD 详解测验

Istio 详解测验

Linkerd 详解测验

Cilium 服务网格详解测验

Cilium 深入解析

Calico 深入解析

指标

日志

追踪

告警

可观测性实验

ArgoCD

流量管理

安全性

可观测性

弹性

高级

对比指南

指标

日志

追踪

告警

Kubernetes 扩展机制 ​

概述 ​

Custom Resource Definitions (CRD) ​

CRD 基本概念 ​

CRD 创建示例 ​

自定义资源实例创建 ​

Custom Controllers ​

Controller 模式 ​

Controller 实现方法 ​

1. 使用 client-go ​

2. 使用 controller-runtime ​

Operator 模式 ​

Operator SDK ​

API Server 扩展 ​

1. Aggregation Layer ​

2. Webhooks ​

Admission Webhooks ​

Scheduler 扩展 ​

1. Scheduler Framework ​

2. Scheduler Extender ​

Network Plugins ​

CNI (Container Network Interface) ​

Storage Plugins ​

CSI (Container Storage Interface) ​

结论 ​

Kubernetes 扩展机制

概述

Custom Resource Definitions (CRD)

CRD 基本概念

CRD 创建示例

自定义资源实例创建

Custom Controllers

Controller 模式

Controller 实现方法

1. 使用 client-go

2. 使用 controller-runtime

Operator 模式

Operator SDK

API Server 扩展

1. Aggregation Layer

2. Webhooks

Admission Webhooks

Scheduler 扩展

1. Scheduler Framework

2. Scheduler Extender

Network Plugins

CNI (Container Network Interface)

Storage Plugins

CSI (Container Storage Interface)

结论