CloudEvents integration

docs/api.md

@@ -16,23 +16,27 @@ for invoking function. By default Events API runs on `:4000` port.
 
 ### Event Definition
 
-All data that passes through the Event Gateway is formatted as an Event, based on our default Event schema:
+All data that passes through the Event Gateway is formatted as an CloudEvent, based on CloudEvent schema:
 
-* `event` - `string` - the event name
-* `id` - `string` - the event's instance universally unique ID (provided by the event gateway)
-* `receivedAt` - `number` - the time (milliseconds) when the Event was received by the Event Gateway (provided by the event gateway)
-* `data` - type depends on `dataType` - the event payload
-* `dataType` - `string` - the mime type of `data` payload
+* `event-type` - `string` - the event name
+* `event-id` - `string` - the event's instance universally unique ID (provided by the event gateway)
+* `cloud-events-version` - `string` - the version of CloudEvent definition
+* `source` - `string` - source for the event
+* `event-time` - `string` - RFC 3339 formatted time when the Event was received by the Event Gateway (provided by the event gateway)
+* `data` - type depends on `content-type` - the event payload
+* `content-type` - `string` - the mime type of `data` payload
 
 Example:
 
 ```json
 {
-  "event": "myapp.user.created",
-  "id": "66dfc31d-6844-42fd-b1a7-a489a49f65f3",
-  "receivedAt": 1500897327098,
+  "event-type": "myapp.user.created",
+  "event-id": "66dfc31d-6844-42fd-b1a7-a489a49f65f3",
+  "cloud-events-version": "0.1",
+  "source": "", // TBD
+  "event-time": "1990-12-31T23:59:60Z",
   "data": { "foo": "bar" },
-  "dataType": "application/json"
+  "content-type": "application/json"
 }
 ```
 

event/event.go

@@ -8,26 +8,34 @@ import (
 	"go.uber.org/zap/zapcore"
 
 	uuid "github.com/satori/go.uuid"
+	"github.com/serverless/event-gateway/internal/zap"
 )
 
 // Event is a default event structure. All data that passes through the Event Gateway is formatted as an Event, based on
 // this schema.
 type Event struct {
-	Type       Type        `json:"event"`
-	ID         string      `json:"id"`
-	ReceivedAt uint64      `json:"receivedAt"`
-	Data       interface{} `json:"data"`
-	DataType   string      `json:"dataType"`
+	EventType          Type                   `json:"event-type"`
+	EventTypeVersion   string                 `json:"event-type-version"`
+	CloudEventsVersion string                 `json:"cloud-events-version"`
+	Source             string                 `json:"source"`
+	EventID            string                 `json:"event-id"`
+	EventTime          time.Time              `json:"event-time"`
+	SchemaURL          string                 `json:"schema-url"`
+	ContentType        string                 `json:"content-type"`
+	Extensions         zap.MapStringInterface `json:"extensions"`
+	Data               interface{}            `json:"data"`
 }
 
 // New return new instance of Event.
 func New(eventType Type, mime string, payload interface{}) *Event {
 	return &Event{
-		Type:       eventType,
-		ID:         uuid.NewV4().String(),
-		ReceivedAt: uint64(time.Now().UnixNano() / int64(time.Millisecond)),
-		DataType:   mime,
-		Data:       payload,
+		EventType:          eventType,
+		CloudEventsVersion: "0.1",
+		Source:             "",
+		EventID:            uuid.NewV4().String(),
+		EventTime:          time.Now(),
+		ContentType:        mime,
+		Data:               payload,
 	}
 }
 
@@ -42,17 +50,22 @@ const TypeHTTP = Type("http")
 
 // MarshalLogObject is a part of zapcore.ObjectMarshaler interface
 func (e Event) MarshalLogObject(enc zapcore.ObjectEncoder) error {
-	enc.AddString("type", string(e.Type))
-	enc.AddString("id", e.ID)
-	enc.AddUint64("receivedAt", e.ReceivedAt)
+	enc.AddString("event-type", string(e.EventType))
+	enc.AddString("event-type-version", e.EventTypeVersion)
+	enc.AddString("cloud-events-version", e.CloudEventsVersion)
+	enc.AddString("source", e.Source)
+	enc.AddString("event-id", e.EventID)
+	enc.AddString("event-time", e.EventTime.String())
+	enc.AddString("schema-url", e.SchemaURL)
+	enc.AddString("content-type", e.ContentType)
+	e.Extensions.MarshalLogObject(enc)
 	payload, _ := json.Marshal(e.Data)
 	enc.AddString("data", string(payload))
-	enc.AddString("dataType", e.DataType)
 
 	return nil
 }
 
 // IsSystem indicates if th event is a system event.
 func (e Event) IsSystem() bool {
-	return strings.HasPrefix(string(e.Type), "gateway.")
+	return strings.HasPrefix(string(e.EventType), "gateway.")
 }

internal/zap/strings.go

@@ -1,14 +1,33 @@
 package zap
 
-import "go.uber.org/zap/zapcore"
+import (
+	"encoding/json"
+	"go.uber.org/zap/zapcore"
+)
 
 // Strings is a string array that implements MarshalLogArray.
 type Strings []string
 
