# Solution - Operator ###### tags: `Assignment`, `Operator`, `Solution` ## Outline 這是[Assignment - Opertor](https://hackmd.io/Nb-PIgHdS4mgf_yaEgFsVg)的解答，是透過**client-go**實現，僅供參考。 [code](https://github.com/StarCoral/K8S_Study_Group_2020/tree/master/Operator) 將會存放於github 接下來我們會分為兩個部分來做介紹： + Part I ： Operator + Part II： the monitor pod program ( 對應於pm-{podname}的image) ## Note 將會用到的package放入$GOPATH/src ```bash= $ cd $GOPATH/src # 下載 $ wget https://lsalab.cs.nthu.edu.tw/~riya/assignment/podMonitorVendor.tar.gz # 解壓縮 $ tar -zxvf ./podMonitorVendor.tar.gz # 將vendor的檔案移出 $ mv ./vendor/* . ``` ## <font color = gray>Part I - Operator </font> Operator 開發主要可以分為兩個階段: 1. Custom Resource Definition 2. Logic implement ### Custom Resource Definition 首先我們先定義一個Custom Resource - **PodMonitor**。 透過CustomResourceDefinition，告訴k8s系統目前有一個新的resource ```bash= $ kubectl apply -f pm-crd.yaml ``` ```yaml apiVersion: apiextensions.k8s.io/v1beta1 kind: CustomResourceDefinition metadata: # name must match the spec fields below, and be in the form: <plural>.<group> name: podmonitors.lsalab.nthu spec: # group name to use for REST API: /apis/<group>/<version> group: lsalab.nthu # version supported by this CRD version: v1 # either Namespaced or Cluster scope: Namespaced names: # kind is normally the CamelCased singular type. Your resource manifests use this. kind: PodMonitor #upper camel case # plural name to be used in the URL: /apis/<group>/<version>/<plural> plural: podmonitors # singular name to be used as an alias on the CLI and for display singular: podmonitor # shortNames allow shorter string to match your resource on the CLI shortNames: - pm ``` 查看是否成功創立 ```bash= $ kubectl get crd ``` ![](https://i.imgur.com/0pcY4kk.png) 但目前為止，只是告知k8s系統有這樣一種資源存在， 但還不知道這種資源有哪些field，所以我們要透過程式碼將該資源註冊到k8s的API中，告訴etcd這個資源要以什麼樣的形式儲存。 接下來我們會在`$GOPATH/src/github.com/NTHU-LSALAB/podMonitor/`這個專案中，撰寫下列檔案： ```bash= src/github.com └── NTHU-LSALAB └── podMonitor ├── deploy │ ├── pm-crd.yaml │ └── test │ ├── pm-example.yaml │ ├── pod-test1.yaml │ └── pod-test2.yaml ├── hack │ └── k8s │ ├── boilerplate.go.txt │ ├── tools.go │ ├── update-codegen.sh (optional) │ └── verify-codegen.sh └── pkg └── apis └── podmonitor ├── register.go └── v1 ├── doc.go ├── register.go └── types.go ``` + pkg/apis/podmonitor/register.go ```go package podmonitor // GroupName is the group name used in this package const ( GroupName = "lsalab.nthu" ) ``` + pkg/apis/podmonitor/v1/doc.go 請根據自己的版本號作變更。 在這個檔案中我們可以看到在package上方，有一個Global tag， deepcopy-gen告訴我們預設為這個package的每個type創建deep-copy的methods。 ```go // +k8s:deepcopy-gen=package // +groupName=lsalab.nthu //Package v1 is the v1 version of the API package v1 // import "github.com/NTHU-LSALAB/podMonitor/pkg/apis/lsalab.nthu/v1" ``` :::warning 特別注意Code generator對於tag的擺放位置，有特殊的要求。 有兩種tags，分別為local tag、global tag。 一般local tag會放置於type declaration上方， 而global tag就如同這個檔案一般，置於package上方。 ::: + pkg/apis/podmonitor/v1/register.go 將Custom Resource加到Scheme中 ```go package v1 import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" ) // SchemeGroupVersion is group version used to register these objects var SchemeGroupVersion = schema.GroupVersion{Group: "lsalab.nthu", Version: "v1"} // Kind takes an unqualified kind and returns back a Group qualified GroupKind func Kind(kind string) schema.GroupKind { return SchemeGroupVersion.WithKind(kind).GroupKind() } // Resource takes an unqualified resource and return a Group qualified GroupResource func Resource(resource string) schema.GroupResource { return SchemeGroupVersion.WithResource(resource).GroupResource() } var( SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes) AddToScheme = SchemeBuilder.AddToScheme ) // Adds the list of known types to Scheme. func addKnownTypes(scheme *runtime.Scheme) error { scheme.AddKnownTypes(SchemeGroupVersion, &PodMonitor{}, &PodMonitorList{}, ) metav1.AddToGroupVersion(scheme,SchemeGroupVersion) return nil } ``` + pkg/apis/podmonitor/v1/types.go 依據需求，定義PodMonitor有哪些field。 如下圖所示，PodMonitorSpec的一些variable會對應yaml的field 因此我們可以想像到原先的yaml只是一個簡單的文字檔， 透過type的定義，我們會去處理讀進來的yaml，來實現對應的邏輯。 ![](https://i.imgur.com/keAYVUZ.png) ```go package v1 import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) // +genclient // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object type PodMonitor struct { metav1.TypeMeta `json:",inline"` metav1.ObjectMeta `json:"metadata,omitempty"` Spec PodMonitorSpec `json:"spec"` Status PodMonitorStatus `json:"status"` } type PodMonitorPhase string const ( PodMonitorNone PodMonitorPhase = "" PodMonitorCreating PodMonitorPhase = "Creating" PodMonitorPending PodMonitorPhase = "Pending" PodMonitorRunning PodMonitorPhase = "Running" PodMonitorTerminating PodMonitorPhase = "Terminating" PodMonitorCompleted PodMonitorPhase = "Completed" PodMonitorFailed PodMonitorPhase = "Failed" // PodMonitorUnknown PodMonitorPhase = "Unknown" ) type PodMonitorSpec struct { // the default second is 30 sec Speed int32 `json:"speed,omitempty"` // the file that store the information about resource usage LogDir string `json:"logdir,omitempty"` } type PodMonitorStatus struct { Phase PodMonitorPhase `json:"phase"` Reason string `json:"reason,omitempty"` } // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object type PodMonitorList struct { metav1.TypeMeta `json:",inline"` metav1.ListMeta `json:"metadata"` Items []PodMonitor `json:"items"` } ``` + hack/k8s/boilerplate.go.txt 這個檔是存放License, code generator產生的檔案會自動在產生的檔案最上方加入License ```go /* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ ``` + hack/k8s/update-codegen.sh 透過script方便我們使用code generator，用以產生clientset﹑informers﹑listers。 ```bash= #!/usr/bin/env bash GO111MODULE=off $GOPATH/src/k8s.io/code-generator/generate-groups.sh all \ github.com/NTHU-LSALAB/podMonitor/pkg/generated \ github.com/NTHU-LSALAB/podMonitor/pkg/apis \ podmonitor:v1 \ --go-header-file $GOPATH/src/github.com/NTHU-LSALAB/podMonitor/hack/k8s/boilerplate.go.txt ``` 執行`hack/k8s/update-codegen.sh` ```bash= $ chmod +x hack/k8s/update-codegen.sh $ chmod +x $GOPATH/src/k8s.io/code-generator/generate-groups.sh $ sh ./hack/k8s/update-codegen.sh ``` 執行完後目錄會如下所示。 ```bash= . ├── deploy │ ├── pm-crd.yaml │ └── test │ ├── pm-example.yaml │ ├── pod-test1.yaml │ └── pod-test2.yaml ├── go.mod ├── go.sum ├── hack │ └── k8s │ ├── boilerplate.go.txt │ ├── tools.go │ ├── update-codegen.sh │ └── verify-codegen.sh └── pkg ├── apis │ └── podmonitor │ ├── register.go │ └── v1 │ ├── doc.go │ ├── register.go │ ├── types.go │ └── zz_generated.deepcopy.go └── generated ├── clientset │ └── versioned │ ├── clientset.go │ ├── doc.go │ ├── fake │ │ ├── clientset_generated.go │ │ ├── doc.go │ │ └── register.go │ ├── scheme │ │ ├── doc.go │ │ └── register.go │ └── typed │ └── podmonitor │ └── v1 │ ├── doc.go │ ├── fake │ │ ├── doc.go │ │ ├── fake_podmonitor_client.go │ │ ├── fake_podmonitor.go │ │ └── fake_podmonitorphase.go │ ├── generated_expansion.go │ ├── podmonitor_client.go │ ├── podmonitor.go │ └── podmonitorphase.go ├── informers │ └── externalversions │ ├── factory.go │ ├── generic.go │ ├── internalinterfaces │ │ └── factory_interfaces.go │ └── podmonitor │ ├── interface.go │ └── v1 │ ├── interface.go │ └── podmonitor.go └── listers └── podmonitor └── v1 ├── expansion_generated.go └── podmonitor.go ``` ### Control Logic + main.go 此為程式的入口，透過下方的程式可以看出這裡是採用out-cluster的寫法， api server會驗證傳遞進來的kubeconfig與masterURL兩個參數。 接著會建立clientset與Informer，然後執行實際的logic。 ```go= package main import ( "flag" "time" kubeinformers "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/clientcmd" "k8s.io/klog" clientset "github.com/NTHU-LSALAB/podMonitor/pkg/generated/clientset/versioned" informers "github.com/NTHU-LSALAB/podMonitor/pkg/generated/informers/externalversions" "github.com/NTHU-LSALAB/podMonitor/pkg/signals" ) var ( masterURL string kubeconfig string ) func main() { klog.InitFlags(nil) flag.Parse() // set up signals so we handle the first shutdown signal gracefully stopCh := signals.SetupSignalHandler() cfg, err := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig) if err != nil { klog.Fatalf("Error building kubeconfig: %s", err.Error()) } // create the actual Kubernetes client set kubeClient, err := kubernetes.NewForConfig(cfg) if err != nil { klog.Fatalf("Error building kubernetes clientset: %s", err.Error()) } pmClient, err := clientset.NewForConfig(cfg) if err != nil { klog.Fatalf("Error building podMonitor clientset: %s",err.Error()) } // Informers are a combination of this event interface and an in-memory cache with indexed lookup. // NewSharedInformerFactory caches all objects of a resource in all namespaces in the store kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30) pmInformerFactory := informers.NewSharedInformerFactory(pmClient, time.Second*30) controller := NewController(kubeClient, pmClient, kubeInformerFactory.Core().V1().Pods(), pmInformerFactory.Lsalab().V1().PodMonitors()) // notice that there is no need to run Start method in a separate goroutine. // (i.e. go kubeInformerFactory.Start(stopCh)) // Start method is non-blocking and runs all registered informers in a dedicated goroutine. kubeInformerFactory.Start(stopCh) pmInformerFactory.Start(stopCh) if err = controller.Run(2, stopCh); err != nil { klog.Fatalf("Error running controller: %s", err.Error()) } } func init() { flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.") flag.StringVar(&masterURL, "master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.") } ``` + controller.go ``` go= package main import ( "fmt" "reflect" "strconv" "time" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" // "k8s.io/apimachinery/pkg/runtime/schema" utilruntime "k8s.io/apimachinery/pkg/util/runtime" "k8s.io/apimachinery/pkg/util/wait" corev1informer "k8s.io/client-go/informers/core/v1" "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" corev1lister "k8s.io/client-go/listers/core/v1" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/record" "k8s.io/client-go/util/workqueue" "k8s.io/klog" lsalabv1 "github.com/NTHU-LSALAB/podMonitor/pkg/apis/podmonitor/v1" clientset "github.com/NTHU-LSALAB/podMonitor/pkg/generated/clientset/versioned" pmscheme "github.com/NTHU-LSALAB/podMonitor/pkg/generated/clientset/versioned/scheme" informers "github.com/NTHU-LSALAB/podMonitor/pkg/generated/informers/externalversions/podmonitor/v1" listers "github.com/NTHU-LSALAB/podMonitor/pkg/generated/listers/podmonitor/v1" ) const controllerAgentName = "PodMonitorController" const ( PodMonitorLogPath = "/var/local" ) // Controller is the controller implementation for PodMonitor resources type Controller struct { kubeClientset kubernetes.Interface pmClientset clientset.Interface podLister corev1lister.PodLister podsSynced cache.InformerSynced pmLister listers.PodMonitorLister pmsSynced cache.InformerSynced // workqueue is a rate limited work queue. This is used to queue work to be // processed instead of performing it as soon as a change happens. This // means we can ensure we only process a fixed amount of resources at a // time, and makes it easy to ensure we are never processing the same item // simultaneously in two different workers. workqueue workqueue.RateLimitingInterface // recorder is an event recorder for recording Event resources to the // Kubernetes API. recorder record.EventRecorder } // NewController returns a new pm controller func NewController( kubeClientset kubernetes.Interface, pmClientset clientset.Interface, podInformer corev1informer.PodInformer, pmInformer informers.PodMonitorInformer) *Controller { // Create event broadcaster // Add PodMonitorController types to the default // kubernetes Scheme so Events can be logged for // PodMonitorController types. utilruntime.Must(pmscheme.AddToScheme(scheme.Scheme)) klog.V(4).Info("Creating event broadcaster") eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(klog.Infof) eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClientset.CoreV1().Events("")}) recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName}) controller := &Controller{ kubeClientset: kubeClientset, pmClientset: pmClientset, podLister: podInformer.Lister(), podsSynced: podInformer.Informer().HasSynced, pmLister: pmInformer.Lister(), pmsSynced: pmInformer.Informer().HasSynced, workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "PodMonitors"), recorder: recorder, } klog.Info("Setting up event handlers") // Set up an event handler for when PodMonitor resources change pmInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: controller.enqueuePodMonitor, UpdateFunc: func(old, new interface{}) { controller.enqueuePodMonitor(new) }, }) podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: controller.enqueuePod, UpdateFunc: func(old, new interface{}) { controller.enqueuePod(new) }, }) return controller } // Run will set up the event handlers for types we are interested in, as well // as syncing informer caches and starting workers. It will block untill stopCh // is closed, at which point it will shutdown the workqueue and wait for // workers to finish processing their current work items. func (c *Controller) Run(threadiness int, stopCh <- chan struct{}) error { defer utilruntime.HandleCrash() defer c.workqueue.ShutDown() // Start the informer factories to begin populating the informer caches klog.Info("Starting PodMonitor controller") // Wait for the caches to be synced before starting workers if ok := cache.WaitForCacheSync(stopCh, c.pmsSynced); !ok { return fmt.Errorf("faild to wait for caches to sync") } if ok := cache.WaitForCacheSync(stopCh, c.podsSynced); !ok { return fmt.Errorf("faild to wait for caches to sync") } klog.Info("Starting workers") // Launch two workers to process PodMonitor resources for i := 0 ; i < threadiness ; i++ { go wait.Until(c.runWorker, time.Second, stopCh) } klog.Info("Started workers") <-stopCh klog.Info("Shutting down workers") return nil } // runWorker is a long-running function that will continually call the // processNextWorkItem function in order to read and process a message on the // workqueue. func (c *Controller) runWorker() { for c.processNextWorkItem() { } } // processNextWorkItem will read a single work item off the workqueue and // attempt to process it, by calling the syncHandler. func (c *Controller) processNextWorkItem() bool { obj, shutdown := c.workqueue.Get() if shutdown{ return false } // We wrap this block in a func so we can defer c.workqueue.Done. err := func(obj interface{}) error { // We call Done here so the workqueue knows we have finished // processing this item. We also must remember to call Forget if we // do not want this work item being re-queued. For example, we do // not call Forget if a transient error occurs, instead the item is // put back on the workqueue and attempted again after a back-off // period. defer c.workqueue.Done(obj) var key string var ok bool // We expect strings to come off the workqueue. These are of the // form namespace/name. We do this as the delayed nature of the // workqueue means the items in the informer cache may actually be // more up to date that when the item was initially put onto the // workqueue. if key, ok = obj.(string) ; !ok { // As the item in the workqueue is actually invalid, we call // Forget here else we'd go into a loop of attempting to // process a work item that is invalid. c.workqueue.Forget(obj) utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj)) return nil } // Run the syncHandler, passing it the namespace/name string of the // PodMonitor resource to be synced. if err := c.syncHandler(key); err != nil{ // Put the item back on the workqueue to handle any transient errors. c.workqueue.AddRateLimited(key) return fmt.Errorf("error syncing '%s': %s,requeuing",key,err.Error()) } // Finally, if no error occurs we Forget this item so it does not // get queued again until another change happens. c.workqueue.Forget(obj) klog.Infof("Successfully Synced '%s'",key) return nil }(obj) if err != nil { utilruntime.HandleError(err) return true } return true } // syncHandler compares the actual state with the desired, and attempts to // converge the two. It then updates the Status block of the PodMonitor resource // with the current status of the resource. It returns how long to wait // until the schedule is due. func (c *Controller) syncHandler(key string) (error) { klog.Infof("------------- Reconciling Resource PodMonitor %s -------------", key) // Convert the namespace/name string into a distinct namespace and name namespace, name, err := cache.SplitMetaNamespaceKey(key) if err != nil { utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key)) return nil } // Get the PodMonitor resource with this namespace/name pm, err := c.pmLister.PodMonitors(namespace).Get(name) if err != nil { // The PodMonitor resource may no longer exist, in which case we stop // processing. if errors.IsNotFound(err) { utilruntime.HandleError(fmt.Errorf("pm '%s' in work queue no longer exists", key)) return nil } } pm_pod := pm.DeepCopy() switch pm.Status.Phase { case lsalabv1.PodMonitorNone: if err := c.changePhase(pm, lsalabv1.PodMonitorPending); err != nil { return err } case lsalabv1.PodMonitorPending, lsalabv1.PodMonitorFailed: klog.Infof("PodMonitor %s: phase=PENDING", key) //pod := createPodForCRD(pm) _, err := c.podLister.Pods(namespace).Get(name) if errors.IsNotFound(err) { klog.Info("Can't find the pod to monitor.") return err } if err := c.changePhase(pm, lsalabv1.PodMonitorRunning); err != nil { return err } case lsalabv1.PodMonitorRunning: klog.Infof("PodMonitor %s: phase=RUNNING", key) pod := createPodForCRD(pm) // Set the pm instance as the owner and controller owner := metav1.NewControllerRef(pm_pod, lsalabv1.SchemeGroupVersion.WithKind("PodMonitor")) pod.ObjectMeta.OwnerReferences = append(pod.ObjectMeta.OwnerReferences, *owner) found, err := c.kubeClientset.CoreV1().Pods(pod.Namespace).Get(pod.Name, metav1.GetOptions{}) if err != nil && errors.IsNotFound(err) { found, err = c.kubeClientset.CoreV1().Pods(pod.Namespace).Create(pod) if err != nil { return err } klog.Infof("PodMonitor %s: pod launched: name=%s", key, pod.Name) }else if err != nil { return err }else if found.Status.Phase == corev1.PodFailed || found.Status.Phase == corev1.PodSucceeded { klog.Infof("PodMonitor %s: container terminated: reason=%q message=%q", key, found.Status.Reason, found.Status.Message) pm_pod.Status.Phase = lsalabv1.PodMonitorCompleted } else { return nil } case lsalabv1.PodMonitorCompleted: klog.Infof("PodMonitor %s: phase=COMPLETED", key) return nil default: klog.Infof("PodMonitor %s: NOP", key) return nil } if !reflect.DeepEqual(pm, pm_pod) { _, err := c.pmClientset.LsalabV1().PodMonitors(pm_pod.Namespace).UpdateStatus(pm_pod) if err != nil { return err } } return nil } // enqueuePodMonitor takes a PodMonitor resource and converts it into a namespace/name // string which is then put onto the work queue. This method should *not* be // passed resources of any type other than PodMonitor. func (c *Controller) enqueuePodMonitor(obj interface{}) { var key string var err error if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil { utilruntime.HandleError(err) return } c.workqueue.Add(key) } func (c *Controller) enqueuePod(obj interface{}) { var pod *corev1.Pod var ok bool if pod, ok = obj.(*corev1.Pod); !ok { tombstone, ok := obj.(cache.DeletedFinalStateUnknown) if !ok { utilruntime.HandleError(fmt.Errorf("error decoding pod, invalid type")) return } pod, ok = tombstone.Obj.(*corev1.Pod) if !ok { utilruntime.HandleError(fmt.Errorf("error decoding pod tombstone, invalid type")) return } klog.V(4).Infof("Recovered deleted pod '%s' from tombstone", pod.GetName()) } if ownerRef := metav1.GetControllerOf(pod); ownerRef != nil { if ownerRef.Kind != "PodMonitor" { return } pm, err := c.pmLister.PodMonitors(pod.GetNamespace()).Get(ownerRef.Name) if err != nil { klog.V(4).Infof("ignoring orphaned pod '%s' of PodMonitor '%s'", pod.GetSelfLink(), ownerRef.Name) return } klog.Infof("enqueuing PodMonitor %s/%s because pod changed", pm.Namespace, pm.Name) c.enqueuePodMonitor(pm) } } func (c *Controller) changePhase(pm *lsalabv1.PodMonitor, phase lsalabv1.PodMonitorPhase) error { // Clone because the pm object is owned by the lister pmCopy := pm.DeepCopy() return c.updateStatus(pmCopy, phase, nil) } func (c *Controller) updateStatus(pm *lsalabv1.PodMonitor, phase lsalabv1.PodMonitorPhase, reason error) error { pm.Status.Reason = "" if reason != nil { pm.Status.Reason = reason.Error() } pm.Status.Phase = phase _, err := c.pmClientset.LsalabV1().PodMonitors(pm.Namespace).Update(pm) return err } func createPodForCRD(pm *lsalabv1.PodMonitor) *corev1.Pod { labels := map[string]string { "podmonitors.lsalab.nthu": pm.Name, "podmonitors": "true", } PodMonitorLogDir := pm.Spec.LogDir //var pod *corev1.Pod pod := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "podmonitor-log", Namespace: pm.Namespace, Labels: labels, }, Spec: corev1.PodSpec{ Containers: []corev1.Container{ { Name: "podmonitor", Image: "riyazhu/podmonitor:latest", // Name: "busybox", // Image: "busybox", }, }, RestartPolicy: corev1.RestartPolicyOnFailure, ServiceAccountName: "podmonitor-metricstest", }, } //klog.Info("There is pod template ", pod) podTem := pod.Spec.DeepCopy() for i := range podTem.Containers{ c := &podTem.Containers[i] c.Env = append( c.Env, corev1.EnvVar{ Name: "PODMONITOR_NAMESPACE", Value: pm.Namespace, }, corev1.EnvVar{ Name: "PODMONITOR_NAME", Value: pm.Name, }, corev1.EnvVar{ Name: "PODMONITOR_SPEED", Value: strconv.Itoa(int(pm.Spec.Speed)), }, corev1.EnvVar{ Name: "PODMONITOR_LOGDIR", Value: pm.Spec.LogDir, }, ) c.VolumeMounts = append(c.VolumeMounts, corev1.VolumeMount{ Name: "podmonitor-log", MountPath: PodMonitorLogDir, }, ) } podTem.Volumes = append(podTem.Volumes, corev1.Volume{ Name: "podmonitor-log", VolumeSource: corev1.VolumeSource{ HostPath: &corev1.HostPathVolumeSource{ Path: PodMonitorLogPath, }, }, }, ) return &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "pm-"+pm.Name, Namespace: pm.Namespace, Labels: labels, }, Spec: *podTem, } } ``` ### Build & Deploy + 編譯成可執行檔 可以直接使用這個執行檔測試，也可透過pod的方式運行 ```bash $ go build -o podMonitor . ``` + 編譯成image - Dockerfile ```shell= FROM golang:1.14.7 AS build-image ADD . /go/src/github.com/NTHU-LSALAB/podMonitor WORKDIR /go/src/github.com/NTHU-LSALAB/podMonitor RUN if [ "$(uname -m)" = "aarch64" ]; then \ CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o podMonitor .; \ else \ CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o podMonitor .; \ fi FROM alpine:3.9 COPY --from=build-image /go/src/github.com/NTHU-LSALAB/podMonitor/podMonitor /usr/bin/podMonitor ENTRYPOINT ["podMonitor"] ``` + 設定權限&部署 ```bash= $ kubectl apply -f deploy.yaml ``` ```yaml= apiVersion: v1 kind: ServiceAccount metadata: name: podmonitor-operator namespace: kube-system --- kind: ClusterRole apiVersion: rbac.authorization.k8s.io/v1 metadata: name: podmonitor-operator rules: - apiGroups: [""] resources: ["*"] verbs: ["*"] - apiGroups: ["lsalab.nthu"] resources: ["*"] verbs: ["*"] --- kind: ClusterRoleBinding apiVersion: rbac.authorization.k8s.io/v1 metadata: name: podmonitor-operator subjects: - kind: ServiceAccount name: podmonitor-operator namespace: kube-system roleRef: kind: ClusterRole name: podmonitor-operator apiGroup: rbac.authorization.k8s.io --- apiVersion: v1 kind: Pod metadata: name: podmonitor-operator namespace: kube-system labels: app: podmonitor-operator spec: serviceAccountName: podmonitor-operator containers: - name: podmonitor-operator image: riyazhu/podmonitor-operator:latest ``` ## <font color = gray>Part II</font> ### 主程式 那這個部分將要另外實現pm-{podname}的程式，透過這個程式蒐集欲觀察的Pod中所有container的CPU與Memory使用情況，並且會將其製作成image。 而這個程式的話，則是使用in-cluster的方式撰寫，因為是透過pod來運行。 ```go package main import ( "fmt" "os" "path/filepath" "time" "strconv" // "k8s.io/client-go/tools/clientcmd" // corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/klog" metrics "k8s.io/metrics/pkg/client/clientset/versioned" ) var ( podmonitor_namespace string = "default" podmonitor_name string podmonitor_logdir string = "./log" podmonitor_speed int32 = 30 podmonitor_logpath string ) func init() { podmonitor_namespace = os.Getenv("PODMONITOR_NAMESPACE") podmonitor_name = os.Getenv("PODMONITOR_NAME") podmonitor_logdir = os.Getenv("PODMONITOR_LOGDIR") speed_string := os.Getenv("PODMONITOR_SPEED") ps, err := strconv.ParseInt(speed_string, 10, 64) podmonitor_speed = int32(ps) if err != nil { klog.Info("podmonitor_speed change error, to be default 30 sec.") } if podmonitor_name == "" { klog.Fatalf("The Environmental variables are losts") } podmonitor_logpath = filepath.Join(podmonitor_logdir,"/",podmonitor_name) klog.Info("===Setting up Environmental variables ===") klog.Info("Pod is ",podmonitor_namespace," ",podmonitor_name) klog.Info("Log file will store in ",podmonitor_logpath) klog.Info("Sampling speed: ", podmonitor_speed, " sec.") } func main() { // create the file os.Mkdir(podmonitor_logdir, 0755) file, err := os.OpenFile(podmonitor_logpath, os.O_WRONLY|os.O_APPEND|os.O_CREATE,0600) if err != nil { fmt.Println("Can't open the file") } defer file.Close() msg := fmt.Sprintf("============ %s / %s ============ \n", podmonitor_namespace, podmonitor_name) klog.Info(msg) file.WriteString(msg) // klog.Info("============", podmonitor_namespace, " / ", podmonitor_name, "============") // get the config by the variables (KUBERNETES_SERVICE_HOST KUBERNETES_SERVICE_PORT) config, err := rest.InClusterConfig() if err != nil { klog.Fatalf("Can't get cluster config: %s", err.Error()) } mc,err := metrics.NewForConfig(config) if err != nil { panic(err.Error()) } clientset, err := kubernetes.NewForConfig(config) if err != nil { panic(err.Error()) } for { pod, err := clientset.CoreV1().Pods(podmonitor_namespace).Get(podmonitor_name, metav1.GetOptions{}) if err != nil { klog.Info(err) continue } msg := fmt.Sprintf("Pod status: %s\n", pod.Status.Phase) // klog.Info(" \t Pod status: ", pod.Status.Phase) klog.Info(msg) file.WriteString(msg) if pod.Status.Phase =="Succeeded" || pod.Status.Phase =="Failed"{ klog.Info("Done.") file.WriteString("Done.\n") break } podMetrics, err := mc.MetricsV1beta1().PodMetricses(podmonitor_namespace).Get(podmonitor_name, metav1.GetOptions{}) if err != nil { klog.Info(err) } for _, container := range podMetrics.Containers{ cpuQuantity, ok := container.Usage.Cpu().AsInt64() memQuantity, ok := container.Usage.Memory().AsInt64() if !ok{ return } msg := fmt.Sprintf("Container Name: %s \t CPU(cores): %d \t MEMORY(bytes): %d", container.Name, cpuQuantity, memQuantity) klog.Infof(msg) // klog.Infof("Container Name: %s \t CPU usage: %d \t Memory usage: %d", container.Name, cpuQuantity, memQuantity) file.WriteString(msg) } time.Sleep(time.Duration(podmonitor_speed)*time.Second) } } ``` ### 設定權限&部署 + 授權 因為我們需要與api溝通，因此需要賦予對應的權限 ```yaml= apiVersion: v1 kind: ServiceAccount metadata: name: podmonitor-metricstest namespace: default --- kind: ClusterRole apiVersion: rbac.authorization.k8s.io/v1 metadata: namespace: default name: podmonitor-metricstest rules: - apiGroups: [""] resources: ["*"] verbs: ["*"] - apiGroups: ["lsalab.nthu"] resources: ["*"] verbs: ["*"] - apiGroups: ["metrics.k8s.io"] resources: ["*"] verbs: ["*"] --- kind: ClusterRoleBinding apiVersion: rbac.authorization.k8s.io/v1 metadata: name: podmonitor-metricstest subjects: - kind: ServiceAccount name: podmonitor-metricstest namespace: default roleRef: kind: ClusterRole name: podmonitor-metricstest apiGroup: rbac.authorization.k8s.io ``` + Dockerfile ```shell= FROM debian COPY ./metric-test /metric-test ENTRYPOINT /metric-test ``` ## Refernece http://www.ruanyifeng.com/blog/2017/11/bash-set.html https://itnext.io/how-to-generate-client-codes-for-kubernetes-custom-resource-definitions-crd-b4b9907769ba