Skip to content
New issue

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

#

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

Already on GitHub? # to your account

go mod: update grpc module to latest 1.59.0 #352

Merged
merged 4 commits into from
Nov 21, 2023
Merged

go mod: update grpc module to latest 1.59.0 #352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8bb2182
go mod: update grpc module to latest 1.59.0
mdibaiee Oct 26, 2023
20100d9
dispatcher_test: remove use of deprecated UpdateSubConnState
mdibaiee Nov 15, 2023
514fc3b
dispatcher: migrate to SubConn.Shutdown and deprecate Addresses
mdibaiee Nov 15, 2023
f034b5c
go fmt
mdibaiee Nov 17, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 20 additions & 10 deletions broker/protocol/dispatcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,11 +29,11 @@ func RegisterGRPCDispatcher(localZone string) {
// passed to a gRPC RPC call. If ProcessSpec_ID is non-zero valued, the RPC is
// dispatched to the specified member. Otherwise, the RPC is dispatched to a
// Route member, preferring:
// * A member not having a currently-broken network connection (eg, due to
// a stale Route or network split).
// * A member which is in the same zone as the caller (potentially reducing
// network traffic costs.
// * A member having a Ready connection (potentially reducing latency).
// - A member not having a currently-broken network connection (eg, due to
// a stale Route or network split).
// - A member which is in the same zone as the caller (potentially reducing
// network traffic costs.
// - A member having a Ready connection (potentially reducing latency).
func WithDispatchRoute(ctx context.Context, rt Route, id ProcessSpec_ID) context.Context {
return context.WithValue(ctx, dispatchRouteCtxKey{}, dispatchRoute{route: rt, id: id})
}
Expand Down Expand Up @@ -110,7 +110,7 @@ func (d *dispatcher) UpdateClientConnState(_ balancer.ClientConnState) error {
// implements its own resolution and selection of an appropriate A record.
func (d *dispatcher) ResolverError(_ error) {}

func (d *dispatcher) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
func (d *dispatcher) updateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
d.mu.Lock()
var id, ok = d.connID[sc]
if !ok {
Expand Down Expand Up @@ -152,6 +152,13 @@ func (d *dispatcher) UpdateSubConnState(sc balancer.SubConn, state balancer.SubC
})
}

// This method has been deprecated but may still be in use. See https://github.com/grpc/grpc-go/pull/6481
// For updates to the logic, apply them to `updateSubConnState` instead which has been integrated with the new
// StateListener interface
func (d *dispatcher) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
d.updateSubConnState(sc, state)
}

// markedSubConn tracks the last mark associated with a SubConn.
// SubConns not used for a complete sweep interval are closed.
type markedSubConn struct {
Expand Down Expand Up @@ -186,9 +193,12 @@ func (d *dispatcher) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
if msc.subConn, err = d.cc.NewSubConn(
[]resolver.Address{{
Addr: d.idToAddr(dr.route, dispatchID),
Type: resolver.Backend,
}},
balancer.NewSubConnOptions{},
balancer.NewSubConnOptions{
StateListener: func(state balancer.SubConnState) {
d.updateSubConnState(msc.subConn, state)
},
},
); err != nil {
return balancer.PickResult{}, err
}
Expand Down Expand Up @@ -305,10 +315,10 @@ func (d *dispatcher) sweep() {
d.mu.Unlock()

for _, sc := range toSweep {
// RemoveSubConn begins SubConn shutdown. We expect to see a
// SubConn.Shutdown begins a shutdown. We expect to see a
// HandleSubConnStateChange with connectivity.Shutdown, at which
// point we'll de-index it.
d.cc.RemoveSubConn(sc)
sc.Shutdown()
}
}

Expand Down
87 changes: 52 additions & 35 deletions broker/protocol/dispatcher_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ func (s *DispatcherSuite) TestContextAdapters(c *gc.C) {
func (s *DispatcherSuite) TestDispatchCases(c *gc.C) {
var cc mockClientConn
var disp = dispatcherBuilder{zone: "local"}.Build(&cc, balancer.BuildOptions{}).(*dispatcher)
cc.disp = disp
close(disp.sweepDoneCh) // Disable async sweeping.

// Case: Called without a dispatchRoute. Expect it panics.
Expand All @@ -58,75 +59,75 @@ func (s *DispatcherSuite) TestDispatchCases(c *gc.C) {
// SubConn to the default service address is started.
var _, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.Equals, balancer.ErrNoSubConnAvailable)
c.Check(cc.created, gc.DeepEquals, []mockSubConn{"default.addr"})
c.Check(cc.created, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "default.addr", disp: disp}})
cc.created = nil

// Case: Default connection transitions to Ready. Expect it's now returned.
disp.UpdateSubConnState(mockSubConn("default.addr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "default.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})

result, err := disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.IsNil)
c.Check(result.Done, gc.IsNil)
c.Check(result.SubConn, gc.Equals, mockSubConn("default.addr"))
c.Check(result.SubConn, gc.Equals, mockSubConn{Name: "default.addr", disp: disp})

