Blogs

开发Kubernetes自定义APIServer

由 SuKai in clusterpedia, golang

April 4, 2022

前面文章介绍了DaoCloud开源的Clusterpedia的业务流程和资源同步机制，Clusterpedia将多个Kubernetes集群的指定资源对象同步保存到MySQL数据库并对外提供查询和检索。今天我们一起看一下Clusterpedia如何开发一个Kubernetes自定义的APIServer，将保存在数据库中的多个Kubernetes业务集群的资源对外提供查询和检索的。 本文内容分以下部分： 1，Kubernetes APIServer基本知识 2，自定义APIServer开发 Kubernetes APIServer基本知识 Kubernetes APIServer是一个实现了RESTful API的WebServer。 API类型 core group，/api/v1为前缀的API接口PATH，core group不需要在apiVersion字段中指定，示例：apiVersion: v1。 named groups，REST path为/apis/$GROUP_NAME/$VERSION，指定apiVersion: $GROUP_NAME/$VERSION，示例apiVersion: batch/v1。 暴露系统状态的API，比如/metrics、/healthz等。 扩展机制 为了增加Kubernetes API的扩展性，Kubernetes提供了两种机制：1，APIExtensions，创建自定义资源CRD，处理CRD/CR的REST请求。2，Aggretgator聚合机制，注册APIService资源，APIServer将对应的API GroupVersion请求代理转发到注册的Service上。 委托调用链 APIServer使用委托模式，通过DelegationTarget接口，把Aggretgator、API Server、APIExtensions链式串联起来，对外提供服务。 当请求Kubernetes对象时，如果在Aggregator中找不到，就去KubeAPIServer中找，最后到APIExtensions中找。 Clusterpedia通过这种委托模式，注册Aggregator服务，优先于所在Kubernetes集群的APIServer提供api和apis两个group的服务，来访问存储在数据库中的多个业务集群的Kubernetes资源对象，kubectl也可以直接执行命令进行下游集群的资源对象查询检索操作。 资源对象操作 在APIServer中，API请求资源对象时，先通过RESTStorage进行REST API处理，在RESTStorage中调用etcd store进行数据存储操作。 Clusterpedia中，与Kubernetes APIServer类似，先通过RESTStorage进行REST API处理，在RESTStorage中调用ResourceStorage进行数据库操作。 自定义APIServer开发 Clusterpedia APIServer基本流程 1，config-complete-new模式构造GenericAPIServer实例kubeResourceAPIServer NewDefaultConfig初始化默认配置。BuildHandlerChain中WithRequestInfo处理API请求信息保存到context中，WithPanicRecovery处理崩溃日志并恢复，RemoveFieldSelectorFromRequest处理URL Query。 complete完善配置，wrapRequestInfoResolverForNamespace封装处理RequestInfo解析器。 New从空委托者构造GenericAPIServer。创建discoveryManager提供自动发现API处理，NonGoRestfulMux.Handle注册/api和/apis的自动发现路由。创建resourceHandler提供资源对象请求API处理，NonGoRestfulMux.HandlePrefix注册/api和/apis为前缀PATH的路由。NewClusterResourceController创建ClusterPedia控制器，监听ClusterPedia CRD事件，处理下游集群的资源，discoveryManager读取controller更新的资源信息提供自动发现API服务。 2，BuildHandlerChain 将WithRequestQuery添加到HandlerChain，WithRequestQuery保存URL Query到context 3，构造GenericAPIServer实例genericServer config.GenericConfig.New从kubeResourceAPIServer委托者构造一个新的GenericAPIServer实例genericServer。所以API资源请求调用链是，先在kubeResourceAPIServer找，再到genericServer中找。 3，InstallAPIGroup注册API对象 genericServer中注册API对象：GroupVersion：clusterpedia.io/v1beta1，Resources资源为resources，collectionresources对象。每个资源对应一个REST storage，用于处理对应资源的请求。 4，AddPostStartHookOrDie添加APIServer启动后调用函数 APIServer启动后，启动Informer，同步Cache。 5，PrepareRun() 准备健康状态检查，存活检查 6，Run() 启动运行APIServer clusterpedia-apiserver命令 在opts.Config()中： 1，根据配置的数据库类型和配置项，创建storage接口，用于操作数据库 2，根据服务器配置，生成SSL自签名证书 3，生成genericapiserver默认配置 通过Config->Complete->New模式构造一个ClusterPediaServer实例server server.Run运行ClusterPediaServer func NewClusterPediaServerCommand(ctx context.Context) *cobra.

