Getting/setting metadata in Python

[Alex Lamaison]

How do you set the request metadata in the client and retrieve it in the server when using Python gRPC?

We are looking at ways to pass trace IDs across RPC calls for Dapper-style tracing. Of course we could add a field to every request proto, but it seems more appropriate to use metadata to avoid cluttering the API. Is that possible?

Many thanks, Alex

[Angus Ma]

yes, it is possible.

client side:

metadata = [(b'client', b'foo'), (b'version', b'bar')]

response = stub.YourGRPCMessage(request, _TIMEOUT_SECONDS, metadata=metadata)

server side:

metadata = context.invocation_metadata()

metadata_dict = {}

for c in metadata:

     metadata_dict[c.key] = c.value

 

[Alex Lamaison]

Amazing.  Thanks, that works nicely.  I couldn't find this documented anywhere, nor any example.

Alex

[Angus Ma]

yes, sometimes test cases in codebase is help.

[weiyu...@gmail.com]

Hi Angus,
What is the "context"? in which package should I include?  Any example?

metadata = context.invocation_metadata()

I now want to get the metadata from client side using python.

Thanks.

[weiyu...@gmail.com]

Hi Alex,

Have you tried retrieving metadata from client side? any example?

Thanks.

[alexander...@gmail.com]

The context is the second parameter to your service method:

Ie.

def MyGrpcMethod(request, context):
    metadata = context.invocation_metadata()

HTH
Alex

[Nathaniel Manista]

On Fri, Apr 1, 2016 at 10:30 PM, <weiyu...@gmail.com> wrote:

Have you tried retrieving metadata from client side?

Addressed over in this other thread.

-Nathaniel

[Tang Weiyu]

Thanks Nathaniel, unfortunately the returned response don't have those 2 methods, though I did get the returned response in the form of stream Type defined in proto file RCP.

Anything missing here?

for data,call in stub.Subscribe(sub_req, _TIMEOUT_SECONDS):
  metadata = data.initial_metadata()

*****************************************
Exception: 'Data' object has no attribute 'initial_metadata'.

Thanks,

Weiyu

On Mon, Apr 4, 2016 at 4:33 PM, Nathaniel Manista <nath...@google.com> wrote:

On Mon, Apr 4, 2016 at 2:59 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

Basically I am now setting up a bi-directional stream grpc.

This is important - response-streaming RPC methods don't accept a "with_call" keyword argument.

When I pass in the parameters with_call = True, i.e. for data,call in stub.Subscribe(sub_req, _TIMEOUT_SECONDS, with_call=True),

it complains:
 Exception: <lambda>() got an unexpected keyword argument 'with_call'

or if I only pass "True" to the third arg, it errors only 2 args expected.

What will be the acceptable call format for carrying "call" obj?

Is the iterator of responses that you got back from the RPC invocation not already a Call object? Does it have "initial_metadata" and "terminal_metadata" methods on it?

-N

[Tang Weiyu]

Hi Nathaniel,

Any other info you need?
Definition of rpc:
  rpc telemetrySubscribe(SubscriptionRequest)     returns (stream OpenConfigData) {}

How to get metadata from client side for streaming type of rpc call?

Appreciate your help.

Thanks,

Weiyu.

[Nathaniel Manista]

On Tue, Apr 5, 2016 at 9:58 AM, Tang Weiyu <weiyu...@gmail.com> wrote:

Definition of rpc:

  rpc telemetrySubscribe(SubscriptionRequest)     returns (stream OpenConfigData) {}

What's important here is that the RPC is response-streaming: it will return zero, or one, or many more responses.

How to get metadata from client side for streaming type of rpc call?

On Mon, Apr 4, 2016 at 5:15 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

Thanks Nathaniel, unfortunately the returned response don't have those 2 methods, though I did get the returned response in the form of stream Type defined in proto file RCP.

Anything missing here?

for data,call in stub.Subscribe(sub_req, _TIMEOUT_SECONDS):
  metadata = data.initial_metadata()

