general refactoring

cmd/main.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"time"
+
+	"github.com/Azure/Kore/pkg/controller"
+	"github.com/Azure/Kore/pkg/kubernetes"
+	"github.com/Azure/Kore/pkg/signals"
+	log "github.com/Sirupsen/logrus"
+)
+
+func main() {
+	koreClient, kubeClient, err := kubernetes.GetClients()
+	if err != nil {
+		panic(err)
+	}
+
+	ctx := signals.Context()
+	controller.NewController(koreClient, kubeClient).Run(ctx)
+
+	shutdownDuration := 5 * time.Second
+	log.Infof("allowing %s for graceful shutdown to complete", shutdownDuration)
+	<-time.After(shutdownDuration)
+}

crd/scaledobject-crd-status-subresource.yaml → deploy/kore.yaml

@@ -9,5 +9,4 @@ spec:
     kind: ScaledObject
     plural: scaledobjects
   scope: Namespaced
-  subresources:
-status: {}
+---

pkg/controller/controller.go

@@ -0,0 +1,95 @@
+package controller
+
+import (
+	"context"
+	"sync"
+
+	kore_v1alpha1 "github.com/Azure/Kore/pkg/apis/kesc/v1alpha1"
+	clientset "github.com/Azure/Kore/pkg/client/clientset/versioned"
+	koreinformer_v1alpha1 "github.com/Azure/Kore/pkg/client/informers/externalversions/kesc/v1alpha1"
+	log "github.com/Sirupsen/logrus"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+type Controller interface {
+	Run(ctx context.Context)
+}
+
+type controller struct {
+	scaledObjectsInformer cache.SharedInformer
+	opsLock               sync.Mutex
+	ctx                   context.Context
+	koreClient            clientset.Interface
+	kubeClient            kubernetes.Interface
+}
+
+func NewController(koreClient clientset.Interface, kubeClient kubernetes.Interface) Controller {
+	c := &controller{
+		koreClient: koreClient,
+		kubeClient: kubeClient,
+		scaledObjectsInformer: koreinformer_v1alpha1.NewScaledObjectInformer(
+			koreClient,
+			meta_v1.NamespaceAll,
+			0,
+			cache.Indexers{},
+		),
+		opsLock: sync.Mutex{},
+	}
+
+	c.scaledObjectsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: c.syncScaledObject,
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			new := newObj.(*kore_v1alpha1.ScaledObject)
+			old := oldObj.(*kore_v1alpha1.ScaledObject)
+			if new.ResourceVersion == old.ResourceVersion {
+				return
+			}
+
+			c.syncScaledObject(newObj)
+		},
+		DeleteFunc: c.syncDeletedScaledObject,
+	})
+
+	return c
+}
+
+//TODO: might need seperate method for updates to reconcile differences when removing/changing properties
+func (c *controller) syncScaledObject(obj interface{}) {
+	c.opsLock.Lock()
+	defer c.opsLock.Unlock()
+
+	scaledObject := obj.(*kore_v1alpha1.ScaledObject)
+	deploymentName := scaledObject.Spec.DeploymentName
+	if deploymentName == "" {
+		log.Infof("Notified about ScaledObject with missing deployment name: %s", scaledObject.GetName())
+		return
+	}
+
+	deployment, err := c.kubeClient.AppsV1().Deployments(scaledObject.GetNamespace()).Get(deploymentName, meta_v1.GetOptions{})
+	if err != nil {
+		log.Errorf("Error getting deployment: %s", err)
+		return
+	}
+
+	log.Infof("Starting autoscalers for: %s. target deployment: %s", scaledObject.GetName(), deployment.GetName())
+}
+
+func (c *controller) syncDeletedScaledObject(obj interface{}) {
+	scaledObject := obj.(*kore_v1alpha1.ScaledObject)
+	log.Infof("Notified about deletion of ScaledObject: %s", scaledObject.GetName())
+}
+
+func (c *controller) Run(ctx context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	c.ctx = ctx
+	go func() {
+		<-ctx.Done()
+		log.Infof("Controller is shutting down")
+	}()
+	log.Infof("Controller is started")
+	c.scaledObjectsInformer.Run(ctx.Done())
+	cancel()
+}

pkg/kubernetes/client.go

@@ -0,0 +1,56 @@
+package kubernetes
+
+import (
+	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
+	// _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	clientset "github.com/Azure/Kore/pkg/client/clientset/versioned"
+	"github.com/kelseyhightower/envconfig"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+const envconfigPrefix = "KUBE"
+
+type config struct {
+	MasterURL      string `envconfig:"MASTER"`
+	KubeConfigPath string `envconfig:"CONFIG"`
+}
+
+func Config() (*rest.Config, error) {
+	c := config{}
+	err := envconfig.Process(envconfigPrefix, &c)
+	if err != nil {
+		return nil, err
+	}
+	var cfg *rest.Config
+	if c.MasterURL == "" && c.KubeConfigPath == "" {
+		cfg, err = rest.InClusterConfig()
+	} else {
+		cfg, err = clientcmd.BuildConfigFromFlags(c.MasterURL, c.KubeConfigPath)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}
+
+func GetClients() (*clientset.Clientset, *kubernetes.Clientset, error) {
+	cfg, err := Config()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	koreClient, err := clientset.NewForConfig(cfg)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	kubeClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return koreClient, kubeClient, nil
+}