Go语⾔中使⽤K8sAPI及⼀些常⽤API整理
Go Client
在进⼊代码之前，理解k8s的go client项⽬是对我们⼜帮助的。它是k8s client中最古⽼的⼀个，因此具有很多特性。 Client-go 没有使⽤Swagger⽣成器，就像前⾯我们介绍的openAPI⼀样。它使⽤的是源于k8s项⽬中的源代码⽣成⼯具，这个⼯具的⽬的是要⽣成k8s风格的对象和序列化程序。
该项⽬是⼀组包的集合，该包能够满⾜从REST风格的原语到复杂client的不同的编程需求。
RESTClient是⼀个基础包，它使⽤api-machinery库中的类型作为⼀组REST原语提供对API的访问。作为对RESTClient之上的抽象，_clientt_将是你创建k8s client⼯具的起点。它暴露了公开化的API资源及其对应的序列化。
注意：在 client-go中还包含了如discovery, dynamic, 和 scale这样的包，虽然本次不介绍这些包，但是了解它们的能⼒还是很重要的。
⼀个简单的k8s client⼯具
让我们再次回顾我们将要构建的⼯具，来说明go client的⽤法。pvcwatch是⼀个简单的命令⾏⼯具，它可以监听集群中声明的PVC容量。当总数到达⼀个阈值的时候，他会采取⼀个action（在这个例⼦中是在屏幕上通知显⽰）
这个例⼦是为了展⽰k8s的go client的以下⼏个⽅⾯： - 如何去连接 - 资源列表的检索和遍历 - 对象监听
连接API Server
我们Go client的第⼀步就是建⽴⼀个于API Server的连接。为了做到这⼀点，我们要使⽤实体包中的clientcmd，如下代码所⽰：
import (
...
"k8s.io/client-go/tools/clientcmd"
)
func main() {
kubeconfig := filepath.Join(
os.Getenv("HOME"), ".kube", "config",
)
config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
if err != nil {
log.Fatal(err)
}
...
}
_Client-go_通过提供实体功能来从不同的上下⽂中获取你的配置，从⽽使之成为⼀个不重要的任务。
从config⽂件
正如上⾯的例⼦所做的那样，你能从kubeconfig⽂件启动配置来连接API rver。当你的代码运⾏在集群之外的时候这是⼀个理想的⽅案。 clientcmd.BuildConfigFromFlags("", configFile)
从集群
当你的代码运⾏在这个集群中的时候，你可以⽤上⾯的函数并且不使⽤任何参数，这个函数就会通过集群的信息去连接api rver。
clientcmd.BuildConfigFromFlags("", "")
或者我们可以通过rest包来创建⼀个使⽤集群中的信息去配置启动的(译者注：k8s⾥所有的Pod都会以Volume的⽅式⾃动挂载k8s⾥⾯默认的ServiceAccount，所以会实⽤默认的ServiceAccount的授权信息)，如下：
import "k8s.io/client-go/rest"
...
rest.InClusterConfig()
创建⼀个clientt
我们需要创建⼀个序列化的client为了让我们获取API对象。在kubernetes包中的Clientt类型定义，提供了去访问公开的API对象的序列化client，如下：
type Clientt struct {
*authenticationv1beta1.AuthenticationV1beta1Client
*authorizationv1.AuthorizationV1Client
...
*corev1.CoreV1Client
}
⼀旦我们有正确的配置连接，我们就能使⽤这个配置去初始化⼀个clientt，如下：
func main() {
config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
...
clientt, err := kubernetes.NewForConfig(config)
if err != nil {
log.Fatal(err)
}
}
对于我们的例⼦，我们使⽤的是v1的API对象。下⼀步，我们要使⽤clientt通过CoreV1()去访问核⼼api资源，如下：
func main() {
...
clientt, err := kubernetes.NewForConfig(config)
if err != nil {
log.Fatal(err)
}
api := clientt.CoreV1()
}
获取集群的PVC列表
我们对clientt执⾏的最基本操作之⼀获取存储的API对象的列表。在我们的例⼦中，我们将要拿到⼀个namespace下⾯的pvc列表，如下：
import (
...
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
func main() {
var ns, label, field string
flag.StringVar(&ns, "namespace", "", "namespace")
flag.StringVar(&label, "l", "", "Label lector")
flag.StringVar(&field, "f", "", "Field lector")
...
api := clientt.CoreV1()
// tup list options
listOptions := metav1.ListOptions{
LabelSelector: label,
FieldSelector: field,
}
pvcs, err := api.PersistentVolumeClaims(ns).List(listOptions)
if err != nil {
log.Fatal(err)
}
printPVCs(pvcs)
...
}
在上⾯的代码中，我们使⽤ListOptions指定 label 和 field lectors （还有namespace）来缩⼩pvc列表的范围，这个结果的返回类型是v1.PeristentVolumeClaimList。下⾯的这个代码展⽰了我们如何去遍历和打印从api rver中获取的pvc列表。
func printPVCs(pvcs *v1.PersistentVolumeClaimList) {
template := "%-32s%-8s%-8s\n"
fmt.Printf(template, "NAME", "STATUS", "CAPACITY")
for _, pvc := range pvcs.Items {
quant := pvc.Spec.Resources.Requests[v1.ResourceStorage]
fmt.Printf(
template,
pvc.Name,
string(pvc.Status.Pha),
quant.String())
}
}
监听集群中pvc
k8s的Go client框架⽀持为指定的API对象在其⽣命周期事件中监听集群的能⼒，包括创建，更新，删除⼀个指定对象时候触发的CREATED,MODIFIED,DELETED事件。对于我们的命令⾏⼯具，我们将要监听在集群中已经声明的PVC的总量。
对于某⼀个namespace，当pvc的容量到达了某⼀个阈值（⽐如说200Gi），我们将会采取某个动作。为了简单起见，我们将要在屏幕上打印个通知。但是在更复杂的实现中，可以使⽤相同的办法触发⼀个⾃动操作。
启动监听功能
现在让我们为PersistentVolumeClaim这个资源通过Watch去创建⼀个监听器。然后这个监听器通过ResultChan从go的channel中访问事件通知。
func main() {
...
api := clientt.CoreV1()
listOptions := metav1.ListOptions{
LabelSelector: label,
FieldSelector: field,
}
watcher, err :=api.PersistentVolumeClaims(ns).
Watch(listOptions)
if err != nil {
log.Fatal(err)
}
ch := watcher.ResultChan()
...
}
循环事件
接下来我们将要处理资源事件。但是在我们处理事件之前，我们先声明resource.Quantity类型的的两个变量为maxClaimsQuant和totalClaimQuant来分别表⽰我们的申请资源阈值（译者注：代表某个ns下集群中运⾏的PVC申请的上限）和运⾏总数。
import(
"k8s.io/apimachinery/pkg/api/resource"
...
)
func main() {
var maxClaims string
flag.StringVar(&maxClaims, "max-claims", "200Gi",
"Maximum total claims to watch")
var totalClaimedQuant resource.Quantity
maxClaimedQuant := resource.MustPar(maxClaims)
...
ch := watcher.ResultChan()
for event := range ch {
pvc, ok := event.Object.(*v1.PersistentVolumeClaim)
if !ok {
log.Fatal("unexpected type")
}
...
}
}
在上⾯的for-range循环中，watcher的channel⽤于处理来⾃服务器传⼊的通知。每个事件赋值给变量event，并且event.Object的类型被声明为PersistentVolumeClaim类型，所以我们能从中提取出来。
处理ADDED事件
当⼀个新的PVC创建的时候，event.Type的值被设置为watch.Added。然后我们⽤下⾯的代码去获取新增的声明的容量（quant），将其添加到正在运⾏的总容量中（totalClaimedQuant）。最后我们去检查是否当前的容量总值⼤于当初设定的最⼤值(maxClaimedQuant)，如果⼤于的话我们就触发⼀个事件。
import(
"k8s.io/apimachinery/pkg/watch"
...
)
func main() {
...
for event := range ch {
pvc, ok := event.Object.(*v1.PersistentVolumeClaim)
if !ok {
log.Fatal("unexpected type")
}
quant := pvc.Spec.Resources.Requests[v1.ResourceStorage]
switch event.Type {
ca watch.Added:
totalClaimedQuant.Add(quant)
log.Printf("PVC %s added, claim size %s\n",
pvc.Name, quant.String())
if totalClaimedQuant.Cmp(maxClaimedQuant) == 1 {
log.Printf(
"\nClaim overage reached: max %s at %s",
maxClaimedQuant.String(),
totalClaimedQuant.String())
// trigger action
log.Println("*** Taking action ***")
}
}
...
}
}
}
处理DELETED事件
代码也会在PVC被删除的时候做出反应，它执⾏相反的逻辑以及把被删除的这个PVC申请的容量在正在运⾏的容量的总值⾥⾯减去。
func main() {
...
for event := range ch {
...
switch event.Type {
ca watch.Deleted:
quant := pvc.Spec.Resources.Requests[v1.ResourceStorage]
totalClaimedQuant.Sub(quant)
log.Printf("PVC %s removed, size %s\n",
pvc.Name, quant.String())
if totalClaimedQuant.Cmp(maxClaimedQuant) <= 0 {
log.Printf("Claim usage normal: max %s at %s",
maxClaimedQuant.String(),
totalClaimedQuant.String(),
)
// trigger action
log.Println("*** Taking action ***")
}
}
...
}
}
运⾏程序
当程序在⼀个运⾏中的集群被执⾏的时候，⾸先会列出PVC的列表。然后开始监听集群中新的PersistentVolumeClaim事件。
$> ./pvcwatch
Using kubeconfig: /Urs/vladimir/.kube/config
--- PVCs ----
NAME STATUS CAPACITY
my-redis-redis Bound 50Gi
my-redis2-redis Bound 100Gi
-----------------------------
Total capacity claimed: 150Gi
-----------------------------
--- PVC Watch (max claims 200Gi) ----
2018/02/13 21:55:03 PVC my-redis2-redis added, claim size 100Gi
2018/02/13 21:55:03
At 50.0% claim capcity (100Gi/200Gi)
2018/02/13 21:55:03 PVC my-redis-redis added, claim size 50Gi
2018/02/13 21:55:03
At 75.0% claim capcity (150Gi/200Gi)
下⾯让我们部署⼀个应⽤到集群中，这个应⽤会申请75Gi容量的存储。（例如，让我们通过去部署⼀个实例）。
helm install --name my-influx \
--abled=true,persistence.size=75Gi stable/influxdb
正如下⾯你看到的，我们的⼯具⽴刻反应出来有个新的声明以及⼀个警告因为当前的运⾏的声明总量已经⼤于我们设定的阈值。
--- PVC Watch (max claims 200Gi) ----
...
2018/02/13 21:55:03
At 75.0% claim capcity (150Gi/200Gi)
2018/02/13 22:01:29 PVC my-influx-influxdb added, claim size 75Gi
2018/02/13 22:01:29
Claim overage reached: max 200Gi at 225Gi
2018/02/13 22:01:29 *** Taking action ***
2018/02/13 22:01:29
At 112.5% claim capcity (225Gi/200Gi)
相反，从集群中删除⼀个PVC的时候，该⼯具会相应展⽰提⽰信息。
...
At 112.5% claim capcity (225Gi/200Gi)
2018/02/14 11:30:36 PVC my-redis2-redis removed, size 100Gi
2018/02/14 11:30:36 Claim usage normal: max 200Gi at 125Gi
2018/02/14 11:30:36 *** Taking action ***
client-go常⽤api
Example1
import (
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/kubernetes"
appsv1beta1 "k8s.io/api/apps/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
apiv1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes/typed/apps/v1beta1"
"flag"
"fmt"
"encoding/json"
)
func main() {
//kubelet.kubeconfig 是⽂件对应地址
kubeconfig := flag.String("kubeconfig", "kubelet.kubeconfig", "(optional) absolute path to the kubeconfig file")
flag.Par()
// 解析到config
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
if err != nil {
panic(err.Error())
}
// 创建连接
clientt, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err.Error())
}
deploymentsClient := clientt.AppsV1beta1().Deployments(apiv1.NamespaceDefault)
//创建deployment
go createDeployment(deploymentsClient)
//监听deployment
startWatchDeployment(deploymentsClient)
}
/
/监听Deployment变化
func startWatchDeployment(deploymentsClient v1beta1.DeploymentInterface) {
w, _ := deploymentsClient.Watch(metav1.ListOptions{})
for {
lect {
ca e, _ := <-w.ResultChan():
fmt.Println(e.Type, e.Object)
}
}
}
//创建deployemnt，需要谨慎按照部署的k8s版本来使⽤api接⼝
func createDeployment(deploymentsClient v1beta1.DeploymentInterface) {
var r apiv1.ResourceRequirements
//资源分配会遇到⽆法设置值的问题，故采⽤json反解析
j := `{"limits": {"cpu":"2000m", "memory": "1Gi"}, "requests": {"cpu":"2000m", "memory": "1Gi"}}`
json.Unmarshal([]byte(j), &r)
deployment := &appsv1beta1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: "engine",
Labels: map[string]string{
"app": "engine",
},
},
Spec: appsv1beta1.DeploymentSpec{
Replicas: int32Ptr2(1),
Template: apiv1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"app": "engine",
},
},
Spec: apiv1.PodSpec{
Containers: []apiv1.Container{
{ Name: "engine",
Image: "my.com/engine:v2",
Resources: r,
},
},
},
},
},
}
fmt.Println("")
result, err := deploymentsClient.Create(deployment)
if err != nil {
panic(err)
}
fmt.Printf("Created deployment %q.\n", result.GetObjectMeta().GetName())
}
func int32Ptr2(i int32) *int32 { return &i }
Example2
package main
import (
"flag"
"fmt"
"k8s.io/client-go/1.4/kubernetes"
"k8s.io/client-go/1.4/pkg/api"
"k8s.io/client-go/1.4/pkg/api/unversioned"
"k8s.io/client-go/1.4/pkg/api/v1"
"k8s.io/client-go/1.4/tools/clientcmd"
"log"
)
var (
kubeconfig = flag.String("kubeconfig", "./config", "absolute path to the kubeconfig file")
)
func main() {
flag.Par()
// us the current context in kubeconfig
config, err := clientcmd.BuildConfigFromFlags("116.213.205.180:8080", *kubeconfig)
if err != nil {
panic(err.Error())
}
// creates the clientt
clientt, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err.Error())
}
/
/ 创建pod
pod := new(v1.Pod)
pod.TypeMeta = unversioned.TypeMeta{Kind: "Pod", APIVersion: "v1"}
pod.ObjectMeta = v1.ObjectMeta{Name: "testapi", Namespace: "default", Labels: map[string]string{"name": "testapi"}} pod.Spec = v1.PodSpec{
RestartPolicy: v1.RestartPolicyAlways,
Containers: []v1.Container{
v1.Container{
Name: "testapi",
Image: "nginx",
Ports: []v1.ContainerPort{
v1.ContainerPort{
ContainerPort: 80,
Protocol: v1.ProtocolTCP,
},
},
},
},
