Created
April 29, 2022 13:12
-
-
Save able8/c809f6daab711e901c882f6983c0b6ed to your computer and use it in GitHub Desktop.
A Basic Controller Sample Code
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
package main | |
import ( | |
"flag" | |
"fmt" | |
"path/filepath" | |
"time" | |
"k8s.io/klog/v2" | |
_ "k8s.io/client-go/plugin/pkg/client/auth" | |
v1 "k8s.io/api/core/v1" | |
"k8s.io/apimachinery/pkg/util/runtime" | |
"k8s.io/apimachinery/pkg/util/wait" | |
"k8s.io/client-go/informers" | |
"k8s.io/client-go/kubernetes" | |
"k8s.io/client-go/tools/cache" | |
"k8s.io/client-go/tools/clientcmd" | |
"k8s.io/client-go/util/homedir" | |
"k8s.io/client-go/util/workqueue" | |
) | |
// Controller demonstrates how to implement a controller with client-go. | |
type Controller struct { | |
clientset kubernetes.Interface | |
podInformer cache.SharedIndexInformer | |
queue workqueue.RateLimitingInterface | |
} | |
// NewController creates a new Controller. | |
func NewController(clientset kubernetes.Interface) *Controller { | |
informerFactory := informers.NewSharedInformerFactory(clientset, 0) | |
podInformer := informerFactory.Core().V1().Pods().Informer() | |
queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()) | |
// register event handlers to fill the queue with pod creations, updates and deletions | |
podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ | |
AddFunc: func(obj interface{}) { | |
key, err := cache.MetaNamespaceKeyFunc(obj) | |
if err == nil { | |
queue.Add(key) | |
} | |
}, | |
UpdateFunc: func(old interface{}, new interface{}) { | |
key, err := cache.MetaNamespaceKeyFunc(new) | |
if err == nil { | |
queue.Add(key) | |
} | |
}, | |
DeleteFunc: func(obj interface{}) { | |
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj) | |
if err == nil { | |
queue.Add(key) | |
} | |
}, | |
}) | |
return &Controller{ | |
clientset: clientset, | |
podInformer: podInformer, | |
queue: queue, | |
} | |
} | |
func (c *Controller) runWorker() { | |
// hot loop until we're told to stop. processNextWorkItem will | |
// automatically wait until there's work available, so we don't worry | |
// about secondary waits | |
for c.processNextWorkItem() { | |
} | |
} | |
// processNextWorkItem deals with one key off the queue. It returns false | |
// when it's time to quit. | |
func (c *Controller) processNextWorkItem() bool { | |
// Wait until there is a new item in the working queue | |
key, quit := c.queue.Get() | |
if quit { | |
return false | |
} | |
// Tell the queue that we are done with processing this key. This unblocks the key for other workers | |
// This allows safe parallel processing because two pods with the same key are never processed in | |
// parallel. | |
defer c.queue.Done(key) | |
// Invoke the method containing the business logic | |
err := c.syncToStdout(key.(string)) | |
// Handle the error if something went wrong during the execution of the business logic | |
c.handleErr(err, key) | |
return true | |
} | |
// syncToStdout is the business logic of the controller. In this controller it simply prints | |
// information about the pod to stdout. In case an error happened, it has to simply return the error. | |
// The retry logic should not be part of the business logic. | |
func (c *Controller) syncToStdout(key string) error { | |
obj, exists, err := c.podInformer.GetIndexer().GetByKey(key) | |
if err != nil { | |
klog.Errorf("Fetching object with key %s from store failed with %v", key, err) | |
return err | |
} | |
if !exists { | |
// Below we will warm up our cache with a Pod, so that we will see a delete for one pod | |
fmt.Printf("Pod %s does not exist anymore\n", key) | |
} else { | |
// Note that you also have to check the uid if you have a local controlled resource, which | |
// is dependent on the actual instance, to detect that a Pod was recreated with the same name | |
fmt.Printf("Sync/Add/Update for Pod %s\n", obj.(*v1.Pod).GetName()) | |
} | |
return nil | |
} | |
// handleErr checks if an error happened and makes sure we will retry later. | |
func (c *Controller) handleErr(err error, key interface{}) { | |
if err == nil { | |
// Forget about the #AddRateLimited history of the key on every successful synchronization. | |
// This ensures that future processing of updates for this key is not delayed because of | |
// an outdated error history. | |
c.queue.Forget(key) | |
return | |
} | |
// This controller retries 3 times if something goes wrong. After that, it stops trying. | |
if c.queue.NumRequeues(key) < 3 { | |
klog.Infof("Error syncing pod %v: %v", key, err) | |
// Re-enqueue the key rate limited. Based on the rate limiter on the | |
// queue and the re-enqueue history, the key will be processed later again. | |
c.queue.AddRateLimited(key) | |
return | |
} | |
c.queue.Forget(key) | |
// Report to an external entity that, even after several retries, we could not successfully process this key | |
runtime.HandleError(err) | |
klog.Infof("Dropping pod %q out of the queue: %v", key, err) | |
} | |
// Run begins watching and syncing. | |
func (c *Controller) Run(workers int, stopCh chan struct{}) { | |
defer runtime.HandleCrash() | |
// Let the workers stop when we are done | |
defer c.queue.ShutDown() | |
klog.Info("Starting Pod controller") | |
go c.podInformer.Run(stopCh) | |
// Wait for all involved caches to be synced, before processing items from the queue is started | |
if !cache.WaitForCacheSync(stopCh, c.podInformer.HasSynced) { | |
runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync")) | |
return | |
} | |
// start up your workers. Some controllers have multiple kinds of workers | |
for i := 0; i < workers; i++ { | |
// runWorker will loop until "something bad" happens. The .Until will | |
// then rekick the worker after one second | |
go wait.Until(c.runWorker, time.Second, stopCh) | |
} | |
// wait until we're told to stop | |
<-stopCh | |
klog.Info("Stopping Pod controller") | |
} | |
func main() { | |
var kubeconfig *string | |
if home := homedir.HomeDir(); home != "" { | |
kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file") | |
} else { | |
kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file") | |
} | |
flag.Parse() | |
// Use the current context in kubeconfig | |
config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig) | |
if err != nil { | |
panic(err) | |
} | |
// create the clientset | |
clientset, err := kubernetes.NewForConfig(config) | |
if err != nil { | |
panic(err) | |
} | |
controller := NewController(clientset) | |
// Now let's start the controller | |
stop := make(chan struct{}) | |
defer close(stop) | |
go controller.Run(1, stop) | |
// Wait forever | |
select {} | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment