k8s kubebuilder系列开发 — 编写自定义资源和Reconciliation循环

原文转自公众号云原生CTO

这是 Kubernetes Operator Dev N 部分系列的第三部分。在本文中，我们将深入探讨如何编写自定义资源和operator/控制器的reconciliation协调循环。

我强烈建议您快速浏览文章/前面的一部分,因为它深入kubebuilder本文建议的这是一个先决条件,在前面的部分,我提出,引导一个operator叫做PostgresWriter在这篇文章中,我们将operator的reconciliation协调循环进行编码。

但是，如果您感到繁琐，我将在下一节为您总结PostgresWriter operator，这样您就可以获得足够的上下文，而不必完全依赖上一篇文章了:)

让我们设置一些上下文

如果您已经了解Reconciliation循环和PostgresWriter自定义资源/operator(我在前一部分中解释过了)，或者您已经阅读了本系列文章的第一部分和第二部分，那么您完全可以跳过本部分。

我们的示例operator，即PostgresWriter同样，如果您已经阅读了本系列文章的第二部分(上一部分)，可以随意跳过这部分。

我们将要构建的operator将被称为PostgresWriter。这个想法很简单。比方说，我们在世界的某个角落有一个Postgres DB。我们的集群将有一个名为“postgresql -writer”的自定义资源。

与“postgresql -writer”资源相关的清单如下:

解析正在创建的传入“postgres_writer”资源的spec，并识别table，name，age和country。

以<传入的postgresql资源>的命名空间/<传入的postgresql资源>的名称(在本例中为default/sample-student)的格式形成一个与上面传入的" postgresql -writer "资源对应的唯一id。

因为在Kubernetes中，对于某种资源类型(在我们的例子中是PostgresWriter)，名称空间/名称组合在集群中总是唯一的。

在名为spec的表中，在Postgres DB中插入一个新行。表和相应的spec.name, spec.age和spec.country字段，主键将是我们形成的上面唯一的id(传入资源的名称空间/名称)。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2gTDsEm7-1646976795925)(data:image/gif;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVQImWNgYGBgAAAABQABh6FO1AAAAABJRU5ErkJggg==)]

同时,每当PostgresWriterresource像上面的将被删除,相应我们的operator将删除的行对应资源从我们PostgresDB以保持我们的PostgresDB和PostgresWriter资源出现在我们的集群相互一致。

对于上面的示例，如果我们使用kubectl删除上面的sample-studentPostgresWriter资源，那么operator将删除id default/sample-student对应的行。

这将确保对于我们集群中的每个PostgresWriter资源，在我们的PostgresDB中都有一行，仅此而已。

Reconciliation 循环

Kubernetes本质上是一个系统，本质上是声明性的，这意味着它由用户提供一个所需的状态/规范，并且它尝试根据当前的现实来匹配所需的状态。协调当前状态和期望状态之间的漂移的艰苦工作是由称为控制器的组件完成的。

这些控制器具有“reconciliation循环”(也称为“控制循环”)，它会不断运行并查看集群的当前状态和期望的状态，每当它们之间发生漂移时，它们就会尝试采取某些行动将当前状态恢复到期望的状态。

在我们的例子中，在Kubernetes看来，我们的operator也只是一个控制器。它正在观察一个名为PostgresWriter的自定义资源，捕获传入的资源规范(期望的状态)，并执行一些神奇的操作(在Postgres DB中写入行)以匹配当前的现实和期望的状态。

例如:我希望您知道Kubernetes资源Deployment。它的关键特性之一是在指定数量的Pods(副本)上运行工作负载/应用程序，并确保始终运行指定数量的Pods。因此，假设部署应该在它的保护伞下运行5个副本，并假设您删除了它的两个pod。作为Reconciliation循环的一部分,部署控制器当前状态之间要注意这个漂移(3pods)和期望状态(5pods),这是需要创建两个新的pod的行动(因为3 + 2 = 5)匹配/协调当前状态所需的状态。

在这篇文章中我们要做什么?

