decoupling Function, Trigger and Runtimes

.gitignore

@@ -145,8 +145,10 @@ artifacts/
 # Kubeless specific
 bats/
 bundles/
-docker/controller/kubeless-controller
+docker/controller-manager/kubeless-controller-manager
+docker/kafka-controller/kafka-controller
 ksonnet-lib/
 kubeless-openshift.yaml
 kubeless-rbac.yaml
 kubeless.yaml
+kafka-zookeeper.yaml

.travis.yml

@@ -20,9 +20,11 @@ env:
     - TEST_TARGET=minikube_kafka
     - TEST_TARGET=GKE
   global:
-    - CONTROLLER_IMAGE_NAME=bitnami/kubeless-controller
+    - CONTROLLER_IMAGE_NAME=bitnami/kubeless-controller-manager
     - CONTROLLER_TAG=${TRAVIS_TAG:-build-$TRAVIS_BUILD_ID}
     - CONTROLLER_IMAGE=${CONTROLLER_IMAGE_NAME}:${CONTROLLER_TAG}
+    - KAFKA_CONTROLLER_IMAGE_NAME=bitnami/kafka-trigger-controller
+    - KAFKA_CONTROLLER_IMAGE=${KAFKA_CONTROLLER_IMAGE_NAME}:${CONTROLLER_TAG}
     - CGO_ENABLED=0
     - TEST_DEBUG=1
     - GKE_VERSION=1.7.12-gke.2
@@ -92,10 +94,12 @@ script:
     if [[ "$SHOULD_TEST" == "1" ]]; then
       make VERSION=${TRAVIS_TAG:-build-$TRAVIS_BUILD_ID} binary
       make controller-image CONTROLLER_IMAGE=$CONTROLLER_IMAGE
+      make kafka-controller-image KAFKA_CONTROLLER_IMAGE=$KAFKA_CONTROLLER_IMAGE
       make all-yaml
       sed -i.bak 's/'":latest"'/'":${CONTROLLER_TAG}"'/g' kubeless.yaml
       sed -i.bak 's/'":latest"'/'":${CONTROLLER_TAG}"'/g' kubeless-rbac.yaml
       sed -i.bak 's/'":latest"'/'":${CONTROLLER_TAG}"'/g' kubeless-openshift.yaml
+      sed -i.bak 's/'":latest"'/'":${CONTROLLER_TAG}"'/g' kafka-zookeeper.yaml
       case $TEST_TARGET in
       unit)
         make test
@@ -112,6 +116,7 @@ script:
       GKE)
         docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD"
         docker push $CONTROLLER_IMAGE
+        docker push $KAFKA_CONTROLLER_IMAGE
         echo "Waiting for the GKE cluster to be ready"
         tail -f $TRAVIS_BUILD_DIR/gke-start.log &
         wait $GKE_START_PID
@@ -152,7 +157,6 @@ before_deploy:
       sed 's/'":${CONTROLLER_TAG}"'/'"@$NEW_DIGEST"'/g' kubeless.yaml > kubeless-${TRAVIS_TAG}.yaml
       sed 's/'":${CONTROLLER_TAG}"'/'"@$NEW_DIGEST"'/g' kubeless-rbac.yaml > kubeless-rbac-${TRAVIS_TAG}.yaml
       sed 's/'":${CONTROLLER_TAG}"'/'"@$NEW_DIGEST"'/g' kubeless-openshift.yaml > kubeless-openshift-${TRAVIS_TAG}.yaml
-      cp kafka-zookeeper.yaml kafka-zookeeper-${TRAVIS_TAG}.yaml
     fi
 
 deploy:

Makefile

@@ -5,7 +5,8 @@ VERSION = dev-$(shell date +%FT%T%z)
 
 KUBECFG = kubecfg
 DOCKER = docker
-CONTROLLER_IMAGE = kubeless-controller:latest
+CONTROLLER_IMAGE = kubeless-controller-manager:latest
+KAFKA_CONTROLLER_IMAGE = kafka-trigger-controller:latest
 OS = linux
 ARCH = amd64
 BUNDLES = bundles
@@ -47,15 +48,24 @@ kubeless-openshift.yaml: kubeless-openshift.jsonnet kubeless-rbac.jsonnet
 
 kafka-zookeeper.yaml: kafka-zookeeper.jsonnet
 
-docker/controller: controller-build
-	cp $(BUNDLES)/kubeless_$(OS)-$(ARCH)/kubeless-controller $@
+docker/controller-manager: controller-build
+	cp $(BUNDLES)/kubeless_$(OS)-$(ARCH)/kubeless-controller-manager $@
 
 controller-build:
 	./script/binary-controller -os=$(OS) -arch=$(ARCH)
 
-controller-image: docker/controller
+controller-image: docker/controller-manager
 	$(DOCKER) build -t $(CONTROLLER_IMAGE) $<
 
