[raft] Form the Raft groups through the admin API

Dockerfile

@@ -1,5 +1,5 @@
 # This Dockerfile builds the InterUSS `dss` image which contains the binary
-# executables for the core-service and the db-manager. It also
+# executables for the core-service, the db-manager and dssctl. It also
 # contains a light weight tool that provides debugging capability. To run a
 # container for this image, the desired binary must be specified (either
 # /usr/bin/core-service or /usr/bin/db-manager).
@@ -31,6 +31,7 @@ FROM alpine:latest
 RUN apk update && apk add ca-certificates curl
 COPY --from=build /go/bin/core-service /usr/bin
 COPY --from=build /go/bin/db-manager /usr/bin
+COPY --from=build /go/bin/dssctl /usr/bin
 COPY --from=build /go/bin/dlv /usr/bin
 COPY build/jwt-public-certs /jwt-public-certs
 COPY build/test-certs /test-certs

build/dev/ci_infra/docker-compose.yaml

@@ -29,8 +29,7 @@ services:
       INTRA_USS_NETEM_CONF: ${INTRA_USS_NETEM_CONF-}
       INTER_USS_NETEM_CONF: ${INTER_USS_NETEM_CONF-}
       CORE_SERVICE_EXTRA_FLAGS: ${CORE_SERVICE_EXTRA_FLAGS-}
-      RAFT_ID: ${RAFT_ID:-1}
-      RAFT_NODES: ${RAFT_NODES-}
+      RAFT_URL: http://${DSS_IP:?}:95${PADDED_NODE_IDX:?}
     healthcheck:
       test: wget -O - 'http://localhost/healthy' || exit 1
       start_period: 120s

build/dev/ci_infra/run_locally.sh

@@ -56,17 +56,6 @@ if [[ "$DC_COMMAND" == up* ]]; then
 fi
 
 if [[ "$DC_COMMAND" == up* ]]; then
