Support pubsub push

[majelbstoat]

I'd like to run an event handling service on a serverless platform, like Cloud Run or Cloud Functions. This requires the service respond to http requests (example).

The current cloudevents sdk doesn't appear to support receiving messages in this way. As a result, my service would get spun down due to no incoming traffic. I can work around this by specifying a minimum number of instances, but it also wouldn't scale up in cases of event spikes.

I'd like to request some sort of framework for receiving messages from pubsub as a blocking http server, instead of streaming pulls.

It would be cool if the same StartReceiver method could be used, and controlled only by an option to pubsub.New().

Perhaps something like:

host := "https://myservice.cloudfunctions.net/events"
t := pubsub.New(ctx, pubsub.WithPushConfig(pubsub.PushConfig{
  Endpoint: host,
}))
c, err := cloudevents.NewClient(t, ce.WithTimeNow())
err := c.StartReceiver(ctx, func(ctx context.Context, event event.Event) error {
  // Same as before.
})

Then, in the presence of a push config, StartReceiver would instead start an http server to accept and process the messages, doing sufficient transformation to allow re-use of the same user-defined receive method.

[majelbstoat]

(I know I could just start an http server, but i'd like to take advantage of auto subscription creation, especially when using a local pubsub emulator in development that is more frequently stopped and started.)

Edit: In fact, I can't just use the standard http receiver for this because pubsub push sends messages in a format that the sdk doesn't understand:

{
  "subscription": "projects\/my-project\/subscriptions\/my-subscription",
  "message": {
    "data": "eyJ1c2VySWQiOiJBQUVCRlRpdXE0UFNEUEJ5UEEifQ==",
    "messageId": "1",
    "attributes": {
      "ce-source": "my-app",
      "ce-specversion": "1.0",
      "ce-type": "test.event.signIn",
      "ce-id": "MFkWEgrOHKs44AEADQ",
      "ce-time": "2020-05-24T18:38:18.7759848Z",
      "Content-Type": "application\/json"
    }
  }
}

[slinkydeveloper]

@nachocano , @yolocs any input?

[n3wscott]

The issue here is that Pub/Sub Push is HTTP and incompatible with CloudEvents in a general way.

The receiver would need to be aware that it is being invoked by Pub/Sub via http post and I am not aware of Pub/Sub allowing you to add or edit any headers to help signal this.

You can handle the raw http message and repack the request into an protocol/http.Message and use the normal methods from there, or reimplement the http protocol to do that unpacking and shuffling via a http protocol fork.

This repo might not be the place to introduce such a protocol, unless there is a known way to determine Pub/Sub is the sender (and as of 6 months ago, there was not from my looking) but perhaps you can poke around https://github.com/google/knative-gcp/ and see if they have any samples. That project will support something like this and at one time we had a working demo of exactly what you are asking for, but that was for sdk v1 and I am unable to find that code now. Looks to be deleted.

[majelbstoat]

i was hoping that an addition of the pushconfig/endpoint parameter to the pubsub protocol would signal data was to be expected in that way, and allow the receiver to do that unmarshalling.

In fact, it seems there's actually a TODO in the code to do just that?

https://github.com/cloudevents/sdk-go/blob/master/protocol/pubsub/v2/internal/connection.go#L217

I currently work around this by defining a custom format (see linked issue). But it seems like a missed opportunity given that serverless is a very natural application for handling events, pubsub is a very popular service for delivering events, and the GCP documentation specifically suggests using pubsub push when writing a serverless handler.

[yolocs]

As @n3wscott pointed out, pubsub push works very differently. The work seems to involve: 1) create a pubsub push subscription 2) start a http server that converts pubsub http message to cloudevent 3) on stop, delete the push subscription. Some existing code in the pubsub/http protocol package could be reused but might not be in a clean way.

I don't think it's totally invalid to include pubsub push in this sdk. One option could be to have a separate package for push based pubsub. A draft PR could tell a lot about whether it could work well.

[n3wscott]

I am sorry, you are correct @yolocs we could host it inside of pubsub or make a pubsub push protocol. I would lean toward making it work inside of the current pubsub with a flag or another class