re-architecting around gateways and sensors

Gopkg.toml

@@ -23,7 +23,7 @@ required = [
 
 [[constraint]]
   name = "github.com/ghodss/yaml"
-  version = "1.0.0"
+  branch = "master"
 
 [[constraint]]
   name = "github.com/minio/minio-go"
@@ -49,6 +49,10 @@ required = [
   name = "github.com/Shopify/sarama"
   version = "1.16.0"
 
+[[constraint]]
+  name = "github.com/stretchr/testify"
+  version = "1.1.4"
+
 [[override]]
   branch = "release-1.10"
   name = "k8s.io/api"

common/util.go

@@ -24,9 +24,7 @@ import (
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
-	"math/rand"
 	"net/http"
-	"time"
 )
 
 // DefaultConfigMapName returns a formulated name for a configmap name based on the sensor-controller deployment name
@@ -96,21 +94,3 @@ func SendErrorResponse(writer http.ResponseWriter) {
 	writer.WriteHeader(http.StatusBadRequest)
 	writer.Write([]byte(ErrorResponse))
 }
-
-// RandomStringGenerator generates a random string
-func RandomStringGenerator() string {
-	var n int
-	for {
-		n = rand.Intn(10)
-		if n != 0 {
-			break
-		}
-	}
-	rand.Seed(time.Now().UnixNano())
-	var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
-	b := make([]rune, n)
-	for i := range b {
-		b[i] = letterRunes[rand.Intn(len(letterRunes))]
-	}
-	return string(b)
-}

controllers/gateway/operator.go

@@ -53,7 +53,7 @@ func (goc *gwOperationCtx) operate() error {
 	goc.log.Info().Msg("operating on the gateway...")
 
 	// performs a basic validation on gateway resource.
-	err := goc.validate()
+	err := Validate(goc.gw)
 	if err != nil {
 		goc.log.Error().Err(err).Msg("gateway validation failed")
 		goc.markGatewayPhase(v1alpha1.NodePhaseError, "validation failed")

controllers/gateway/validate.go

@@ -6,19 +6,20 @@ import (
 )
 
 // Validates the gateway resource.
-func (goc *gwOperationCtx) validate() error {
-	if goc.gw.Spec.DeploySpec == nil {
+// Exporting this function so that external APIs can use this to validate gateway resource.
+func Validate(gw *v1alpha1.Gateway) error {
+	if gw.Spec.DeploySpec == nil {
 		return fmt.Errorf("gateway deploy specification is not specified")
 	}
-	if goc.gw.Spec.Type == "" {
+	if gw.Spec.Type == "" {
 		return fmt.Errorf("gateway type is not specified")
 	}
-	if goc.gw.Spec.Version == "" {
+	if gw.Spec.Version == "" {
 		return fmt.Errorf("gateway version is not specified")
 	}
-	switch goc.gw.Spec.DispatchMechanism {
+	switch gw.Spec.DispatchMechanism {
 	case v1alpha1.HTTPGateway:
-		if goc.gw.Spec.Watchers == nil || (goc.gw.Spec.Watchers.Gateways == nil && goc.gw.Spec.Watchers.Sensors == nil) {
+		if gw.Spec.Watchers == nil || (gw.Spec.Watchers.Gateways == nil && gw.Spec.Watchers.Sensors == nil) {
 			return fmt.Errorf("no associated watchers with gateway")
 		}
 	case v1alpha1.NATSGateway:

controllers/gateway/validate_test.go

@@ -10,6 +10,6 @@ func TestGwOperationCtx_validate(t *testing.T) {
 	gateway, err := getGateway()
 	assert.Nil(t, err)
 	goc := newGatewayOperationCtx(gateway, fakeController)
-	err = goc.validate()
+	err = Validate(goc.gw)
 	assert.Nil(t, err)
 }

controllers/sensor/operator.go

@@ -75,7 +75,7 @@ func (soc *sOperationCtx) operate() error {
 		// non nil err indicates failed validation
 		// we do not want to requeue a sensor in this case
 		// since validation will fail every time
-		err := validateSensor(soc.s)
+		err := ValidateSensor(soc.s)
 		if err != nil {
 			soc.log.Error().Err(err).Msg("failed to validate sensor")
 			soc.markSensorPhase(v1alpha1.NodePhaseError, true, err.Error())

controllers/sensor/validate.go

@@ -25,10 +25,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// validateSensor accepts a sensor and performs validation against it
+// ValidateSensor accepts a sensor and performs validation against it
 // we return an error so that it can be logged as a message on the sensor status
 // the error is ignored by the operation context as subsequent re-queues would produce the same error.
-func validateSensor(s *v1alpha1.Sensor) error {
+// Exporting this function so that external APIs can use this to validate sensor resource.
+func ValidateSensor(s *v1alpha1.Sensor) error {
 	if err := validateSignals(s.Spec.Signals); err != nil {
 		return err
 	}

controllers/sensor/validate_test.go

@@ -24,6 +24,6 @@ import (
 func Test_validateSensor(t *testing.T) {
 	sensor, err := getSensor()
 	assert.Nil(t, err)
-	err = validateSensor(sensor)
+	err = ValidateSensor(sensor)
 	assert.Nil(t, err)
 }

examples/sensors/calendar.yaml

@@ -10,7 +10,6 @@ spec:
   repeat: true
   signals:
     - name: calendar-gateway/calendar.fooConfig
-    - name:
   triggers:
     - name: calendar-workflow-trigger
       resource:

gateways/core/artifact/s3.go

@@ -101,7 +101,7 @@ func (s3ce *s3ConfigExecutor) StartConfig(config *gateways.ConfigContext) error
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-name", config.Data.Src).Msg("parsing configuration...")
 

gateways/core/calendar/calendar.go

@@ -63,7 +63,7 @@ func (ce *calendarConfigExecutor) StartConfig(config *gateways.ConfigContext) er
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-name", config.Data.Src).Msg("parsing configuration...")
 

gateways/core/resource/resource.go

@@ -63,7 +63,7 @@ func (rce *resourceConfigExecutor) StartConfig(config *gateways.ConfigContext) e
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-key", config.Data.Src).Msg("parsing configuration...")
 

gateways/core/stream/amqp/amqp.go

@@ -48,7 +48,7 @@ func (ace *amqpConfigExecutor) StartConfig(config *gateways.ConfigContext) error
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-key", config.Data.Src).Msg("parsing configuration...")
 

gateways/core/stream/kafka/kafka.go

@@ -46,7 +46,7 @@ func (kce *kafkaConfigExecutor) StartConfig(config *gateways.ConfigContext) erro
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	var s *stream.Stream
 	err = yaml.Unmarshal([]byte(config.Data.Config), &s)

gateways/core/stream/mqtt/mqtt.go

@@ -47,7 +47,7 @@ func (mce *mqttConfigExecutor) StartConfig(config *gateways.ConfigContext) error
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-key", config.Data.Src).Msg("parsing configuration...")
 	var wg sync.WaitGroup

gateways/core/stream/nats/nats.go

@@ -46,7 +46,7 @@ func (nce *natsConfigExecutor) StartConfig(config *gateways.ConfigContext) error
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-key", config.Data.Src).Msg("parsing configuration...")
 

gateways/core/webhook/webhook.go

@@ -69,7 +69,7 @@ func (wce *webhookConfigExecutor) StartConfig(config *gateways.ConfigContext) er
 	var errMessage string
 
 	// mark final gateway state
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	gatewayConfig.Log.Info().Str("config-name", config.Data.Src).Msg("parsing configuration...")
 

gateways/gateway.go

@@ -691,16 +691,16 @@ func (gc *GatewayConfig) GetK8Event(reason string, action v1alpha1.NodePhase, co
 }
 
 // GatewayCleanup marks configuration as non-active and marks final gateway state
-func (gc *GatewayConfig) GatewayCleanup(config *ConfigContext, errMessage string, err error) {
+func (gc *GatewayConfig) GatewayCleanup(config *ConfigContext, errMessage *string, err error) {
 	var event *corev1.Event
 	// mark configuration as deactivated so gateway processor client won't run configStopper in case if there
 	// was configuration error.
 	config.Active = false
 	// check if gateway configuration is in error condition.
 	if err != nil {
-		gc.Log.Error().Err(err).Str("config-key", config.Data.Src).Msg(errMessage)
+		gc.Log.Error().Err(err).Str("config-key", config.Data.Src).Msg(*errMessage)
 		// create k8 event for error state
-		event = gc.GetK8Event(errMessage, v1alpha1.NodePhaseError, config.Data)
+		event = gc.GetK8Event(*errMessage, v1alpha1.NodePhaseError, config.Data)
 	} else {
 		// gateway successfully completed/deactivated this configuration.
 		gc.Log.Info().Str("config-key", config.Data.Src).Msg("configuration completed")

gateways/grpc/gateway.go

@@ -53,7 +53,7 @@ func (gce *grpcConfigExecutor) StartConfig(config *gateways.ConfigContext) error
 	var err error
 	var errMessage string
 
-	defer gatewayConfig.GatewayCleanup(config, errMessage, err)
+	defer gatewayConfig.GatewayCleanup(config, &errMessage, err)
 
 	opts := []grpc.DialOption{
 		grpc.WithInsecure(),

gateways/rest/calendar/calendar.go

@@ -72,7 +72,7 @@ func runGateway(config *gateways.ConfigContext) error {
 	var errMessage string
 
 	// mark final gateway state
-	defer httpGatewayServerConfig.GwConfig.GatewayCleanup(config, errMessage, err)
+	defer httpGatewayServerConfig.GwConfig.GatewayCleanup(config, &errMessage, err)
 
 	httpGatewayServerConfig.GwConfig.Log.Info().Str("config-name", config.Data.Src).Msg("parsing calendar schedule...")
 

gateways/rest/gateway.go

@@ -26,7 +26,8 @@ func (hce *httpConfigExecutor) StartConfig(config *gateways.ConfigContext) error
 		Config: config.Data.Config,
 	}, httpGatewayServerConfig.ConfigActivateEndpoint)
 	if err != nil {
-		httpGatewayServerConfig.GwConfig.GatewayCleanup(config, "failed to send new configuration to gateway processor server", err)
+		errMsg := "failed to send new configuration to gateway processor server"
+		httpGatewayServerConfig.GwConfig.GatewayCleanup(config, &errMsg, err)
 	}
 	return err
 }
@@ -38,7 +39,8 @@ func (hce *httpConfigExecutor) StopConfig(config *gateways.ConfigContext) error
 		Config: config.Data.Config,
 	}, httpGatewayServerConfig.ConfigurationDeactivateEndpoint)
 	if err != nil {
-		httpGatewayServerConfig.GwConfig.GatewayCleanup(config, "failed to send configuration to stop to gateway processor server", err)
+		errMsg := "failed to send configuration to stop to gateway processor server"
+		httpGatewayServerConfig.GwConfig.GatewayCleanup(config, &errMsg, err)
 	}
 	return err
 }