Kubernetes Operator 如何处理并发？

最初发布在我的博客

默认情况下，使用 Kubebuilder 和控制器运行时构建的操作符一次处理一个协调请求。这是一个明智的设置，因为运算符开发人员可以更轻松地推理和调试应用程序中的逻辑。它还限制了从控制器到 ectd 和 API 服务器等核心 Kubernetes 资源的吞吐量。

但是，如果您的工作队列开始备份，并且由于请求留在队列中等待处理而导致平均协调时间增加，该怎么办？对我们来说幸运的是，控制器运行时 Controller 结构包含一个 MaxConcurrentReconciles 字段（正如我之前在 Kubebuilder Tips 文章中提到的）。此选项允许您设置在单个控制器中运行的并发协调循环的数量。因此，如果值大于 1，您可以同时协调多个 Kubernetes 资源。

在我的 Operator 旅程的早期，我遇到的一个问题是我们如何保证同一资源不会同时被 2 个或更多 Goroutine 协调？将 MaxConcurrentReconciles 设置为高于 1 时，这可能会导致各种竞争条件和不良行为，因为协调循环内的对象状态可能会因外部源（在不同线程中运行的协调循环）的副作用而发生变化.

我对此思考了一段时间，甚至实现了一种基于sync.Map的方法，该方法允许goroutine获取给定资源的锁（基于其命名空间/名称）。

事实证明，所有这些努力都是徒劳的，因为我最近（在 k8s slack 通道中）了解到控制器工作队列已经包含此功能！尽管实现更简单。

这是一个关于 k8s 控制器的工作队列如何保证唯一资源按顺序协调的简短故事。因此，即使 MaxConcurrentReconciles 设置为高于 1，您也可以确信一次只有一个协调函数作用于任何给定资源。

客户端/util

Controller-runtime 使用 client-go/util/workqueue 库来实现其底层协调队列。在包的 doc.go 文件中，注释指出工作队列支持以下属性：

• 公平：按添加顺序处理项目。
• 吝啬：单个item不会被同时处理多次，如果一个item在被处理之前被多次添加，那么它只会被处理一次。
• 多个消费者和生产者。特别是，允许​​在处理项目时重新排队。
• 关闭通知。

等一下...我的答案就在第二个项目符号中，“吝啬”属性！根据这些文档，队列将自动为我处理这个并发问题，而无需编写一行代码。让我们来看看具体的实现。

工作队列如何工作？

workqueue 结构体有 3 个主要方法，Add、Get 和 Done。在控制器内部，通知者会将协调请求（通用 k8s 资源的命名空间名称）添加到工作队列中。在单独的 goroutine 中运行的协调循环将从队列中获取下一个请求（如果队列为空则阻塞）。该循环将执行控制器中编写的任何自定义逻辑，然后控制器将调用队列上的 Done，并将协调请求作为参数传递。这将再次开始该过程，并且协调循环将调用 Get 来检索下一个工作项。

这类似于在 RabbitMQ 中处理消息，工作人员从队列中弹出一个项目，对其进行处理，然后将“Ack”发送回消息代理，表明处理已完成并且可以安全地从队列中删除该项目队列。

不过，我有一个在生产环境中运行的操作员为 QuestDB Cloud 的基础设施提供支持，并且希望确保工作队列按照宣传的那样工作。因此，a 编写了一个快速测试来验证其行为。

一点测试

这是一个验证“Stingy”属性的简单测试：

package main_test

import (
    "testing"

    "github.com/stretchr/testify/assert"

    "k8s.io/client-go/util/workqueue"
)

func TestWorkqueueStingyProperty(t *testing.T) {

    type Request int

    // Create a new workqueue and add a request
    wq := workqueue.New()
    wq.Add(Request(1))
    assert.Equal(t, wq.Len(), 1)

    // Subsequent adds of an identical object
    // should still result in a single queued one
    wq.Add(Request(1))
    wq.Add(Request(1))
    assert.Equal(t, wq.Len(), 1)

    // Getting the object should remove it from the queue
    // At this point, the controller is processing the request
    obj, _ := wq.Get()
    req := obj.(Request)
    assert.Equal(t, wq.Len(), 0)

    // But re-adding an identical request before it is marked as "Done"
    // should be a no-op, since we don't want to process it simultaneously
    // with the first one
    wq.Add(Request(1))
    assert.Equal(t, wq.Len(), 0)

    // Once the original request is marked as Done, the second
    // instance of the object will be now available for processing
    wq.Done(req)
    assert.Equal(t, wq.Len(), 1)

    // And since it is available for processing, it will be
    // returned by a Get call
    wq.Get()
    assert.Equal(t, wq.Len(), 0)
}

由于工作队列在底层使用互斥体，因此这种行为是线程安全的。因此，即使我编写了更多使用多个 goroutine 同时高速读取和写入队列的测试来试图破坏它，工作队列的实际行为也将与我们的单线程测试相同。

一切并没有失去

Kubernetes 标准库中隐藏着很多像这样的小东西，其中一些位于不太明显的地方（比如在 go 客户端包中找到的控制器运行时工作队列）。尽管有这个发现，以及我过去所做的其他类似的发现，我仍然觉得我之前解决这些问题的尝试并不是完全浪费时间。它们迫使您批判性地思考分布式系统计算中的基本问题，并帮助您更多地了解幕后发生的事情。因此，当我发现“Kubernetes 做到了”时，我松了一口气，因为我可以简化我的代码库，或许还可以删除一些不必要的单元测试。

以上是Kubernetes Operator 如何处理并发？的详细内容。更多信息请关注PHP中文网其他相关文章！