Unpacking the iterator returned by your RPC invocation appears to be a mistake here. I suspect that this should be something like:

my_response_iterator = stub.Subscribe(sub_req, _TIMEOUT_SECONDS)

my_initial_metadata = my_response_iterator.initial_metadata()

for my_response in my_response_iterator:

  <take some action for each response value>

my_terminal_metadata = my_response_iterator.terminal_metadata()

Again, the critical part is that the RPC is response-streaming - the code for a response-unary RPC would look very different.

-N

[Tang Weiyu]

Still says initial_metadata method not found:

Exception: '_CancellableIterator' object has no attribute 'initial_metadata'
E0406 05:55:24.957187913     591 call.c:456]                 Invalid entry in accept encoding metadata: ' gzip'. Ignoring.

Code:         
data_itr = stub.telemetrySubscribe(sub_req, _TIMEOUT_SECONDS)
          metadata = data_itr.initial_metadata()
          print metadata
          for data in data_itr :
            if not data:
              print "Server returned NULL"
              return
            msg = ""
            lmsg = ""
            log(separator,logfile_handle)

            print data
          metadata = data_itr.print_metadata()
          print metadata

[Tang Weiyu]

Code:
          data_itr = stub.telemetrySubscribe(sub_req, _TIMEOUT_SECONDS)
          metadata = data_itr.initial_metadata()
          print metadata
          for data in data_itr :
            if not data:
              print "Server returned NULL"
              return
            msg = ""
            lmsg = ""
            log(separator,logfile_handle)

            print data

          metadata = data_itr.terminal_metadata()

Exception: '_CancellableIterator' object has no attribute 'initial_metadata'

E0406 05:58:46.269377873     606 call.c:456]                 Invalid entry in accept encoding metadata: ' gzip'. Ignoring.

[Nathaniel Manista]

On Tue, Apr 5, 2016 at 10:57 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

Exception: '_CancellableIterator' object has no attribute 'initial_metadata'

The appearance of _CancellableIterator in the error indicates that you're using a version of gRPC Python that was released before metadata support was implemented. Do you happen to know what version you are running? If you installed it with pip, what is the output of "pip freeze"?

-Nathaniel

[Tang Weiyu]

Thanks Nathaniel for help, the version looks like 0.13.0, does that version support metadata?

root@b69d8f6a744c:/home/grpc/lspmon# pip freeze
argparse==1.2.1
chardet==2.0.1
colorama==0.2.5
enum34==1.1.2
futures==3.0.5
grpcio==0.13.0
html5lib==0.999
mosquitto==0.15
paho-mqtt==1.1
pluggy==0.3.1
protobuf==3.0.0a4
py==1.4.31
pytz==2016.3
requests==2.2.1
six==1.10.0
tox==2.3.1
urllib3==1.7.1
virtualenv==15.0.1
wheel==0.24.0
wsgiref==0.1.2

[Nathaniel Manista]

On Wed, Apr 6, 2016 at 11:52 AM, Tang Weiyu <weiyu...@gmail.com> wrote:

Thanks Nathaniel for help, the version looks like 0.13.0, does that version support metadata?

It does, but you're still hitting the code path that doesn't. When was the last time you re-ran the code generator to generate your _pb2.py file(s)?

-Nathaniel

[Tang Weiyu]

I just re-ran generated code lib some 5 mins ago, but same failure with latest _pb2.

[Nathaniel Manista]

On Wed, Apr 6, 2016 at 1:33 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

I just re-ran generated code lib some 5 mins ago, but same failure with latest _pb2.

Is the newly generated code different than the generated code that you had before? Are there new methods to call that weren't there before?

-N

[Tang Weiyu]

No diff, I checked. Actually the previous one was generated just about 1 week ago on the same docker machine.

[Nathaniel Manista]

On Wed, Apr 6, 2016 at 1:43 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

No diff, I checked. Actually the previous one was generated just about 1 week ago on the same docker machine.

