Filter is applied on the metadata of the resource
+Filter is applied on the metadata of the resource +If you apply filter, then the internal event informer will only monitor objects that pass the filter.
eventType
+eventTypes
-ResourceEventType
+[]ResourceEventType
Type is the event type. -If not provided, the gateway will watch all events for a resource.
+EventTypes is the list of event type to watch. +Possible values are - ADD, UPDATE and DELETE.
-Filter is applied on the metadata of the resource +Filter is applied on the metadata of the resource If you apply filter, +then the internal event informer will only monitor objects that pass the +filter.
@@ -3646,20 +3648,18 @@ Group of the resourceeventType
- ResourceEventType
+eventTypes
+ \[\]ResourceEventType
-Type is the event type. If not provided, the gateway will watch all -events for a resource. +EventTypes is the list of event type to watch. Possible values are - +ADD, UPDATE and DELETE.
@@ -4775,5 +4775,4 @@ all types of events will be processed. More info at Generated withgen-crd-api-reference-docs on git
commit 1782e0f.
-
diff --git a/api/gateway.md b/api/gateway.md
index 3f2be472d8..5f91474494 100644
--- a/api/gateway.md
+++ b/api/gateway.md
@@ -1322,5 +1322,4 @@ NATS refers to the subscribers over NATS protocol.
Generated with gen-crd-api-reference-docs on git
commit 1782e0f.
-
diff --git a/api/sensor.md b/api/sensor.md
index 8fb0603526..3a124f7f78 100644
--- a/api/sensor.md
+++ b/api/sensor.md
@@ -6240,5 +6240,4 @@ VerifyCert decides whether the connection is secure or not
Generated with gen-crd-api-reference-docs on git
commit 1782e0f.
-
diff --git a/docs/setup/resource.md b/docs/setup/resource.md
index a3089ff9da..e7dbe1a07c 100644
--- a/docs/setup/resource.md
+++ b/docs/setup/resource.md
@@ -59,7 +59,7 @@ The structure of an event dispatched by the gateway to the sensor looks like fol
metadata:
name: my-workflow
labels:
- app: my-workflow
+ name: my-workflow
spec:
entrypoint: whalesay
templates:
diff --git a/gateways/server/resource/start.go b/gateways/server/resource/start.go
index e848b5d07f..a8f50e2567 100644
--- a/gateways/server/resource/start.go
+++ b/gateways/server/resource/start.go
@@ -111,7 +111,7 @@ func (listener *EventListener) listenEvents(eventSource *gateways.EventSource, c
}
if resourceEventSource.Filter != nil && resourceEventSource.Filter.Fields != nil {
- sel, err := LabelSelector(resourceEventSource.Filter.Fields)
+ sel, err := FieldSelector(resourceEventSource.Filter.Fields)
if err != nil {
return errors.Wrapf(err, "failed to create the field selector for the event source %s", eventSource.Name)
}
@@ -128,66 +128,88 @@ func (listener *EventListener) listenEvents(eventSource *gateways.EventSource, c
informer := factory.ForResource(gvr)
informerEventCh := make(chan *InformerEvent)
+ stopCh := make(chan struct{})
go func() {
logger.Infoln("listening to resource events...")
- for event := range informerEventCh {
- objBody, err := json.Marshal(event.Obj)
- if err != nil {
- logger.WithError(err).Errorln("failed to marshal the resource, rejecting the event...")
- continue
- }
+ for {
+ select {
+ case event := <-informerEventCh:
+ objBody, err := json.Marshal(event.Obj)
+ if err != nil {
+ logger.WithError(err).Errorln("failed to marshal the resource, rejecting the event...")
+ continue
+ }
- eventData := &events.ResourceEventData{
- EventType: string(event.Type),
- Body: (*json.RawMessage)(&objBody),
- Group: resourceEventSource.Group,
- Version: resourceEventSource.Version,
- Resource: resourceEventSource.Resource,
- }
- eventBody, err := json.Marshal(eventData)
- if err != nil {
- logger.WithError(err).Errorln("failed to marshal the event. rejecting the event...")
- continue
- }
- if err := passFilters(event, resourceEventSource.Filter, resourceEventSource.EventType); err != nil {
- logger.WithError(err).Warnln("failed to apply the filter, rejecting the event...")
- continue
+ eventData := &events.ResourceEventData{
+ EventType: string(event.Type),
+ Body: (*json.RawMessage)(&objBody),
+ Group: resourceEventSource.Group,
+ Version: resourceEventSource.Version,
+ Resource: resourceEventSource.Resource,
+ }
+ eventBody, err := json.Marshal(eventData)
+ if err != nil {
+ logger.WithError(err).Errorln("failed to marshal the event. rejecting the event...")
+ continue
+ }
+ if err := passFilters(event, resourceEventSource.Filter); err != nil {
+ logger.WithError(err).Warnln("failed to apply the filter, rejecting the event...")
+ continue
+ }
+ channels.Data <- eventBody
+ case <-stopCh:
+ return
}
- channels.Data <- eventBody
}
}()
- sharedInformer := informer.Informer()
- sharedInformer.AddEventHandler(
- cache.ResourceEventHandlerFuncs{
- AddFunc: func(obj interface{}) {
+ handlerFuncs := cache.ResourceEventHandlerFuncs{}
+
+ for _, eventType := range resourceEventSource.EventTypes {
+ switch eventType {
+ case v1alpha1.ADD:
+ handlerFuncs.AddFunc = func(obj interface{}) {
+ logger.Infoln("detected create event")
informerEventCh <- &InformerEvent{
Obj: obj,
Type: v1alpha1.ADD,
}
- },
- UpdateFunc: func(oldObj, newObj interface{}) {
+ }
+ case v1alpha1.UPDATE:
+ handlerFuncs.UpdateFunc = func(oldObj, newObj interface{}) {
+ logger.Infoln("detected update event")
informerEventCh <- &InformerEvent{
Obj: newObj,
OldObj: oldObj,
Type: v1alpha1.UPDATE,
}
- },
- DeleteFunc: func(obj interface{}) {
+ }
+ case v1alpha1.DELETE:
+ handlerFuncs.DeleteFunc = func(obj interface{}) {
+ logger.Infoln("detected delete event")
informerEventCh <- &InformerEvent{
Obj: obj,
Type: v1alpha1.DELETE,
}
- },
- },
- )
+ }
+ default:
+ stopCh <- struct{}{}
+ return errors.Errorf("unknown event type: %s", string(eventType))
+ }
+ }
+
+ sharedInformer := informer.Informer()
+ sharedInformer.AddEventHandler(handlerFuncs)
doneCh := make(chan struct{})
+
+ logger.Infoln("running informer...")
sharedInformer.Run(doneCh)
<-channels.Done
doneCh <- struct{}{}
+ stopCh <- struct{}{}
logger.Infoln("event source is stopped")
close(informerEventCh)
@@ -231,7 +253,7 @@ func FieldSelector(fieldSelectors map[string]string) (fields.Selector, error) {
}
// helper method to check if the object passed the user defined filters
-func passFilters(event *InformerEvent, filter *v1alpha1.ResourceFilter, eventType v1alpha1.ResourceEventType) error {
+func passFilters(event *InformerEvent, filter *v1alpha1.ResourceFilter) error {
uObj := event.Obj.(*unstructured.Unstructured)
// no filters are applied.
if filter == nil {
@@ -244,8 +266,5 @@ func passFilters(event *InformerEvent, filter *v1alpha1.ResourceFilter, eventTyp
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
}
diff --git a/gateways/server/resource/start_test.go b/gateways/server/resource/start_test.go
index fa24110812..6383f942c7 100644
--- a/gateways/server/resource/start_test.go
+++ b/gateways/server/resource/start_test.go
@@ -64,13 +64,13 @@ func TestFilter(t *testing.T) {
err = passFilters(&InformerEvent{
Obj: &unstructured.Unstructured{Object: outmap},
Type: "ADD",
- }, resourceEventSource.Filter, v1alpha1.ADD)
+ }, resourceEventSource.Filter)
convey.So(err, convey.ShouldBeNil)
err = passFilters(&InformerEvent{
Obj: &unstructured.Unstructured{Object: outmap},
Type: "ADD",
- }, resourceEventSource.Filter, v1alpha1.UPDATE)
+ }, resourceEventSource.Filter)
convey.So(err, convey.ShouldNotBeNil)
})
}
diff --git a/gateways/server/resource/validate.go b/gateways/server/resource/validate.go
index 727dec6652..1a0bb5133f 100644
--- a/gateways/server/resource/validate.go
+++ b/gateways/server/resource/validate.go
@@ -68,5 +68,8 @@ func validate(eventSource *v1alpha1.ResourceEventSource) error {
if eventSource.Resource == "" {
return fmt.Errorf("resource must be specified")
}
+ if eventSource.EventTypes == nil {
+ return fmt.Errorf("event types must be specified")
+ }
return nil
}
diff --git a/pkg/apis/eventsources/v1alpha1/openapi_generated.go b/pkg/apis/eventsources/v1alpha1/openapi_generated.go
index 5ac0f4785e..7b9f610453 100644
--- a/pkg/apis/eventsources/v1alpha1/openapi_generated.go
+++ b/pkg/apis/eventsources/v1alpha1/openapi_generated.go
@@ -1450,7 +1450,7 @@ func schema_pkg_apis_eventsources_v1alpha1_ResourceEventSource(ref common.Refere
},
"filter": {
SchemaProps: spec.SchemaProps{
- Description: "Filter is applied on the metadata of the resource",
+ Description: "Filter is applied on the metadata of the resource If you apply filter, then the internal event informer will only monitor objects that pass the filter.",
Ref: ref("github.com/argoproj/argo-events/pkg/apis/eventsources/v1alpha1.ResourceFilter"),
},
},
@@ -1472,15 +1472,22 @@ func schema_pkg_apis_eventsources_v1alpha1_ResourceEventSource(ref common.Refere
Format: "",
},
},
- "eventType": {
+ "eventTypes": {
SchemaProps: spec.SchemaProps{
- Description: "Type is the event type. If not provided, the gateway will watch all events for a resource.",
- Type: []string{"string"},
- Format: "",
+ Description: "EventTypes is the list of event type to watch. Possible values are - ADD, UPDATE and DELETE.",
+ Type: []string{"array"},
+ Items: &spec.SchemaOrArray{
+ Schema: &spec.Schema{
+ SchemaProps: spec.SchemaProps{
+ Type: []string{"string"},
+ Format: "",
+ },
+ },
+ },
},
},
},
- Required: []string{"namespace", "group", "version", "resource"},
+ Required: []string{"namespace", "group", "version", "resource", "eventTypes"},
},
},
Dependencies: []string{
diff --git a/pkg/apis/eventsources/v1alpha1/types.go b/pkg/apis/eventsources/v1alpha1/types.go
index 78eb26ab6f..a83e66ea16 100644
--- a/pkg/apis/eventsources/v1alpha1/types.go
+++ b/pkg/apis/eventsources/v1alpha1/types.go
@@ -140,14 +140,14 @@ type ResourceEventSource struct {
// Namespace where resource is deployed
Namespace string `json:"namespace" protobuf:"bytes,1,name=namespace"`
// Filter is applied on the metadata of the resource
+ // If you apply filter, then the internal event informer will only monitor objects that pass the filter.
// +optional
Filter *ResourceFilter `json:"filter,omitempty" protobuf:"bytes,2,opt,name=filter"`
// Group of the resource
metav1.GroupVersionResource `json:",inline"`
- // Type is the event type.
- // If not provided, the gateway will watch all events for a resource.
- // +optional
- EventType ResourceEventType `json:"eventType,omitempty" protobuf:"bytes,3,opt,name=eventType"`
+ // EventTypes is the list of event type to watch.
+ // Possible values are - ADD, UPDATE and DELETE.
+ EventTypes []ResourceEventType `json:"eventTypes" protobuf:"bytes,3,name=eventTypes"`
}
// ResourceFilter contains K8 ObjectMeta information to further filter resource event objects
diff --git a/pkg/apis/eventsources/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/eventsources/v1alpha1/zz_generated.deepcopy.go
index f4945e49c1..ad5630492c 100644
--- a/pkg/apis/eventsources/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/eventsources/v1alpha1/zz_generated.deepcopy.go
@@ -664,6 +664,11 @@ func (in *ResourceEventSource) DeepCopyInto(out *ResourceEventSource) {
(*in).DeepCopyInto(*out)
}
out.GroupVersionResource = in.GroupVersionResource
+ if in.EventTypes != nil {
+ in, out := &in.EventTypes, &out.EventTypes
+ *out = make([]ResourceEventType, len(*in))
+ copy(*out, *in)
+ }
return
}