pubsub messages are twice base64 encoded when using GAX

[ericbuehl]

I am using a recent master (85a0c1a)

Python 2.7.11 (v2.7.11:6d1b6a68f775, Dec  5 2015, 12:54:16) 
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.

$ pip show google-cloud-pubsub

---
Metadata-Version: 2.0
Name: google-cloud-pubsub
Version: 0.20.0+85a0c1a
Summary: Python Client for Google Cloud Pub/Sub
Home-page: https://github.com/GoogleCloudPlatform/google-cloud-python
Author: Google Cloud Platform
Author-email: jjg+google-cloud-python@google.com
Installer: pip
License: Apache 2.0
Location: /Users/eric/env/lib/python2.7/site-packages
Requires: google-cloud-core, grpc-google-pubsub-v1, gapic-google-pubsub-v1, grpcio, google-gax
Classifiers:
  Development Status :: 4 - Beta
  Intended Audience :: Developers
  License :: OSI Approved :: Apache Software License
  Operating System :: OS Independent
  Programming Language :: Python :: 2
  Programming Language :: Python :: 2.7
  Programming Language :: Python :: 3
  Programming Language :: Python :: 3.4
  Programming Language :: Python :: 3.5
  Topic :: Internet

---

To demonstrate this issue, I set up a simple pubsub push endpoint that prints the posted data.

$ python
>>> from google.cloud import pubsub
>>> pubsub.Client("myproject").topic("mytopic").publish("blah")

Data received at push endpoint:

{u'message': {u'attributes': {},
              u'data': u'WW14aGFBPT0=',
...

---

$ GOOGLE_CLOUD_DISABLE_GRPC=1 python
>>> from google.cloud import pubsub
>>> pubsub.Client("myproject").topic("mytopic").publish("blah")

Data received at push endpoint:

{u'message': {u'attributes': {},
              u'data': u'YmxhaA==',
...

As you can see, the data in the first version is a twice-encoded version of the message "blah" while GAX is in use. I would expect to see the latter regardless of GAX being enabled or not.

---

(Added by @dhermes)

>>> import base64
>>> base64.b64decode(u'WW14aGFBPT0=')
'YmxhaA=='
>>> base64.b64decode('YmxhaA==')
'blah'

[daspecster]

@ericbuehl thanks for the report!

I'm checking on this right now.

[daspecster]

@dhermes & @tseaver, I'm not seeing how this is happening in our lib. I also don't see it happening in the GAPIC code either.

This seems to be the only place we encode the message.

message_b = base64.b64encode(message).decode('ascii')

[ericbuehl]

If I had to guess, I would say that protobuf is responsible for the second encoding

[daspecster]

@ericbuehl I was thinking the same thing.

@bjwatson do you happen to know if that changed recently?

[bjwatson]

@daspecster We haven't changed our version of gRPC or protobuf in more than a month (probably not for two months). I'm not aware of any changes in GAPIC or GAX that would explain this. Do you have any thoughts @geigerj?

[geigerj]

I am not aware of any base64 encoding happening at the GAPIC/GAX layer.

[dhermes]

@bjwatson It may not be a change, it could just be something we've never noticed. From google.protobuf.json_format

      if field.type == descriptor.FieldDescriptor.TYPE_BYTES:
        # Use base64 Data encoding for bytes
        return base64.b64encode(value).decode('utf-8')
      else:
        return value

[Fkawala]

As a side note the double base64 did not occurs with version 0.18.3.

[dhermes]

@Fkawala It most likely was, did you try running with gRPC installed / enabled. gRPC for Pub/Sub was supported in 0.18.3 and the double encoding happened because google.protobuf was automatically converting bytes to base64. google.protobuf has been doing that longer than this project has existed.

[Fkawala]

@dhermes I made a MWE (see this gist) with gcloud 0.18.3, the subscription has a push endpoint, but I don't think its changes anything.

producer code:

from gcloud import pubsub
import ujson

if __name__ == '__main__':

    client = pubsub.Client()
    topic = client.topic('debug-topic')
    if not topic.exists():
        topic.create()
    subscription = pubsub.subscription.Subscription(name='debug-subscription', topic=topic, push_endpoint='https://random.ngrok.io/')
    if not subscription.exists():
        subscription.create()
    batch = topic.batch()
    for i in range(10):
        batch.publish(ujson.encode({'a_string': 'my-val', 'an_int': i}), an_attr=str('a value'))

    batch.commit()
    print('topic {0}'.format(topic))
    print('subscription {0}'.format(subscription))
    print('message_ids {0}'.format(batch.message_ids))

Consumer code:

from webapp2 import Route, WSGIApplication, RequestHandler
import ujson 
import urllib
import base64
import logging

class Handler(RequestHandler):
    def post(self):

        message = ujson.decode(urllib.unquote(self.request.body).rstrip('='))['message']
        attributes = message.get('attributes', {})
        data = message.get('data', {})

        try:
            print "data >>", data
            print "b64decode >>", base64.b64decode(str(message['data']))
            print "+" * 72
        except Exception as e:
            print logging.exception(e)

    def get(self):
        return self.post()

application = WSGIApplication(debug=True, routes=[Route('/', Handler)])

Result

data >> eyJhbl9pbnQiOjQsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":4,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjgsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":8,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjcsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":7,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjYsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":6,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjIsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":2,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjEsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":1,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjAsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":0,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjMsImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":3,"a_string":"my-val"}
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
data >> eyJhbl9pbnQiOjksImFfc3RyaW5nIjoibXktdmFsIn0=
b64decode >> {"an_int":9,"a_string":"my-val"}

[dhermes]

@Fkawala Thanks for the great example, but it doesn't ever switch between transports. Check out this repro that uses them side-by-side. First

$ virtualenv venv
$ source venv/bin/activate
(venv) $ venv/bin/pip install 'gcloud==0.18.3'
(venv) $ venv/bin/python

then in the interpreter create two clients (one for each transport) by lying to the gcloud.pubsub.client module:

>>> from gcloud.pubsub import client
>>> from gcloud import pubsub
>>>
>>>
>>> client._USE_GAX
False
>>> client_http = pubsub.Client()
>>> # Note: It's required to access these properties since they rely on the
>>> #       value of _USE_GAX at the first access
>>> client_http.publisher_api
<gcloud.pubsub.connection._PublisherAPI object at 0x7faf0c4ec210>
>>> client_http.subscriber_api
<gcloud.pubsub.connection._SubscriberAPI object at 0x7faf08911e10>
>>>
>>>
>>> client._USE_GAX = True
>>> client_gax = pubsub.Client()
>>> client_gax.publisher_api
<gcloud.pubsub._gax._PublisherAPI object at 0x7faf089330d0>
>>> client_gax.subscriber_api
<gcloud.pubsub._gax._SubscriberAPI object at 0x7faf089331d0>

then create a topic/subscription pair bound to each client and create the topic / subscription and publish to the topic

>>> gax_topic = client_gax.topic('mytopic-foo')
>>> gax_sub = gax_topic.subscription('foo-bar')
>>> same_topic = client_http.topic(gax_topic.name)
>>> same_sub = same_topic.subscription(gax_sub.name)
>>>
>>>
>>> gax_topic.create()
>>> gax_sub.create()
>>>
>>>
>>> gax_topic.publish('blah')

Finally, read the published message from each client's subscription to see the difference

>>> (_, gax_message), = gax_sub.pull(return_immediately=True)
>>> # Wait 5-10s so that you can pull the same message
>>> import time
>>> time.sleep(10)
>>> (_, http_message), = same_sub.pull(return_immediately=True)
>>>
>>>
>>> gax_message.data
'blah'
>>> http_message.data
'YmxhaA=='
>>> import base64
>>> base64.b64decode(http_message.data)
'blah'

(Also don't forget to gax_topic.delete(), gax_sub.delete() after.)

[tseaver]

0.18.3 had the GAX transport as optional (off by default); 0.19.0 enabled it by default.

[Fkawala]

@dhermes Sorry I would have add a "how to run" file if it wasn't for DYN DDOS. Anyway, thanks to you and @tseaver for this useful explantation. Is there a chance for #2578 to be shipped in the next release ? Is there a schedule for the next release ?