Does your generated _pb2.py file contain functions and classes with names containing the words "early adopter"? Does it contain functions and classes with names containing the word "beta"? How certain are you that you're running a recent version of the code generator?

-N

[Tang Weiyu]

Yup, it does contain.

I used below func to generate stub:

def early_adopter_create_Agent_stub(host, port, metadata_transformer=None, secure=False, root_certificates=None, private_key=None, certificate_chain=None, server_host_override=None):
  import agent_pb2
  import agent_pb2
  import agent_pb2
  import agent_pb2
  import agent_pb2
  import agent_pb2
  import agent_pb2
  import agent_pb2
  method_invocation_descriptions = {
    "telemetryOperationalStateGet": alpha_utilities.unary_unary_invocation_description(
      agent_pb2.GetRequest.SerializeToString,
      agent_pb2.OpenConfigData.FromString,
    ),
    "telemetrySubscribe": alpha_utilities.unary_stream_invocation_description(
      agent_pb2.SubscriptionRequest.SerializeToString,
      agent_pb2.OpenConfigData.FromString,
    ),
    "telemetrySubscriptionsGet": alpha_utilities.unary_unary_invocation_description(
      agent_pb2.GetRequest.SerializeToString,
      agent_pb2.OpenConfigData.FromString,
    ),
    "telemetryUnSubscribe": alpha_utilities.unary_unary_invocation_description(
      agent_pb2.UnSubscribeRequest.SerializeToString,
      agent_pb2.Reply.FromString,
    ),
  }
  return early_adopter_implementations.stub("agent.Agent", method_invocation_descriptions, host, port, metadata_transformer=metadata_transformer, secure=secure, root_certificates=root_certificates, private_key=private_key, certificate_chain=certificate_chain, server_host_override=server_host_override)


root@b69d8f6a744c:/home/grpc/lspmon# protoc --version
libprotoc 3.0.0

[Nathaniel Manista]

On Wed, Apr 6, 2016 at 1:56 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

I used below func to generate stub:

def early_adopter_create_Agent_stub(host, port, metadata_transformer=None, secure=False, root_certificates=None, private_key=None, certificate_chain=None, server_host_override=None):

This is where your problems start - the stub object returned by an early_adopter function doesn't know anything about metadata and isn't going to be able to tell you about metadata. You'll need to switch code that you have using the early_adopter functions to using beta functions.

-Nathaniel

[Tang Weiyu]

huh, interesting. I just did following steps trying to make the env up to date.

Pull github grpc and install gpb from grpc/third_party/protobuf, including python libs.

Then make install grpc from the latest pull.

When I recompile the proto file and generate _pb2.py, I did see difference, in that the early_adopter_ funcs are no longer present.

Then I changed the code to use beta func:
agent_pb2.beta_create_Agent_stub(grpc_server["ip"], grpc_server["port"]) as stub:

But it seems incorrect args passed in, beta_create_stub takes "channel" as first arg.

How do I define the channel in this case?

i.e.
def beta_create_Agent_stub(channel, host=None, metadata_transformer=None, pool=None, pool_size=None):

Thanks.

 

[Nathaniel Manista]

On Wed, Apr 6, 2016 at 3:46 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

Then I changed the code to use beta func:

agent_pb2.beta_create_Agent_stub(grpc_server["ip"], grpc_server["port"]) as stub:

But it seems incorrect args passed in, beta_create_stub takes "channel" as first arg.

How do I define the channel in this case?

You'll want to use one of grpc.beta.insecure_channel or grpc.beta.secure_channel.

-N

[Tang Weiyu]

Thanks Nathan, I will let you know how it goes.
Now the docker seems got messed up after I reinstalled gpb/grpc, will re-create tomorrow.

[Tang Weiyu]

HI Nathaniel,

I am now able to get the metadata from streaming response grpc client. The problem is actually in the docker creation script, when we were pulling grpc 0.11 for some specific reason. And later I installed grpcio based on that docker already had 0.11 grpc in place, as you said 0.11 not supporting initial_metadata() yet.