+docker/kafka-controller: kafka-controller-build
+	cp $(BUNDLES)/kubeless_$(OS)-$(ARCH)/kafka-controller $@
+
+kafka-controller-build:
+	./script/kafka-controller.sh -os=$(OS) -arch=$(ARCH)
+
+kafka-controller-image: docker/kafka-controller
+	$(DOCKER) build -t $(KAFKA_CONTROLLER_IMAGE) $<
+
 update:
 	./hack/update-codegen.sh
 
@@ -102,4 +112,4 @@ bootstrap: bats ksonnet-lib
 	fi
 
 build_and_test:
-	./script/start-test-environment.sh "make binary && make controller-image CONTROLLER_IMAGE=bitnami/kubeless-controller:latest && make integration-tests"
+	./script/start-test-environment.sh "make binary && make controller-image CONTROLLER_IMAGE=bitnami/kubeless-controller-manager:latest && make integration-tests"

cmd/kafka-trigger-controller/kafka-controller.go

@@ -0,0 +1,68 @@
+/*
+Copyright (c) 2016-2017 Bitnami
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/kubeless/kubeless/pkg/controller"
+	"github.com/kubeless/kubeless/pkg/utils"
+	"github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+)
+
+const (
+	globalUsage = `` //TODO: adding explanation
+)
+
+var rootCmd = &cobra.Command{
+	Use:   "kafka-controller",
+	Short: "Kafka controller",
+	Long:  globalUsage,
+	Run: func(cmd *cobra.Command, args []string) {
+		kubelessClient, err := utils.GetFunctionClientInCluster()
+		if err != nil {
+			logrus.Fatalf("Cannot get kubeless client: %v", err)
+		}
+
+		kafkaTriggerCfg := controller.KafkaTriggerConfig{
+			TriggerClient: kubelessClient,
+		}
+
+		kafkaTriggerController := controller.NewKafkaTriggerController(kafkaTriggerCfg)
+
+		stopCh := make(chan struct{})
+		defer close(stopCh)
+
+		go kafkaTriggerController.Run(stopCh)
+
+		sigterm := make(chan os.Signal, 1)
+		signal.Notify(sigterm, syscall.SIGTERM)
+		signal.Notify(sigterm, syscall.SIGINT)
+		<-sigterm
+	},
+}
+
+func main() {
+	if err := rootCmd.Execute(); err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+}

cmd/kubeless-controller/kubeless-controller.go → cmd/kubeless-controller-manager/kubeless-controller-manager.go

@@ -26,6 +26,7 @@ import (
 	"syscall"
 
 	monitoringv1alpha1 "github.com/coreos/prometheus-operator/pkg/client/monitoring/v1alpha1"
+	"github.com/kubeless/kubeless/pkg/client/informers/externalversions"
 	"github.com/kubeless/kubeless/pkg/controller"
 	"github.com/kubeless/kubeless/pkg/utils"
 	"github.com/sirupsen/logrus"
@@ -47,10 +48,18 @@ var rootCmd = &cobra.Command{
 			logrus.Fatalf("Cannot get kubeless client: %v", err)
 		}
 
-		cfg := controller.Config{
+		functionCfg := controller.Config{
 			KubeCli:        utils.GetClient(),
 			FunctionClient: kubelessClient,
 		}
+		httpTriggerCfg := controller.HTTPTriggerConfig{
+			KubeCli:       utils.GetClient(),
+			TriggerClient: kubelessClient,
+		}
+		cronJobTriggerCfg := controller.CronJobTriggerConfig{
+			KubeCli:       utils.GetClient(),
+			TriggerClient: kubelessClient,
+		}
 
 		restCfg, err := rest.InClusterConfig()
 		if err != nil {
@@ -62,11 +71,18 @@ var rootCmd = &cobra.Command{
 			logrus.Fatal(err)
 		}
 
-		c := controller.New(cfg, smclient)
+		sharedInformerFactory := externalversions.NewSharedInformerFactory(kubelessClient, 0)
+
+		functionController := controller.NewFunctionController(functionCfg, smclient)
+		httpTriggerController := controller.NewHTTPTriggerController(httpTriggerCfg, sharedInformerFactory)
+		cronJobTriggerController := controller.NewCronJobTriggerController(cronJobTriggerCfg, sharedInformerFactory)
+
 		stopCh := make(chan struct{})
 		defer close(stopCh)
 
-		go c.Run(stopCh)
+		go functionController.Run(stopCh)
+		go httpTriggerController.Run(stopCh)
+		go cronJobTriggerController.Run(stopCh)
 
 		sigterm := make(chan os.Signal, 1)
 		signal.Notify(sigterm, syscall.SIGTERM)

cmd/kubeless/autoscale/autoscaleCreate.go

@@ -24,11 +24,6 @@ var autoscaleCreateCmd = &cobra.Command{
 			ns = utils.GetDefaultNamespace()
 		}
 
-		function, err := utils.GetFunction(funcName, ns)
-		if err != nil {
-			logrus.Fatalf("Unable to find the function %s. Received %s: ", funcName, err)
-		}
-
 		min, err := cmd.Flags().GetInt32("min")
 		if err != nil {
 			logrus.Fatal(err)
@@ -55,18 +50,24 @@ var autoscaleCreateCmd = &cobra.Command{
 			logrus.Fatal(err)
 		}
 
-		hpa, err := getHorizontalAutoscaleDefinition(funcName, ns, metric, min, max, value, function.ObjectMeta.Labels)
+		kubelessClient, err := utils.GetKubelessClientOutCluster()
 		if err != nil {
 			logrus.Fatal(err)
 		}
-		function.Spec.HorizontalPodAutoscaler = hpa
 
-		kubelessClient, err := utils.GetFunctionClientOutCluster()
+		function, err := utils.GetFunctionCustomResource(kubelessClient, funcName, ns)
+		if err != nil {
+			logrus.Fatalf("Unable to find the function %s. Received %s: ", funcName, err)
+		}
+
+		hpa, err := getHorizontalAutoscaleDefinition(funcName, ns, metric, min, max, value, function.ObjectMeta.Labels)
 		if err != nil {
 			logrus.Fatal(err)
 		}
+		function.Spec.HorizontalPodAutoscaler = hpa
+
 		logrus.Infof("Adding autoscaling rule to the function...")
-		err = utils.UpdateK8sCustomResource(kubelessClient, &function)
+		err = utils.UpdateFunctionCustomResource(kubelessClient, function)
 		if err != nil {
 			logrus.Fatal(err)
 		}

cmd/kubeless/autoscale/autoscaleDelete.go

@@ -38,23 +38,24 @@ var autoscaleDeleteCmd = &cobra.Command{
 			ns = utils.GetDefaultNamespace()
 		}
 
-		function, err := utils.GetFunction(funcName, ns)
+		kubelessClient, err := utils.GetKubelessClientOutCluster()
+		if err != nil {
+			logrus.Fatal(err)
+		}
+
+		function, err := utils.GetFunctionCustomResource(kubelessClient, funcName, ns)
 		if err != nil {
 			logrus.Fatalf("Unable to find the function %s. Received %s: ", funcName, err)
 		}
 
 		if function.Spec.HorizontalPodAutoscaler.Name != "" {
 			function.Spec.HorizontalPodAutoscaler = v2beta1.HorizontalPodAutoscaler{}
-			kubelessClient, err := utils.GetFunctionClientOutCluster()
-			if err != nil {
-				logrus.Fatal(err)
-			}
-			logrus.Infof("Removing autoscaling rule to the function...")
-			err = utils.UpdateK8sCustomResource(kubelessClient, &function)
+			logrus.Infof("Removing autoscaling rule from the function...")
+			err = utils.UpdateFunctionCustomResource(kubelessClient, function)
 			if err != nil {
 				logrus.Fatal(err)
 			}
-			logrus.Infof("Removed Autoscaling rule for %s", funcName)
+			logrus.Infof("Removed Autoscaling rule from %s", funcName)
 		} else {
 			logrus.Fatalf("Not found an auto scale definition for %s", funcName)
 		}

cmd/kubeless/function/call.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/kubeless/kubeless/pkg/utils"
 	"github.com/sirupsen/logrus"
@@ -83,6 +84,15 @@ var callCmd = &cobra.Command{
 			// So we need to manually build the URL
 			req = req.AbsPath(svc.ObjectMeta.SelfLink + ":" + port + "/proxy/")
 		}
+		timestamp := time.Now().UTC()
+		eventID, err := utils.GetRandString(11)
+		if err != nil {
+			logrus.Fatalf("Unable to generate ID %v", err)
+		}
+		req.SetHeader("event-id", eventID)
+		req.SetHeader("event-type", "application/json")
+		req.SetHeader("event-time", timestamp.String())
+		req.SetHeader("event-namespace", "cli.kubeless.io")
 		res, err := req.Do().Raw()
 		if err != nil {
 			// Properly interpret line breaks

cmd/kubeless/function/delete.go

@@ -40,12 +40,12 @@ var deleteCmd = &cobra.Command{
 			ns = utils.GetDefaultNamespace()
 		}
 
-		kubelessClient, err := utils.GetFunctionClientOutCluster()
+		kubelessClient, err := utils.GetKubelessClientOutCluster()
 		if err != nil {
 			logrus.Fatal(err)
 		}
 
-		err = utils.DeleteK8sCustomResource(kubelessClient, funcName, ns)
+		err = utils.DeleteFunctionCustomResource(kubelessClient, funcName, ns)
 		if err != nil {
 			logrus.Fatal(err)
 		}