go - 当使用 k8s.io/client-go 库更改 kubernetes 部署时，获得通知的最佳方式是什么？

Question

上下文

我正在编写一个使用 k8s.io/client-go 的脚本库 ( godocs here ) 来操作部署。特别是，我想为集群中的每个 Deployment 添加一个标签选择器。部署标签选择器是 immutable .所以我的方法是:

1. 为每个 Deployment 创建一个副本，唯一的区别是名称以“-temp”为后缀。这是为了最大限度地减少现有部署的停机时间。
2. 删除原来的部署。
3. 重新创建原始 Deployment，唯一的区别是多了一个标签选择器。
4. 删除临时部署。

我不能只使用 client-go 库按顺序执行步骤 1-4，因为我只想在 API 服务器认为上一步已完成时继续下一步。例如，在 API 服务器说原始 Deployments 已被删除之前，我不想执行第 3 步。否则，我会得到同名 Deployment 已存在的错误。

问题

使用 client-go 库检测 Deployment 何时创建和删除以及附加回调函数的最佳方法是什么？我遇到了以下软件包。

• watch
• informers
• cache/informers

但我不确定它们之间有什么区别以及使用哪一个。

我阅读了 watch here 的示例和 informer here .这是 two related所以问题。

更新

好像watch提供了一种较低级别的方法来监视资源的更改并接收有关更改的事件。好像在使用 SharedInformerFactory创建一个 SharedInformer 是可行的方法。

目前为止

import (
    "encoding/json"
    "errors"
    "flag"
    "fmt"
    "io/ioutil"
    "k8s.io/api/apps/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/informers"
    "k8s.io/client-go/kubernetes"
    typedv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
    "k8s.io/client-go/tools/cache"
    "path/filepath"
    "strings"

    // We need this import to load the GCP auth plugin which is required to authenticate against GKE clusters.
    _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
    "k8s.io/client-go/tools/clientcmd"
    "log"
    "os"
)

func main() {

...

    factory := informers.NewSharedInformerFactory(kubeclient, 0)
    informer := factory.Apps().V1().Deployments().Informer()
    stopper := make(chan struct{})
    defer close(stopper)
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            d := obj.(v1.Deployment)
            fmt.Printf("Created deployment in namespace %s, name %s.\n", d.GetNamespace(), d.GetName())

            if _, ok := d.GetLabels()[tempLabelKey]; ok {
                fmt.Printf("Detected temporary deployment created in namespace %s, name %s.\n", d.GetNamespace(), d.GetName())
                deploymentToDelete := strings.Replace(d.GetName(), tempSuffix, "", -1)
                fmt.Printf("Now deleting previous deployment in namespace %s, name %s.\n", d.GetNamespace(), deploymentToDelete)
                deleteDeployment(deploymentToDelete, d.GetNamespace(), kubeclient)
            }
        },
        DeleteFunc: func(obj interface{}) {
            d := obj.(v1.Deployment)
            fmt.Printf("Deleted deployment in namespace %s, name %s.\n", d.GetNamespace(), d.GetName())

            if _, ok := d.GetLabels()[stageLabelKey]; !ok {
                fmt.Printf("Detected deployment without stage label was deleted in namespace %s, name %s.\n", d.GetNamespace(), d.GetName())
                fmt.Printf("Now creating normal deployment with stage label in namespace %s, name %s.\n", d.GetNamespace(), d.GetName())
                deployment := createDeploymentWithNewLabel(stageLabelKey, "production", d)
                createDeploymentsOnApi(deployment, kubeclient)
            }
        },
    })
    informer.Run(stopper)
}

Answer 1

我最终使用了 SharedInformer .

这些资源很有帮助。

• https://jvns.ca/blog/2017/07/27/how-does-the-kubernetes-scheduler-work/
• https://github.com/kubernetes/community/blob/8decfe4/contributors/devel/controllers.md

.

package main

import (
    "encoding/json"
    "errors"
    "flag"
    "fmt"
    "io/ioutil"
    "k8s.io/api/apps/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"
    "k8s.io/client-go/informers"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/cache"
    "path/filepath"
    "strings"
    // We need this import to load the GCP auth plugin which is required to authenticate against GKE clusters.
    _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
    "k8s.io/client-go/tools/clientcmd"
    "log"
    "os"
)

const manifestsDir = "manifests"

// Use an empty string to run on all namespaces
const namespace = ""
const newLabelKey = "new-label-to-add"
const tempLabelKey = "temporary"
const tempSuffix = "-temp"
const componentLabelKey = "component"