Today I created another docker using 0.13 grpc, and the client worked as the way you mentioned.

I'd like to sincerely thank you for helping me nail the issue down, really appreciate it.

-Weiyu

Post the snippet for future reference:

with agent_pb2.beta_create_Agent_stub(channel) as stub: 
  data_itr = stub.method(request, _TIMEOUT_SECONDS)
  metadata = data_itr.initial_metadata()
  for data in data_itr :

    print data

  metadata = data_itr.terminal_metadata()

root@2b3d189dbe67# pip freeze

argparse==1.2.1
chardet==2.0.1
colorama==0.2.5
enum34==1.1.2
futures==3.0.5

grpcio==0.13.1
html5lib==0.999
pluggy==0.3.1
protobuf==3.0.0b2.post2
py==1.4.31
pytz==2016.3
requests==2.2.1
six==1.10.0
ssh-import-id==3.21

tox==2.3.1
urllib3==1.7.1
virtualenv==15.0.1
wheel==0.24.0
wsgiref==0.1.2

[Tang Weiyu]

Hi Nathan,

Another question regarding response-streaming RPC:

in the example code, it says

for feature in stub.ListFeatures(rectangle, timeout_in_seconds):

How do I understand the streaming concept? does it mean the stream will always up, and in a waiting-data status?

The reason I am asking is, in order to get metadata, I have to first assign RPC call to an iterator varialble, then loop thru this iterator for each data it has.

And if I assume the stream is already up and the server sends the data after some time, say 1 second, how do I get the second data?
Calling stub.ListFeatures(rectangle, timeout_in_seconds) twice seems not right from server perspective.

I found out in the cpp version client, it creates a Reader object and you can continuously poll this reader to see if there is new data.

e.g. 

while (reader->Read(&feature)) {
      std::cout << "Found feature called "
                << feature.name() << " at "
                << feature.location().latitude()/kCoordFactor_ << ", "
                << feature.location().longitude()/kCoordFactor_ << std::endl;
    }

So, for python, it seems not the case, how do I do the similar for streaming type RPC?
 i.e. call streaming RPC once, get the metadata,  then keep checking data buffer and read.

Thanks,

Weiyu

[Nathaniel Manista]

On Fri, Apr 8, 2016 at 4:21 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

Another question regarding response-streaming RPC:

in the example code, it says

for feature in stub.ListFeatures(rectangle, timeout_in_seconds):

How do I understand the streaming concept? does it mean the stream will always up, and in a waiting-data status?

The reason I am asking is, in order to get metadata, I have to first assign RPC call to an iterator varialble, then loop thru this iterator for each data it has.

And if I assume the stream is already up and the server sends the data after some time, say 1 second, how do I get the second data?
Calling stub.ListFeatures(rectangle, timeout_in_seconds) twice seems not right from server perspective.

I found out in the cpp version client, it creates a Reader object and you can continuously poll this reader to see if there is new data.

e.g. 

while (reader->Read(&feature)) {
      std::cout << "Found feature called "
                << feature.name() << " at "
                << feature.location().latitude()/kCoordFactor_ << ", "
                << feature.location().longitude()/kCoordFactor_ << std::endl;
    }

So, for python, it seems not the case, how do I do the similar for streaming type RPC?
 i.e. call streaming RPC once, get the metadata,  then keep checking data buffer and read.

I don't think I fully understand your question. But answering the question that I think you are asking: messages are emitted from the iterator as they are received from the server. If a server sends five messages, each one second apart, then the iterator present on the client will emit those five messages each one second apart. The client-side iterator will not block for five seconds and then emit all five messages as fast as the client-side application will process them.

Does that help?

-Nathaniel

[Tang Weiyu]

so, will the stream be always up between server and client in this case? for example, if the server wants to send data to the client after 5 seconds of sending 5 messages, can the server still use that "stream" to send another 5 msgs and can the same iterator on client still emit those 5 new msgs?

Or, the client has to call the RPC func the second time to get the data?

Thanks,

Weiyu.