在前一篇文章中，我们在kubebuilder的帮助下，用一些样板代码引导了PostgresWriter operator。因此，在本文中，我们将做以下事情: 我们将访问api/v1/postgreswriter_types。然后设置与PostgresWriter资源对应的结构体。

我们将编写所有与PostgresDB连接和插入/删除相关的辅助函数，因为这是我们的operator将执行的核心功能。

最后，我们将深入到controllers/postgreswriter_controller.go，编辑Reconcile()方法，并编写出我们operator背后的主要最终逻辑/代码。

重要提示:在本文中，我不会编写reconciliation循环来标识PostgresWriter资源的删除，并足够聪明地从Postgres DB中相应地删除该行(对应于已删除资源)。

我将在下一篇文章中专门讨论这个问题，因为这需要深入解释谓词、终结器以及将多个协调器连接到同一个控制器/operator，我希望在单独的一篇文章中顺利地专门讨论这些主题。所以，请原谅我，❤

就让我们一探究竟吧!

剖析我们的定制资源

首先，让我们把我们期望从oeprator那里得到的一切。

我们的自定义资源的实例看起来像这样:

所以这些是以下要求:

它的“Group”是demo.yash.com，“Version”是v1。
它的“kind”是PostgresWriter。
它下面有一个必需的节规范。
spec部分下面有四个必需字段:table(string)、name(string)、age(整数)和country(string)。

我在前一部分即本系列文章的第二部分解释了GVK即Group-Version-Kind。

转到api/v1/postgreswriter_types.go。这个文件将包含与我们的自定义资源将要遵循的结构相对应的所有结构。我们去自上而下。让我们首先在自定义资源的根中定义所需的spec字段。

//+kubebuilder:object:root=true
//+kubebuilder:subresource:status// PostgresWriter is the Schema for the postgreswriters API
type PostgresWriter struct {metav1.TypeMeta   `json:",inline"`metav1.ObjectMeta `json:"metadata,omitempty"`Spec   PostgresWriterSpec   `json:"spec"`Status PostgresWriterStatus `json:"status,omitempty"`
}

现在，让我们在spec下定义需要的table , name , age 和 country .

// PostgresWriterSpec defines the desired state of PostgresWriter
//+kubebuilder:validation:Required
type PostgresWriterSpec struct {//+kubebuilder:validation:Required//+kubebuilder:validation:Type=stringTable string `json:"table"`//+kubebuilder:validation:Required//+kubebuilder:validation:Type=integer//+kubebuilder:validation:Minimum=0Age int `json:"age"`//+kubebuilder:validation:Required//+kubebuilder:validation:Type=stringName string `json:"name"`//+kubebuilder:validation:Required//+kubebuilder:validation:Type=stringCountry string `json:"country"`
}

所有这些注释看起来像//+kubebuilder:validation....被称为CRD验证标记，kubebuilder在生成我们的CRD时将识别这些标记，并相应地将相关过滤器和条件应用到我们的最终CRD。要查看所有其他CRD验证标记

请参阅 :https://book.kubebuilder.io/reference/markers/crd-validation.html

我们已经定义了PostgresWriter自定义资源的结构，现在让我们生成CRD。

运行如下命令生成CRD。

make manifests

这将在config/CRD/bases/处生成你的CRD您将注意到生成的CRD已经指定了spec和age,country, name, table作为必需字段，并且它已经正确地定义了这些单独字段的类型，甚至在age字段下，还指定了最小值:0。这都要感谢Kubebuilder的CRD验证标记

编写Postgres的代码

本质上，我们的operator的核心是执行两个任务:

当创建PostgresWriter实例时，在PostgresDB上运行INSERT查询。
当PostgresWriter的一个实例被删除时，在PostgresDB上运行一个DELETEquery。

为了让它能够执行与Postgres相关的功能，让我们创建一个pkg/psql/psql.go文件和编码出所有与Postgres相关的核心功能。

让我们从声明包并导入所有相关库开始。