func main() {
    var kubeconfig *string
    if home := homeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
    }
    flag.Parse()

    // use the current context in kubeconfig
    // TODO (dxia) How can I specify a masterUrl or even better a kubectl context?
    cfg, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    exitOnErr(err)

    kubeclient, err := kubernetes.NewForConfig(cfg)
    exitOnErr(err)

    fmt.Printf("Getting deployments with '%s' label.\n", componentLabelKey)
    deployments, err := kubeclient.AppsV1().Deployments(namespace).List(metav1.ListOptions{
        LabelSelector: componentLabelKey,
    })
    fmt.Printf("Got %d deployments.\n", len(deployments.Items))
    exitOnErr(err)

    deployments = processDeployments(deployments)
    fmt.Println("Saving deployment manifests to disk as backup.")
    err = saveDeployments(deployments)
    exitOnErr(err)

    tempDeployments := appendToDeploymentName(deployments, tempSuffix)
    tempDeployments = createDeploymentsWithNewLabel(tempLabelKey, "true", tempDeployments)

    factory := informers.NewSharedInformerFactory(kubeclient, 0)
    informer := factory.Apps().V1().Deployments().Informer()
    stopper := make(chan struct{})
    defer close(stopper)
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            d := obj.(*v1.Deployment)
            labels := d.GetLabels()

            if _, ok := labels[tempLabelKey]; ok {
                labelsStr := joinLabelKeyVals(labels)
                fmt.Printf("2: Temporary deployment created in namespace %s, name %s, labels '%s'.\n", d.GetNamespace(), d.GetName(), labelsStr)
                deploymentToDelete := strings.Replace(d.GetName(), tempSuffix, "", -1)

                deployment := getDeployment(d.GetNamespace(), deploymentToDelete, componentLabelKey, kubeclient)

                if deployment != nil {
                    fmt.Printf("3: Now deleting previous deployment in namespace %s, name %s.\n", d.GetNamespace(), deploymentToDelete)
                    if err := deleteDeployment(d.GetNamespace(), deploymentToDelete, kubeclient); err != nil {
                        exitOnErr(err)
                    }
                } else {
                    fmt.Printf("4: Didn't find deployment in namespace %s, name %s, label %s. Skipping.\n", d.GetNamespace(), deploymentToDelete, componentLabelKey)
                }
            } else if labelVal, ok := labels[newLabelKey]; ok && labelVal == "production" {
                fmt.Printf("Normal deployment with '%s' label created in namespace %s, name %s.\n", newLabelKey, d.GetNamespace(), d.GetName())
                deploymentToDelete := d.GetName() + tempSuffix
                fmt.Printf("6: Now deleting temporary deployment in namespace %s, name %s.\n", d.GetNamespace(), deploymentToDelete)
                if err := deleteDeployment(d.GetNamespace(), deploymentToDelete, kubeclient); err != nil {
                    exitOnErr(err)
                }
            }
        },
        DeleteFunc: func(obj interface{}) {
            d := obj.(*v1.Deployment)
            labels := d.GetLabels()

            if _, ok := labels[newLabelKey]; !ok {
                if _, ok := labels[tempLabelKey]; !ok {
                    fmt.Printf("Deployment without '%s' or '%s' label deleted in namespace %s, name %s.\n", newLabelKey, tempLabelKey, d.GetNamespace(), d.GetName())
                    fmt.Printf("5: Now creating normal deployment with '%s' label in namespace %s, name %s.\n", newLabelKey, d.GetNamespace(), d.GetName())
                    deploymentToCreate := createDeploymentWithNewLabel(newLabelKey, "production", *d)
                    if err := createDeploymentOnApi(deploymentToCreate, kubeclient); err != nil {
                        exitOnErr(err)
                    }
                }
            }
        },
    })

    fmt.Println("1: Creating temporary Deployments.")
    err = createDeploymentsOnApi(tempDeployments, kubeclient)
    exitOnErr(err)

    informer.Run(stopper)
}

func getDeployment(namespace string, name string, labelKey string, client *kubernetes.Clientset) *v1.Deployment {
    d, err := client.AppsV1().Deployments(namespace).Get(name, metav1.GetOptions{})
    if err != nil {
        return nil
    }

    if _, ok := d.GetLabels()[labelKey]; !ok {
        return nil
    }

    return d
}

