mirror of
https://github.com/stakater/Reloader.git
synced 2026-08-27 14:37:17 +00:00
Update controller to remove unnecessary code
This commit is contained in:
faizanahmad055
@@ -1,32 +1,23 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"strings"
|
||||
"time"
|
||||
"fmt"
|
||||
|
||||
"github.com/stakater/Reloader/internal/pkg/actions"
|
||||
"github.com/golang/glog"
|
||||
"github.com/pkg/errors"
|
||||
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
|
||||
clientset "k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/apimachinery/pkg/util/runtime"
|
||||
informerruntime "k8s.io/apimachinery/pkg/runtime"
|
||||
"github.com/stakater/Reloader/internal/pkg/config"
|
||||
"k8s.io/apimachinery/pkg/fields"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
"k8s.io/kubernetes/pkg/watch"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
errorHandler "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/client-go/tools/cache"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
|
||||
"sort"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
updateOnChangeAnnotation = "configmap.fabric8.io/update-on-change"
|
||||
updateOnChangeAnnotation = "reloader.stakater.com.io/update-on-change"
|
||||
// AllNamespaces as our controller will be looking for events in all namespaces
|
||||
AllNamespaces = ""
|
||||
)
|
||||
@@ -41,90 +32,40 @@ type Event struct {
|
||||
|
||||
// Controller for checking events
|
||||
type Controller struct {
|
||||
clientset clientset.Interface
|
||||
client kubernetes.Interface
|
||||
indexer cache.Indexer
|
||||
queue workqueue.RateLimitingInterface
|
||||
informer cache.Controller
|
||||
controllerConfig config.Controller
|
||||
Actions []actions.Action
|
||||
|
||||
stopCh chan struct{}
|
||||
}
|
||||
|
||||
// NewController for initializing a Controller
|
||||
func NewController(
|
||||
clientset clientset.Interface,
|
||||
resyncPeriod time.Duration, controllerConfig config.Controller, objType informerruntime.Object) (*Controller, error) {
|
||||
client kubernetes.Interface,
|
||||
controllerConfig config.Controller, objType informerruntime.Object) (*Controller, error) {
|
||||
|
||||
c := Controller{
|
||||
clientset: clientset,
|
||||
client: client,
|
||||
controllerConfig: controllerConfig,
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
|
||||
queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
|
||||
listWatcher := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), controllerConfig.Type, AllNamespaces, fields.Everything())
|
||||
listWatcher := cache.NewListWatchFromClient(client.CoreV1().RESTClient(), controllerConfig.Type, AllNamespaces, fields.Everything())
|
||||
|
||||
indexer, informer := cache.NewIndexerInformer(listWatcher, objType, 0, cache.ResourceEventHandlerFuncs {
|
||||
AddFunc: c.Add,
|
||||
UpdateFunc: c.Update,
|
||||
DeleteFunc: c.Delete,
|
||||
}, cache.Indexers{})
|
||||
|
||||
/*c.cmLister.Store, c.cmController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: configMapListFunc(c.client, namespace),
|
||||
WatchFunc: configMapWatchFunc(c.client, namespace),
|
||||
},
|
||||
&api.ConfigMap{},
|
||||
resyncPeriod,
|
||||
framework.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
newCM := obj.(*api.ConfigMap)
|
||||
//typeOfMaster, err := util.TypeOfMaster(kubeClient)
|
||||
if err != nil {
|
||||
glog.Fatalf("failed to create REST client config: %s", err)
|
||||
}
|
||||
err = rollingUpgradeDeployments(newCM, kubeClient)
|
||||
if err != nil {
|
||||
glog.Errorf("failed to update Deployment: %v", err)
|
||||
}
|
||||
|
||||
},
|
||||
UpdateFunc: func(oldObj interface{}, newObj interface{}) {
|
||||
oldM := oldObj.(*api.ConfigMap)
|
||||
newCM := newObj.(*api.ConfigMap)
|
||||
|
||||
if oldM.ResourceVersion != newCM.ResourceVersion {
|
||||
//typeOfMaster, err := util.TypeOfMaster(kubeClient)
|
||||
if err != nil {
|
||||
glog.Fatalf("failed to create REST client config: %s", err)
|
||||
}
|
||||
err = rollingUpgradeDeployments(newCM, kubeClient)
|
||||
if err != nil {
|
||||
glog.Errorf("failed to update Deployment: %v", err)
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
)*/
|
||||
c.indexer = indexer
|
||||
c.informer = informer
|
||||
c.queue = queue
|
||||
return &c, nil
|
||||
}
|
||||
|
||||
// Run starts the controller.
|
||||
/*func (c *Controller) Run() {
|
||||
glog.Infof("starting reloader")
|
||||
|
||||
<-c.stopCh
|
||||
}*/
|
||||
|
||||
// Stop stops the controller.
|
||||
/*func (c *Controller) Stop() {
|
||||
glog.Infof("stopping reloader")
|
||||
|
||||
close(c.stopCh)
|
||||
}*/
|
||||
|
||||
// Add function to add a 'create' event to the queue in case of creating a pod
|
||||
func (c *Controller) Add(obj interface{}) {
|
||||
key, err := cache.MetaNamespaceKeyFunc(obj)
|
||||
@@ -143,6 +84,8 @@ func (c *Controller) Update(old interface{}, new interface{}) {
|
||||
var event Event
|
||||
|
||||
if err == nil {
|
||||
event.key = key
|
||||
event.eventType = "update"
|
||||
c.queue.Add(event)
|
||||
}
|
||||
}
|
||||
@@ -155,7 +98,7 @@ func (c *Controller) Delete(obj interface{}) {
|
||||
//Run function for controller which handles the queue
|
||||
func (c *Controller) Run(threadiness int, stopCh chan struct{}) {
|
||||
|
||||
glog.Infof("Starting Controller for type ", c.controllerConfig.Type)
|
||||
logrus.Infof("Starting Controller for type ", c.controllerConfig.Type)
|
||||
defer errorHandler.HandleCrash()
|
||||
|
||||
// Let the workers stop when we are done
|
||||
@@ -174,7 +117,7 @@ func (c *Controller) Run(threadiness int, stopCh chan struct{}) {
|
||||
}
|
||||
|
||||
<-stopCh
|
||||
glog.Infof("Stopping Controller for type ", c.controllerConfig.Type)
|
||||
logrus.Infof("Stopping Controller for type ", c.controllerConfig.Type)
|
||||
}
|
||||
|
||||
func (c *Controller) runWorker() {
|
||||
@@ -205,16 +148,16 @@ func (c *Controller) takeAction(event Event) error {
|
||||
|
||||
obj, _, err := c.indexer.GetByKey(event.key)
|
||||
if err != nil {
|
||||
glog.Infof("Fetching object with key %s from store failed with %v", event.key, err)
|
||||
logrus.Infof("Fetching object with key %s from store failed with %v", event.key, err)
|
||||
}
|
||||
if obj == nil {
|
||||
glog.Infof("Error in Action")
|
||||
logrus.Infof("Error in Action")
|
||||
} else {
|
||||
glog.Infof("Detected changes in object %s", obj)
|
||||
/*glog.Infof("Resource block not found, performing actions")
|
||||
logrus.Infof("Detected changes in object %s", obj)
|
||||
/*logrus.Infof("Resource block not found, performing actions")
|
||||
// process events based on its type
|
||||
for index, action := range c.Actions {
|
||||
glog.Infof("Performing '%s' action for controller of type '%s'", c.controllerConfig.Actions[index].Name, c.controllerConfig.Type)
|
||||
gllogrusog.Infof("Performing '%s' action for controller of type '%s'", c.controllerConfig.Actions[index].Name, c.controllerConfig.Type)
|
||||
switch event.eventType {
|
||||
case "create":
|
||||
action.ObjectCreated(obj)
|
||||
@@ -241,7 +184,7 @@ func (c *Controller) handleErr(err error, key interface{}) {
|
||||
|
||||
// This controller retries 5 times if something goes wrong. After that, it stops trying.
|
||||
if c.queue.NumRequeues(key) < 5 {
|
||||
log.Printf("Error syncing events %v: %v", key, err)
|
||||
logrus.Infof("Error syncing events %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.
|
||||
@@ -252,122 +195,5 @@ func (c *Controller) handleErr(err error, key interface{}) {
|
||||
c.queue.Forget(key)
|
||||
// Report to an external entity that, even after several retries, we could not successfully process this key
|
||||
runtime.HandleError(err)
|
||||
log.Printf("Dropping the key %q out of the queue: %v", key, err)
|
||||
}
|
||||
|
||||
/*func configMapListFunc(c *client.Client, ns string) func(api.ListOptions) (runtime.Object, error) {
|
||||
return func(opts api.ListOptions) (runtime.Object, error) {
|
||||
return c.ConfigMaps(ns).List(opts)
|
||||
}
|
||||
}
|
||||
|
||||
func configMapWatchFunc(c *client.Client, ns string) func(options api.ListOptions) (watch.Interface, error) {
|
||||
return func(options api.ListOptions) (watch.Interface, error) {
|
||||
return c.ConfigMaps(ns).Watch(options)
|
||||
}
|
||||
}
|
||||
|
||||
func rollingUpgradeDeployments(cm *api.ConfigMap, c *client.Client) error {
|
||||
ns := cm.Namespace
|
||||
configMapName := cm.Name
|
||||
configMapVersion := convertConfigMapToToken(cm)
|
||||
|
||||
deployments, err := c.Deployments(ns).List(api.ListOptions{})
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to list deployments")
|
||||
}
|
||||
for _, d := range deployments.Items {
|
||||
containers := d.Spec.Template.Spec.Containers
|
||||
// match deployments with the correct annotation
|
||||
annotationValue, _ := d.ObjectMeta.Annotations[updateOnChangeAnnotation]
|
||||
if annotationValue != "" {
|
||||
values := strings.Split(annotationValue, ",")
|
||||
matches := false
|
||||
for _, value := range values {
|
||||
if value == configMapName {
|
||||
matches = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if matches {
|
||||
updateContainers(containers, annotationValue, configMapVersion)
|
||||
|
||||
// update the deployment
|
||||
_, err := c.Deployments(ns).Update(&d)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "update deployment failed")
|
||||
}
|
||||
glog.Infof("Updated Deployment %s", d.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}*/
|
||||
|
||||
// lets convert the configmap into a unique token based on the data values
|
||||
func convertConfigMapToToken(cm *api.ConfigMap) string {
|
||||
values := []string{}
|
||||
for k, v := range cm.Data {
|
||||
values = append(values, k+"="+v)
|
||||
}
|
||||
sort.Strings(values)
|
||||
text := strings.Join(values, ";")
|
||||
// we could zip and base64 encode
|
||||
// but for now we could leave this easy to read so that its easier to diagnose when & why things changed
|
||||
return text
|
||||
}
|
||||
|
||||
func updateContainers(containers []api.Container, annotationValue, configMapVersion string) bool {
|
||||
// we can have multiple configmaps to update
|
||||
answer := false
|
||||
configmaps := strings.Split(annotationValue, ",")
|
||||
for _, cmNameToUpdate := range configmaps {
|
||||
configmapEnvar := "FABRIC8_" + convertToEnvVarName(cmNameToUpdate) + "_CONFIGMAP"
|
||||
|
||||
for i := range containers {
|
||||
envs := containers[i].Env
|
||||
matched := false
|
||||
for j := range envs {
|
||||
if envs[j].Name == configmapEnvar {
|
||||
matched = true
|
||||
if envs[j].Value != configMapVersion {
|
||||
glog.Infof("Updating %s to %s", configmapEnvar, configMapVersion)
|
||||
envs[j].Value = configMapVersion
|
||||
answer = true
|
||||
}
|
||||
}
|
||||
}
|
||||
// if no existing env var exists lets create one
|
||||
if !matched {
|
||||
e := api.EnvVar{
|
||||
Name: configmapEnvar,
|
||||
Value: configMapVersion,
|
||||
}
|
||||
containers[i].Env = append(containers[i].Env, e)
|
||||
answer = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return answer
|
||||
}
|
||||
|
||||
// convertToEnvVarName converts the given text into a usable env var
|
||||
// removing any special chars with '_'
|
||||
func convertToEnvVarName(text string) string {
|
||||
var buffer bytes.Buffer
|
||||
lower := strings.ToUpper(text)
|
||||
lastCharValid := false
|
||||
for i := 0; i < len(lower); i++ {
|
||||
ch := lower[i]
|
||||
if (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') {
|
||||
buffer.WriteString(string(ch))
|
||||
lastCharValid = true
|
||||
} else {
|
||||
if lastCharValid {
|
||||
buffer.WriteString("_")
|
||||
}
|
||||
lastCharValid = false
|
||||
}
|
||||
}
|
||||
return buffer.String()
|
||||
}
|
||||
logrus.Infof("Dropping the key %q out of the queue: %v", key, err)
|
||||
}
|
||||
Reference in New Issue
Block a user