creating extensible signal plugin interface (#36)

* creating extensible signal plugin interface

* making streams plugin and others builtin

* adding multiple signal binaries to controller

* making signal plugin name constant

Gopkg.toml

@@ -6,9 +6,20 @@ required = [
   "k8s.io/code-generator/cmd/defaulter-gen",
   "k8s.io/code-generator/cmd/openapi-gen",
   "k8s.io/code-generator/cmd/go-to-protobuf",
-  "k8s.io/code-generator/cmd/go-to-protobuf/protoc-gen-gogo"
+  "k8s.io/code-generator/cmd/go-to-protobuf/protoc-gen-gogo",
+  "github.com/golang/protobuf/protoc-gen-go",
+  "github.com/gogo/protobuf/protoc-gen-gofast",
+  "github.com/gogo/protobuf/protoc-gen-gogofast",
 ]
 
+[[constraint]]
+  name = "google.golang.org/grpc"
+  version = "1.9.2"
+
+[[constraint]]
+  name = "github.com/golang/protobuf"
+  version = "v1.1.0"
+
 [[constraint]]
   name = "k8s.io/api"
   branch = "release-1.9"
@@ -65,6 +76,10 @@ required = [
   name = "github.com/Shopify/sarama"
   version = "1.16.0"
 
+[[constraint]]
+  branch = "master"
+  name = "github.com/hashicorp/go-plugin"
+
 [prune]
   go-tests = true
   unused-packages = true

Makefile

@@ -1,12 +1,14 @@
-PACKAGE=github.com/argoproj/argo-events
+
 CURRENT_DIR=$(shell pwd)
 DIST_DIR=${CURRENT_DIR}/dist
+PLUGIN_DIR=${DIST_DIR}/plugins
 
 VERSION=$(shell cat ${CURRENT_DIR}/VERSION)
 BUILD_DATE=$(shell date -u +'%Y-%m-%dT%H:%M:%SZ')
 GIT_COMMIT=$(shell git rev-parse HEAD)
 GIT_TAG=$(shell if [ -z "`git status --porcelain`" ]; then git describe --exact-match --tags HEAD 2>/dev/null; fi)
 GIT_TREE_STATE=$(shell if [ -z "`git status --porcelain`" ]; then echo "clean" ; else echo "dirty"; fi)
+PLUGINS=$(shell find . \( -type d -and -path '*/signals/stream/builtin/*' \))
 
 override LDFLAGS += \
   -X ${PACKAGE}.version=${VERSION} \
@@ -34,9 +36,9 @@ IMAGE_PREFIX=${IMAGE_NAMESPACE}/
 endif
 
 # Build the project
-.PHONY: all controller controller-image executor-job executor-job-image clean test
+.PHONY: all controller controller-image clean test
 
-all: controller-image executor-job-image
+all: controller-image
 
 # Sensor controller
 controller:
@@ -45,19 +47,19 @@ controller:
 controller-linux:
 	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make controller
 
-controller-image: controller-linux
+controller-image: controller-linux stream-plugins-linux
 	docker build -t $(IMAGE_PREFIX)sensor-controller:$(IMAGE_TAG) -f ./controller/Dockerfile .
 	@if [ "$(DOCKER_PUSH)" = "true" ] ; then docker push $(IMAGE_PREFIX)sensor-controller:$(IMAGE_TAG) ; fi
 
-# Sensor executor
-executor-job:
-	go build -v -ldflags '${LDFLAGS}' -o ${DIST_DIR}/sensor-executor ./cmd/sensor-job
-
-executor-job-linux:
-	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make executor-job
+# Plugins
+stream-plugins:
+	go build -v -ldflags '${LDFLAGS}' -o ${PLUGIN_DIR}/nats ./signals/stream/builtin/nats
+	go build -v -ldflags '${LDFLAGS}' -o ${PLUGIN_DIR}/mqtt ./signals/stream/builtin/mqtt
+	go build -v -ldflags '${LDFLAGS}' -o ${PLUGIN_DIR}/kafka ./signals/stream/builtin/kafka
+	go build -v -ldflags '${LDFLAGS}' -o ${PLUGIN_DIR}/amqp ./signals/stream/builtin/amqp
 
-executor-job-image: executor-job-linux
-	docker build -t $(IMAGE_PREFIX)sensor-executor:$(IMAGE_TAG) -f ./job/Dockerfile .
+stream-plugins-linux:
+	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make stream-plugins
 
 test:
 	go test $(shell go list ./... | grep -v /vendor/) -race -short -v

cmd/sensor-controller/main.go

@@ -18,41 +18,44 @@ package main
 
 import (
 	"context"
-	"log"
 	"os"
 
 	"go.uber.org/zap"
-	"k8s.io/client-go/kubernetes"
 
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/controller"
-	sensorclientset "github.com/argoproj/argo-events/pkg/client/clientset/versioned"
 )
 
 func main() {
+	// kubernetes configuration
 	kubeConfig, _ := os.LookupEnv(common.EnvVarKubeConfig)
 	restConfig, err := common.GetClientConfig(kubeConfig)
 	if err != nil {
-		log.Fatal(err)
+		panic(err)
 	}
 
-	kubeClientset := kubernetes.NewForConfigOrDie(restConfig)
-	sensorClientset := sensorclientset.NewForConfigOrDie(restConfig)
+	// controller configuration
+	configMap, ok := os.LookupEnv(common.EnvVarConfigMap)
+	if !ok {
+		configMap = common.DefaultConfigMapName(common.DefaultSensorControllerDeploymentName)
+	}
 
+	// logger
 	logger, err := zap.NewDevelopment()
 	if err != nil {
-		panic(err.Error())
+		panic(err)
 	}
 
-	configMap, ok := os.LookupEnv(common.EnvVarConfigMap)
-	if !ok {
-		configMap = common.DefaultConfigMapName(common.DefaultSensorControllerDeploymentName)
+	// stream signal plugins
+	pluginMgr, err := controller.NewPluginManager()
+	if err != nil {
+		panic(err)
 	}
 
-	controller := controller.NewSensorController(restConfig, kubeClientset, sensorClientset, logger.Sugar(), configMap)
+	controller := controller.NewSensorController(restConfig, configMap, pluginMgr, logger.Sugar())
 	err = controller.ResyncConfig()
 	if err != nil {
-		log.Fatalf("%+v", err)
+		panic(err)
 	}
 
 	go controller.Run(context.Background(), 1, 1)