// Case: Specific remote peer is dispatched to.
ctx = WithDispatchRoute(context.Background(),
buildRouteFixture(), ProcessSpec_ID{Zone: "remote", Suffix: "primary"})

result, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.Equals, balancer.ErrNoSubConnAvailable)
c.Check(cc.created, gc.DeepEquals, []mockSubConn{"remote.addr"})
c.Check(cc.created, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "remote.addr", disp: disp}})
cc.created = nil

disp.UpdateSubConnState(mockSubConn("remote.addr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "remote.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})

result, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.IsNil)
c.Check(result.Done, gc.IsNil)
c.Check(result.SubConn, gc.Equals, mockSubConn("remote.addr"))
c.Check(result.SubConn, gc.Equals, mockSubConn{Name: "remote.addr", disp: disp})

// Case: Route allows for multiple members. A local one is now dialed.
ctx = WithDispatchRoute(context.Background(), buildRouteFixture(), ProcessSpec_ID{})

_, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.Equals, balancer.ErrNoSubConnAvailable)
c.Check(cc.created, gc.DeepEquals, []mockSubConn{"local.addr"})
c.Check(cc.created, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "local.addr", disp: disp}})
cc.created = nil

disp.UpdateSubConnState(mockSubConn("local.addr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "local.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})

result, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.IsNil)
c.Check(result.Done, gc.IsNil)
c.Check(result.SubConn, gc.Equals, mockSubConn("local.addr"))
c.Check(result.SubConn, gc.Equals, mockSubConn{Name: "local.addr", disp: disp})

// Case: One local addr is marked as failed. Another is dialed.
disp.UpdateSubConnState(mockSubConn("local.addr"), balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
mockSubConn{Name: "local.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})

_, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.Equals, balancer.ErrNoSubConnAvailable)
c.Check(cc.created, gc.DeepEquals, []mockSubConn{"local.otherAddr"})
c.Check(cc.created, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "local.otherAddr", disp: disp}})
cc.created = nil

disp.UpdateSubConnState(mockSubConn("local.otherAddr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "local.otherAddr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})

result, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.IsNil)
c.Check(result.Done, gc.IsNil)
c.Check(result.SubConn, gc.Equals, mockSubConn("local.otherAddr"))
c.Check(result.SubConn, gc.Equals, mockSubConn{Name: "local.otherAddr", disp: disp})

