CRD是一个内建的API, 它提供了一个简单的方式来创建自定义资源。 部署一个CRD到集群中使Kubernetes API服务端开始为你指定的自定义资源服务。
这使你不必再编写自己的API服务端来处理自定义资源，但是这种实现的一般性意味着比你使用API server aggregation缺乏灵活性。
如果只是想添加资源到集群，可以考虑使用 customer resource define，简称CRD，CRD需要更少的编码和重用， 在这里阅读更多有关自定义资源和扩展api之间的差异
cat > EOF | kubectl create -f
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: podtoips.example.com
spec:
group: example.com
version: v1
scope: Namespaced
names:
plural: podtoips
singular: podtoip
kind: Podtoip
shortNames:
- ptp
EOF
controller 一般会有 1 个或者多个 informer，来跟踪某一个 resource，
跟 APIserver 保持通讯，把最新的状态反映到本地的 cache 中。
只要这些资源有变化，informal 会调用 callback。
这些 callbacks 只是做一些非常简单的预处理，把不关心的的变化过滤掉，
然后把关心的变更的 Object 放到 workqueue 里面。
其实真正的 business logic 都是在 worker 里面，
一般 1 个 Controller 会启动很多 goroutines 跑 Workers，
处理 workqueue 里的 items。它会计算用户想要达到的状态和当前的状态有多大的区别，
然后通过 clients 向 APIserver 发送请求，来驱动这个集群向用户要求的状态演化。
图里面蓝色的是 client-go 的原件，红色是自己写 controller 时填的代码
下面是我实现的一个简单的 poptoip controller
package main
import (
"flag"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/runtime"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/tools/cache"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/client-go/util/workqueue"
"time"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/fields"
"fmt"
"k8s.io/apimachinery/pkg/util/wait"
"log"
"k8s.io/client-go/kubernetes/scheme"
)
func init(){
log.SetFlags(log.Lshortfile)
}
func main(){
kubeconfig := flag.String("kubeconfig", "./kubeconfig", "Path to a kube config. Only required if out-of-cluster.")
flag.Parse()
pic:=newPodipcontroller(*kubeconfig)
var stopCh <-chan struct{}
pic.Run(2, stopCh)
}
type Podipcontroller struct {
kubeClient *kubernetes.Clientset
crdclient *Podipclient
podStore cache.Store
//cache.AppendFunc
podController cache.Controller
podtoip *PodToIp
podsQueue workqueue.RateLimitingInterface
}
func (p2p *Podipcontroller) Run(workers int, stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
fmt.Println("Starting Controller")
//p2p.registerTPR()
go p2p.podController.Run(stopCh)
//go p2p.endpointController.Run(stopCh)
// wait for the controller to List. This help avoid churns during start up.
if !cache.WaitForCacheSync(stopCh, p2p.podController.HasSynced) {
return
}
for i := 0; i < workers; i++ {
go wait.Until(p2p.podWorker, time.Second, stopCh)
}
<-stopCh
fmt.Printf("Shutting down Controller")
p2p.podsQueue.ShutDown()
}
func (p2p *Podipcontroller) podWorker() {
workFunc := func() bool {
key, quit := p2p.podsQueue.Get()
log.Println(key)
if quit {
return true
}
defer p2p.podsQueue.Done(key)
p2p.podStore.Resync()
obj, exists, err := p2p.podStore.GetByKey(key.(string))
log.Printf("%#v",obj)
if !exists {
fmt.Printf("Pod has been deleted %v\n", key)
return false
}
if err != nil {
fmt.Printf("cannot get pod: %v\n", key)
return false
}
pod := obj.(*v1.Pod)
if pod.DeletionTimestamp!=nil{
log.Println(p2p.crdclient.Delete(pod.Name,pod.Namespace))
return false
}
log.Println(p2p.crdclient.Create(&PodToIp{
Metadata: metav1.ObjectMeta{
Name: pod.ObjectMeta.Name,
Namespace: pod.Namespace,
},
PodName: pod.ObjectMeta.Name,
PodAddress: pod.Status.PodIP,
}))
return false
}
for {
if quit := workFunc(); quit {
fmt.Printf("pod worker shutting down")
return
}
}
}
func newPodipcontroller(kubeconfig string) *Podipcontroller{
p2p:=&Podipcontroller{
kubeClient:getClientsetOrDie(kubeconfig),
crdclient: getCRDClientOrDie(kubeconfig),
podsQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pods"),
}
watchList:=cache.NewListWatchFromClient(p2p.kubeClient.CoreV1().RESTClient(),"pods",v1.NamespaceAll,fields.Everything())
p2p.podStore,p2p.podController=cache.NewInformer(
watchList,
&v1.Pod{},
time.Second*30,
cache.ResourceEventHandlerFuncs{
AddFunc: p2p.enqueuePod,
UpdateFunc: p2p.updatePod,
},
)
return p2p
}
func (p2p *Podipcontroller) enqueuePod(obj interface{}) {
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
if err != nil {
fmt.Printf("Couldn't get key for object %+v: %v", obj, err)
return
}
p2p.podsQueue.Add(key)
}
func (p2p *Podipcontroller) updatePod(oldObj, newObj interface{}) {
oldPod := oldObj.(*v1.Pod)
newPod := newObj.(*v1.Pod)
if newPod.Status.PodIP == oldPod.Status.PodIP {
return
}
p2p.enqueuePod(newObj)
}
type Podipclient struct {
rest *rest.RESTClient
}
type PodToIp struct {
metav1.TypeMeta `json:",inline"`
Metadata metav1.ObjectMeta `json:"metadata"`
PodName string `json:"podName"`
PodAddress string `json:"podAddress"`
}
func getClientsetOrDie(kubeconfig string) *kubernetes.Clientset {
// Create the client config. Use kubeconfig if given, otherwise assume in-cluster.
config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
if err != nil {
panic(err)
}
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err)
}
//clientset.CoreV1().Pods("111").Get("",nil)
return clientset
}
func (c *Podipclient) Create(body *PodToIp) (*PodToIp, error) {
var ret PodToIp
err := c.rest.Post().
Resource("podtoips").
Namespace(body.Metadata.Namespace).
Body(body).
Do().Into(&ret)
return &ret, err
}
func (c *Podipclient) Update(body *PodToIp) (*PodToIp, error) {
var ret PodToIp
err := c.rest.Put().
Resource("podtoips").
Namespace(body.Metadata.Namespace).
Name(body.Metadata.Name).
Body(body).
Do().Into(&ret)
return &ret, err
}
func (c *Podipclient) Get(name string,namespace string) (*PodToIp, error) {
var ret PodToIp
err := c.rest.Get().
Resource("podtoips").
Namespace(namespace).
Name(name).
Do().Into(&ret)
return &ret, err
}
func (c *Podipclient) Delete(name string,namespace string) (*PodToIp, error) {
var ret PodToIp
err := c.rest.Delete().
Resource("podtoips").
Namespace(namespace).
Name(name).
Do().Into(&ret)
return &ret, err
}
func getCRDClientOrDie(kubeconfig string) *Podipclient {
config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
if err != nil {
panic(err)
}
configureClient(config)
rest, err := rest.RESTClientFor(config)
if err != nil {
panic(err)
}
return &Podipclient{rest}
}
func configureClient(config *rest.Config) {
groupversion := schema.GroupVersion{
Group: "example.com",
Version: "v1",
}
config.GroupVersion = &groupversion
config.APIPath = "/apis"
// Currently TPR only supports JSON
config.ContentType = runtime.ContentTypeJSON
config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: scheme.Codecs}
schemeBuilder := runtime.NewSchemeBuilder(
func(scheme *runtime.Scheme) error {
scheme.AddKnownTypes(
groupversion,
&PodToIp{},
&PodToIpList{},
&metav1.ListOptions{},
&metav1.DeleteOptions{},
)
return nil
})
schemeBuilder.AddToScheme(scheme.Scheme)
}
type PodToIpList struct {
metav1.TypeMeta `json:",inline"`
Metadata metav1.ListMeta `json:"metadata"`
Items []PodToIp `json:"items"`
}
func (in *PodToIp) DeepCopy() *PodToIp {
if in == nil {
return nil
}
out := new(PodToIp)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *PodToIp) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
} else {
return nil
}
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *PodToIp) DeepCopyInto(out *PodToIp) {
*out = *in
return
}
func (in *PodToIpList) DeepCopy() *PodToIpList {
if in == nil {
return nil
}
out := new(PodToIpList)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *PodToIpList) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
} else {
return nil
}
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *PodToIpList) DeepCopyInto(out *PodToIpList) {
*out = *in
return
}