-
Notifications
You must be signed in to change notification settings - Fork 885
PLT-1125: Bound concurrent in-flight RPCs and connections per IP on the gRPC query plane #4078
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
Merged
Merged
Changes from all commits
Commits
Show all changes
4 commits
Select commit
Hold shift + click to select a range
4ef28e0
PLT-1125: Bound concurrent in-flight RPCs and connections per IP on gRPC
amir-deris b1b4d2e
Point the CHANGELOG entries at the PR they landed in
amir-deris e6fcbf8
Default max-connections-per-ip to 0 (unlimited)
amir-deris 41a7c6a
Addressing AI feedback
amir-deris File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package ratelimiter | ||
|
|
||
| import ( | ||
| "context" | ||
| "net" | ||
| "sync" | ||
|
|
||
| "go.opentelemetry.io/otel/attribute" | ||
| "go.opentelemetry.io/otel/metric" | ||
| ) | ||
|
|
||
| // ConnLimitListener returns a listener that bounds the number of simultaneously | ||
| // open connections from any one client address to maxPerIP. A non-positive | ||
| // maxPerIP returns inner unchanged. | ||
| // | ||
| // It bounds the layer below the RPC — accepted sockets, TLS handshakes, HTTP/2 | ||
| // frame state — which a per-RPC counter cannot see, and it caps the per-address | ||
| // share of the global connection budget. Wrap the raw listener with this before | ||
| // the global cap, so a connection this rejects never spends a global slot. | ||
| // | ||
| // Addresses are keyed on the TCP peer, normalized the way rate-limit buckets | ||
| // are: a client rotating within an IPv6 /64 does not get a fresh allowance per | ||
| // address, and every client sharing an address shares one allowance. | ||
| // trusted-proxy-cidrs does not apply here, so on a node behind a proxy or load | ||
| // balancer maxPerIP bounds that proxy as a whole rather than its clients. | ||
| func ConnLimitListener(inner net.Listener, plane string, maxPerIP int) net.Listener { | ||
| if maxPerIP <= 0 { | ||
| return inner | ||
| } | ||
| return &connLimitListener{ | ||
| Listener: inner, | ||
| plane: plane, | ||
| counter: newInflightCounter(maxPerIP), | ||
| } | ||
| } | ||
|
|
||
| // connLimitListener is the listener ConnLimitListener returns. | ||
| type connLimitListener struct { | ||
| net.Listener | ||
|
|
||
| plane string | ||
| counter *inflightCounter | ||
| } | ||
|
|
||
| // Accept returns the next connection whose client address is under the cap, | ||
| // closing and counting the ones that are not. | ||
| // | ||
| // Over-cap connections are dropped here rather than surfaced as an Accept error, | ||
| // because an Accept error stops the serving loop and would turn one client's | ||
| // excess into an outage for every other client. | ||
| func (l *connLimitListener) Accept() (net.Conn, error) { | ||
| for { | ||
| conn, err := l.Listener.Accept() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| // The peer address is all a listener has: nothing has been read off the | ||
| // socket yet, so there is no X-Forwarded-For to resolve a client behind | ||
| // a trusted proxy to. | ||
| key := bucketKey(stripPort(conn.RemoteAddr().String())) | ||
| if l.counter.acquire(key) { | ||
| return &limitedConn{Conn: conn, release: func() { l.counter.release(key) }}, nil | ||
| } | ||
| registryMetrics.connRejectedCounter.Add( | ||
| context.Background(), | ||
| 1, | ||
| metric.WithAttributes(attribute.String("plane", l.plane)), | ||
| ) | ||
| _ = conn.Close() | ||
| } | ||
| } | ||
|
|
||
| // limitedConn returns its address's connection slot when it is closed. | ||
| type limitedConn struct { | ||
| net.Conn | ||
|
|
||
| once sync.Once | ||
| release func() | ||
| } | ||
|
|
||
| // Close closes the underlying connection and returns its slot. It is safe to | ||
| // call more than once: net/http and grpc-go both close a connection from more | ||
| // than one path, and a double release would hand the address a free slot. | ||
| func (c *limitedConn) Close() error { | ||
| err := c.Conn.Close() | ||
| c.once.Do(c.release) | ||
| return err | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| package ratelimiter | ||
|
|
||
| import ( | ||
| "io" | ||
| "net" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| // acceptedConns runs a listener's Accept loop into a channel so a test can see | ||
| // which connections the per-IP cap let through. | ||
| func acceptedConns(t *testing.T, ln net.Listener) <-chan net.Conn { | ||
| t.Helper() | ||
| out := make(chan net.Conn, 16) | ||
| go func() { | ||
| defer close(out) | ||
| for { | ||
| conn, err := ln.Accept() | ||
| if err != nil { | ||
| return | ||
| } | ||
| out <- conn | ||
| } | ||
| }() | ||
| return out | ||
| } | ||
|
|
||
| func dialTo(t *testing.T, ln net.Listener) net.Conn { | ||
| t.Helper() | ||
| conn, err := net.Dial("tcp", ln.Addr().String()) | ||
| require.NoError(t, err) | ||
| t.Cleanup(func() { _ = conn.Close() }) | ||
| return conn | ||
| } | ||
|
|
||
| // requireClosedByServer pins that the server hung up on conn rather than serving | ||
| // it, which is how an over-cap connection is refused. | ||
| func requireClosedByServer(t *testing.T, conn net.Conn) { | ||
| t.Helper() | ||
| require.NoError(t, conn.SetReadDeadline(time.Now().Add(5*time.Second))) | ||
| _, err := conn.Read(make([]byte, 1)) | ||
| require.ErrorIs(t, err, io.EOF) | ||
| } | ||
|
|
||
| func TestConnLimitListenerCapsConnectionsPerIP(t *testing.T) { | ||
| raw, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| ln := ConnLimitListener(raw, PlaneGRPC, 2) | ||
| t.Cleanup(func() { _ = ln.Close() }) | ||
| accepted := acceptedConns(t, ln) | ||
|
|
||
| first := dialTo(t, ln) | ||
| second := dialTo(t, ln) | ||
| served := []net.Conn{<-accepted, <-accepted} | ||
|
|
||
| // The third is dropped, and the loop keeps serving rather than failing. | ||
| requireClosedByServer(t, dialTo(t, ln)) | ||
| require.Empty(t, accepted) | ||
|
|
||
| // Closing a served connection returns its slot. | ||
| require.NoError(t, served[0].Close()) | ||
| require.NoError(t, first.Close()) | ||
| dialTo(t, ln) | ||
| require.NotNil(t, <-accepted) | ||
|
|
||
| _ = second.Close() | ||
| _ = served[1].Close() | ||
| } | ||
|
|
||
| // TestConnLimitListenerDoubleCloseReturnsOneSlot pins the idempotent release: | ||
| // net/http and grpc-go both close a connection from more than one path, and a | ||
| // second release would hand the address a slot it is not holding. | ||
| func TestConnLimitListenerDoubleCloseReturnsOneSlot(t *testing.T) { | ||
| raw, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| ln := ConnLimitListener(raw, PlaneGRPC, 1) | ||
| t.Cleanup(func() { _ = ln.Close() }) | ||
| accepted := acceptedConns(t, ln) | ||
|
|
||
| dialTo(t, ln) | ||
| served := <-accepted | ||
| require.NoError(t, served.Close()) | ||
| _ = served.Close() | ||
|
|
||
| limiter, ok := ln.(*connLimitListener) | ||
| require.True(t, ok) | ||
| require.Equal(t, 0, limiter.counter.heldFor("127.0.0.1")) | ||
|
|
||
| // One slot came back, not two. | ||
| dialTo(t, ln) | ||
| require.NotNil(t, <-accepted) | ||
| requireClosedByServer(t, dialTo(t, ln)) | ||
| } | ||
|
|
||
| // TestConnLimitListenerDisabledReturnsInner pins that a non-positive cap adds no | ||
| // wrapper at all, so the listener behaves exactly as before. | ||
| func TestConnLimitListenerDisabledReturnsInner(t *testing.T) { | ||
| raw, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| t.Cleanup(func() { _ = raw.Close() }) | ||
|
|
||
| require.Same(t, raw, ConnLimitListener(raw, PlaneGRPC, 0)) | ||
| require.Same(t, raw, ConnLimitListener(raw, PlaneGRPC, -1)) | ||
| } | ||
|
|
||
| // TestConnLimitListenerAcceptStopsOnListenerError pins that a closed listener | ||
| // still ends the Accept loop rather than spinning inside it. | ||
| func TestConnLimitListenerAcceptStopsOnListenerError(t *testing.T) { | ||
| raw, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| ln := ConnLimitListener(raw, PlaneGRPC, 1) | ||
| require.NoError(t, ln.Close()) | ||
|
|
||
| _, err = ln.Accept() | ||
| require.Error(t, err) | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
[suggestion] This rewrite makes a claim the code does not support. Four
grpckeys are read with noIsSetguard and no clamp:grpc.enable,grpc.address,grpc.keepalive-permit-without-stream, andgrpc.rate-limiting-enabled(sei-cosmos/server/config/config.go:731,:732,:742,:746). The connection-age durations are not "the exception" — they are the exception among the unguarded keys in that their absent-key value coincides with the declared default.That matters most for
grpc.enable, which is the key the paragraph directly above this one is about: it is declaredtruefor a full node and an archive node andfalsefor a validator and a seed, and because it is read unconditionally an absent key casts tofalseand clobbers the declaredtrue. A reader who takes this paragraph at face value concludes the mode rule survives an absent key, which is the opposite of what happens. The text this replaced accounted for those four by arithmetic ("Nine of these fifteen keys... Two more are durations..."), so the gap was at least visible; the new phrasing closes it with a false generalization.The two keys this PR adds are both
IsSet-guarded, so the honest update is to keep the shape of the old claim rather than broaden it — name the unguarded set, and say which members of it are rescued by a coinciding zero and which are not.