// Case: otherAddr is also failed. Expect that an error is returned,
// rather than dispatch to remote addr. (Eg we prefer to wait for a
// local replica to recover or the route to change, vs using a remote
// endpoint which incurs more networking cost).
disp.UpdateSubConnState(mockSubConn("local.otherAddr"), balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
mockSubConn{Name: "local.otherAddr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})

_, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.Equals, balancer.ErrTransientFailure)

// Case: local.addr is Ready again. However, primary is required and has failed.
disp.UpdateSubConnState(mockSubConn("local.addr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
disp.UpdateSubConnState(mockSubConn("remote.addr"), balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
mockSubConn{Name: "local.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "remote.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})

ctx = WithDispatchRoute(context.Background(),
buildRouteFixture(), ProcessSpec_ID{Zone: "remote", Suffix: "primary"})
Expand All @@ -150,7 +151,7 @@ func (s *DispatcherSuite) TestDispatchCases(c *gc.C) {
result, err = disp.Pick(balancer.PickInfo{Ctx: ctx})
c.Check(err, gc.IsNil)
c.Check(result.Done, gc.NotNil)
c.Check(result.SubConn, gc.Equals, mockSubConn("local.addr"))
c.Check(result.SubConn, gc.Equals, mockSubConn{Name: "local.addr", disp: disp})

// Closure callback with an Unavailable error (only) will trigger an invalidation.
result.Done(balancer.DoneInfo{Err: nil})
Expand All @@ -164,6 +165,7 @@ func (s *DispatcherSuite) TestDispatchCases(c *gc.C) {
func (s *DispatcherSuite) TestDispatchMarkAndSweep(c *gc.C) {
var cc mockClientConn
var disp = dispatcherBuilder{zone: "local"}.Build(&cc, balancer.BuildOptions{}).(*dispatcher)
cc.disp = disp
defer disp.Close()

var err error
Expand All @@ -177,11 +179,11 @@ func (s *DispatcherSuite) TestDispatchMarkAndSweep(c *gc.C) {
_, err = disp.Pick(balancer.PickInfo{Ctx: localCtx})
c.Check(err, gc.Equals, balancer.ErrNoSubConnAvailable)

c.Check(cc.created, gc.DeepEquals, []mockSubConn{"remote.addr", "local.addr"})
c.Check(cc.created, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "remote.addr", disp: disp}, mockSubConn{Name: "local.addr", disp: disp}})
cc.created = nil

disp.UpdateSubConnState(mockSubConn("remote.addr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
disp.UpdateSubConnState(mockSubConn("local.addr"), balancer.SubConnState{ConnectivityState: connectivity.Connecting})
mockSubConn{Name: "remote.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "local.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})

disp.sweep()
c.Check(cc.removed, gc.IsNil)
Expand All @@ -205,14 +207,14 @@ func (s *DispatcherSuite) TestDispatchMarkAndSweep(c *gc.C) {

// This time, expect that local.addr is swept.
disp.sweep()
c.Check(cc.removed, gc.DeepEquals, []mockSubConn{"local.addr"})
c.Check(cc.removed, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "local.addr", disp: disp}})
cc.removed = nil
disp.UpdateSubConnState(mockSubConn("local.addr"), balancer.SubConnState{ConnectivityState: connectivity.Shutdown})
mockSubConn{Name: "local.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Shutdown})

disp.sweep() // Now remote.addr is swept.
c.Check(cc.removed, gc.DeepEquals, []mockSubConn{"remote.addr"})
c.Check(cc.removed, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "remote.addr", disp: disp}})
cc.removed = nil
disp.UpdateSubConnState(mockSubConn("remote.addr"), balancer.SubConnState{ConnectivityState: connectivity.Shutdown})
mockSubConn{Name: "remote.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Shutdown})

// No connections remain.
c.Check(disp.idConn, gc.HasLen, 0)
Expand All @@ -223,10 +225,10 @@ func (s *DispatcherSuite) TestDispatchMarkAndSweep(c *gc.C) {
_, err = disp.Pick(balancer.PickInfo{Ctx: localCtx})
c.Check(err, gc.Equals, balancer.ErrNoSubConnAvailable)

c.Check(cc.created, gc.DeepEquals, []mockSubConn{"local.addr"})
c.Check(cc.created, gc.DeepEquals, []mockSubConn{mockSubConn{Name: "local.addr", disp: disp}})
cc.created = nil

disp.UpdateSubConnState(mockSubConn("local.addr"), balancer.SubConnState{ConnectivityState: connectivity.Ready})
mockSubConn{Name: "local.addr", disp: disp}.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
_, err = disp.Pick(balancer.PickInfo{Ctx: localCtx})
c.Check(err, gc.IsNil)
}
Expand All @@ -235,27 +237,42 @@ type mockClientConn struct {
err error
created []mockSubConn
removed []mockSubConn
disp *dispatcher
}

type mockSubConn string
type mockSubConn struct {
Name string
disp *dispatcher
}

func (s1 mockSubConn) Equal(s2 mockSubConn) bool {
return s1.Name == s2.Name
}

func (s mockSubConn) UpdateAddresses([]resolver.Address) {}
func (s mockSubConn) Connect() {}
func (s mockSubConn) UpdateAddresses([]resolver.Address) { panic("deprecated") }
func (s mockSubConn) UpdateState(state balancer.SubConnState) { s.disp.updateSubConnState(s, state) }
func (s mockSubConn) Connect() {}
func (s mockSubConn) GetOrBuildProducer(balancer.ProducerBuilder) (balancer.Producer, func()) {
return nil, func() {}
}
func (s mockSubConn) Shutdown() {
var c = s.disp.cc.(*mockClientConn)
c.removed = append(c.removed, s)
}

func (c *mockClientConn) NewSubConn(a []resolver.Address, _ balancer.NewSubConnOptions) (balancer.SubConn, error) {
var sc = mockSubConn(a[0].Addr)
var sc = mockSubConn{Name: a[0].Addr, disp: c.disp}
c.created = append(c.created, sc)
return sc, c.err
}

func (c *mockClientConn) RemoveSubConn(sc balancer.SubConn) {
c.removed = append(c.removed, sc.(mockSubConn))
}

func (c *mockClientConn) UpdateAddresses(balancer.SubConn, []resolver.Address) {}
func (c *mockClientConn) UpdateAddresses(balancer.SubConn, []resolver.Address) { panic("deprecated") }
func (c *mockClientConn) UpdateState(balancer.State) {}
func (c *mockClientConn) ResolveNow(resolver.ResolveNowOptions) {}
func (c *mockClientConn) Target() string { return "default.addr" }
func (c *mockClientConn) RemoveSubConn(sc balancer.SubConn) {
sc.Shutdown()
}

type mockRouter struct{ invalidated string }

Expand Down
51 changes: 29 additions & 22 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
module go.gazette.dev/core

go 1.18
go 1.19

require (
cloud.google.com/go/storage v1.16.1
cloud.google.com/go/storage v1.30.1
github.com/Azure/azure-pipeline-go v0.2.3
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.6.1
github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.3.0
Expand All @@ -14,9 +14,9 @@ require (
github.com/dustin/go-humanize v1.0.0
github.com/dustinkirkland/golang-petname v0.0.0-20191129215211-8e5a1ed0cff0
github.com/gogo/protobuf v1.3.2
github.com/golang/protobuf v1.5.2
github.com/golang/protobuf v1.5.3
github.com/golang/snappy v0.0.4
github.com/google/uuid v1.3.0
github.com/google/uuid v1.3.1
github.com/gorilla/schema v1.2.0
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
github.com/hashicorp/golang-lru v0.5.4
Expand All @@ -34,36 +34,40 @@ require (
github.com/sirupsen/logrus v1.8.1
github.com/soheilhy/cmux v0.1.5
github.com/spf13/afero v1.6.0
github.com/stretchr/testify v1.7.1
github.com/stretchr/testify v1.8.1
go.etcd.io/etcd/api/v3 v3.5.0
go.etcd.io/etcd/client/v3 v3.5.0
golang.org/x/net v0.8.0
golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
google.golang.org/api v0.56.0
google.golang.org/grpc v1.40.0
google.golang.org/protobuf v1.27.1
golang.org/x/net v0.14.0
golang.org/x/oauth2 v0.11.0
golang.org/x/sync v0.3.0
google.golang.org/api v0.126.0
google.golang.org/grpc v1.59.0
google.golang.org/protobuf v1.31.0
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c
gopkg.in/yaml.v2 v2.4.0
k8s.io/api v0.0.0-20190620073856-dcce3486da33
)

require (
cloud.google.com/go v0.94.0 // indirect
cloud.google.com/go v0.110.7 // indirect
cloud.google.com/go/compute v1.23.0 // indirect
cloud.google.com/go/compute/metadata v0.2.3 // indirect
cloud.google.com/go/iam v1.1.1 // indirect
github.com/Azure/azure-sdk-for-go/sdk/internal v1.3.0 // indirect
github.com/AzureAD/microsoft-authentication-library-for-go v1.0.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.1.2 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/cockroachdb/apd v1.1.0 // indirect
github.com/coreos/go-semver v0.3.0 // indirect
github.com/coreos/go-systemd/v22 v22.3.2 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96 // indirect
github.com/golang-jwt/jwt/v4 v4.5.0 // indirect
github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e // indirect
github.com/google/go-cmp v0.5.6 // indirect
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
github.com/google/gofuzz v1.0.0 // indirect
github.com/googleapis/gax-go/v2 v2.1.0 // indirect
github.com/google/s2a-go v0.1.4 // indirect
github.com/googleapis/enterprise-certificate-proxy v0.2.3 // indirect
github.com/googleapis/gax-go/v2 v2.11.0 // indirect
github.com/googleapis/gnostic v0.0.0-20170729233727-0c5108395e2d // indirect
github.com/imdario/mergo v0.3.5 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
Expand All @@ -85,17 +89,20 @@ require (
github.com/satori/go.uuid v1.2.0 // indirect
github.com/spf13/pflag v1.0.3 // indirect
go.etcd.io/etcd/client/pkg/v3 v3.5.0 // indirect
go.opencensus.io v0.23.0 // indirect
go.opencensus.io v0.24.0 // indirect
go.uber.org/atomic v1.9.0 // indirect
go.uber.org/multierr v1.7.0 // indirect
go.uber.org/zap v1.19.0 // indirect
golang.org/x/crypto v0.7.0 // indirect
golang.org/x/sys v0.6.0 // indirect
golang.org/x/term v0.6.0 // indirect
golang.org/x/text v0.8.0 // indirect
golang.org/x/crypto v0.12.0 // indirect
golang.org/x/sys v0.11.0 // indirect
golang.org/x/term v0.11.0 // indirect
golang.org/x/text v0.12.0 // indirect
golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
google.golang.org/appengine v1.6.7 // indirect
google.golang.org/genproto v0.0.0-20210831024726-fe130286e0e2 // indirect
google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
gopkg.in/inf.v0 v0.9.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
k8s.io/apimachinery v0.0.0-20190620073744-d16981aedf33 // indirect
Expand Down
Loading
Loading