status-im
/
consul
mirror of https://github.com/status-im/consul.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Various bug-fixes and improvements (#20866) 

* Shuffle the list of servers returned by `pbserverdiscovery.WatchServers`.

This randomizes the list of servers to help reduce the chance of clients
all connecting to the same server simultaneously. Consul-dataplane is one
such client that does not randomize its own list of servers.

* Fix potential goroutine leak in xDS recv loop.

This commit ensures that the goroutine which receives xDS messages from
proxies will not block forever if the stream's context is cancelled but
the `processDelta()` function never consumes the message (due to being
terminated).

* Add changelog.
This commit is contained in:
Derek Menteer 2024-03-15 13:10:48 -05:00 committed by GitHub
parent 94c0d783b8
commit eabff257d7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 19 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.changelog/20866.txt Normal file
View File

@ -0,0 +1,7 @@
```release-note:improvement
api: Randomize the returned server list for the WatchServers gRPC endpoint.
```
```release-note:bug
connect: Fix potential goroutine leak in xDS stream handling.
```

5
agent/grpc-external/services/serverdiscovery/watch_servers.go
View File

@ -6,6 +6,7 @@ package serverdiscovery
import (
"context"
"errors"
"math/rand"
"github.com/hashicorp/go-hclog"
"google.golang.org/grpc/codes"
@ -130,6 +131,10 @@ func eventToResponse(req *pbserverdiscovery.WatchServersRequest, event stream.Ev
})
}
// Shuffle servers so that consul-dataplane doesn't consistently choose the same connections on startup.
rand.Shuffle(len(servers), func(i, j int) {
servers[i], servers[j] = servers[j], servers[i]
})
return &pbserverdiscovery.WatchServersResponse{
Servers: servers,
}, nil

6
agent/grpc-external/services/serverdiscovery/watch_servers_test.go
View File

@ -166,7 +166,7 @@ func TestWatchServers_StreamLifeCycle(t *testing.T) {
// 4. See the corresponding message sent back through the stream.
rsp = mustGetServers(t, rspCh)
require.NotNil(t, rsp)
prototest.AssertDeepEqual(t, twoServerResponse, rsp)
prototest.AssertElementsMatch(t, twoServerResponse.Servers, rsp.Servers)
// 5. Send a NewCloseSubscriptionEvent for the token secret.
publisher.Publish([]stream.Event{
@ -176,7 +176,7 @@ func TestWatchServers_StreamLifeCycle(t *testing.T) {
// 6. Observe another snapshot message
rsp = mustGetServers(t, rspCh)
require.NotNil(t, rsp)
prototest.AssertDeepEqual(t, twoServerResponse, rsp)
prototest.AssertElementsMatch(t, twoServerResponse.Servers, rsp.Servers)
// 7. Publish another event to move to 3 servers.
publisher.Publish([]stream.Event{
@ -192,7 +192,7 @@ func TestWatchServers_StreamLifeCycle(t *testing.T) {
// seen after stream reinitialization.
rsp = mustGetServers(t, rspCh)
require.NotNil(t, rsp)
prototest.AssertDeepEqual(t, threeServerResponse, rsp)
prototest.AssertElementsMatch(t, threeServerResponse.Servers, rsp.Servers)
}
func TestWatchServers_ACLToken_AnonymousToken(t *testing.T) {

5
agent/xds/delta.go
View File

@ -79,7 +79,10 @@ func (s *Server) DeltaAggregatedResources(stream ADSDeltaStream) error {
close(reqCh)
return
}
reqCh <- req
select {
case <-stream.Context().Done():
case reqCh <- req:
}
}
}()