package psqlimport ("database/sql""fmt"_ "github.com/lib/pq" // no use on our end, just sets up the relevant drivers to be used by "database/sql" package
)

让我们定义一个名为PostgresDBClient的结构体，它的实例将是我们的operator和DB之间的通信点。

type PostgresDBClient struct {host         stringport         intdbname       stringuser         stringpassword     stringDbConnection *sql.DB
}

现在，除了DbConnection. 该对象（sql.DB实际上是指向该对象的指针）将用于与我们的数据库对话并对其执行查询。

但是为了在一开始设置它，甚至在需要时返回它(DbConnection)，让我们定义一个方法setupAndReturnDbConnection()

// If need be, setups a new DbConnection connection for a PostgresDBClient object (and returns it), else just returns the existing one functional one
// Not setting up a new connection blindly everything this method is called so as to maintain idempotency
func (pc *PostgresDBClient) setupAndReturnDbConnection() (*sql.DB, error) {// if the .DbConnection attribute is nil, then we clearly need to setup a new connection as none exists presently// else if it exists but has some other issues (basically, some error occurred while "Ping"ing the DB with current connection), then setup a new connection// PS: expiry of connection won't be a concern because .Ping() not only checks the connection but resets the connections too automatically with new connections, if need be.// Ref: https://cs.opensource.google/go/go/+/refs/tags/go1.17.1:src/database/sql/sql.go;l=868setupNewConnection := falseif pc.DbConnection == nil {setupNewConnection = true} else if err := pc.DbConnection.Ping(); err != nil {setupNewConnection = true}if setupNewConnection {psqlInfo := fmt.Sprintf("host=%s port=%d user=%s "+"password=%s dbname=%s sslmode=disable",pc.host, pc.port, pc.user, pc.password, pc.dbname)newDbConnection, err := sql.Open("postgres", psqlInfo)if err != nil {return nil, err}pc.DbConnection = newDbConnection}return pc.DbConnection, nil
}

这个方法是幂等的，因为它不会在每次调用时盲目地创建新的DB连接。这个方法计算PostgresDBClient对象并检查其现有的DbConnection属性，当且仅当该属性不存在或有其他问题时，将建立一个新的连接并返回。因此，即使重复调用它也不会导致不必要地创建多个不同的结果/连接。

现在，让我们定义一个用于执行INSERT查询的方法。

// Insert inserts a row into the DB to which the receiver PostgresDBClient points
func (pc *PostgresDBClient) Insert(id, table, name string, age int, country string) error {dbConnection, err := pc.setupAndReturnDbConnection()if err != nil {return err}insertQuery := fmt.Sprintf("INSERT INTO %s (id, name, age, country) VALUES ('%s', '%s', %d, '%s') ON CONFLICT (id) DO NOTHING;", table, id, name, age, country)if _, err := dbConnection.Exec(insertQuery); err != nil {return err}return nil
}

让我们定义一个用于执行DELETE查询的方法。

// Delete deletes row from the DB to which the receiver PostgresDBClient points
func (pc *PostgresDBClient) Delete(id, table string) error {dbConnection, err := pc.setupAndReturnDbConnection()if err != nil {return err}deleteQuery := fmt.Sprintf("DELETE FROM %s WHERE id='%s';", table, id)if _, err := dbConnection.Exec(deleteQuery); err != nil {return err}return nil
}

最后，让我们定义一个构造函数(kind)，它将返回一个成熟的PostgresDBClient对象，并提供必要的参数。让我们定义一个Close()方法，它将被用来优雅地清理PostgresDBClient对象，以避免任何潜在的内存泄漏。

// NewPostgresDBClient acts like a constructor to generating a new object of PostgresDBClient with a DbConnection as well
func NewPostgresDBClient(host string, port int, dbname, user, password string) (*PostgresDBClient, error) {pc := &PostgresDBClient{host:     host,port:     port,dbname:   dbname,user:     user,password: password,}if _, err := pc.setupAndReturnDbConnection(); err != nil {return nil, err}return pc, nil
}// Close is used to gracefully close and wrap up the DbConnection associated with the PostgresDBClient object so as to avoid any associated memory leaks
func (pc *PostgresDBClient) Close() {pc.DbConnection.Close()
}