[Nathaniel Manista]

On Wed, Apr 13, 2016 at 2:53 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

so, will the stream be always up between server and client in this case? for example, if the server wants to send data to the client after 5 seconds of sending 5 messages, can the server still use that "stream" to send another 5 msgs and can the same iterator on client still emit those 5 new msgs?

The stream stays up until the application code on the server decides to close it (or the stream closes because the client cancelled the RPC, or the RPC timed out, or the network failed, or something like that). The server-side application can choose to send one message per second for five seconds, then send nothing for sixty seconds, then send one message per second for five seconds, and then indicate the end of the stream. The client-side iterator will not indicate the end of messages until after the server-side application has indicated the end of the stream.

Or, the client has to call the RPC func the second time to get the data?

As long as the server doesn't indicate the end of its responses (and thus the end of the RPC), the stream will stay open (excepting cancellation, timeout, network fault, and so forth). What the server-side application cannot do is indicate the end of an RPC and then later on say "oh wait no I want to send some extra messages to that client".

Does that help? If not, would you be able to share more about the particular application you're developing?

-N

[Tang Weiyu]

Thanks Nathan, this helps a lot.

Basically I am developing a client to "subscribe" a service on a grpc server, when the subscription successes, the server will keep sending data to the client.

In order to get the metadata, I have to call itr = stub.Subscribe(request, __TIMEOUT) first, do itr.initial_metadata(), then loop thru the itr for the real data.

My concern is that if I code "for data in itr:" in this example, will the client be able to get the data from server continously, if the server keeps sending data.

e.g.
         data_itr = stub.Subscribe(sub_req, _TIMEOUT_SECONDS)           <<<<< does this mean "the client cancelled the RPC"
            metadata = data_itr.initial_metadata()
            for data in data_itr :                                   <<<<<< will this loop always continues as long as server keeps sending data?

              if not data:
                print "Server returned NULL"

                next
              oc_data = Parse(data)

Thanks,

Weiyu

[Nathaniel Manista]

On Wed, Apr 13, 2016 at 4:18 PM, Tang Weiyu <weiyu...@gmail.com> wrote:

Basically I am developing a client to "subscribe" a service on a grpc server, when the subscription successes, the server will keep sending data to the client.

In order to get the metadata, I have to call itr = stub.Subscribe(request, __TIMEOUT) first, do itr.initial_metadata(), then loop thru the itr for the real data.

My concern is that if I code "for data in itr:" in this example, will the client be able to get the data from server continously, if the server keeps sending data.

e.g.
         data_itr = stub.Subscribe(sub_req, _TIMEOUT_SECONDS)           <<<<< does this mean "the client cancelled the RPC"

The timeout is the maximum duration in seconds to allow for the RPC. If the RPC lasts longer than the timeout, the RPC ends, but it isn't interpreted as client cancellation. Client cancellation would be something like:

data_itr = stub.Subscribe(sub_req, _TIMEOUT_SECONDS)

initial_metadata = data_itr.initial_metadata()

first_message = next(data_itr)

second_message = next(data_itr)

data_itr.cancel()

. This would be for a case in which the client-side application code wants to only take the first two messages and then unsubscribe.

            metadata = data_itr.initial_metadata()
            for data in data_itr :                                   <<<<<< will this loop always continues as long as server keeps sending data?

Yes, this loop will always continue as long as the server keeps sending data.

-Nathaniel

[Tang Weiyu]

Thanks Nathan, now it's clear, let me try it out.

-Weiyu.

[Amit Saha]

On Thursday, March 3, 2016 at 3:11:15 PM UTC+11, Angus Ma wrote:

yes, it is possible.

client side:

metadata = [(b'client', b'foo'), (b'version', b'bar')]

response = stub.YourGRPCMessage(request, _TIMEOUT_SECONDS, metadata=metadata)

server side:

metadata = context.invocation_metadata()

metadata_dict = {}

for c in metadata:

     metadata_dict[c.key] = c.value

