PubSub IO cannot specify custom timestamp

[kevinsookocheff-wf]

Hi,

I'm defining a pipeline to read logs exported from App Engine to PubSub using the Log Export feature. I wrote a function that parses the LogEntry. Each Log has its own timestamp and I would like to use that timestamp for downstream Windowing operations.

The problem I'm encountering is that PubsubIO attaches its own timestamp to the incoming data here and the timestamp I'm wanting to use from the LogEntry is earlier in time. Attempting to set the timestamp to that of the LogEntry results in an error:

java.lang.IllegalArgumentException: Cannot output with timestamp 2015-10-09T20:06:21.991Z. Output timestamps must be no earlier than the timestamp of the current input (2015-10-09T20:06:25.669Z) minus the allowed skew (0 milliseconds). See the DoFn#getAllowedTimestmapSkew() Javadoc for details on changing the allowed skew.

So, I can change the allowed skew but what I would really like is to allow PubSub IO to optionally not add a timestamp so that we can always use the timestamp from the LogEntry without having to manually specify the skew.

[dhalperi]

@kevinsookocheff-wf – this is a fantastic report.

It sounds like what we really want is for Cloud Logging to provide a labeled timestamp when they export to Pub/Sub, so that when you call timestampLabel("timestampLabel"), PubStubIO would use it instead of attaching generating a timestamp with Instant.now().

Would that resolve your problem?

[kevinsookocheff-wf]

That would absolutely solve my problem. I did a quick test and the Cloud Logging export feature currently only sets Pubsub message metadata for the data below. if we could get the logging timestamp data on there it would definitely help.

{
 "receivedMessages": [
  {
   "ackId": "GxBhCARETS...",
   "message": {
    "data": "redacted",
    "attributes": {
     "appengine.googleapis.com/clone_id": "00c61b117c6ca36e2041beff6f4684e7bea5cbea",
     "appengine.googleapis.com/module_id": "default",
     "appengine.googleapis.com/request_id": "5618273d00ff0a66da55ef65450001737e776b2d6465760001313030392d38366132393561000100",
     "appengine.googleapis.com/version_id": "1009-86a295a"
    },
    "messageId": "4132207616144"
   }
  }
 ]
}

[kevinsookocheff-wf]

Although this doesn't really help the general case of consuming arbirtrary pubsub input and specifying your own timestamp.

[dhalperi]

I acknowledge your larger point. Unfortunately, in the near-term our PubSubIO implementation requires either a timestampLabel or it uses the default of now(). We optimistically hope this isn't so bad -- after all, given that Dataflow integration is a good Pub/Sub use case, hopefully produces of Cloud Pub/Sub channels will include such an attribute.

I've filed FRs internally for both adding the timestampLabel to Cloud Logging-generated PubSubMessages, and, longer-term, for supporting arbitrary timestamp parsing. The former is progressing.

[dhalperi]

Hi Kevin,

It took a little while to get all the moving parts linked together, but

1. Cloud Logging has added timestamps as named attributed to their messages, in RFC 3339 format. As I type this, I believe the time is 20151201T21:48:12.345Z.
2. PubsubIO now supports parsing timestamps in RFC 3339 format.

You should now be able to directly extract message timestamps from Cloud Logging (and any other source that publishes timestamps in this format) using .timestampLabel("logging.googleapis.com/timestamp")

Thanks!

[kevinsookocheff-wf]

Thanks a lot. This is very helpful.