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.

Sign up for GitHub

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

Jump to bottom

fix: ensure the xds grpc server is properly stopped #1860

Merged
merged 4 commits into from
Sep 12, 2023
Merged

fix: ensure the xds grpc server is properly stopped #1860

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
5e6e46e
fix: ensure the xds grpc server is properly stopped
shawnh2 Sep 1, 2023
2887e63
fix ci test and simplify the grpc serve expression
shawnh2 Sep 1, 2023
d68e8f1
Merge branch 'main' into fix-grpc-close
Xunzhuo Sep 5, 2023
7649614
Merge branch 'main' into fix-grpc-close
zirain Sep 6, 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
15 changes: 9 additions & 6 deletions internal/globalratelimit/runner/runner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ func (r *Runner) Start(ctx context.Context) error {
discoveryv3.RegisterAggregatedDiscoveryServiceServer(r.grpc, serverv3.NewServer(ctx, r.cache, cb))

// Start and listen xDS gRPC config Server.
go r.serverXdsConfigServer(ctx)
go r.serveXdsConfigServer(ctx)

// Start message Subscription.
go r.subscribeAndTranslate(ctx)
Expand All @@ -90,20 +90,23 @@ func (r *Runner) Start(ctx context.Context) error {
return nil
}

func (r *Runner) serverXdsConfigServer(ctx context.Context) {
func (r *Runner) serveXdsConfigServer(ctx context.Context) {
addr := net.JoinHostPort(XdsGrpcSotwConfigServerAddress, strconv.Itoa(ratelimit.XdsGrpcSotwConfigServerPort))
l, err := net.Listen("tcp", addr)
if err != nil {
r.Logger.Error(err, "failed to listen on address", "address", addr)
return
}

go func() {
<-ctx.Done()
r.Logger.Info("grpc server shutting down")
r.grpc.Stop()
}()

if err = r.grpc.Serve(l); err != nil {
r.Logger.Error(err, "failed to start grpc based xds config server")
}

<-ctx.Done()
r.Logger.Info("grpc config server shutting down")
r.grpc.Stop()
}

func (r *Runner) subscribeAndTranslate(ctx context.Context) {
Expand Down
22 changes: 12 additions & 10 deletions internal/xds/server/runner/runner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,18 +102,20 @@
r.Logger.Error(err, "failed to listen on address", "address", addr)
return
}
err = r.grpc.Serve(l)
if err != nil {

go func() {
<-ctx.Done()
r.Logger.Info("grpc server shutting down")
// We don't use GracefulStop here because envoy
// has long-lived hanging xDS requests. There's no
// mechanism to make those pending requests fail,
// so we forcibly terminate the TCP sessions.
r.grpc.Stop()
}()

Check warning on line 114 in internal/xds/server/runner/runner.go

View check run for this annotation

Codecov / codecov/patch

internal/xds/server/runner/runner.go#L106-L114 

Added lines #L106 - L114 were not covered by tests

if err = r.grpc.Serve(l); err != nil {

Check warning on line 116 in internal/xds/server/runner/runner.go

View check run for this annotation

Codecov / codecov/patch

internal/xds/server/runner/runner.go#L116 

Added line #L116 was not covered by tests
r.Logger.Error(err, "failed to start grpc based xds server")
}

<-ctx.Done()
r.Logger.Info("grpc server shutting down")
// We don't use GracefulStop here because envoy
// has long-lived hanging xDS requests. There's no
// mechanism to make those pending requests fail,
// so we forcibly terminate the TCP sessions.
r.grpc.Stop()
}

// registerServer registers the given xDS protocol Server with the gRPC
Expand Down