-  RAFT_NODES=""
-  for ((i=1; i<=NUM_USS; i++)); do
-    for ((j=1; j<=NUM_NODES; j++)); do
-      NODE_IDX=$(( (i-1) * NUM_NODES + j ))
-      PADDED_NODE_IDX=$(printf "%02d" "$NODE_IDX")
-      NODE_IP="172.27.${i}.${j}"
-      RAFT_NODES="${RAFT_NODES},${NODE_IDX}=http://${NODE_IP}:95${PADDED_NODE_IDX}"
-    done
-  done
-  export RAFT_NODES=${RAFT_NODES#,}
-
   echo "=== Local DSS pool configuration ==="
   echo "NUM_USS=${NUM_USS}"
   echo "NUM_NODES=${NUM_NODES}"
@@ -82,7 +71,6 @@ if [[ "$DC_COMMAND" == up* ]]; then
       export USS_IDX=$i
       export USS_NODE_IDX=$j
       NODE_IDX=$(( (i-1) * NUM_NODES + j ))
-      export RAFT_ID=$NODE_IDX
       PADDED_NODE_IDX=$(printf "%02d" "$NODE_IDX")
       export PADDED_NODE_IDX
 
@@ -155,6 +143,44 @@ if [[ "$DC_COMMAND" == up* ]]; then
   check_and_connect "ci_infra_1-1-oauth-1" "interop_ecosystem_network"
   echo "Network verification complete."
 
+  # Form the Raft groups: the first node bootstraps them, then the other nodes are added to them one at a time.
+  # This is done before waiting for the containers to be healthy since a node is only healthy once it is ready in all
+  # of its Raft groups.
+  first_node="ci_infra_1-1-dss-1"
+  for ((i=1; i<=NUM_USS; i++)); do
+    for ((j=1; j<=NUM_NODES; j++)); do
+      container="ci_infra_${i}-${j}-dss-1"
+      # The admin API is started as soon as the node runs, retry until then.
+      nodes=""
+      for _ in $(seq 1 60); do
+        if nodes=$(docker exec "${container}" dssctl raft node 2>/dev/null); then
+          break
+        fi
+        nodes=""
+        sleep 2
+      done
+      if [[ -z "${nodes}" ]]; then
+        echo "Error: the admin API of ${container} is not available."
+        exit 1
+      fi
+      # The comma-separated Raft groups in which the node is waiting. A node that is not waiting in any group is
+      # already a member, e.g. if the pool was already up.
+      groups=$(echo "${nodes}" | awk '$2 == "waiting" {sub(":", "", $1); print $1}' | paste -s -d, -)
+      if [[ -z "${groups}" ]]; then
+        continue
+      fi
+      if [[ "${container}" == "${first_node}" ]]; then
+        echo "Bootstrapping the ${groups} Raft groups on ${container}..."
+        docker exec "${first_node}" dssctl raft bootstrap --groups "${groups}"
+      else
+        echo "Adding ${container} to the ${groups} Raft groups..."
+        padded_node_idx=$(printf "%02d" "$(( (i-1) * NUM_NODES + j ))")
+        docker exec "${first_node}" dssctl raft add --url "http://172.27.${i}.${j}:95${padded_node_idx}" --groups "${groups}"
+      fi
+    done
+  done
+  echo "Raft groups formed."
+
   echo "Waiting for all containers to become healthy..."
   timeout=240
   interval=5

build/dev/ci_infra/startup/core_service.sh

@@ -39,7 +39,7 @@ if [ -n "$INTRA_USS_NETEM_CONF" ] || [ -n "$INTER_USS_NETEM_CONF" ]; then
   fi
 fi
 
-DATASTORE_CONNECTION="-store_type raft -raft_node_id=${RAFT_ID} -raft_peers=${RAFT_NODES} -raft_datadir /raftdata -raft_insecure"
+DATASTORE_CONNECTION="-store_type raft -raft_url ${RAFT_URL} -raft_datadir /raftdata -raft_insecure"
 
 if [ "$DEBUG_ON" = "1" ]; then
   echo "Debug Mode: on"

build/dev/docker-compose_dss.yaml

@@ -112,8 +112,10 @@ services:
       - $PWD/startup/core_service.sh:/startup/core_service.sh:ro
       - $PWD/startup/coverdata:/startup/coverdata:rw  # we will save coverage info here
       - raftdata:/raftdata
+      - adminsocket:/run/dss
     environment:
       COMPOSE_PROFILES: ${COMPOSE_PROFILES}
+      RAFT_URL: http://local-dss-core-service:9011
       # Note: requires the Dockerfile to have been built with "-cover" in the EXTRA_GO_INSTALL_FLAGS var
       GOCOVERDIR: "/startup/coverdata"
       OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-http://local-dss-jaeger:4317}
@@ -156,9 +158,10 @@ services:
       - $PWD/../test-certs:/var/test-certs:ro
       - $PWD/startup/core_service.sh:/startup/core_service.sh:ro
       - raftdata2:/raftdata
+      - adminsocket2:/run/dss
     environment:
       COMPOSE_PROFILES: ${COMPOSE_PROFILES}
-      RAFT_NODE_ID: 2
+      RAFT_URL: http://local-dss-core-service-2:9011
       OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-http://local-dss-jaeger:4317}
       CORE_SERVICE_EXTRA_FLAGS: ${CORE_SERVICE_EXTRA_FLAGS-}
     command: /startup/core_service.sh 0
@@ -174,9 +177,10 @@ services:
       - $PWD/../test-certs:/var/test-certs:ro
       - $PWD/startup/core_service.sh:/startup/core_service.sh:ro
       - raftdata3:/raftdata
+      - adminsocket3:/run/dss
     environment:
       COMPOSE_PROFILES: ${COMPOSE_PROFILES}
-      RAFT_NODE_ID: 3
+      RAFT_URL: http://local-dss-core-service-3:9011
       OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-http://local-dss-jaeger:4317}
       CORE_SERVICE_EXTRA_FLAGS: ${CORE_SERVICE_EXTRA_FLAGS-}
     command: /startup/core_service.sh 0
@@ -186,6 +190,31 @@ services:
       - dss_sandbox_default_network
     profiles: ["with-raft-cluster"]
 
+  # Forms the Raft groups of the core services, which wait for them on their first start and only become healthy once
+  # they are formed. It shares the directory of the Unix socket on which each core service serves its admin API.
+  local-dss-raft-bootstrapper:
+    image: interuss-local/dss
+    entrypoint: /startup/raft_bootstrapper.sh
+    volumes:
+      - $PWD/startup/raft_bootstrapper.sh:/startup/raft_bootstrapper.sh:ro
+      - adminsocket:/run/dss
+      - adminsocket2:/run/dss2
+      - adminsocket3:/run/dss3
+    environment:
+      COMPOSE_PROFILES: ${COMPOSE_PROFILES}
+    depends_on:
+      local-dss-core-service:
+        condition: service_started
+      local-dss-core-service-2:
+        condition: service_started
+        required: false
+      local-dss-core-service-3:
+        condition: service_started
+        required: false
+    networks:
+      - dss_sandbox_default_network
+    profiles: ["with-raft", "with-raft-cluster"]
+
   local-dss-dummy-oauth:
     build:
       context: ../..
@@ -257,3 +286,6 @@ volumes:
   raftdata:
   raftdata2:
   raftdata3:
+  adminsocket:
+  adminsocket2:
+  adminsocket3:

build/dev/startup/core_service.sh

@@ -12,11 +12,7 @@ if [ "${COMPOSE_PROFILES#*"with-yugabyte"}" != "${COMPOSE_PROFILES}" ]; then
 else
     if [ "${COMPOSE_PROFILES#*"with-raft"}" != "${COMPOSE_PROFILES}" ]; then
       echo "Using raft"
-      RAFT_PEERS="1=http://127.0.0.1:9011"
-      if [ "${COMPOSE_PROFILES#*"with-raft-cluster"}" != "${COMPOSE_PROFILES}" ]; then
-        RAFT_PEERS="1=http://local-dss-core-service:9011,2=http://local-dss-core-service-2:9011,3=http://local-dss-core-service-3:9011"
-      fi
-      DATASTORE_CONNECTION="-store_type raft -raft_node_id=${RAFT_NODE_ID:-1} -raft_peers=${RAFT_PEERS} -raft_datadir /raftdata -raft_insecure"
+      DATASTORE_CONNECTION="-store_type raft -raft_url ${RAFT_URL} -raft_datadir /raftdata -raft_insecure"
     else
       echo "Using CockroachDB"
       DATASTORE_CONNECTION="-datastore_host local-dss-crdb"

build/dev/startup/raft_bootstrapper.sh

@@ -0,0 +1,52 @@
+#!/bin/sh
+
+# This startup script is meant to be invoked from within a Docker container
+# started by docker-compose_dss.yaml, not on a local system.
+#
+# It forms the Raft groups of the core services: the first one bootstraps them, then, with the with-raft-cluster
+# profile, the others are added to them one at a time.
+
+set -e
+
+# The admin socket of each core service, then the URL at which the others reach it.
+NODES="/run/dss/admin.sock=http://local-dss-core-service:9011"
+if [ "${COMPOSE_PROFILES#*"with-raft-cluster"}" != "${COMPOSE_PROFILES}" ]; then
+  NODES="${NODES} /run/dss2/admin.sock=http://local-dss-core-service-2:9011 /run/dss3/admin.sock=http://local-dss-core-service-3:9011"
+fi
+
+first_socket=""
+for node in ${NODES}; do
+  socket="${node%%=*}"
+  url="${node#*=}"
+  if [ -z "${first_socket}" ]; then
+    first_socket="${socket}"
+  fi
+
+  # The admin API is served as soon as the core service runs, retry until then.
+  nodes=""
+  for _ in $(seq 1 120); do
+    if nodes=$(dssctl raft node --admin_socket "${socket}" 2>/dev/null); then
+      break
+    fi
+    nodes=""
+    sleep 1
+  done
+  if [ -z "${nodes}" ]; then
+    echo "Error: the admin API at ${socket} is not reachable." >&2
+    exit 1
+  fi
+
+  # The groups in which the core service is waiting: it is already a member of the others, e.g. after a restart.
+  groups=$(echo "${nodes}" | awk '$2 == "waiting" {sub(":", "", $1); print $1}' | paste -s -d, -)
+  if [ -z "${groups}" ]; then
+    continue
+  fi
+  if [ "${socket}" = "${first_socket}" ]; then
+    echo "Bootstrapping the ${groups} Raft groups..."
+    dssctl raft bootstrap --admin_socket "${first_socket}" --groups "${groups}"
+  else
+    echo "Adding ${url} to the ${groups} Raft groups..."
+    dssctl raft add --admin_socket "${first_socket}" --url "${url}" --groups "${groups}"
+  fi
+done
+echo "Raft groups formed."

cmds/core-service/main.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"cloud.google.com/go/profiler"
+	adminserver "github.com/interuss/dss/pkg/admin/server"
 	"github.com/interuss/dss/pkg/api"
 	apiauxv1 "github.com/interuss/dss/pkg/api/auxv1"
 	apiridv1 "github.com/interuss/dss/pkg/api/ridv1"
@@ -22,15 +23,19 @@ import (
 	apiversioningv1 "github.com/interuss/dss/pkg/api/versioningv1"
 	"github.com/interuss/dss/pkg/auth"
 	aux "github.com/interuss/dss/pkg/aux_"
+	auxrepos "github.com/interuss/dss/pkg/aux_/repos"
 	auxs "github.com/interuss/dss/pkg/aux_/store"
 	"github.com/interuss/dss/pkg/build"
 	dsserr "github.com/interuss/dss/pkg/errors"
 	"github.com/interuss/dss/pkg/logging"
 	"github.com/interuss/dss/pkg/raftstore"
+	"github.com/interuss/dss/pkg/raftstore/consensus"
+	ridrepos "github.com/interuss/dss/pkg/rid/repos"
 	rid_v1 "github.com/interuss/dss/pkg/rid/server/v1"
 	rid_v2 "github.com/interuss/dss/pkg/rid/server/v2"
 	rids "github.com/interuss/dss/pkg/rid/store"
 	"github.com/interuss/dss/pkg/scd"
+	scdrepos "github.com/interuss/dss/pkg/scd/repos"
 	scds "github.com/interuss/dss/pkg/scd/store"
 	"github.com/interuss/dss/pkg/store/params"
 	"github.com/interuss/dss/pkg/timestamp"
@@ -62,6 +67,7 @@ var (
 	enableMetrics           = flag.Bool("enable_metrics", false, "Enable metric endpoint")
 	enableTracing           = flag.Bool("enable_tracing", false, "Enable tracing")
 	metricsListeningAddress = flag.String("metrics_addr", ":8079", "Address and port that the for the prometheus-compatible metric service binds to and listens on for incoming connections")
+	adminSocket             = flag.String("admin_socket", "/run/dss/admin.sock", "Path of the Unix socket on which the admin API (used by dssctl) is served, when the core-service uses the raft store. Its directory is created if needed.")
 
 	pkFile                  = flag.String("public_key_files", "", "Path to public Keys to use for JWT decoding, separated by commas.")
 	jwksEndpoint            = flag.String("jwks_endpoint", "", "URL pointing to an endpoint serving JWKS")
@@ -410,6 +416,27 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st
 		multiRouter.Routers = append(multiRouter.Routers, &scdV1Router)
 	}
 
+	if params.GetStoreParameters().StoreType == params.RaftStoreType {
+		raftGroups := map[string]*consensus.Consensus{
+			"aux": auxStore.(*raftstore.Store[auxrepos.Repository]).Consensus,
+		}
+		if *enableRID {
+			raftGroups["rid"] = ridStore.(*raftstore.Store[ridrepos.Repository]).Consensus
+		}
+		if *enableSCD {
+			raftGroups["scd"] = scdStore.(*raftstore.Store[scdrepos.Repository]).Consensus
+		}
+		adminServer, err := adminserver.Start(ctx, logger.With(zap.String("component", "admin"), zap.String("socket", *adminSocket)), *adminSocket, *timeout, raftGroups)
+		if err != nil {
+			return stacktrace.Propagate(err, "Failed to start the admin server")
+		}
+		defer func() {
+			if err := adminServer.Shutdown(context.Background()); err != nil {
+				logger.Warn("failed to shut down admin server", zap.Error(err))
+			}
+		}()
+	}
+
 	// the middlewares are wrapped and, therefore, executed in the opposite order
 	handler := healthyEndpointMiddleware(logger, readiness, &multiRouter)
 	handler = authorizer.TokenMiddleware(handler)

cmds/dssctl/main.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	"github.com/interuss/dss/cmds/dssctl/raft"
+	"github.com/spf13/cobra"
+)
+
+var (
+	dssctlCmd = &cobra.Command{
+		Use:   "dssctl",
+		Short: "DSS node admin command line tool",
+		// The error is printed by main.
+		SilenceErrors: true,
+		// We expect most issues to be related to membership validation and, therefore, unrelated to the usage of the command.
+		SilenceUsage: true,
+	}
+)
+
+func init() {
+	dssctlCmd.PersistentFlags().AddGoFlagSet(flag.CommandLine)
+	dssctlCmd.AddCommand(raft.RaftCmd)
+}
+
+func main() {
+	if err := dssctlCmd.Execute(); err != nil {
+		// %#s formats stacktrace errors without the stack.
+		fmt.Fprintf(os.Stderr, "Error: %#s\n", err)
+		os.Exit(1)
+	}
+}

cmds/dssctl/raft/add.go

@@ -0,0 +1,41 @@
+package raft
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+)
+
+var (
+	addCmd = &cobra.Command{
+		Use:   "add",
+		Short: "Add a node to the Raft groups of the DSS node",
+		Long: `Add a node to the Raft groups of the DSS node, all of them unless selected with --groups.
+
+The node to add must be running and waiting, i.e. started with an empty data directory: as soon as it is added, it
+counts in the quorum of the group. The DSS node reads its ID in each group from the node itself, at --url.`,
+		RunE: add,
+	}
+
+	addURL string
+)
+
+func init() {
+	addCmd.Flags().StringVar(&addURL, "url", "", "URL of the node to add, i.e. its --raft_url (required)")
+	cobra.CheckErr(addCmd.MarkFlagRequired("url"))
+}
+
+func add(cmd *cobra.Command, _ []string) error {
+	groups, err := selectedGroups()
+	if err != nil {
+		return err
+	}
+
+	return forEachGroup(groups, func(g namedGroup) (string, error) {
+		node, err := g.AddNode(cmd.Context(), addURL)
+		if err != nil {
+			return "", err
+		}
+		return fmt.Sprintf("added node %s (%s)", node.Id, node.Url), nil
+	})
+}