Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

examples: add new example to show updating metadata in interceptors #5788

Merged
merged 16 commits into from Dec 6, 2022
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
46 changes: 46 additions & 0 deletions Documentation/grpc-metadata.md
Expand Up @@ -223,3 +223,49 @@ func (s *server) SomeStreamingRPC(stream pb.Service_SomeStreamingRPCServer) erro
stream.SetTrailer(trailer)
}
```

## Updating metadata in the interceptor - server side
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about replacing this with:

## Updating metadata from a server interceptor

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it


Server side metadata updating in the interceptor examples are available [here](../examples/features/metadata_interceptor/server/main.go).
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please wrap these lines at 80 cols width. Thanks.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you please update this line to look like:

An example for updating metadata from a server interceptor is 
available [here](../examples/features/metadata_interceptor/server/main.go).

Also, please ensure that this line is also wrapped to 80-cols width. Thanks.


#### Unary interceptor
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we move these sections to the new README instead, to avoid clutter here and also to put the documentation near the example that illustrates it?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dfawley Thank you very much for your response. Do you mean moving the Updating metadata from a server interceptor part of grpc-metadata.md to the README.md under the folder examples/features/metadata_interceptor? Please correct me if anything I am misunderstanding or missing.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I mean move everything from #### Unary interceptor down into examples/features/metadata_interceptor/README.md.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@dfawley , thank you very much for your response. I have updated the document per your review comment. Could you please review it again?


To read and update metadata in the interceptor under unary call, the server needs to retrieve it from RPC context and update it through this RPC context.

```go
func SomeInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
// get the metadata from context
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: Maybe replace with:

// Get the incoming metadata from the RPC context, and add a new 
// key-value pair to it.
md, ok := metadata.FromIncomingContext(ctx)
md.Append("key1", "value1")

// Create a context with the new metadata and pass it to handler.
ctx = metadata.NewIncomingContext(ctx, md)
return handler(ctx, req)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, sorry for my bad statement in the comment

md, ok := metadata.FromIncomingContext(ctx)

// update metadata
md.Append("key1", "value1")

// set the metadata to context
ctx = metadata.NewIncomingContext(ctx, md)
```

#### Streaming interceptor

Since there is no direct way to set RPC context in the stream interceptor, the `wrappedStream` could be created and override `Context()` method with owned `context`. Then pass this `wrappedStream` to the stream interceptor.

```go
type wrappedStream struct {
grpc.ServerStream
ctx context.Context
}

func (s *wrappedStream) Context() context.Context {
return s.ctx
}

func streamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
// get the metadata from context
md, ok := metadata.FromIncomingContext(ss.Context())
// update metadata
md.Append("key1", "value1")
// set the metadata to context
ctx := metadata.NewIncomingContext(ss.Context(), md)