This seems to have now changed. The metadata now is now simply a list of of tuples.

Is there a reason why this was changed?

 

On Thursday, 3 March 2016 07:30:26 UTC+8, Alex Lamaison wrote:

How do you set the request metadata in the client and retrieve it in the server when using Python gRPC?

We are looking at ways to pass trace IDs across RPC calls for Dapper-style tracing. Of course we could add a field to every request proto, but it seems more appropriate to use metadata to avoid cluttering the API. Is that possible?

Many thanks, Alex

[Nathaniel Manista]

On Sun, Aug 27, 2017 at 6:05 AM, Amit Saha <amits...@gmail.com> wrote:

On Thursday, March 3, 2016 at 3:11:15 PM UTC+11, Angus Ma wrote:

yes, it is possible.

client side:

metadata = [(b'client', b'foo'), (b'version', b'bar')]

response = stub.YourGRPCMessage(request, _TIMEOUT_SECONDS, metadata=metadata)

server side:

metadata = context.invocation_metadata()

metadata_dict = {}

for c in metadata:

     metadata_dict[c.key] = c.value

This seems to have now changed. The metadata now is now simply a list of of tuples.

Huh. I didn't think so at first but having looked into it I think you're right.

Is there a reason why this was changed?

I can think of a few specific reasons that are likely to have inadvertently caused the behavior change but the general overall reason is probably "because we neglected to have test coverage that would have prevented it". I've filed issue 12301 to track possibly adding the capability back.

-Nathaniel

[Amit Saha]

Thanks for confirming. I have subscribed to the issue.

 

-Nathaniel

[cyc19...@gmail.com]

It doesn't work if an interceptor intercepts the responses.

Considering there is such an interceptor:

def wrapper(iterator):
  for entry in iterator:
    yield entry
class MyInterceptor(grpc.StreamStreamClientInterceptor):
  def intercept_stream_stream(self, continuation, client_call_details, request_iterator):
    return wrapper(continuation(client_call_details, request_iterator))

Well I know I could delegate a method `initial_metadata` to the original iterator, but unfortunately, I've seen many interceptors written in the way above, causing that I couldn't retrieve the initial metadata from last interceptor.

Any idea?

在 2016年4月6日星期三 UTC+8上午9:08:36，Nathaniel Manista写道：

On Tue, Apr 5, 2016 at 9:58 AM, Tang Weiyu <weiy...@gmail.com> wrote:

Definition of rpc:

  rpc telemetrySubscribe(SubscriptionRequest)     returns (stream OpenConfigData) {}

What's important here is that the RPC is response-streaming: it will return zero, or one, or many more responses.

How to get metadata from client side for streaming type of rpc call?

On Mon, Apr 4, 2016 at 5:15 PM, Tang Weiyu <weiy...@gmail.com> wrote:

Thanks Nathaniel, unfortunately the returned response don't have those 2 methods, though I did get the returned response in the form of stream Type defined in proto file RCP.

Anything missing here?

for data,call in stub.Subscribe(sub_req, _TIMEOUT_SECONDS):
  metadata = data.initial_metadata()

[Lidi Zheng]

Your observation is right that if an interceptor returns a generator instead of a grpc.Call object, downstream applications will lose access to initial_metadata, trailing_metadata, etc..

However, in the API reference [1], the streaming interceptors define the return value as:

    > An object that is both a Call for the RPC and an iterator of response values. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status.

So, returning a generator without wrapping it as grpc.Call is not fully supported. Given that, this support can be implementWe always welcome contributors.

[1] https://grpc.github.io/grpc/python/grpc.html#grpc.StreamStreamClientInterceptor

--
You received this message because you are subscribed to the Google Groups "grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email to grpc-io+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/grpc-io/e93d5c60-693c-4790-b107-8a851f42df9fo%40googlegroups.com.

[cyc19...@gmail.com]

It seems that I'm getting OT on this thread. I've created a new one.

To unsubscribe from this group and stop receiving emails from it, send an email to grp...@googlegroups.com.