func createDeploymentWithNewLabel(key string, val string, deployment v1.Deployment) v1.Deployment {
    newDeployment := deployment.DeepCopy()
    labels := newDeployment.GetLabels()
    if labels == nil {
        labels = make(map[string]string)
        newDeployment.SetLabels(labels)
    }
    labels[key] = val

    podTemplateSpecLabels := newDeployment.Spec.Template.GetLabels()
    if podTemplateSpecLabels == nil {
        podTemplateSpecLabels = make(map[string]string)
        newDeployment.Spec.Template.SetLabels(podTemplateSpecLabels)
    }
    podTemplateSpecLabels[key] = val

    labelSelectors := newDeployment.Spec.Selector.MatchLabels
    if labelSelectors == nil {
        labelSelectors = make(map[string]string)
        newDeployment.Spec.Selector.MatchLabels = labelSelectors
    }
    labelSelectors[key] = val

    return *newDeployment
}

func createDeploymentsWithNewLabel(key string, val string, deployments *v1.DeploymentList) *v1.DeploymentList {
    newDeployments := &v1.DeploymentList{}
    for _, d := range deployments.Items {
        newDeployment := createDeploymentWithNewLabel(key, val, d)
        newDeployments.Items = append(newDeployments.Items, newDeployment)
    }

    return newDeployments
}

func setAPIVersionAndKindForDeployment(d v1.Deployment, apiVersion string, kind string) {
    // These fields are empty strings.
    // Looks like an open issue: https://github.com/kubernetes/kubernetes/issues/3030.
    d.APIVersion = apiVersion
    d.Kind = kind
}

func processDeployments(deployments *v1.DeploymentList) *v1.DeploymentList {
    newDeployments := &v1.DeploymentList{}
    for _, d := range deployments.Items {
        // Set APIVersion and Kind until https://github.com/kubernetes/kubernetes/issues/3030 is fixed
        setAPIVersionAndKindForDeployment(d, "apps/v1", "Deployment")
        d.Status = v1.DeploymentStatus{}
        d.SetUID(types.UID(""))
        d.SetSelfLink("")
        d.SetGeneration(0)
        d.SetCreationTimestamp(metav1.Now())
        newDeployments.Items = append(newDeployments.Items, d)
    }
    return newDeployments
}

func saveDeployments(deployments *v1.DeploymentList) error {
    for _, d := range deployments.Items {
        if err := saveManifest(d); err != nil {
            return err
        }
    }

    return nil
}

func saveManifest(resource interface{}) error {
    var path = manifestsDir
    var name string
    var err error

    switch v := resource.(type) {
    case v1.Deployment:
        path = fmt.Sprintf("%s%s/%s/%s", path, v.GetClusterName(), v.GetNamespace(), "deployments")
        name = v.GetName()
    default:
        return errors.New(fmt.Sprintf("Got an unknown resource kind: %v", resource))
    }

    bytes, err := json.MarshalIndent(resource, "", "  ")
    if err != nil {
        return err
    }

    err = os.MkdirAll(path, 0755)
    if err != nil {
        return err
    }

    err = ioutil.WriteFile(fmt.Sprintf("%s/%s", path, name), bytes, 0644)
    if err != nil {
        return err
    }

    return nil
}

func deleteDeployment(namespace string, name string, client *kubernetes.Clientset) error {
    if err := client.AppsV1().Deployments(namespace).Delete(name, &metav1.DeleteOptions{}); err != nil {
        return err
    }

    return nil
}

func appendToDeploymentName(deployments *v1.DeploymentList, suffix string) *v1.DeploymentList {
    newDeployments := &v1.DeploymentList{}
    for _, d := range deployments.Items {
        d.SetName(fmt.Sprintf("%s%s", d.GetName(), suffix))
        newDeployments.Items = append(newDeployments.Items, d)
    }
    return newDeployments
}

func createDeploymentOnApi(d v1.Deployment, client *kubernetes.Clientset) error {
    d.SetResourceVersion("")

    if _, err := client.AppsV1().Deployments(d.GetNamespace()).Create(&d); err != nil {
        return err
    }

    return nil
}

func createDeploymentsOnApi(deployments *v1.DeploymentList, client *kubernetes.Clientset) error {
    for _, d := range deployments.Items {
        if err := createDeploymentOnApi(d, client); err != nil {
            return err
        }
    }
    return nil
}

func joinLabelKeyVals(labels map[string]string) string {
    labelKeyVals := make([]string, 0, len(labels))
    for k, v := range labels {
        labelKeyVals = append(labelKeyVals, fmt.Sprintf("%v=%v", k, v))
    }
    return strings.Join(labelKeyVals, ", ")
}

func homeDir() string {
    if h := os.Getenv("HOME"); h != "" {
        return h
    }
    return os.Getenv("USERPROFILE") // windows
}

func exitOnErr(err error) {
    if err != nil {
        log.Fatal(err)
    }
}