fix(resource-gateway): fix event type filter (#498)

gateways/server/resource/start.go

@@ -156,7 +156,7 @@ func (listener *EventListener) listenEvents(eventSource *gateways.EventSource, c
 					logger.WithError(err).Errorln("failed to marshal the event. rejecting the event...")
 					continue
 				}
-				if err := passFilters(event.Obj.(*unstructured.Unstructured), resourceEventSource.Filter); err != nil {
+				if err := passFilters(event, resourceEventSource.Filter, resourceEventSource.EventType); err != nil {
 					logger.WithError(err).Warnln("failed to apply the filter, rejecting the event...")
 					continue
 				}
@@ -238,17 +238,21 @@ func FieldSelector(fieldSelectors map[string]string) (fields.Selector, error) {
 }
 
 // helper method to check if the object passed the user defined filters
-func passFilters(obj *unstructured.Unstructured, filter *v1alpha1.ResourceFilter) error {
+func passFilters(event *InformerEvent, filter *v1alpha1.ResourceFilter, eventType v1alpha1.ResourceEventType) error {
+	uObj := event.Obj.(*unstructured.Unstructured)
 	// no filters are applied.
 	if filter == nil {
 		return nil
 	}
-	if !strings.HasPrefix(obj.GetName(), filter.Prefix) {
-		return errors.Errorf("resource name does not match prefix. resource-name: %s, prefix: %s", obj.GetName(), filter.Prefix)
+	if !strings.HasPrefix(uObj.GetName(), filter.Prefix) {
+		return errors.Errorf("resource name does not match prefix. resource-name: %s, prefix: %s", uObj.GetName(), filter.Prefix)
 	}
-	created := obj.GetCreationTimestamp()
+	created := uObj.GetCreationTimestamp()
 	if !filter.CreatedBy.IsZero() && created.UTC().After(filter.CreatedBy.UTC()) {
 		return errors.Errorf("resource is created after filter time. creation-timestamp: %s, filter-creation-timestamp: %s", created.UTC().String(), filter.CreatedBy.UTC().String())
 	}
+	if eventType != "" && event.Type != eventType {
+		return errors.Errorf("resource event type mismatch. expected: %s, actual: %s", string(eventType), string(event.Type))
+	}
 	return nil
 }

gateways/server/resource/start_test.go

@@ -17,14 +17,14 @@ limitations under the License.
 package resource
 
 import (
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"testing"
 
 	"github.com/argoproj/argo-events/pkg/apis/eventsources/v1alpha1"
 	"github.com/mitchellh/mapstructure"
 	"github.com/smartystreets/goconvey/convey"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/kubernetes/fake"
 )
 
@@ -61,9 +61,16 @@ func TestFilter(t *testing.T) {
 		err = mapstructure.Decode(pod, &outmap)
 		convey.So(err, convey.ShouldBeNil)
 
-		err = passFilters(&unstructured.Unstructured{
-			Object: outmap,
-		}, resourceEventSource.Filter)
+		err = passFilters(&InformerEvent{
+			Obj:  &unstructured.Unstructured{Object: outmap},
+			Type: "ADD",
+		}, resourceEventSource.Filter, v1alpha1.ADD)
 		convey.So(err, convey.ShouldBeNil)
+
+		err = passFilters(&InformerEvent{
+			Obj:  &unstructured.Unstructured{Object: outmap},
+			Type: "ADD",
+		}, resourceEventSource.Filter, v1alpha1.UPDATE)
+		convey.So(err, convey.ShouldNotBeNil)
 	})
 }