Graceful shutdown of the coordinator (#320)

* Graceful shutdown of the coordinator

* Add change to CHANGELOG

Author: informalict

CHANGELOG.md

@@ -6,6 +6,7 @@
 - Add tests for 3.8 ArangoDB and remove tests for 3.5.
 - Add Plan support in Query execution.
 - Change Golang version from 1.13.4 to 1.16.6.
+- Add graceful shutdown for the coordinators.
 
 ## [1.1.1](https://github.com/arangodb/go-driver/tree/1.1.1) (2020-11-13)
 - Add Driver V2 in Alpha version

Makefile

@@ -359,6 +359,7 @@ __test_go_test:
 		-e TEST_BACKUP_REMOTE_REPO=$(TEST_BACKUP_REMOTE_REPO) \
 		-e TEST_BACKUP_REMOTE_CONFIG='$(TEST_BACKUP_REMOTE_CONFIG)' \
 		-e TEST_DEBUG='$(TEST_DEBUG)' \
+		-e TEST_ENABLE_SHUTDOWN=$(TEST_ENABLE_SHUTDOWN) \
 		-e GODEBUG=tls13=1 \
 		-e CGO_ENABLED=$(CGO_ENABLED) \
 		-w /usr/code/ \
@@ -380,6 +381,7 @@ __test_v2_go_test:
 		-e TEST_BACKUP_REMOTE_REPO=$(TEST_BACKUP_REMOTE_REPO) \
 		-e TEST_BACKUP_REMOTE_CONFIG='$(TEST_BACKUP_REMOTE_CONFIG)' \
 		-e TEST_DEBUG='$(TEST_DEBUG)' \
+		-e TEST_ENABLE_SHUTDOWN=$(TEST_ENABLE_SHUTDOWN) \
 		-e GODEBUG=tls13=1 \
 		-e CGO_ENABLED=$(CGO_ENABLED) \
 		-w /usr/code/v2/ \

client_server_admin.go

@@ -41,6 +41,12 @@ type ClientServerAdmin interface {
 	// Statistics queries statistics from a specific server
 	Statistics(ctx context.Context) (ServerStatistics, error)
 
+	// ShutdownV2 shuts down a specific coordinator, optionally removing it from the cluster with a graceful manner.
+	ShutdownV2(ctx context.Context, removeFromCluster, graceful bool) error
+
+	// ShutdownInfoV2 queries information about shutdown progress.
+	ShutdownInfoV2(ctx context.Context) (ShutdownInfo, error)
+
 	// Logs retrieve logs from server in ArangoDB 3.8.0+ format
 	Logs(ctx context.Context) (ServerLogs, error)
 }

client_server_admin_impl.go

@@ -34,6 +34,28 @@ type serverModeRequest struct {
 	Mode ServerMode `json:"mode"`
 }
 
+// ShutdownInfo stores information about shutdown of the coordinator.
+type ShutdownInfo struct {
+	// AQLCursors stores a number of AQL cursors that are still active.
+	AQLCursors int `json:"AQLcursors"`
+	// Transactions stores a number of ongoing transactions.
+	Transactions int `json:"transactions"`
+	// PendingJobs stores a number of ongoing asynchronous requests.
+	PendingJobs int `json:"pendingJobs"`
+	// DoneJobs stores a number of finished asynchronous requests, whose result has not yet been collected.
+	DoneJobs int `json:"doneJobs"`
+	// PregelConductors stores a number of ongoing Pregel jobs.
+	PregelConductors int `json:"pregelConductors"`
+	// LowPrioOngoingRequests stores a number of ongoing low priority requests.
+	LowPrioOngoingRequests int `json:"lowPrioOngoingRequests"`
+	// LowPrioQueuedRequests stores a number of queued low priority requests.
+	LowPrioQueuedRequests int `json:"lowPrioQueuedRequests"`
+	// AllClear is set if all operations are closed.
+	AllClear bool `json:"allClear"`
+	// SoftShutdownOngoing describes whether a soft shutdown of the Coordinator is in progress.
+	SoftShutdownOngoing bool `json:"softShutdownOngoing"`
+}
+
 // ServerMode returns the current mode in which the server/cluster is operating.
 // This call needs ArangoDB 3.3 and up.
 func (c *client) ServerMode(ctx context.Context) (ServerMode, error) {
@@ -138,3 +160,48 @@ func (c *client) Statistics(ctx context.Context) (ServerStatistics, error) {
 	}
 	return data, nil
 }
+
+// ShutdownV2 shuts down a specific coordinator, optionally removing it from the cluster with a graceful manner.
+// When `graceful` is true then run soft shutdown process and the `ShutdownInfoV2` can be used to check the progress.
+// It is available since versions: v3.7.12, v3.8.1, v3.9.0.
+func (c *client) ShutdownV2(ctx context.Context, removeFromCluster, graceful bool) error {
+	req, err := c.conn.NewRequest("DELETE", "_admin/shutdown")
+	if err != nil {
+		return WithStack(err)
+	}
+	if removeFromCluster {
+		req.SetQuery("remove_from_cluster", "1")
+	}
+	if graceful {
+		req.SetQuery("soft", "true")
+	}
+	resp, err := c.conn.Do(ctx, req)
+	if err != nil {
+		return WithStack(err)
+	}
+	if err := resp.CheckStatus(200); err != nil {
+		return WithStack(err)
+	}
+	return nil
+}
+
+// ShutdownInfoV2 returns information about shutdown progress.
+// It is available since versions: v3.7.12, v3.8.1, v3.9.0.
+func (c *client) ShutdownInfoV2(ctx context.Context) (ShutdownInfo, error) {
+	req, err := c.conn.NewRequest("GET", "_admin/shutdown")
+	if err != nil {
+		return ShutdownInfo{}, WithStack(err)
+	}
+	resp, err := c.conn.Do(ctx, req)
+	if err != nil {
+		return ShutdownInfo{}, WithStack(err)
+	}
+	if err := resp.CheckStatus(200); err != nil {
+		return ShutdownInfo{}, WithStack(err)
+	}
+	data := ShutdownInfo{}
+	if err := resp.ParseBody("", &data); err != nil {
+		return ShutdownInfo{}, WithStack(err)
+	}
+	return data, nil
+}

test/server_shutdown_test.go

@@ -0,0 +1,92 @@
+//
+// DISCLAIMER
+//
+// Copyright 2021 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Tomasz Mielech
+//
+
+package test
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestServerShutdown tests the graceful shutdown on the coordinator.
+func TestServerShutdown(t *testing.T) {
+	enabled := os.Getenv("TEST_ENABLE_SHUTDOWN")
+
+	if enabled != "on" && enabled != "1" {
+		t.Skipf("TEST_ENABLE_SHUTDOWN is not set")
+	}
+
+	c := createClientFromEnv(t, true)
+	ctx := context.Background()
+	testing.Short()
+
+	// It must be a cluster
+	versionCheck := EnsureVersion(t, ctx, c).Cluster()
+
+	// Check required version.
+	if !isGracefulShutdownAvailable(versionCheck) {
+		t.Skipf("Skipping because version %s is not sufficient", versionCheck.version)
+	}
+
+	// Shutdown the coordinator.
+	err := c.ShutdownV2(ctx, false, true)
+	require.NoError(t, err, "can not shutdown the coordinator")
+
+	// Wait one minute for the coordinator shutdown.
+	ctxTimeout, _ := context.WithTimeout(ctx, time.Minute)
+	for {
+		info, err := c.ShutdownInfoV2(ctxTimeout)
+		require.NoError(t, err, "can not fetch shutdown progress information")
+		if info.AllClear {
+			break
+		}
+		require.NoError(t, ctx.Err(), "shutdown coordinator timeout")
+	}
+}
+
+// isGracefulShutdownAvailable returns true since versions: v3.7.12, v3.8.1, v3.9.0.
+func isGracefulShutdownAvailable(versionCheck VersionCheck) bool {
+	if versionCheck.version.Major() > 3 {
+		return true
+	}
+
+	minor := versionCheck.version.Minor()
+	if minor < 7 {
+		return false
+	}
+
+	if minor == 7 {
+		if versionCheck.version.CompareTo("3.7.12") < 0 {
+			return false
+		}
+	} else if minor == 8 {
+		if versionCheck.version.CompareTo("3.8.1") < 0 {
+			return false
+		}
+	}
+
+	return true
+}