Repository: yorkie-team/yorkie
Pull request #512 (2023-04-12 15:30:22)

Add gRPC MaxConnectionAge & MaxConnectionAgeGrace options
<!--  Thanks for sending a pull request! -->

**What this PR does / why we need it**:

Add gRPC `MaxConnectionAge` & `MaxConnectionAgeGrace` options.

This is to avoid "split-brain" of long-lived gRPC stream connection when performing hash-based load balancing in Yorkie cluster.

For more information, follow: https://github.com/envoyproxy/envoy/issues/26459

gRPC suggest to use `MAX_CONNECTION_AGE` to avoid long-lived RPC issue on load balancing(https://youtu.be/Naonb2XD_2Q?t=136), but stream is not actually closed even `MAX_CONNECTION_AGE` is configured.

This is because after `MAX_CONNECTION_AGE`, server sends GoAway to client, but GoAway is not a signal to close connection instantly. It's just a signal to tell client not to send additional request to server(https://github.com/grpc/grpc-java/issues/8770)

Therefore, `MAX_CONNECTION_AGE_GRACE` is also introduced to forcibly close stream.

So this is how it works:
1. After stream is established, stream is available for `maxConnectionAge`.
2. After `maxConnectionAge`, GoAway frame is sent to client to gracefully close the stream. But stream is not closed at this point.
3. After `MaxConnectionAgeGrace`, stream is forcibly closed.

Reference: https://pkg.go.dev/google.golang.org/grpc/keepalive

Hence, `maxConnectionAge` and `maxConnectionAgeGrace` options are added as gRPC server's keepalive option parameters.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #

**Special notes for your reviewer**:

As I researched some information about long-lived connection, graceful close, etc, I now have good explanation about why `close_connections_on_host_set_change`(gracefully draining connections) are not working for long-lived gRPC connections.

It is because envoy sends HTTP2 GoAway frame on draining sequence, but GoAway is not for instant connection close. Therefore, no connection close are made after draining, and `close_connections_on_host_set_change` option will not work as well.

References: https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/operations/draining

Also, I found some interesting issue related to our issue on long-lived connection close: https://github.com/grpc/grpc/issues/26703.

Maybe it will be useful to introduce `force_close_connections_on_host_set_change` on envoy:: https://github.com/envoyproxy/envoy/issues/26459

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note

```

**Additional documentation**:

<!--
This section can be blank if this pull request does not require a release note.

Please use the following format for linking documentation:
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```

**Checklist**:
- [x] Added relevant tests or not required
- [x] Didn't break anything


codecov[bot]: ## [Codecov](https://codecov.io/gh/yorkie-team/yorkie/pull/512?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) Report
> Merging [#512](https://codecov.io/gh/yorkie-team/yorkie/pull/512?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) (653649d) into [main](https://codecov.io/gh/yorkie-team/yorkie/commit/efe3e87da1e726858408209142476b0a75b7b9d1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) (efe3e87) will **increase** coverage by `0.84%`.
> The diff coverage is `23.07%`.

```diff
@@            Coverage Diff             @@
##             main     #512      +/-   ##
==========================================
+ Coverage   50.22%   51.06%   +0.84%     
==========================================
  Files          64       64              
  Lines        5671     5697      +26     
==========================================
+ Hits         2848     2909      +61     
+ Misses       2481     2429      -52     
- Partials      342      359      +17     
```


| [Impacted Files](https://codecov.io/gh/yorkie-team/yorkie/pull/512?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team) | Coverage Δ | |
|---|---|---|
| [server/config.go](https://codecov.io/gh/yorkie-team/yorkie/pull/512?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team#diff-c2VydmVyL2NvbmZpZy5nbw==) | `43.33% <0.00%> (-2.02%)` | :arrow_down: |
| [server/rpc/config.go](https://codecov.io/gh/yorkie-team/yorkie/pull/512?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team#diff-c2VydmVyL3JwYy9jb25maWcuZ28=) | `42.85% <0.00%> (-57.15%)` | :arrow_down: |
| [server/rpc/server.go](https://codecov.io/gh/yorkie-team/yorkie/pull/512?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team#diff-c2VydmVyL3JwYy9zZXJ2ZXIuZ28=) | `67.79% <60.00%> (-1.60%)` | :arrow_down: |

... and [1 file with indirect coverage changes](https://codecov.io/gh/yorkie-team/yorkie/pull/512/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team)

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yorkie-team)


Patch:
diff --git a/cmd/yorkie/server.go b/cmd/yorkie/server.go
index 56057c30..272ac130 100644
--- a/cmd/yorkie/server.go
+++ b/cmd/yorkie/server.go
@@ -191,6 +191,18 @@ func init() {
 		server.DefaultRPCMaxRequestsBytes,
 		"Maximum client request size in bytes the server will accept.",
 	)
+	cmd.Flags().StringVar(
+		&conf.RPC.MaxConnectionAge,
+		"rpc-max-connection-age",
+		server.DefaultRPCMaxConnectionAge.String(),
+		"Maximum duration of connection may exist before it will be closed by sending a GoAway.",
+	)
+	cmd.Flags().StringVar(
+		&conf.RPC.MaxConnectionAgeGrace,
+		"rpc-max-connection-age-grace",
+		server.DefaultRPCMaxConnectionAgeGrace.String(),
+		"Additional grace period after MaxConnectionAge after which connections will be forcibly closed.",
+	)
 	cmd.Flags().IntVar(
 		&conf.Profiling.Port,
 		"profiling-port",
diff --git a/server/config.go b/server/config.go
index bcbf41a9..1a9dd8a1 100644
--- a/server/config.go
+++ b/server/config.go
@@ -34,8 +34,10 @@ import (
 
 // Below are the values of the default values of Yorkie config.
 const (
-	DefaultRPCPort             = 11101
-	DefaultRPCMaxRequestsBytes = 4 * 1024 * 1024 // 4MiB
+	DefaultRPCPort                  = 11101
+	DefaultRPCMaxRequestsBytes      = 4 * 1024 * 1024 // 4MiB
+	DefaultRPCMaxConnectionAge      = 50 * time.Second
+	DefaultRPCMaxConnectionAgeGrace = 10 * time.Second
 
 	DefaultProfilingPort = 11102
 
@@ -152,6 +154,14 @@ func (c *Config) ensureDefaultValue() {
 		c.RPC.MaxRequestBytes = DefaultRPCMaxRequestsBytes
 	}
 
+	if c.RPC.MaxConnectionAge == "" {
+		c.RPC.MaxConnectionAge = DefaultRPCMaxConnectionAge.String()
+	}
+
+	if c.RPC.MaxConnectionAgeGrace == "" {
+		c.RPC.MaxConnectionAgeGrace = DefaultRPCMaxConnectionAgeGrace.String()
+	}
+
 	if c.Profiling.Port == 0 {
 		c.Profiling.Port = DefaultProfilingPort
 	}
diff --git a/server/config.sample.yml b/server/config.sample.yml
index 0561712f..f878b744 100644
--- a/server/config.sample.yml
+++ b/server/config.sample.yml
@@ -6,6 +6,14 @@ RPC:
   # MaxRequestBytes is the maximum client request size in bytes the server will accept (default: 4194304, 4MiB).
   MaxRequestBytes: 4194304
 
+  # MaxConnectionAge is a duration for the maximum amount of time a connection may exist
+  # before it will be closed by sending a GoAway.
+  MaxConnectionAge: "50s"
+
+  # MaxConnectionAgeGrace is a duration for the amount of time after receiving a GoAway
+  # for pending RPCs to complete before forcibly closing connections.
+  MaxConnectionAgeGrace: "10s"
+
   # CertFile is the file containing the TLS certificate.
   CertFile: ""
 
diff --git a/server/rpc/config.go b/server/rpc/config.go
index 213822f7..ac7b624f 100644
--- a/server/rpc/config.go
+++ b/server/rpc/config.go
@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"time"
 )
 
 var (
@@ -29,6 +30,10 @@ var (
 	ErrInvalidCertFile = errors.New("invalid cert file for RPC server")
 	// ErrInvalidKeyFile occurs when the key file is invalid.
 	ErrInvalidKeyFile = errors.New("invalid key file for RPC server")
+	// ErrInvalidMaxConnectionAge occurs when the max connection age is invalid.
+	ErrInvalidMaxConnectionAge = errors.New("invalid max connection age for RPC server")
+	// ErrInvalidMaxConnectionAgeGrace occurs when the max connection age grace is invalid.
+	ErrInvalidMaxConnectionAgeGrace = errors.New("invalid max connection age grace for RPC server")
 )
 
 // Config is the configuration for creating a Server instance.
@@ -44,6 +49,14 @@ type Config struct {
 
 	// MaxRequestBytes is the maximum client request size in bytes the server will accept.
 	MaxRequestBytes uint64 `yaml:"MaxRequestBytes"`
+
+	// MaxConnectionAge is a duration for the maximum amount of time a connection may exist
+	// before it will be closed by sending a GoAway.
+	MaxConnectionAge string `yaml:"MaxConnectionAge"`
+
+	// MaxConnectionAgeGrace is a duration for the amount of time after receiving a GoAway
+	// for pending RPCs to complete before forcibly closing connections.
+	MaxConnectionAgeGrace string `yaml:"MaxConnectionAgeGrace"`
 }
 
 // Validate validates the port number and the files for certification.
@@ -65,5 +78,21 @@ func (c *Config) Validate() error {
 		}
 	}
 
+	if _, err := time.ParseDuration(c.MaxConnectionAge); err != nil {
+		return fmt.Errorf(
+			"%s: %w",
+			c.MaxConnectionAge,
+			ErrInvalidMaxConnectionAge,
+		)
+	}
+
+	if _, err := time.ParseDuration(c.MaxConnectionAgeGrace); err != nil {
+		return fmt.Errorf(
+			"%s: %w",
+			c.MaxConnectionAgeGrace,
+			ErrInvalidMaxConnectionAgeGrace,
+		)
+	}
+
 	return nil
 }
diff --git a/server/rpc/server.go b/server/rpc/server.go
index 71e4c13c..97f4ee52 100644
--- a/server/rpc/server.go
+++ b/server/rpc/server.go
@@ -23,12 +23,14 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"time"
 
 	grpcmiddleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/health"
 	healthpb "google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/keepalive"
 
 	api "github.com/yorkie-team/yorkie/api/yorkie/v1"
 	"github.com/yorkie-team/yorkie/server/backend"
@@ -73,9 +75,23 @@ func NewServer(conf *Config, be *backend.Backend) (*Server, error) {
 		opts = append(opts, grpc.Creds(creds))
 	}
 
+	maxConnectionAge, err := time.ParseDuration(conf.MaxConnectionAge)
+	if err != nil {
+		return nil, fmt.Errorf("parse max connection age: %w", err)
+	}
+
+	maxConnectionAgeGrace, err := time.ParseDuration(conf.MaxConnectionAgeGrace)
+	if err != nil {
+		return nil, fmt.Errorf("parse max connection age grace: %w", err)
+	}
+
 	opts = append(opts, grpc.MaxRecvMsgSize(int(conf.MaxRequestBytes)))
 	opts = append(opts, grpc.MaxSendMsgSize(math.MaxInt32))
 	opts = append(opts, grpc.MaxConcurrentStreams(math.MaxUint32))
+	opts = append(opts, grpc.KeepaliveParams(keepalive.ServerParameters{
+		MaxConnectionAge:      maxConnectionAge,
+		MaxConnectionAgeGrace: maxConnectionAgeGrace,
+	}))
 
 	yorkieServiceCtx, yorkieServiceCancel := context.WithCancel(context.Background())
 


Test patch:
diff --git a/server/rpc/server_test.go b/server/rpc/server_test.go
index 7b4d9e04..0990b97d 100644
--- a/server/rpc/server_test.go
+++ b/server/rpc/server_test.go
@@ -23,6 +23,7 @@ import (
 	"log"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc"
@@ -90,8 +91,10 @@ func TestMain(m *testing.M) {
 	}
 
 	testRPCServer, err = rpc.NewServer(&rpc.Config{
-		Port:            helper.RPCPort,
-		MaxRequestBytes: helper.RPCMaxRequestBytes,
+		Port:                  helper.RPCPort,
+		MaxRequestBytes:       helper.RPCMaxRequestBytes,
+		MaxConnectionAge:      helper.RPCMaxConnectionAge.String(),
+		MaxConnectionAgeGrace: helper.RPCMaxConnectionAgeGrace.String(),
 	}, be)
 	if err != nil {
 		log.Fatal(err)
@@ -622,6 +625,50 @@ func TestRPCServerBackend(t *testing.T) {
 		)
 		assert.Equal(t, codes.FailedPrecondition, status.Convert(err).Code())
 	})
+
+	t.Run("watch document test", func(t *testing.T) {
+		activateResp, err := testClient.ActivateClient(
+			context.Background(),
+			&api.ActivateClientRequest{ClientKey: t.Name()},
+		)
+		assert.NoError(t, err)
+
+		packWithNoChanges := &api.ChangePack{
+			DocumentKey: helper.TestDocKey(t).String(),
+			Checkpoint:  &api.Checkpoint{ServerSeq: 0, ClientSeq: 0},
+		}
+
+		resPack, err := testClient.AttachDocument(
+			context.Background(),
+			&api.AttachDocumentRequest{
+				ClientId:   activateResp.ClientId,
+				ChangePack: packWithNoChanges,
+			},
+		)
+		assert.NoError(t, err)
+
+		// watch document
+		watchResp, err := testClient.WatchDocument(
+			context.Background(),
+			&api.WatchDocumentRequest{
+				Client:     &api.Client{Id: activateResp.ClientId, Presence: &api.Presence{}},
+				DocumentId: resPack.DocumentId,
+			},
+		)
+		assert.NoError(t, err)
+
+		// check if stream is open
+		_, err = watchResp.Recv()
+		assert.NoError(t, err)
+
+		// wait for MaxConnectionAge + MaxConnectionAgeGrace
+		time.Sleep(helper.RPCMaxConnectionAge + helper.RPCMaxConnectionAgeGrace)
+
+		// check if stream has closed by server (EOF)
+		_, err = watchResp.Recv()
+		assert.Equal(t, codes.Unavailable, status.Code(err))
+		assert.Contains(t, err.Error(), "EOF")
+	})
 }
 
 func TestConfig_Validate(t *testing.T) {
@@ -633,9 +680,23 @@ func TestConfig_Validate(t *testing.T) {
 		{config: &rpc.Config{Port: 11101, CertFile: "noSuchCertFile"}, expected: rpc.ErrInvalidCertFile},
 		{config: &rpc.Config{Port: 11101, KeyFile: "noSuchKeyFile"}, expected: rpc.ErrInvalidKeyFile},
 		// not to use tls
-		{config: &rpc.Config{Port: 11101, CertFile: "", KeyFile: ""}, expected: nil},
+		{config: &rpc.Config{
+			Port:                  11101,
+			CertFile:              "",
+			KeyFile:               "",
+			MaxConnectionAge:      "50s",
+			MaxConnectionAgeGrace: "10s",
+		},
+			expected: nil},
 		// pass any file existing
-		{config: &rpc.Config{Port: 11101, CertFile: "server_test.go", KeyFile: "server_test.go"}, expected: nil},
+		{config: &rpc.Config{
+			Port:                  11101,
+			CertFile:              "server_test.go",
+			KeyFile:               "server_test.go",
+			MaxConnectionAge:      "50s",
+			MaxConnectionAgeGrace: "10s",
+		},
+			expected: nil},
 	}
 	for _, scenario := range scenarios {
 		assert.ErrorIs(t, scenario.config.Validate(), scenario.expected, "provided config: %#v", scenario.config)
diff --git a/test/helper/helper.go b/test/helper/helper.go
index 8e8b4fe8..e8af6d5d 100644
--- a/test/helper/helper.go
+++ b/test/helper/helper.go
@@ -45,8 +45,10 @@ var testStartedAt int64
 
 // Below are the values of the Yorkie config used in the test.
 var (
-	RPCPort            = 21101
-	RPCMaxRequestBytes = uint64(4 * 1024 * 1024)
+	RPCPort                  = 21101
+	RPCMaxRequestBytes       = uint64(4 * 1024 * 1024)
+	RPCMaxConnectionAge      = 4 * gotime.Second
+	RPCMaxConnectionAgeGrace = 1 * gotime.Second
 
 	ProfilingPort = 21102
 
@@ -54,7 +56,7 @@ var (
 
 	AdminUser                             = server.DefaultAdminUser
 	AdminPassword                         = server.DefaultAdminPassword
-	HousekeepingInterval                  = 1 * gotime.Second
+	HousekeepingInterval                  = 10 * gotime.Second
 	HousekeepingCandidatesLimitPerProject = 10
 
 	ClientDeactivateThreshold  = "10s"
@@ -113,8 +115,10 @@ func TestConfig() *server.Config {
 	portOffset += 100
 	return &server.Config{
 		RPC: &rpc.Config{
-			Port:            RPCPort + portOffset,
-			MaxRequestBytes: RPCMaxRequestBytes,
+			Port:                  RPCPort + portOffset,
+			MaxRequestBytes:       RPCMaxRequestBytes,
+			MaxConnectionAge:      RPCMaxConnectionAge.String(),
+			MaxConnectionAgeGrace: RPCMaxConnectionAgeGrace.String(),
 		},
 		Profiling: &profiling.Config{
 			Port: ProfilingPort + portOffset,