return handler(srv, &wrappedStream{ss, ctx})
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar comments here as the one above.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}
```
3 changes: 1 addition & 2 deletions examples/features/metadata/server/main.go
Expand Up @@ -31,10 +31,9 @@ import (

"google.golang.org/grpc"
"google.golang.org/grpc/codes"
pb "google.golang.org/grpc/examples/features/proto/echo"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"

pb "google.golang.org/grpc/examples/features/proto/echo"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need this change?

Although we don't do it at all places, we try (as much as possible) to group the proto imports separately.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for the typo, it was caused by the Goland. I will fix it.

)

var port = flag.Int("port", 50051, "the port to serve on")
Expand Down
18 changes: 18 additions & 0 deletions examples/features/metadata_interceptor/README.md
@@ -0,0 +1,18 @@
# Metadata interceptor example

This example shows how to update metadata in RPC unary interceptor and streaming interceptor.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about:

This example shows how to update metadata from unary and streaming interceptors on the server.
Please see
[grpc-metadata.md](https://github.com/grpc/grpc-go/blob/master/Documentation/grpc-metadata.md)
for more information.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will update the document

Please see
[grpc-metadata.md](https://github.com/grpc/grpc-go/blob/master/Documentation/grpc-metadata.md)
for more information.

## Start the server

```
go run server/main.go
```

## Run the client

```
go run client/main.go
```
90 changes: 90 additions & 0 deletions examples/features/metadata_interceptor/client/main.go
@@ -0,0 +1,90 @@
/*
*
* Copyright 2018 gRPC authors.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Update the year on the copyright notice please.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for that, I will update it.

*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

// Binary client is an example client.
package main

import (
"context"
"flag"
"fmt"
"io"
"log"
"time"

"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
pb "google.golang.org/grpc/examples/features/proto/echo"
)

var addr = flag.String("addr", "localhost:50051", "the address to connect to")

const message = "hello world"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: Get rid of this const and move it inline to the calls to callUnaryEcho and callBidiStreamingEcho.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. Thank you for your view


func callUnaryEcho(client pb.EchoClient, message string) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
resp, err := client.UnaryEcho(ctx, &pb.EchoRequest{Message: message})
if err != nil {
log.Fatalf("client.UnaryEcho(_) = _, %v: ", err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know we do this in many (or all) examples, but I think we can print a simpler/nicer error message. Something like:

log.Fatalf("UnaryEcho: %v", err)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}
fmt.Println("UnaryEcho: ", resp.Message)
}

func callBidiStreamingEcho(client pb.EchoClient, message string) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

c, err := client.BidirectionalStreamingEcho(ctx)
if err != nil {
return
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

log.Fatal here instead of simply returning?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}

if err := c.Send(&pb.EchoRequest{Message: message}); err != nil {
log.Fatalf("failed to send request due to error: %v", err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

log.Fatalf("Sending echo request: %v", err)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}
c.CloseSend()

for {
resp, err := c.Recv()
if err == io.EOF {
break
}
if err != nil {
log.Fatalf("failed to receive response due to error: %v", err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

log.Fatalf("Receiving echo response: %v", err)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}
fmt.Println("BidiStreaming Echo: ", resp.Message)
}
}

func main() {
flag.Parse()
// Set up a connection to the server.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: remove this comment as it is saying the obvious.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
log.Fatalf("did not connect: %v", err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

log.Fatalf("grpc.Dial(%q): %v", *addr, err)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}
defer conn.Close()

ec := pb.NewEchoClient(conn)

callUnaryEcho(ec, message)
time.Sleep(1 * time.Second)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need this sleep?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will remove this sleep.


callBidiStreamingEcho(ec, message)
}
147 changes: 147 additions & 0 deletions examples/features/metadata_interceptor/server/main.go
@@ -0,0 +1,147 @@
/*
*
* Copyright 2018 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

// Binary server is an example server.
package main

import (
"context"
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems like we need some reordering of the imports here to pass vet tests. Try

goimports examples/features/metadata_interceptor/server/main.go

to see how the file should look like

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@arvindbr8 , thank you very much for your response, I have updated the reordering of the imports through goimports.

"flag"
"fmt"
"io"
"log"
"math/rand"
"net"
"time"

"google.golang.org/grpc"
"google.golang.org/grpc/codes"
pb "google.golang.org/grpc/examples/features/proto/echo"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)

var port = flag.Int("port", 50051, "the port to serve on")

var errMissingMetadata = status.Errorf(codes.InvalidArgument, "missing metadata")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s/missing metadata/no incoming metadata in rpc context/

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it


type server struct {
pb.UnimplementedEchoServer
}

func unaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, errMissingMetadata
}

md.Append("key1", "value1")
ctx = metadata.NewIncomingContext(ctx, md)

return handler(ctx, req)
}

func (s *server) UnaryEcho(ctx context.Context, in *pb.EchoRequest) (*pb.EchoResponse, error) {
fmt.Printf("--- UnaryEcho ---\n")

md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, status.Errorf(codes.DataLoss, "UnaryEcho: failed to get metadata")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use Internal status code instead. And use a more descriptive error message: missing incoming metadata in rpc context

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it. I will fix it

}

// Read metadata from interceptor
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Change this to Read and print metadata added by the interceptor.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it

if v, ok := md["key1"]; ok {
fmt.Printf("key1 from metadata: \n")
for i, e := range v {
fmt.Printf(" %d. %s\n", i, e)
}
}

fmt.Printf("request received: %v, sending echo\n", in)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please remove this print. Thanks.

Also, get rid of unnecessary newlines.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.


return &pb.EchoResponse{Message: in.Message}, nil
}

type wrappedStream struct {
grpc.ServerStream
ctx context.Context
}

func (s *wrappedStream) Context() context.Context {
return s.ctx
}

func streamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
md, ok := metadata.FromIncomingContext(ss.Context())
if !ok {
return errMissingMetadata
}

md.Append("key1", "value1")
ctx := metadata.NewIncomingContext(ss.Context(), md)

return handler(srv, &wrappedStream{ss, ctx})
}

func (s *server) BidirectionalStreamingEcho(stream pb.Echo_BidirectionalStreamingEchoServer) error {
fmt.Printf("--- BidirectionalStreamingEcho ---\n")

md, ok := metadata.FromIncomingContext(stream.Context())
if !ok {
return status.Errorf(codes.DataLoss, "BidirectionalStreamingEcho: failed to get metadata")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comment here about the status code and the error message.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.

}

// Read metadata from interceptor.
if v, ok := md["key1"]; ok {
fmt.Printf("key1 from metadata: \n")
for i, e := range v {
fmt.Printf(" %d. %s\n", i, e)
}
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comment here like the one in the unary handler.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.


// Read requests and send responses.
for {
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
fmt.Printf("request received %v, sending echo\n", in)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove this print please.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.

if err := stream.Send(&pb.EchoResponse{Message: in.Message}); err != nil {
return err
}
}
}

func main() {
flag.Parse()
rand.Seed(time.Now().UnixNano())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems to be unused. Could you please remove it.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.

lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
if err != nil {
log.Fatalf("failed to listen: %v", err)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

 log.Fatalf("net.Listen() failed: %v", err)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.

}
fmt.Printf("server listening at %v\n", lis.Addr())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s/server/Server/

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got it, I will fix it.


s := grpc.NewServer(
grpc.UnaryInterceptor(unaryInterceptor),
grpc.StreamInterceptor(streamInterceptor))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please make this a single line. See: https://google.github.io/styleguide/go/guide#line-length

pb.RegisterEchoServer(s, &server{})
s.Serve(lis)
}