Support for passing complete event payload from signal to trigger

CONTRIBUTING.md

@@ -40,7 +40,7 @@ minikube start
 eval $(minikube docker-env)
 ```
 
-#### 5. Build the project & Docker images
+#### 5. Build the project
 ```
 make all
 ```

Makefile

@@ -35,13 +35,13 @@ endif
 
 # Build the project images
 .DELETE_ON_ERROR:
-all: sensor-linux sensor-controller-linux gateway-controller-linux gateway-transformer-linux webhook-linux calendar-linux artifact-linux nats-linux kafka-linux amqp-linux mqtt-linux
+all: sensor-linux sensor-controller-linux gateway-controller-linux gateway-http-transformer-linux webhook-linux calendar-linux resource-linux artifact-linux nats-linux kafka-linux amqp-linux mqtt-linux gateway-processor-grpc-client-linux calendar-grpc-linux gateway-processor-http-client-linux calendar-http-linux
 
-all-images: sensor-image sensor-controller-image gateway-controller-image gateway-http-transformer-image webhook-image calendar-image artifact-image nats-image kafka-image amqp-image mqtt-image gateway-processor-grpc-client-image calendar-grpc-image gateway-processor-http-client-image calendar-http-image
+all-images: sensor-image sensor-controller-image gateway-controller-image gateway-http-transformer-image webhook-image calendar-image resource-image artifact-image nats-image kafka-image amqp-image mqtt-image gateway-processor-grpc-client-image calendar-grpc-image gateway-processor-http-client-image calendar-http-image
 
 all-controller-images: sensor-controller-image gateway-controller-image
 
-all-gateway-images: webhook-image calendar-image artifact-image nats-image kafka-image amqp-image mqtt-image
+all-core-gateway-images: webhook-image calendar-image artifact-image nats-image kafka-image amqp-image mqtt-image resource-image
 
 .PHONY: all clean test
 
@@ -115,6 +115,18 @@ calendar-image: calendar-linux
 	@if [ "$(DOCKER_PUSH)" = "true" ] ; then  docker push $(IMAGE_PREFIX)calendar-gateway:$(IMAGE_TAG) ; fi
 
 
+resource:
+	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -v -ldflags '${LDFLAGS}' -o ${DIST_DIR}/resource-gateway ./gateways/core/resource/
+
+resource-linux:
+	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make resource
+
+resource-image: resource-linux
+	 docker build -t $(IMAGE_PREFIX)resource-gateway:$(IMAGE_TAG) -f ./gateways/core/resource/Dockerfile .
+	@if [ "$(DOCKER_PUSH)" = "true" ] ; then  docker push $(IMAGE_PREFIX)resource-gateway:$(IMAGE_TAG) ; fi
+
+
+
 artifact:
 	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -v -ldflags '${LDFLAGS}' -o ${DIST_DIR}/artifact-gateway ./gateways/core/artifact/
 

README.md

@@ -17,7 +17,7 @@ Argo Events is an open source event-based dependency manager for Kubernetes. The
 - Runtime agnostic. The first runtime and package agnostic event framework for Kubernetes.
 - Containers. Designed from the ground-up as Kubernetes-native. 
 - Extremely lightweight. All gateways, with the exception of calendar-based gateways, are event-driven, meaning that there is no polling involved.
-- Configurable. Select gateways you want to support, deploy those to Kubernetes and configure them on the fly
+- Configurable. Configure gateways at runtime
 - Scalable & Resilient.
 - Simple or Complex dependencies. Manage everything from simple, linear, real-time dependencies to complex, multi-source, batch job dependencies.
 

cmd/controllers/gateway/main.go

@@ -23,7 +23,7 @@ func main() {
 
 	namespace, ok := os.LookupEnv(common.GatewayNamespace)
 	if !ok {
-		namespace = common.DefaultGatewayControllerNamespace
+		namespace = common.DefaultControllerNamespace
 	}
 
 	// create new gateway controller

cmd/controllers/sensor/main.go

@@ -39,7 +39,7 @@ func main() {
 
 	namespace, ok := os.LookupEnv(common.SensorNamespace)
 	if !ok {
-		namespace = common.DefaultGatewayControllerNamespace
+		namespace = common.DefaultControllerNamespace
 	}
 
 	// create a new sensor controller

common/common.go

@@ -40,6 +40,9 @@ const (
 
 	// StandardTimeFormat is time format reference for golang
 	StandardTimeFormat = "2006-01-02 15:04:05"
+
+	// StandardYYYYMMDDFormat formats date in yyyy-mm-dd format
+	StandardYYYYMMDDFormat = "2006-01-02"
 )
 
 // SENSOR CONTROLLER CONSTANTS

common/namespace.go

@@ -25,15 +25,11 @@ import (
 const namespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
 
 var (
-	// DefaultSensorControllerNamespace is the default namespace where the sensor sensor-controller is installed
-	DefaultSensorControllerNamespace = "argo-events"
-
-	// Todo: Does this even have to be separate
-	// DefaultGatewayControllerNamespace is the default namespace where the sensor sensor-controller is installed
-	DefaultGatewayControllerNamespace = "argo-events"
+	// DefaultControllerNamespace is the default namespace where the sensor and gateways controllers are installed
+	DefaultControllerNamespace = "argo-events"
 
 	// ErrReadNamespace occurs when the namespace cannot be read from a Kubernetes pod's service account token
-	ErrReadNamespace = errors.New("Could not read namespace from service account secret")
+	ErrReadNamespace = errors.New("could not read namespace from service account secret")
 )
 
 func init() {
@@ -61,16 +57,16 @@ func RefreshNamespace() {
 	// 1 - env variable
 	nm, ok := os.LookupEnv(EnvVarNamespace)
 	if ok {
-		DefaultSensorControllerNamespace = nm
+		DefaultControllerNamespace = nm
 		return
 	}
 
 	// 2 - pod service account token
 	nm, err := detectNamespace()
 	if err == nil {
-		DefaultSensorControllerNamespace = nm
+		DefaultControllerNamespace = nm
 	}
 
-	// 3 - use the DefaultSensorControllerNamespace
+	// 3 - use the DefaultControllerNamespace
 	return
 }

common/namespace_test.go

@@ -25,8 +25,7 @@ import (
 func TestResolveNamespace(t *testing.T) {
 	defer os.Unsetenv(EnvVarNamespace)
 
-	RefreshNamespace()
-	assert.Equal(t, "argo-events", DefaultSensorControllerNamespace)
+	assert.Equal(t, "argo-events", DefaultControllerNamespace)
 
 	// TODO: now write the namespace file
 
@@ -37,5 +36,5 @@ func TestResolveNamespace(t *testing.T) {
 	}
 
 	RefreshNamespace()
-	assert.Equal(t, "test", DefaultSensorControllerNamespace)
+	assert.Equal(t, "test", DefaultControllerNamespace)
 }

controllers/gateway/config.go

@@ -98,7 +98,7 @@ func (c *GatewayController) updateConfig(cm *apiv1.ConfigMap) error {
 		return err
 	}
 	if config.Namespace == "" {
-		config.Namespace = common.DefaultGatewayControllerNamespace
+		config.Namespace = common.DefaultControllerNamespace
 	}
 	c.Config = config
 	return nil

controllers/gateway/config_test.go

@@ -56,5 +56,5 @@ func TestGatewayController_ResyncConfig(t *testing.T) {
 	assert.NotNil(t, cm)
 	assert.NotNil(t, gc.Config)
 	assert.NotEqual(t, gc.Config.Namespace, gc.ConfigMapNS)
-	assert.Equal(t, gc.Config.Namespace, common.DefaultGatewayControllerNamespace)
+	assert.Equal(t, gc.Config.Namespace, common.DefaultControllerNamespace)
 }

controllers/sensor/config.go

@@ -114,7 +114,7 @@ func (c *SensorController) updateConfig(cm *apiv1.ConfigMap) error {
 		return err
 	}
 	if config.Namespace == "" {
-		config.Namespace = common.DefaultSensorControllerNamespace
+		config.Namespace = common.DefaultControllerNamespace
 	}
 	c.Config = config
 	return nil

controllers/sensor/controller.go

@@ -20,20 +20,18 @@ import (
 	"context"
 	"errors"
 	"time"
-
+	"fmt"
+	"log"
 
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
-
 	base "github.com/argoproj/argo-events"
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
 	sensorclientset "github.com/argoproj/argo-events/pkg/client/sensor/clientset/versioned"
-	"fmt"
-	"log"
 )
 
 const (

controllers/sensor/notification-handler.go

@@ -166,7 +166,6 @@ func (se *sensorExecutionCtx) processSignal(gwEventWrapper *sensorEventWrapper)
 
 	// apply filters if any.
 	// error is thrown if some problem occurs during filtering the signal
-	// apply filters if any
 	ok, err := se.filterEvent(gwEventWrapper.signal.Filters, gwEventWrapper.event)
 	if err != nil {
 		se.log.Error().Err(err).Str("signal-name", gwEventWrapper.event.Context.Source.Host).Err(err).Msg("failed to apply filter")

controllers/sensor/signal-filter.go

@@ -91,13 +91,9 @@ func (se *sensorExecutionCtx) filterTime(timeFilter *v1alpha1.TimeFilter, eventT
 	if timeFilter != nil {
 		se.log.Info().Str("event-time", eventTime.String()).Msg("event time")
 		currentT := time.Now().UTC()
-		se.log.Info().Str("current-time", currentT.String()).Msg("current time")
-		currentMonth := fmt.Sprintf("%d", int(currentT.Month()))
-		if int(currentT.Month()) < 10 {
-			currentMonth = "0" + currentMonth
-		}
-		currentTStr := fmt.Sprintf("%d-%s-%d", currentT.Year(), currentMonth, currentT.Day())
-
+		currentT = time.Date(currentT.Year(), currentT.Month(), currentT.Day(), 0, 0, 0, 0, time.UTC)
+		currentTStr := currentT.Format(common.StandardYYYYMMDDFormat)
+		se.log.Info().Str("date", currentTStr).Msg("current date")
 		if timeFilter.Start != "" && timeFilter.Stop != "" {
 			se.log.Info().Str("start time format", currentTStr+" "+timeFilter.Start).Msg("start time format")
 			startTime, err := time.Parse(common.StandardTimeFormat, currentTStr+" "+timeFilter.Start)

controllers/sensor/signal-filter_test.go

@@ -16,7 +16,6 @@ limitations under the License.
 package sensor
 
 import (
-	"fmt"
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
 	"github.com/stretchr/testify/assert"
@@ -36,12 +35,10 @@ func Test_filterTime(t *testing.T) {
 		Start: "10:11:00",
 	}
 	event := getCloudEvent()
+
 	currentT := time.Now().UTC()
-	currentMonth := fmt.Sprintf("%d", int(currentT.Month()))
-	if int(currentT.Month()) < 10 {
-		currentMonth = "0" + currentMonth
-	}
-	currentTStr := fmt.Sprintf("%d-%s-%d", currentT.Year(), currentMonth, currentT.Day())
+	currentT = time.Date(currentT.Year(), currentT.Month(), currentT.Day(), 0, 0, 0, 0, time.UTC)
+	currentTStr := currentT.Format(common.StandardYYYYMMDDFormat)
 	parsedTime, err := time.Parse(common.StandardTimeFormat, currentTStr+" 16:36:34")
 	assert.Nil(t, err)
 	event.Context.EventTime = metav1.MicroTime{

controllers/sensor/trigger-params.go

@@ -55,6 +55,10 @@ func resolveParamValue(src *v1alpha1.ResourceParameterSource, events map[string]
 			}
 			return "", err
 		}
+		// check if complete payload needs to be passed to the trigger
+		if src.Path == "" {
+			return string(js), nil
+		}
 		res := gjson.GetBytes(js, src.Path)
 		if res.Exists() {
 			return res.String(), nil

controllers/sensor/validate.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // ValidateSensor accepts a sensor and performs validation against it
@@ -83,8 +82,9 @@ func validateSignalFilter(filter v1alpha1.SignalFilter) error {
 }
 
 func validateSignalTimeFilter(tFilter *v1alpha1.TimeFilter) error {
-	currentT := metav1.Time{Time: time.Now().UTC()}
-	currentTStr := fmt.Sprintf("%d-%d-%d", currentT.Year(), int(currentT.Month()), currentT.Day())
+	currentT := time.Now().UTC()
+	currentT = time.Date(currentT.Year(), currentT.Month(), currentT.Day(), 0, 0, 0, 0, time.UTC)
+	currentTStr := currentT.Format(common.StandardYYYYMMDDFormat)
 	if tFilter.Start != "" && tFilter.Stop != "" {
 		startTime, err := time.Parse(common.StandardTimeFormat, currentTStr+" "+tFilter.Start)
 		if err != nil {
@@ -104,7 +104,7 @@ func validateSignalTimeFilter(tFilter *v1alpha1.TimeFilter) error {
 			return err
 		}
 		stopTime = stopTime.UTC()
-		if stopTime.Before(currentT.Time) {
+		if stopTime.Before(currentT.UTC()) {
 			return fmt.Errorf("invalid signal time filter: stop '%s' is before the current time '%s'", tFilter.Stop, currentT)
 		}
 	}

docs/artifact-guide.md

@@ -1,7 +1,5 @@
 # Artifact Guide
-This is a guide for help in utilizing artifacts within Argo Events. Sensors use artifacts for two purposes:
-1. Object notifications for use in `Artifact` signals. (currently S3 bucket notifications are only supported)
-2. A Resource Object store for use in `Resource` triggers
+This is a guide for help in utilizing artifacts within Argo Events. Sensors use artifacts for Resource Object store for use in `Resource` triggers
 
 ## Inline
 Inlined artifacts are included directly within the sensor resource and decoded as a string.