继续阅读

开发Kubernetes自已的Informer

由 SuKai in clusterpedia, golang

March 30, 2022

我们在Kubernetes自定义资源CRD控制器开发中都使用过Informer，Informer主要提供了两个功能：1，同步数据到本地缓存。2，监听Kubernetes资源的事件，根据对应的事件操作类型，触发事先注册好的ResourceEventHandle。 前面文章介绍了DaoCloud开源的Clusterpedia通过Informer机制实现了同步多个Kubernetes集群资源到MySQL等数据库。那么如何开发一个简化的自定义的Informer呢？下面我们一起看一下DaoCloud开源的Clusterpedia如何实现Kubernetes资源版本Informer的。 文章内容分以下几部分： 1，Kubernetes原生的Informer机制与实现 2，Clusterpedia的ResourceVersionInformer与原生的Informer比较 3，ResourceVersionInformer代码实现 Kubernetes原生的Informer机制 Informer组件 Reflector：反射器，通过Kubernetes的List/Watch API监控指定类型的资源对象。 DeltaFIFO Queue：将Reflector监控到的变化的对象存放在这个FIFO队列中。 LocalStore：Informer的本地缓存，缓存Kubernetes资源对象，可以被Lister的List/Get方法访问，减少对APIServer的访问压力。 WorkQueue：DeltaFIFO中的事件更新完Store后保存到WorkQueue中，Controller处理WorkQueue中的资源对象事件调用对应的回调函数。 SharedInformer实现 这里可以看到： 1，NewSharedIndexInformer中，初始化了processor，这个processor用于回调用户注册的事件处理回调函数，NewIndexer初始化了一个indexer，这个indexer是作为informer的store本地缓存。listerWatcher指定ListerWatcher接口lw，这个是Kubernetes的client连接用于监控Kubernetes资源对象。objectType指定监控的Kubernetes资源类型。 2，SharedInformer的Run中，NewDeltaFIFOWithOptions构造了一个DeltaFIFO实例fifo，构造了一个controller，这个controller使用fifo队列，使用listerWatcher监控资源对象，ObjectType指定资源类型，Process指定controller处理队列的函数。s.processor.run启动监听通知，调用用户注册的回调函数处理。 type SharedInformer interface { AddEventHandler(handler ResourceEventHandler) AddEventHandlerWithResyncPeriod(handler ResourceEventHandler, resyncPeriod time.Duration) GetStore() Store GetController() Controller Run(stopCh <-chan struct{}) HasSynced() bool LastSyncResourceVersion() string SetWatchErrorHandler(handler WatchErrorHandler) error } func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) SharedIndexInformer { realClock := &clock.RealClock{} sharedIndexInformer := &sharedIndexInformer{ processor: &sharedProcessor{clock: realClock}, indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers), listerWatcher: lw, objectType: exampleObject, resyncCheckPeriod: defaultEventHandlerResyncPeriod, defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod, cacheMutationDetector: NewCacheMutationDetector(fmt.

继续阅读

Golang事件队列开发

由 SuKai in clusterpedia, golang

March 28, 2022

前面文章介绍了DaoCloud开源的Clusterpedia通过Informer机制监控Kubernetes资源事件，保存事件对象到队列中，等待事件处理函数将Kubernetes资源对象保存到数据库中。今天一起学习一下DaoCloud开源的Clusterpedia如何实现事件队列的。 下面我们文章首先介绍Clusterpedia如何使用EventQueue的，再来介绍EventQueue如何实现的。主要分以下部分： 1，Clusterpedia事件处理函数HandleDeltas，调用informer.handler，这个handler实现了ResourceEventHandler接口，有OnAdd，OnUpdate，OnDelete方法。 2，实现了ResourceEventHandler接口的ResourceSynchro，ResourceSynchro调用EventQueue接口将对象写入队列 3，实现了EventQueue接口的pressurequeue，pressurequeue负责队列的操作。 事件处理函数HandleDeltas Clusterpedia直接基于client-go底层的cache.Controller开发了自己的resourceVersionInformer，cache.Controller的Process函数调用HandleDeltas来处理资源事件。根据cache.Deltas的操作类型，调用informer.storage和informer.handler的对应操作。这里的handler指的是ResourceEventHandler。 func (informer *resourceVersionInformer) HandleDeltas(deltas cache.Deltas) error { for _, d := range deltas { switch d.Type { case cache.Replaced, cache.Added, cache.Updated: version, exists, err := informer.storage.Get(d.Object) if err != nil { return err } if !exists { if err := informer.storage.Add(d.Object); err != nil { return err } informer.handler.OnAdd(d.Object) break } if d.Type == cache.Replaced { if v := compareResourceVersion(d.Object, version); v <= 0 { if v == 0 { informer.

继续阅读

DaoCloud Clusterpedia多Kubernetes集群资源同步与检索

由 SuKai in clusterpedia, golang

March 27, 2022

Clusterpedia是DaoCloud开源的支持同步多个集群的指定资源，将资源对象保存到MySQL等数据库，可以通过Clusterpedia检索排序分页资源。 今天我们一起来看一下Clusterpedia的资源同步控制器代码， 了解他的业务流程。 Clusterpedia架构 Clusterpedia在一个Kubernetes集群中，运行Clusterpedia APIServer和ClusterSynchro Manager，提供聚合API，管理和同步多个Kubernetes集群的资源。Clusterpedia所在的集群可理解为管理集群或者控制集群。Clusterpeida将同步的业务集群的资源保存到MySQL/PostgreSQL数据库中，这部分由代码中的StorageFactory实现。 主要业务流程： 1，在管理集群中通过自定义资源PediaCluster CRD来保存、管理和调谐下调业务集群。 2，每一个PediaCluster资源，对应一个ClusterSynchro，也就是一个业务集群会有一个ClusterSynchro实例，用来保存下游业务集群的RestConfig，storage数据库工厂接口，resourceSynchros同步资源GVR对应的ResourceSynchro。 3，每个集群的每个GVR资源对应一个ResourceSynchro，用来保存GVR/GVK信息，事件队列等信息。 4，在PediaCluster资源调谐过程中，主要负责 a，创建对应的ClusterSynchro，将ClusterSynchro中的同步资源对应的Informer启动，将Informer事件调用ResourceSynchro的事件处理函数进行处理。ResourceSynchro实现了ResourceEventHandler接口。 b，ResourceSynchro的事件处理函数，主要实现将对象放入ResourceSynchro队列queue中。 c，开启worker线程池，等待处理队列中的事件，将对象同步到数据库中。 代码 clustersynchro-manager配置项 1，storage.NewStorageFactory生成存储工厂，支持数据库：MySQL, PostgreSQL 2，生成管理集群的Kubeconfig，创建kubernetes以及自定义资源PediaCluster的clientset连接。 func (o *Options) Config() (*config.Config, error) { if err := o.Validate(); err != nil { return nil, err } storagefactory, err := storage.NewStorageFactory(o.Storage.Name, o.Storage.ConfigPath) if err != nil { return nil, err } kubeconfig, err := clientcmd.BuildConfigFromFlags(o.Master, o.Kubeconfig) if err != nil { return nil, err } client, err := clientset.

继续阅读

Golang实现WorkerPool文件上传

由 SuKai in golang

March 21, 2022

在前面文章中介绍了golang开发rpc命令行工具，今天继续后续功能的介绍。 场景描述： 1，客户端运行一个daemon程序，执行文件上传任务，使用boltdb数据库记录任务执行状态。支持继续上传未完成文件。 2，客户端命令行通过rpc调用客户端daemon程序API，进行文件上传任务管理，包括任务创建，查看，启停。 通过这个功能开发，我们将golang并发编程的goroutine协程，channel通道，context上下文, Mutex互斥锁, WaitGroup协程同步等技术得到应用。 基本知识： WaiGroup 等待一组协程执行完成后继续向下执行，WaitGroup内部有一个计数器，从0开始计数，有3个方法：Add()，Done(), Wait()。Add()添加计数，Done()减掉一个计数，Wait()执行阻塞，直到WaitGroup数量变成0。 Select select和channel配合使用，通过select可以监听多个channel的I/O读写事件。 select { case <-ctx.Done(): p.log.Error("the context error \n") return context.Canceled default: } 如果没有default分支，select会阻塞在多个channel上，对多个channel进行监控。如果有default分支，多个channel都没有满足，则执行default分支。 Context Golang的Context称之为上下文，用来跟踪goroutine关系链，传递通知，达到控制他们的目的。主要用法是，传递取消信号，传递数据。 下面是一个传递取消信号的使用过程，首先context.Background()返回一个空的Context，一般用于整个context tree的根节点。context.WithCancel()返回一个ctx可取消的Sub Context，作为Run的参数传入goroutine，这样可以使用ctx跟踪这个Goroutine。cancel()调用Sub Context的取消函数，向关联的Goroutine发送一个"取消"通知。在Run函数中，接收ctx.Done的cancel通知，做相关清理后退出。 ctx, cancel := context.WithCancel(context.Background()) go pool.Run(ctx) cancel() for { select { case <-ctx.Done(): return } } Mutext互斥锁 Mutext互斥锁在同一时间只被一个goroutine访问，不区分读写。有两个方法：Lock()和Unlock()。当一个goroutine申请了Lock()，那么另一个goroutine申请Lock()时会阻塞等待直到Unlock()释放锁。 multipart/form请求 multipart/form请求是http Post方法，可以发送文件和消息，在请求的Header中包含一个特殊头信息Content-Type: multipart/form-data; boundary=，boundary的值为随机计算生成的值，用于分隔上传多个form-data的间隔。 POST /raw/v1alpha2/rawdatas/6214cb78ff3f2903536f5751/images HTTP/1.1 Host: 127.0.0.1:9090 Content-Length: 546 Content-Type: multipart/form-data; boundary=----WebKitFormBoundary7MA4YWxkTrZu0gW ----WebKitFormBoundary7MA4YWxkTrZu0gW Content-Disposition: form-data; name="files"; filename="/C:/Users/ycsk0/Pictures/ubuntu-install.png" Content-Type: image/png (data) ----WebKitFormBoundary7MA4YWxkTrZu0gW Content-Disposition: form-data; name="files"; filename="/C:/Users/ycsk0/Pictures/windows store.

继续阅读

Golang操作BoltDB数据库

由 SuKai in golang, boltdb

March 2, 2022

boltdb是一个go语言开发的嵌入式kv数据库。 今天我们一起看看golang如何操作BoltDB，再看看Etcd是如何操作BoltDB的，Etcd对BoltDB做了很多优化，复杂度提高很多。 Etcd是一个基于Raft开发的分布式key-value存储。Kubernetes使用Etcd来作为集群数据的存储。Etcd存储层包含预写日志(WAL)、快照(Snapshot)、boltdb。其中WAL与Snapshot实现了故障恢复和数据回滚或重做，让数据尽量不丢失，boltdb则保存了集群元数据和用户写入的数据。 使用场景 在自动驾驶汽车算法模型开发中，需要大量的2D图片和3D点云图片，我们将这些图片上传到打标系统，进行数据标注，标注的数据用于算法模型的数据集。 这些图片文件的上传，需要有一个有状态的客户端来完成文件上传工作，客户端可以在本地创建上传任务，记录上传位置，任务启停等功能。 这里我们使用go来开发客户端，选择boltdb这种简单的嵌入式kv数据库作为存储。 BoltDB基本概念 DB数据库文件 BoltDB的数据库只有一个文件，一个进程打开此文件后，其他进程无法使用，只有等这个进程释放文件锁。对应关系型数据库中database。 Bucket 对应关系型数据库中的table，操作有CreateBucket, CreateBucketIfNotExists, DeleteBucket。 Key/Value键值对 boltdb中键值对都是使用字节数组值进行存储，keys以字节排序的顺序存储在一个bucket中。 Transaction事务 boltdb在某一时刻，只允许一个读写事务或者允许多个只读事务。用户可以通过db的Begin方法启动一个事务，通过Rollback和Commit方法自己控制提交和回滚，Close关闭事务。同时boltdb还提供了内置隐式事务Update, View, Batch方法。 读写事务 db.Update()启动一个读写事务，可重复读的事务。return error，回滚整个修改，return nil提交修改。 只读事务 db.View()打开一个只读事务，无法做写入操作。 批处理读写事务 db.Batch()多次读写事务合并为一次事务，使用上和Update读写事务相同，boltdb自动将其分批，分批写入磁盘减少并发读写事务等待磁盘I/O的开销。 代码学习 普通使用案例 定义Client接口，实现CreateBucketIfNotExists创建Bucket，Put写入更新数据，Delete删除数据，Range查询单条数据或数据列表，ForEach遍历数据。可以看到只使用了两种不同的事务Update和View来完成操作。 主要操作有三种Get, Put, Delete，这里只用到了Put,Delete操作，通过Range获取单条或者多条数据，另外通过Cursor游标，Seek偏移，Next来遍历数据。 package boltdb import ( "bytes" "math" "go.etcd.io/bbolt" bolt "go.etcd.io/bbolt" "go.uber.org/zap" ) type client struct { db *bbolt.DB log *zap.Logger } type Client interface { CreateBucketIfNotExists(bucket Bucket) Put(bucketType Bucket, key, value []byte) Delete(bucketType Bucket, key []byte) Range(bucketType Bucket, key, endKey []byte, limit int64) (keys [][]byte, vs [][]byte) ForEach(bucketType Bucket, visitor func(k, v []byte) error) Close() error } func NewClient(path string, log *zap.

继续阅读

Kubernetes应用系统API用户认证与鉴权 -- 鉴权篇

由 SuKai in Kubernetes, Go, Api

February 14, 2022

上篇文章讲了通过LDAP和Dex实现两种不同的API用户认证的方式，用户认证是识别API请求的来源，让系统知道请求的访问者是谁。应用系统可以根据请求的身份，来判断该用户可以访问哪些API接口获得数据，这就是用户鉴权。下面我们一起来看一下，在Kubernetes平台上，如何通过RBAC来实现资源访问限制，同样我们还是通过Kubesphere的代码来学习。 鉴权实现的过程： 1，构建一个HandlerChain，HandlerChain中包含了：WithAuthentication, WithRequestInfo和WithAuthorization用户认证和鉴权的三个Filter以及WithAuditing Filter用于审计日志。Filter可以理解为web 框架的Middleware。 2，filters.WithAuthentication上篇文章介绍过，完成用户身份识别和分发Token，将用户信息写入Context。 3，filters.WithRequestInfo处理API请求信息，将请求信息保存到context中。请求信息包括：Path, Verb, APIPrefix, APIGroup, APIVersion, Namespace, Resource, Name这些RBAC鉴权需要用到的资源信息，以及客户端信息等。 4，filters.WithAuthorization根据RequestInfo和RBAC规则，判断请求的合法性，终止或者允许请求继续下去。 代码实现： 1，HandlerChain处理请求 这里可以看到有两个鉴权，一个是pathAuthorizer，一个是RBACAuthorizer，pathAuthorizer定义了哪些URL Path不需要鉴权，RBACAuthorizer顾名思义就是通过RBAC鉴权，后面我们详细介绍这两个鉴权。 var authorizers authorizer.Authorizer switch s.Config.AuthorizationOptions.Mode { case authorization.AlwaysAllow: authorizers = authorizerfactory.NewAlwaysAllowAuthorizer() case authorization.AlwaysDeny: authorizers = authorizerfactory.NewAlwaysDenyAuthorizer() default: fallthrough case authorization.RBAC: excludedPaths := []string{"/oauth/*", "/kapis/config.kubesphere.io/*", "/kapis/version", "/kapis/metrics"} pathAuthorizer, _ := path.NewAuthorizer(excludedPaths) amOperator := am.NewReadOnlyOperator(s.InformerFactory, s.DevopsClient) authorizers = unionauthorizer.New(pathAuthorizer, rbac.NewRBACAuthorizer(amOperator)) } handler = filters.WithAuthorization(handler, authorizers) handler = filters.WithAuthentication(handler, authn) handler = filters.WithRequestInfo(handler, requestInfoResolver) s.

继续阅读

Kubernetes应用系统API用户认证与鉴权 -- 认证篇

由 SuKai in Kubernetes, Go, Api

February 13, 2022

前面文章介绍了Kubernetes应用系统用户管理，实现了外部用户存储(LDAP)与Kubernetes用户映射，下面以Kubesphere为参考，一起看一下如何实现Kubernetes应用系统用户的认证。 应用平台用户认证主要实现两个功能： 1，用户身份鉴别 用户身份认证可以通过LDAP进行用户名和密码认证，也可以通过第三方认证服务进行OAuth认证。 2，Token管理 Client端认证通过后，带上获取Token请求API。应用系统根据Token提取用户ID，进行请求权限鉴别。同时应用系统负责Token的生命周期管理。 基本概念： OpenID Connect(OIDC)是基于OAuth 2.0身份认证协议，增加了OAuth 2.0中末定义的规范，例如scope, Claim用户信息字段。 代码实现 基本认证方式 API路由 // legacy auth API legacy := &restful.WebService{} legacy.Path("/aiapis/iam.aiscope/v1alpha2/login"). Consumes(restful.MIME_JSON). Produces(restful.MIME_JSON) legacy.Route(legacy.POST(""). To(handler.login). Deprecate(). Doc("Aiscope APIs support token-based authentication via the Authtoken request header. The POST Login API is used to retrieve the authentication token. After the authentication token is obtained, it must be inserted into the Authtoken header for all requests."). Reads(LoginRequest{}). Returns(http.StatusOK, api.StatusOK, oauth.Token{}). Metadata(restfulspec.

继续阅读

Kubernetes应用平台API开发实战

由 SuKai in kubernetes, 机器学习

January 27, 2022

前面文章介绍了通过基于go-restful框架开发API, client-go生成clientset, informers, listers来读取和写入自定义资源，基于kubebuilder开发CRD资源控制器。今天我们通过一个实例来看一下API整个开发过程。 开发步骤 1，通过kubebuilder来开发CRD控制器 2，通过client-gen，lister-gen，informer-gen生成clientset, informers, listers代码 3，开发models实现CRD资源的kubernets读写操作，读取列表时的排序，分页，过滤 4，开发handler实现CRD资源的API处理 CRD资源的读操作 CRD资源的读操作通过Informer来读取，减少API和Etcd集群的压力。 Informer的主要工作原理为：通过Reflector反射来监听Kubernetes对资源的操作事件，把资源对象和操作类型写入到一个DeltaFIFO的队列中。Reflector消费队列，将资源对象存储到indexer，indexer与Etcd集群的数据完全保持一致。 CRD资源的写操作 CRD资源的写通过client-go的clientset来完成对资源的Create，Update，Patch操作。 代码开发 这里我们需要实现上一篇讲的TrackingServer自定义资源的API。TrackingServer主要用于管理机器学习实验跟踪MLflow在k8s里的实例资源。 创建Informer informerFactories结构体实现了InformerFactory接口，这个接口有两个SharedInformerFactory，一个为Kubernetes资源的informerFactory，一个为本项目自定义资源的aiInformerFactory。aiscopeinformers.NewSharedInformerFactory创建了一个aiInformerFactory实例，这里的NewSharedInformerFactory为代码生成器生成的方法。通过代码生成器创建clientset客户端aiClient，作为参数来创建InformerFactory type InformerFactory interface { KubernetesSharedInformerFactory() k8sinformers.SharedInformerFactory AIScopeSharedInformerFactory() aiscopeinformers.SharedInformerFactory // Start shared informer factory one by one if they are not nil Start(stopCh <-chan struct{}) } type informerFactories struct { informerFactory k8sinformers.SharedInformerFactory aiInformerFactory aiscopeinformers.SharedInformerFactory } func NewInformerFactories(client kubernetes.Interface, aiClient versioned.Interface) InformerFactory { factory := &informerFactories{} if client != nil { factory.informerFactory = k8sinformers.

继续阅读

Kubebuilder开发MLflow实验跟踪控制器

由 SuKai in kubernetes, 机器学习

January 23, 2022

前面文章讲了client-go, go-restful开发Kubernetes应用平台，今天给大家看看在这个应用平台中添加一个自定义资源控制器的开发。 需求场景： 在多租户机器学习平台中，开发一个Kubernetes控制器，实现CRD(自定义资源) TrackingServer的调谐，完成Kubernetes中对应的PersistentVolumeClaim, TLS Secret, Service, Ingress, Deployment资源管理。 功能描述： 1，当CR实例的参数中指定了VolumeSize和StorageClassName，则创建对应的PersistentVolumeClaim用于MLflow的local database sqllite的数据存储目录。当未指定时，不创建或者删除已经创建的PersistentVolumeClaim。 2，当CR实例的参数中指定了Cert和Key数据，则创建对应的TLS类型的Secret，用于Ingress的TLS证书。当未指定时，不创建或删除对应Secret。 3，查找对应命名空间和名称的Secret，如果有Ingress配置对应的TLS证书。 4，根据CR实例的参数管理Service和Deployment的创建和修改。 5，删除CR实例后，对应清理K8S资源。当删除资源时，判断被删除资源是否为CR实例的附属资源。 代码实现： 整个业务代码开发分为几个大的步骤： 1，Kubebuilder生成代码和部署文件 2，在Controller Manager中注册控制器 3，在控制器调谐代码中，实现业务逻辑 Kubebuilder中创建API 指定GVK，这里TrackingServer为我们需要的MLflow资源。 kubebuilder create api --group experiment --version v1alpha2 --kind TrackingServer kubebuilder create api --group experiment --version v1alpha2 --kind JupyterNotebook kubebuilder create api --group experiment --version v1alpha2 --kind CodeServer 自定义资源TrackingServer定义 定义VolumeSize, Cert, Key字段为omitempty，表示非必须字段。 +genclient表示代码生成器生成clientset，informer, lister代码。 printcolumn表示kubectl get资源时展示字段 // TrackingServerSpec defines the desired state of TrackingServer type TrackingServerSpec struct { // INSERT ADDITIONAL SPEC FIELDS - desired state of cluster // Important: Run "make" to regenerate code after modifying this file Size int32 `json:"size"` Image string `json:"image"` S3_ENDPOINT_URL string `json:"s3_endpoint_url"` AWS_ACCESS_KEY string `json:"aws_access_key"` AWS_SECRET_KEY string `json:"aws_secret_key"` ARTIFACT_ROOT string `json:"artifact_root"` BACKEND_URI string `json:"backend_uri"` URL string `json:"url"` VolumeSize string `json:"volumeSize,omitempty"` StorageClassName string `json:"storageClassName,omitempty"` Cert string `json:"cert,omitempty"` Key string `json:"key,omitempty"` } // TrackingServerStatus defines the observed state of TrackingServer type TrackingServerStatus struct { // INSERT ADDITIONAL STATUS FIELD - define observed state of cluster // Important: Run "make" to regenerate code after modifying this file } // +genclient // +kubebuilder:object:root=true // +kubebuilder:subresource:status // +kubebuilder:printcolumn:name="S3_ENDPOINT_URL",type="string",JSONPath=".

继续阅读

搜索

分类

标签