现在，我们已经完成了围绕Postgres的核心功能的设置，可以开始编写operator了。但在此之前，让我们快速了解一下围绕operator的现有引导代码(仅是有用的部分)。

理解我们的引导控制器/`operator`

让我们探索文件的controllers/postgreswriter_controller.go某些部分，main.go看看我们的operator是如何被实例化和执行的。

让我们从controllers/postgreswriter_controller.go开始

你会注意到PostgresWriterReconciler结构。这个结构实际上代表了我们的operator。它将实现协调器接口和Reconciliation循环，当PostgresWritercustom资源发生什么事情时，Reconciliation循环将被触发。它还能够与PostgresDB通信，并运行相应的INSERT或DELETE查询。根据其拥有的权限，它可以与集群通信、访问甚至operator源。

// PostgresWriterReconciler reconciles a PostgresWriter object
type PostgresWriterReconciler struct {client.ClientScheme           *runtime.Scheme
}

client.Client — 只要我们的operator (PostgresWriterReconciler) 想要与我们的 Kubernetes 集群对话并对其执行任何 CRUD，就会使用此属性。因此，诸如getting/updating/deleting/creating之类的任何事情都将由client.Client.
Scheme — 该属性用于通过管理器将 struct/type 注册PostgresWriter到其对应的 GVK（Group: demo.yash.com, V: v1, Kind: PostgresWriter）。它主要在幕后被不同的客户端和缓存使用，以将 Go 类型/结构与其对应的 GVK 关联和关联。

现在，继续进行，您会注意到该Reconcile()方法。该方法将包含我们operator的reconciliation循环背后的源代码。这是我们将在需要时对我们的operator进行编程以运行 PostgresDB 查询的地方。每当 PostgresWriter 自定义资源发生诸如创建/更新/删除之类的事情时，就会触发此方法。

但是谁能确保在我们的集群中当PostgresWriter资源发生问题时调用这个方法?

接下来是SetupWithManager方法。SetupWithManager(mgr ctrl.Manager)——仔细查看代码中是如何定义的。这几乎不言自明。它确保了无论什么时候PostgresWrite资源(For(&demov1.PostgresWriter{}))发生任何事情，调用我们的operator/PostgresWriterReconciler (Complete(r)，其中r是*PostgresWriterReconciler)的Reconcile()方法。

// SetupWithManager sets up the controller with the Manager.
func (r *PostgresWriterReconciler) SetupWithManager(mgr ctrl.Manager) error {return ctrl.NewControllerManagedBy(mgr).For(&demov1.PostgresWriter{}).Complete(r)
}

我们已经完成了controllers/postgreswriter_controller.go，所以让我们深入了解一下main.go

func main() {......if err = (&controllers.PostgresWriterReconciler{Client:           mgr.GetClient(),Scheme:           mgr.GetScheme(),}).SetupWithManager(mgr); err != nil {setupLog.Error(err, "unable to create controller", "controller", "PostgresWriter")os.Exit(1)}//+kubebuilder:scaffold:builder......
}

我们的控制器/operator ( PostgresWriterReconciler) 正在实例化并与管理器连接/设置。

因此，无论何时，它执行时，我们的operator也将作为它的一部分启动。

与我们的`operator`一起设置`PostgresDB`客户端策略

从高效的角度来看，我们想要的是，对于我们的 operator 的一个实例，PostgresDBClient应该只创建一个object实例并用于处理与我们的 PostgresDB的所有通信，而不是盲目地为每个协调循环创建一个 PostgresDBClient和连接.

因此，让我们从修改引导operator的一段代码开始，以符合上述每当operator被实例化时实例化一个 Postgres 客户端的情况。

您可以在代码中对 Postgres执行大量进一步调整，例如管理连接池、连接超时、空闲超时等。但我们不会深入研究这些内容。本文重点介绍 Kubernetes operator dev