feat: Replace gatewayName with eventSourceName in Sensor dependencies (…

…#687)

* feat: Replace gatewayName with eventSourceName in Sensor dependencies

This is the first step of merging Gateway to EventSource.

* fix validation

* test case

* still test case

* codegen

* revert example changes

* revert docs

* still doc

api/openapi-spec/swagger.json

@@ -1813,19 +1813,24 @@
       "required": [
         "name",
         "gatewayName",
+        "eventSourceName",
         "eventName"
       ],
       "properties": {
         "eventName": {
           "description": "EventName is the name of the event",
           "type": "string"
         },
+        "eventSourceName": {
+          "description": "EventSourceName is the name of EventSource that Sensor depends on",
+          "type": "string"
+        },
         "filters": {
           "description": "Filters and rules governing toleration of success and constraints on the context and data of an event",
           "$ref": "#/definitions/io.argoproj.sensor.v1alpha1.EventDependencyFilter"
         },
         "gatewayName": {
-          "description": "GatewayName is the name of the gateway from whom the event is received",
+          "description": "GatewayName is the name of the gateway from whom the event is received DEPRECATED: Use EventSourceName instead.",
           "type": "string"
         },
         "name": {

api/sensor.html

@@ -819,7 +819,19 @@ <h3 id="argoproj.io/v1alpha1.EventDependency">EventDependency
 </em>
 </td>
 <td>
-<p>GatewayName is the name of the gateway from whom the event is received</p>
+<p>GatewayName is the name of the gateway from whom the event is received
+DEPRECATED: Use EventSourceName instead.</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>eventSourceName</code></br>
+<em>
+string
+</em>
+</td>
+<td>
+<p>EventSourceName is the name of EventSource that Sensor depends on</p>
 </td>
 </tr>
 <tr>

api/sensor.md

@@ -1638,6 +1638,27 @@ Name is a unique name of this dependency
 <p>
 
 GatewayName is the name of the gateway from whom the event is received
+DEPRECATED: Use EventSourceName instead.
+
+</p>
+
+</td>
+
+</tr>
+
+<tr>
+
+<td>
+
+<code>eventSourceName</code></br> <em> string </em>
+
+</td>
+
+<td>
+
+<p>
+
+EventSourceName is the name of EventSource that Sensor depends on
 
 </p>
 

controllers/sensor/validate.go

@@ -453,9 +453,9 @@ func validateDependencies(eventDependencies []v1alpha1.EventDependency) error {
 		if dep.Name == "" {
 			return errors.New("event dependency must define a name")
 		}
-
-		if dep.GatewayName == "" {
-			return errors.New("event dependency must define the gateway name")
+		// TODO: GatewayName will be deprecated
+		if dep.EventSourceName == "" && dep.GatewayName == "" {
+			return errors.New("event dependency must define the EventSource name")
 		}
 
 		if dep.EventName == "" {

docs/demo/notebooks.md

@@ -94,7 +94,7 @@ In this demo, we are going to set up an image processing pipeline using 2 notebo
             serviceAccountName: argo-events-sa
           dependencies:
             - name: test-dep
-              gatewayName: webhook-gateway
+              eventSourceName: webhook-event-source
               eventName: example
           subscription:
             http:

gateways/client/cloud-events.go

@@ -138,7 +138,7 @@ func (gatewayContext *GatewayContext) transformEvent(gatewayEvent *gateways.Even
 	event.SetID(fmt.Sprintf("%x", uuid.New()))
 	event.SetSpecVersion(cloudevents.VersionV03)
 	event.SetType(string(gatewayContext.gateway.Spec.Type))
-	event.SetSource(gatewayContext.gateway.Name)
+	event.SetSource(gatewayContext.gateway.Spec.EventSourceRef.Name)
 	event.SetDataContentType("application/json")
 	event.SetSubject(gatewayEvent.Name)
 	event.SetTime(time.Now())

gateways/client/cloud-events_test.go

@@ -37,13 +37,16 @@ func TestTransformEvent(t *testing.T) {
 			},
 			Spec: v1alpha1.GatewaySpec{
 				Type: "webhook",
+				EventSourceRef: &v1alpha1.EventSourceRef{
+					Name: "test-event-source",
+				},
 			},
 		},
 	}
 	cloudevent, err := ctx.transformEvent(event)
 	assert.Nil(t, err)
 	assert.NotNil(t, cloudevent.Context.AsV03())
-	assert.Equal(t, "test-gateway", cloudevent.Source())
+	assert.Equal(t, "test-event-source", cloudevent.Source())
 	assert.Equal(t, "hello", cloudevent.Subject())
 	assert.Equal(t, "webhook", cloudevent.Type())
 

gateways/client/event-source_test.go

@@ -48,6 +48,9 @@ func getGatewayContext() *GatewayContext {
 				Namespace: "fake-namespace",
 			},
 			Spec: v1alpha1.GatewaySpec{
+				EventSourceRef: &v1alpha1.EventSourceRef{
+					Name: "fake-event-source",
+				},
 				Subscribers: &v1alpha1.Subscribers{
 					HTTP: []string{},
 				},

pkg/apis/sensor/v1alpha1/types.go

@@ -207,11 +207,14 @@ type EventDependency struct {
 	// Name is a unique name of this dependency
 	Name string `json:"name" protobuf:"bytes,1,opt,name=name"`
 	// GatewayName is the name of the gateway from whom the event is received
-	GatewayName string `json:"gatewayName" protobuf:"bytes,2,opt,name=gatewayName"`
+	// DEPRECATED: Use EventSourceName instead.
+	GatewayName string `json:"gatewayName" protobuf:"bytes,2,name=gatewayName"`
+	// EventSourceName is the name of EventSource that Sensor depends on
+	EventSourceName string `json:"eventSourceName" protobuf:"bytes,3,name=eventSourceName"`
 	// EventName is the name of the event
-	EventName string `json:"eventName" protobuf:"bytes,3,opt,name=eventName"`
+	EventName string `json:"eventName" protobuf:"bytes,4,name=eventName"`
 	// Filters and rules governing toleration of success and constraints on the context and data of an event
-	Filters *EventDependencyFilter `json:"filters,omitempty" protobuf:"bytes,4,opt,name=filters"`
+	Filters *EventDependencyFilter `json:"filters,omitempty" protobuf:"bytes,5,opt,name=filters"`
 }
 
 // DependencyGroup is the group of dependencies

sensors/dependencies/resolution.go

@@ -18,24 +18,42 @@ package dependencies
 
 import (
 	"github.com/gobwas/glob"
+	"github.com/sirupsen/logrus"
 
 	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
 )
 
 // ResolveDependency resolves a dependency based on Event and gateway name
-func ResolveDependency(dependencies []v1alpha1.EventDependency, events *v1alpha1.Event) *v1alpha1.EventDependency {
+func ResolveDependency(dependencies []v1alpha1.EventDependency, event *v1alpha1.Event, logger *logrus.Logger) *v1alpha1.EventDependency {
 	for _, dependency := range dependencies {
-		gatewayNameGlob, err := glob.Compile(dependency.GatewayName)
+		eventNameGlob, err := glob.Compile(dependency.EventName)
 		if err != nil {
 			continue
 		}
-		eventNameGlob, err := glob.Compile(dependency.EventName)
+		if !eventNameGlob.Match(event.Context.Subject) {
+			continue
+		}
+		var sourceGlob glob.Glob
+		switch {
+		case len(dependency.EventSourceName) > 0:
+			sourceGlob, err = glob.Compile(dependency.EventSourceName)
+		case len(dependency.GatewayName) > 0:
+			// DEPRECATED:
+			logger.WithFields(logrus.Fields{
+				"source":  event.Context.Source,
+				"subject": event.Context.Subject,
+			}).Warn("spec.dependencies.gatewayName is DEPRECATED, it will be unsupported soon, please use spec.dependencies.eventSourceName")
+			sourceGlob, err = glob.Compile(dependency.GatewayName)
+		default:
+			continue
+		}
 		if err != nil {
 			continue
 		}
-		if gatewayNameGlob.Match(events.Context.Source) && eventNameGlob.Match(events.Context.Subject) {
-			return &dependency
+		if !sourceGlob.Match(event.Context.Source) {
+			continue
 		}
+		return &dependency
 	}
 	return nil
 }

sensors/dependencies/resolution_test.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/pkg/apis/sensor/v1alpha1"
 )
 
@@ -35,7 +36,7 @@ func TestResolveDependency(t *testing.T) {
 		{
 			name: "unknown dependency",
 			updateFunc: func() {
-				obj.Spec.Dependencies[0].GatewayName = "fake-gateway"
+				obj.Spec.Dependencies[0].GatewayName = "fake-source"
 				obj.Spec.Dependencies[0].EventName = "example"
 			},
 			testFunc: func(dep *v1alpha1.EventDependency) {
@@ -45,7 +46,7 @@ func TestResolveDependency(t *testing.T) {
 		{
 			name: "known dependency",
 			updateFunc: func() {
-				obj.Spec.Dependencies[0].GatewayName = "fake-gateway"
+				obj.Spec.Dependencies[0].GatewayName = "fake-source"
 				obj.Spec.Dependencies[0].EventName = "example-1"
 			},
 			testFunc: func(dep *v1alpha1.EventDependency) {
@@ -66,7 +67,7 @@ func TestResolveDependency(t *testing.T) {
 
 	event := &v1alpha1.Event{
 		Context: &v1alpha1.EventContext{
-			Source:  "fake-gateway",
+			Source:  "fake-source",
 			Subject: "example-1",
 		},
 		Data: nil,
@@ -75,7 +76,7 @@ func TestResolveDependency(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			test.updateFunc()
-			dependency := ResolveDependency(obj.Spec.Dependencies, event)
+			dependency := ResolveDependency(obj.Spec.Dependencies, event, common.NewArgoEventsLogger())
 			test.testFunc(dependency)
 		})
 	}

sensors/listener.go

@@ -172,7 +172,7 @@ func (sensorCtx *SensorContext) handleEvent(eventBody []byte) error {
 
 	// Resolve Dependency
 	// validate whether the event is from gateway that this sensor is watching
-	if eventDependency := dependencies.ResolveDependency(sensorCtx.Sensor.Spec.Dependencies, internalEvent); eventDependency != nil {
+	if eventDependency := dependencies.ResolveDependency(sensorCtx.Sensor.Spec.Dependencies, internalEvent, sensorCtx.Logger); eventDependency != nil {
 		sensorCtx.NotificationQueue <- &types.Notification{
 			Event:            internalEvent,
 			EventDependency:  eventDependency,