+// MapStringInterface is a map that implements MarshalLogObject.
+type MapStringInterface map[string]interface{}
+
 // MarshalLogArray implementation
 func (ss Strings) MarshalLogArray(enc zapcore.ArrayEncoder) error {
 	for _, s := range ss {
 		enc.AppendString(s)
 	}
 	return nil
 }
+
+// MarshalLogObject implementation
+func (msi MapStringInterface) MarshalLogObject(enc zapcore.ObjectEncoder) error {
+	for key, val := range msi {
+		v, err := json.Marshal(val)
+		if err != nil {
+			enc.AddString(key, string(v))
+		} else {
+			return err
+		}
+	}
+	return nil
+}

plugin/example/plugin.go

@@ -29,10 +29,10 @@ func (s *Simple) Subscriptions() []plugin.Subscription {
 
 // React is called for every event that plugin subscribed to.
 func (s *Simple) React(instance event.Event) error {
-	switch instance.Type {
+	switch instance.EventType {
 	case event.SystemEventReceivedType:
 		received := instance.Data.(event.SystemEventReceivedData)
-		log.Printf("received gateway.received.event for event: %q", received.Event.Type)
+		log.Printf("received gateway.received.event for event: %q", received.Event.EventType)
 		break
 	}
 

plugin/manager.go

@@ -88,7 +88,7 @@ func (m *Manager) Kill() {
 func (m *Manager) React(event *event.Event) error {
 	for _, plugin := range m.Plugins {
 		for _, subscription := range plugin.Subscriptions {
-			if subscription.EventType == event.Type {
+			if subscription.EventType == event.EventType {
 				err := plugin.Reacter.React(*event)
 				if err != nil {
 					m.Log.Debug("Plugin returned error.",

router/event.go

@@ -80,7 +80,7 @@ func (router *Router) eventFromRequest(r *http.Request) (*eventpkg.Event, string
 		}
 	}
 
-	if event.Type == eventpkg.TypeHTTP {
+	if event.EventType == eventpkg.TypeHTTP {
 		event.Data = &eventpkg.HTTPEvent{
 			Headers: headers,
 			Query:   r.URL.Query(),

router/router.go

@@ -87,7 +87,7 @@ func (router *Router) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 
-			if event.Type == eventpkg.TypeInvoke {
+			if event.EventType == eventpkg.TypeInvoke {
 				functionID := function.ID(r.Header.Get(headerFunctionID))
 				space := r.Header.Get(headerSpace)
 				if space == "" {
@@ -100,7 +100,7 @@ func (router *Router) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 				metricEventsProcessed.WithLabelValues(space, "invoke").Inc()
 			} else if !event.IsSystem() {
-				reportReceivedEvent(event.ID)
+				reportReceivedEvent(event.EventID)
 
 				router.enqueueWork(path, event)
 				w.WriteHeader(http.StatusAccepted)
@@ -447,9 +447,9 @@ func (router *Router) loop() {
 
 // processEvent call all functions subscribed for an event
 func (router *Router) processEvent(e backlogEvent) {
-	reportEventOutOfQueue(e.event.ID)
+	reportEventOutOfQueue(e.event.EventID)
 
-	subscribers := router.targetCache.SubscribersOfEvent(e.path, e.event.Type)
+	subscribers := router.targetCache.SubscribersOfEvent(e.path, e.event.EventType)
 	for _, subscriber := range subscribers {
 		router.callFunction(subscriber.Space, subscriber.ID, e.event)
 	}

router/router_test.go

@@ -155,6 +155,44 @@ func TestRouterServeHTTP_Encoding(t *testing.T) {
 	}
 }
 
+func TestRouterServeHTTP_CloudEvents(t *testing.T) {
+	var contentType = "application/json"
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	testListServer := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			testevent := event.Event{}
+			json.NewDecoder(r.Body).Decode(&testevent)
+			defer r.Body.Close()
+
+			assert.Equal(t, testevent.EventType, event.TypeHTTP)
+			assert.Equal(t, testevent.CloudEventsVersion, "0.1")
+			assert.NotEqual(t, testevent.Source, nil)
+			assert.NotEqual(t, len(testevent.EventID), 0)
+			assert.NotEqual(t, testevent.EventTime, nil)
+			assert.Equal(t, testevent.ContentType, contentType)
+		}))
+	defer testListServer.Close()
+	target := mock.NewMockTargeter(ctrl)
+	someFunc := function.Function{
+		Space:        "",
+		ID:           "somefunc",
+		ProviderType: httpprovider.Type,
+		Provider: httpprovider.HTTP{
+			URL: testListServer.URL,
+		},
+	}
+	target.EXPECT().HTTPBackingFunction(http.MethodPost, "/").Return("", &someFunc.ID, pathtree.Params{}, nil)
+	target.EXPECT().Function("", someFunc.ID).Return(&someFunc)
+	target.EXPECT().SubscribersOfEvent(gomock.Any(), gomock.Any()).Return([]router.FunctionInfo{}).MaxTimes(3)
+	router := testrouter(target)
+
+	req, _ := http.NewRequest(http.MethodPost, "/", strings.NewReader(`{"some": "thing"}`))
+	req.Header.Set("content-type", contentType)
+	recorder := httptest.NewRecorder()
+	router.ServeHTTP(recorder, req)
+}
+
 func TestRouterServeHTTP_ErrorOnCustomEventEmittedWithNonPostMethod(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()