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

标签 go kubernetes client-go

上下文

我正在编写一个使用 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 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)
}

最佳答案

我最终使用了 SharedInformer .

这些资源很有帮助。

.

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)
    }
}

关于go - 当使用 k8s.io/client-go 库更改 kubernetes 部署时，获得通知的最佳方式是什么？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/53200785/

上一篇：ssl - 从 Go 服务器发送分块的 HTTPS 响应

下一篇：amazon-web-services - 在 golang 中订阅 SNS 主题和/或 SQS 队列？

相关文章：

go - 我试图理解为什么这个案子总是失败

kubernetes - kubernetes容器中的hostPID和hostIPC选项是什么意思？

azure - 无法访问 kubernetes v1.2.0 上的 apiserver

docker - Kubernetes - 是否有适用于 centos 的存储库？

go - 如何获取集群外go Kubernetes客户端的当前命名空间？

kubernetes - 什么情况导致 Pod 日志读取器返回 EOF

go - 'For'循环前后空语句

go - 如何使用 Twitter API v2 使用 Go 发布推文？

go - 如何在亚马逊 eks 服务上使用 kubernetes go-client？

function - 声明和定义中不同的函数参数类型

©2024 IT工具网  联系我们