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
balancer: support injection of per-call metadata from LB policies #5853
Changes from 3 commits
83983c7
296c477
64db6af
17b0110
23184d9
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change | ||
---|---|---|---|---|
|
@@ -438,7 +438,7 @@ func (a *csAttempt) getTransport() error { | |||
cs := a.cs | ||||
|
||||
var err error | ||||
a.t, a.done, err = cs.cc.getTransport(a.ctx, cs.callInfo.failFast, cs.callHdr.Method) | ||||
a.t, a.pickResult, err = cs.cc.getTransport(a.ctx, cs.callInfo.failFast, cs.callHdr.Method) | ||||
if err != nil { | ||||
if de, ok := err.(dropError); ok { | ||||
err = de.error | ||||
|
@@ -455,7 +455,23 @@ func (a *csAttempt) getTransport() error { | |||
func (a *csAttempt) newStream() error { | ||||
cs := a.cs | ||||
cs.callHdr.PreviousAttempts = cs.numRetries | ||||
s, err := a.t.NewStream(a.ctx, cs.callHdr) | ||||
|
||||
// Merge metadata stored in PickResult, if any, with existing call metadata. | ||||
ctx := a.ctx | ||||
if a.pickResult.Metatada != nil { | ||||
// We currently do not have a function it the metadata package which | ||||
// merges given metadata with existing metadata in a context. Existing | ||||
// function `AppendToOutgoingContext()` takes a variadic argument of key | ||||
// value pairs. | ||||
// | ||||
// TODO: Make it possible to retrieve key value pairs from metadata.MD | ||||
// in a form passable to AppendToOutgoingContext(). | ||||
md, _ := metadata.FromOutgoingContext(ctx) | ||||
md = metadata.Join(md, a.pickResult.Metatada) | ||||
ctx = metadata.NewOutgoingContext(a.ctx, md) | ||||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should this not update There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
I feel that by not modifying the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Each attempt creates a new There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Oh yes, that's right.
When will we ever use it again? From what I see, the context is used in I'm not opposed to updating the attempt's context in place with metadata from the LB policy. But, I thought it was safer to derive one from it and pass it to There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I'm not sure we ever will use it again. But if There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Line 701 in e53d28f
And we use the modified context (with all outgoing metadata) when creating the transport stream. So, it looks like we are good here too. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Honestly it's still feeling wrong to not update the attempt's context when the metadata is updated by the LB policy. Unless there's a strong reason not to do this, then I'd rather do it and be safe vs. sorry. The concerns about it affecting subsequent attempts are a misunderstanding of the design of attempts (everything in an attempt should be local to that one attempt), and not a good reason to not update in place. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Done. |
||||
} | ||||
|
||||
s, err := a.t.NewStream(ctx, cs.callHdr) | ||||
if err != nil { | ||||
nse, ok := err.(*transport.NewStreamError) | ||||
if !ok { | ||||
|
@@ -529,12 +545,12 @@ type clientStream struct { | |||
// csAttempt implements a single transport stream attempt within a | ||||
// clientStream. | ||||
type csAttempt struct { | ||||
ctx context.Context | ||||
cs *clientStream | ||||
t transport.ClientTransport | ||||
s *transport.Stream | ||||
p *parser | ||||
done func(balancer.DoneInfo) | ||||
ctx context.Context | ||||
cs *clientStream | ||||
t transport.ClientTransport | ||||
s *transport.Stream | ||||
p *parser | ||||
pickResult balancer.PickResult | ||||
|
||||
finished bool | ||||
dc Decompressor | ||||
|
@@ -1103,12 +1119,12 @@ func (a *csAttempt) finish(err error) { | |||
tr = a.s.Trailer() | ||||
} | ||||
|
||||
if a.done != nil { | ||||
if a.pickResult.Done != nil { | ||||
br := false | ||||
if a.s != nil { | ||||
br = a.s.BytesReceived() | ||||
} | ||||
a.done(balancer.DoneInfo{ | ||||
a.pickResult.Done(balancer.DoneInfo{ | ||||
Err: err, | ||||
Trailer: tr, | ||||
BytesSent: a.s != nil, | ||||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Maybe add to this sentence:
..., or create a version of AppendToOutgoingContext that accepts a metadata.MD.
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done.