Skip to content

Commit

Permalink
server,sql: implement connection_wait for graceful draining
Browse files Browse the repository at this point in the history
This commit is to add a phase to current draining process.
At this phase, the server waits for SQL connections to be closed.
New SQL connections are not allowed now. Once all SQL connections are closed,
the server proceeds to draining the range leases. The maximum duration of this
phase is determined by the cluster setting `server.shutdown.connection_wait`

The duration can be set similarly to the other 3 existing draining phases:
```
SET CLUSTER SETTING server.shutdown.connection_wait = '40s'
```

Resolves cockroachdb#66319

Release note (ops change):  add `server.shutdown.connection_wait` to the
draining process configuration. This provides a workaround when customers
encountered intermittent blips and failed requests when they were performing
operations that are related to restarting nodes.

Release justification: Low risk, high benefit changes to existing functionality
(optimize the node draining process).
  • Loading branch information
ZhouXing19 committed Mar 1, 2022
1 parent 12c968f commit be3412f
Show file tree
Hide file tree
Showing 7 changed files with 439 additions and 60 deletions.
3 changes: 2 additions & 1 deletion docs/generated/settings/settings-for-tenants.txt
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,8 @@ server.oidc_authentication.provider_url string sets OIDC provider URL ({provide
server.oidc_authentication.redirect_url string https://localhost:8080/oidc/v1/callback sets OIDC redirect URL via a URL string or a JSON string containing a required `redirect_urls` key with an object that maps from region keys to URL strings (URLs should point to your load balancer and must route to the path /oidc/v1/callback)
server.oidc_authentication.scopes string openid sets OIDC scopes to include with authentication request (space delimited list of strings, required to start with `openid`)
server.rangelog.ttl duration 720h0m0s if nonzero, range log entries older than this duration are deleted every 10m0s. Should not be lowered below 24 hours.
server.shutdown.drain_wait duration 0s the amount of time a server waits in an unready state before proceeding with a drain (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)
server.shutdown.connection_wait duration 0s the maximum amount of time a server waits for all SQL connections to be closed before proceeding with a drain. When all SQL connections are closed before times out, the server early exits and proceeds to draining range leases. (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)
server.shutdown.drain_wait duration 0s the amount of time a server waits in an unready state before proceeding with a drain (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting. --drain-wait is to specify the duration of the whole draining process, while server.shutdown.drain_wait is to set the wait time for health probes to notice that the node is not ready.)
server.shutdown.lease_transfer_wait duration 5s the timeout for a single iteration of the range lease transfer phase of draining (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)
server.shutdown.query_wait duration 10s the timeout for waiting for active queries to finish during a drain (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)
server.time_until_store_dead duration 5m0s the time after which if there is no new gossiped information about a store, it is considered dead
Expand Down
3 changes: 2 additions & 1 deletion docs/generated/settings/settings.html
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,8 @@
<tr><td><code>server.oidc_authentication.redirect_url</code></td><td>string</td><td><code>https://localhost:8080/oidc/v1/callback</code></td><td>sets OIDC redirect URL via a URL string or a JSON string containing a required `redirect_urls` key with an object that maps from region keys to URL strings (URLs should point to your load balancer and must route to the path /oidc/v1/callback) </td></tr>
<tr><td><code>server.oidc_authentication.scopes</code></td><td>string</td><td><code>openid</code></td><td>sets OIDC scopes to include with authentication request (space delimited list of strings, required to start with `openid`)</td></tr>
<tr><td><code>server.rangelog.ttl</code></td><td>duration</td><td><code>720h0m0s</code></td><td>if nonzero, range log entries older than this duration are deleted every 10m0s. Should not be lowered below 24 hours.</td></tr>
<tr><td><code>server.shutdown.drain_wait</code></td><td>duration</td><td><code>0s</code></td><td>the amount of time a server waits in an unready state before proceeding with a drain (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)</td></tr>
<tr><td><code>server.shutdown.connection_wait</code></td><td>duration</td><td><code>0s</code></td><td>the maximum amount of time a server waits for all SQL connections to be closed before proceeding with a drain. When all SQL connections are closed before times out, the server early exits and proceeds to draining range leases. (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)</td></tr>
<tr><td><code>server.shutdown.drain_wait</code></td><td>duration</td><td><code>0s</code></td><td>the amount of time a server waits in an unready state before proceeding with a drain (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting. --drain-wait is to specify the duration of the whole draining process, while server.shutdown.drain_wait is to set the wait time for health probes to notice that the node is not ready.)</td></tr>
<tr><td><code>server.shutdown.lease_transfer_wait</code></td><td>duration</td><td><code>5s</code></td><td>the timeout for a single iteration of the range lease transfer phase of draining (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)</td></tr>
<tr><td><code>server.shutdown.query_wait</code></td><td>duration</td><td><code>10s</code></td><td>the timeout for waiting for active queries to finish during a drain (note that the --drain-wait parameter for cockroach node drain may need adjustment after changing this setting)</td></tr>
<tr><td><code>server.time_until_store_dead</code></td><td>duration</td><td><code>5m0s</code></td><td>the time after which if there is no new gossiped information about a store, it is considered dead</td></tr>
Expand Down
1 change: 1 addition & 0 deletions pkg/cmd/roachtest/tests/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ go_library(
"disk_stall.go",
"django.go",
"django_blocklist.go",
"drain.go",
"drop.go",
"drt.go",
"encryption.go",
Expand Down
236 changes: 236 additions & 0 deletions pkg/cmd/roachtest/tests/drain.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,236 @@
// Copyright 2022 The Cockroach Authors.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.txt.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0, included in the file
// licenses/APL.txt.

package tests

import (
"context"
gosql "database/sql"
"fmt"
"math/rand"
"path/filepath"
"time"

"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/cluster"
"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/option"
"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/registry"
"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
"github.com/cockroachdb/cockroach/pkg/roachprod/install"
"github.com/cockroachdb/cockroach/pkg/util/timeutil"
"github.com/cockroachdb/errors"
"github.com/stretchr/testify/require"
)

func registerDrain(r registry.Registry) {
{
r.Add(registry.TestSpec{
Name: "drain/conn-wait",
Owner: registry.OwnerSQLExperience,
Cluster: r.MakeClusterSpec(1),
Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
runConnectionWait(ctx, t, c)
},
},
)
}
}

func runConnectionWait(ctx context.Context, t test.Test, c cluster.Cluster) {
var err error

err = c.PutE(ctx, t.L(), t.Cockroach(), "./cockroach", c.All())
require.NoError(t, err, "cannot mount cockroach binary")

// Verify that draining proceeds immediately after connections are closed client-side.
{
const (
// Set the duration of each phase of the draining period.
drainWaitDuration = 10 * time.Second
connectionWaitDuration = 100 * time.Second
queryWaitDuration = 10 * time.Second
// pokeDuringConnWaitTimestamp is the timestamp after the server
// starts waiting for SQL connections to close (with the start of the whole
// draining process marked as timestamp 0). It should be set larger than
// drainWaitDuration, but smaller than (drainWaitDuration +
// connectionWaitDuration).
pokeDuringConnWaitTimestamp = 45 * time.Second
connMaxLifetime = 30 * time.Second
connMaxCount = 5
nodeToDrain = 1
)
totalWaitDuration := drainWaitDuration + connectionWaitDuration + queryWaitDuration

prepareCluster(ctx, t, c, drainWaitDuration, connectionWaitDuration, queryWaitDuration)

db := c.Conn(ctx, t.L(), nodeToDrain)
defer db.Close()

db.SetConnMaxLifetime(connMaxLifetime)
db.SetMaxOpenConns(connMaxCount)

var conns []*gosql.Conn

// Get two connections from the connection pools.
for j := 0; j < 2; j++ {
conn, err := db.Conn(ctx)

require.NoError(t, err, "failed to a SQL connection from the connection pool")

conns = append(conns, conn)
}

// Start draining the node.
m := c.NewMonitor(ctx, c.Node(nodeToDrain))

m.Go(func(ctx context.Context) error {
t.Status(fmt.Sprintf("start draining node %d", nodeToDrain))
return c.RunE(ctx,
c.Node(nodeToDrain),
fmt.Sprintf("./cockroach node drain --insecure --drain-wait=%fs",
totalWaitDuration.Seconds()))
})

drainStartTimestamp := timeutil.Now()

// Sleep till the server is in the status of waiting for users to close SQL
// connections. Verify that the server is rejecting new SQL connections now.
time.Sleep(pokeDuringConnWaitTimestamp)
_, err = db.Conn(ctx)
if err != nil {
t.Status(fmt.Sprintf("%s after draining starts, server is rejecting "+
"new SQL connections: %v", pokeDuringConnWaitTimestamp, err))
} else {
t.Fatal(errors.New("new SQL connections should not be allowed when the server " +
"starts waiting for the user to close SQL connections"))
}

require.Equalf(t, db.Stats().OpenConnections, 2, "number of open connections should be 2")

t.Status("number of open connections: ", db.Stats().OpenConnections)

randConn := conns[rand.Intn(len(conns))]

// When server is waiting clients to close connections, verify that SQL
// queries do not fail.
_, err = randConn.ExecContext(ctx, "SELECT 1;")

require.NoError(t, err, "expected query not to fail before the "+
"server starts draining SQL connections")

for _, conn := range conns {
err := conn.Close()
require.NoError(t, err,
"expected connection to be able to be successfully closed client-side")
}

t.Status("all SQL connections are put back to the connection pool")

err = m.WaitE()
require.NoError(t, err, "error waiting for the draining to finish")

drainEndTimestamp := timeutil.Now()
actualDrainDuration := drainEndTimestamp.Sub(drainStartTimestamp).Seconds()

t.L().Printf("the draining lasted %f seconds", actualDrainDuration)

if actualDrainDuration >= float64(totalWaitDuration)-10 {
t.Fatal(errors.New("the draining process didn't early exit " +
"when waiting for server to close all SQL connections"))
}

// Fully quit the draining node so that we can restart it for the next test.
quitNode(ctx, t, c, nodeToDrain)
}

// Verify a warning exists in the case that connectionWait expires.
{
const (
// Set the duration of the draining period.
drainWaitDuration = 0 * time.Second
connectionWaitDuration = 10 * time.Second
queryWaitDuration = 20 * time.Second
nodeToDrain = 1
)

totalWaitDuration := drainWaitDuration + connectionWaitDuration + queryWaitDuration

prepareCluster(ctx, t, c, drainWaitDuration, connectionWaitDuration, queryWaitDuration)

db := c.Conn(ctx, t.L(), nodeToDrain)
defer db.Close()

// Get a connection from the connection pool.
_, err = db.Conn(ctx)

require.NoError(t, err, "cannot get a SQL connection from the connection pool")

m := c.NewMonitor(ctx, c.Node(nodeToDrain))
m.Go(func(ctx context.Context) error {
t.Status(fmt.Sprintf("draining node %d", nodeToDrain))
return c.RunE(ctx,
c.Node(nodeToDrain),
fmt.Sprintf("./cockroach node drain --insecure --drain-wait=%fs",
totalWaitDuration.Seconds()))
})

err = m.WaitE()
require.NoError(t, err, "error waiting for the draining to finish")

logFile := filepath.Join("logs", "*.log")
err = c.RunE(ctx, c.Node(nodeToDrain),
"grep", "-q", "'proceeding to drain SQL connections'", logFile)
require.NoError(t, err, "warning is not logged in the log file")
}

}

// prepareCluster is to start the server on nodes in the given cluster, and set
// the cluster setting for duration of each phase of the draining process.
func prepareCluster(
ctx context.Context,
t test.Test,
c cluster.Cluster,
drainWait time.Duration,
connectionWait time.Duration,
queryWait time.Duration,
) {

c.Start(ctx, t.L(), option.DefaultStartOpts(), install.MakeClusterSettings(), c.All())

db := c.Conn(ctx, t.L(), 1)
defer db.Close()

_, err := db.ExecContext(ctx, `
SET CLUSTER SETTING server.shutdown.drain_wait = $1;
SET CLUSTER SETTING server.shutdown.connection_wait = $2;
SET CLUSTER SETTING server.shutdown.query_wait = $3;`,
drainWait.Seconds(),
connectionWait.Seconds(),
queryWait.Seconds(),
)
require.NoError(t, err)

}

func quitNode(ctx context.Context, t test.Test, c cluster.Cluster, node int) {
args := append([]string{
"./cockroach", "quit", "--insecure", "--logtostderr=INFO",
fmt.Sprintf("--port={pgport:%d}", node)})
result, err := c.RunWithDetailsSingleNode(ctx, t.L(), c.Node(node), args...)
output := result.Stdout + result.Stderr
t.L().Printf("cockroach quit:\n%s\n", output)
require.NoError(t, err, "cannot quit cockroach")

stopOpts := option.DefaultStopOpts()
stopOpts.RoachprodOpts.Sig = 0
stopOpts.RoachprodOpts.Wait = true
c.Stop(ctx, t.L(), stopOpts, c.All())
t.L().Printf("stopped cluster")
}
1 change: 1 addition & 0 deletions pkg/cmd/roachtest/tests/registry.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ func RegisterTests(r registry.Registry) {
registerDiskFull(r)
RegisterDiskStalledDetection(r)
registerDjango(r)
registerDrain(r)
registerDrop(r)
registerEncryption(r)
registerEngineSwitch(r)
Expand Down
60 changes: 55 additions & 5 deletions pkg/server/drain.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ package server

import (
"context"
"fmt"
"io"
"strings"
"time"
Expand All @@ -32,21 +33,38 @@ import (

var (
queryWait = settings.RegisterDurationSetting(
settings.TenantWritable,
settings.TenantReadOnly,
"server.shutdown.query_wait",
"the timeout for waiting for active queries to finish during a drain "+
"(note that the --drain-wait parameter for cockroach node drain may need adjustment "+
"after changing this setting)",
10*time.Second,
settings.NonNegativeDurationWithMaximum(10*time.Hour),
).WithPublic()

drainWait = settings.RegisterDurationSetting(
settings.TenantWritable,
settings.TenantReadOnly,
"server.shutdown.drain_wait",
"the amount of time a server waits in an unready state before proceeding with a drain "+
"(note that the --drain-wait parameter for cockroach node drain may need adjustment "+
"after changing this setting. --drain-wait is to specify the duration of the "+
"whole draining process, while server.shutdown.drain_wait is to set the "+
"wait time for health probes to notice that the node is not ready.)",
0*time.Second,
settings.NonNegativeDurationWithMaximum(10*time.Hour),
).WithPublic()

connectionWait = settings.RegisterDurationSetting(
settings.TenantReadOnly,
"server.shutdown.connection_wait",
"the maximum amount of time a server waits for all SQL connections to "+
"be closed before proceeding with a drain. "+
"When all SQL connections are closed before times out, the server early "+
"exits and proceeds to draining range leases. "+
"(note that the --drain-wait parameter for cockroach node drain may need adjustment "+
"after changing this setting)",
0*time.Second,
settings.NonNegativeDurationWithMaximum(10*time.Hour),
).WithPublic()
)

Expand Down Expand Up @@ -309,19 +327,33 @@ func (s *drainServer) drainClients(
s.grpc.setMode(modeDraining)
s.sqlServer.isReady.Set(false)

// Log the number of connections periodically.
if err := s.logOpenConns(ctx); err != nil {
log.Ops.Warningf(ctx, "error showing alive SQL connections: %v", err)
}

// Wait the duration of drainWait.
// This will fail load balancer checks and delay draining so that client
// traffic can move off this node.
// Note delay only happens on first call to drain.
if shouldDelayDraining {
log.Ops.Info(ctx, "waiting for health probes to notice that the node "+
"is not ready for new sql connections...")
s.drainSleepFn(drainWait.Get(&s.sqlServer.execCfg.Settings.SV))
}

// Drain all SQL connections.
// The queryWait duration is a timeout for waiting on clients
// to self-disconnect. If the timeout is reached, any remaining connections
// Wait for users to close the existing SQL connections.
// During this phase, the server is rejecting new SQL connections.
// The server exit this phase either once all SQL connections are closed,
// or it reaches the connectionMaxWait timeout, whichever earlier.
s.sqlServer.pgServer.WaitForSQLConnsToClose(ctx, connectionWait.Get(&s.sqlServer.execCfg.Settings.SV))

// Drain any remaining SQL connections.
// The queryWait duration is a timeout for waiting for SQL queries to finish.
// If the timeout is reached, any remaining connections
// will be closed.
queryMaxWait := queryWait.Get(&s.sqlServer.execCfg.Settings.SV)

if err := s.sqlServer.pgServer.Drain(ctx, queryMaxWait, reporter); err != nil {
return err
}
Expand Down Expand Up @@ -357,3 +389,21 @@ func (s *drainServer) drainNode(
// Mark the stores of the node as "draining" and drain all range leases.
return s.kvServer.node.SetDraining(true /* drain */, reporter, verbose)
}

// logOpenConns logs the number of open SQL connections every 3 seconds.
func (s *drainServer) logOpenConns(ctx context.Context) error {
return s.stopper.RunAsyncTask(ctx, "log-open-conns", func(ctx context.Context) {
ticker := time.NewTicker(3 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
log.Ops.Info(ctx, fmt.Sprintf("number of open connections: %d\n", s.sqlServer.pgServer.GetConnCancelMapLen()))
case <-s.stopper.ShouldQuiesce():
return
case <-ctx.Done():
return
}
}
})
}
Loading

0 comments on commit be3412f

Please sign in to comment.