diff --git a/Dockerfile b/Dockerfile index dc43d6810..c490b1165 100644 --- a/Dockerfile +++ b/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 diff --git a/build/dev/ci_infra/docker-compose.yaml b/build/dev/ci_infra/docker-compose.yaml index 403c45c05..aa73b93da 100644 --- a/build/dev/ci_infra/docker-compose.yaml +++ b/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 diff --git a/build/dev/ci_infra/run_locally.sh b/build/dev/ci_infra/run_locally.sh index c1c102753..ed9860c0d 100755 --- a/build/dev/ci_infra/run_locally.sh +++ b/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 diff --git a/build/dev/ci_infra/startup/core_service.sh b/build/dev/ci_infra/startup/core_service.sh index d778e657d..f86abd4a5 100755 --- a/build/dev/ci_infra/startup/core_service.sh +++ b/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" diff --git a/build/dev/docker-compose_dss.yaml b/build/dev/docker-compose_dss.yaml index 29c0fa623..527ffe8a2 100644 --- a/build/dev/docker-compose_dss.yaml +++ b/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: diff --git a/build/dev/startup/core_service.sh b/build/dev/startup/core_service.sh index 736011856..75037cc67 100755 --- a/build/dev/startup/core_service.sh +++ b/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" diff --git a/build/dev/startup/raft_bootstrapper.sh b/build/dev/startup/raft_bootstrapper.sh new file mode 100755 index 000000000..e573360ea --- /dev/null +++ b/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." diff --git a/cmds/core-service/main.go b/cmds/core-service/main.go index 7e6661f1b..9e06df0ac 100644 --- a/cmds/core-service/main.go +++ b/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) diff --git a/cmds/dssctl/main.go b/cmds/dssctl/main.go new file mode 100644 index 000000000..e826c5416 --- /dev/null +++ b/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) + } +} diff --git a/cmds/dssctl/raft/add.go b/cmds/dssctl/raft/add.go new file mode 100644 index 000000000..b99a8a30a --- /dev/null +++ b/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 + }) +} diff --git a/cmds/dssctl/raft/bootstrap.go b/cmds/dssctl/raft/bootstrap.go new file mode 100644 index 000000000..03f8ddc26 --- /dev/null +++ b/cmds/dssctl/raft/bootstrap.go @@ -0,0 +1,33 @@ +package raft + +import ( + "fmt" + + "github.com/spf13/cobra" +) + +var bootstrapCmd = &cobra.Command{ + Use: "bootstrap", + Short: "Bootstrap new Raft groups with the DSS node as their only member", + Long: `Bootstrap new Raft groups with the DSS node as their only member, all of them unless selected with --groups. + +The DSS node must be waiting, i.e. started with an empty data directory. The other nodes are then added to the groups +with dssctl raft add.`, + RunE: bootstrap, +} + +func bootstrap(cmd *cobra.Command, _ []string) error { + groups, err := selectedGroups() + if err != nil { + return err + } + + return forEachGroup(groups, func(g namedGroup) (string, error) { + membership, err := g.Bootstrap(cmd.Context()) + if err != nil { + return "", err + } + // Bootstrap returns once the DSS node is the leader of the new group. + return fmt.Sprintf("bootstrapped with node %s as leader", membership.LeaderId), nil + }) +} diff --git a/cmds/dssctl/raft/list.go b/cmds/dssctl/raft/list.go new file mode 100644 index 000000000..1686de552 --- /dev/null +++ b/cmds/dssctl/raft/list.go @@ -0,0 +1,81 @@ +package raft + +import ( + "cmp" + "fmt" + "slices" + "strings" + "text/tabwriter" + + restapi "github.com/interuss/dss/pkg/api/adminv1" + "github.com/spf13/cobra" +) + +var listCmd = &cobra.Command{ + Use: "list", + Short: "List the members of the Raft groups of the DSS node", + Long: "List the members of the Raft groups of the DSS node, all of them unless selected with --groups, as known by the node serving each admin API.", + RunE: list, +} + +func list(cmd *cobra.Command, _ []string) error { + groups, err := selectedGroups() + if err != nil { + return err + } + + return forEachGroup(groups, func(g namedGroup) (string, error) { + members, err := g.Members(cmd.Context()) + if err != nil { + return "", err + } + + return formatMembers(members) + }) +} + +// formatMembers formats m as a table of the members, followed by the removed IDs. +func formatMembers(m *restapi.RaftGroup) (string, error) { + var builder strings.Builder + builder.WriteString("\n") + if m.LeaderId == "" { + builder.WriteString(" no leader known\n") + } + + table := tabwriter.NewWriter(&builder, 0, 0, 2, ' ', 0) + _, err := fmt.Fprintln(table, " ID\tURL\tROLE") + if err != nil { + return "", err + } + var members []restapi.RaftNode + var removed []string + for _, node := range m.Nodes { + if node.State == stateRemoved { + removed = append(removed, node.Id) + continue + } + members = append(members, node) + } + // The URLs are sorted rather than the IDs, which differ between groups, so that the groups list the nodes in the same order. + slices.SortFunc(members, func(a, b restapi.RaftNode) int { return cmp.Compare(a.Url, b.Url) }) + for _, node := range members { + _, err := fmt.Fprintf(table, " %s\t%s\t%s\n", node.Id, node.Url, memberRole(m, node.Id)) + if err != nil { + return "", err + } + } + _ = table.Flush() + + if len(removed) > 0 { + fmt.Fprintf(&builder, " removed IDs: %v\n", removed) + } + return strings.TrimSuffix(builder.String(), "\n"), nil +} + +func memberRole(m *restapi.RaftGroup, id string) string { + if id == m.LeaderId { + return "leader" + } + // all the members of a Raft group are voters + return "voter" +} diff --git a/cmds/dssctl/raft/node.go b/cmds/dssctl/raft/node.go new file mode 100644 index 000000000..1680519e6 --- /dev/null +++ b/cmds/dssctl/raft/node.go @@ -0,0 +1,30 @@ +package raft + +import ( + "fmt" + + "github.com/spf13/cobra" +) + +var nodeCmd = &cobra.Command{ + Use: "node", + Short: "Describe the DSS node in its Raft groups", + Long: `Describe the DSS node in its Raft groups, all of them unless selected with --groups: its state (waiting, member or +removed), ID and URL in each group.`, + RunE: node, +} + +func node(cmd *cobra.Command, _ []string) error { + groups, err := selectedGroups() + if err != nil { + return err + } + + return forEachGroup(groups, func(g namedGroup) (string, error) { + n, err := g.Node(cmd.Context()) + if err != nil { + return "", err + } + return fmt.Sprintf("%s (ID %s, %s)", n.State, n.Id, n.Url), nil + }) +} diff --git a/cmds/dssctl/raft/raft.go b/cmds/dssctl/raft/raft.go new file mode 100644 index 000000000..55b975c34 --- /dev/null +++ b/cmds/dssctl/raft/raft.go @@ -0,0 +1,116 @@ +package raft + +import ( + "errors" + "fmt" + "slices" + "strings" + "sync" + "time" + + "github.com/interuss/dss/pkg/admin/client" + "github.com/interuss/stacktrace" + "github.com/spf13/cobra" +) + +// groupNames are the Raft groups of a DSS node, in the order in which they are processed. +var groupNames = []string{"aux", "rid", "scd"} + +// The states of a node in a Raft group, as returned by the admin API. +const ( + stateMember = "member" + stateRemoved = "removed" +) + +var ( + RaftCmd = &cobra.Command{ + Use: "raft", + Short: "Manage the membership of the Raft groups (aux, rid, scd) of a DSS node", + Long: `Manage the membership of the Raft groups (aux, rid, scd) of a DSS node, through its admin API. + +A change is requested to each selected group independently: it may succeed for some groups and fail for others, +in which case it can be retried for the failed groups only with --groups.`, + } + + groups []string + adminSocket string + adminTimeout time.Duration +) + +func init() { + RaftCmd.PersistentFlags().StringSliceVar(&groups, "groups", groupNames, "Comma-separated Raft groups to act on, all of them if not set") + RaftCmd.PersistentFlags().StringVar(&adminSocket, "admin_socket", "/run/dss/admin.sock", "Path of the Unix socket of the admin API of the DSS node, matching its --admin_socket") + RaftCmd.PersistentFlags().DurationVar(&adminTimeout, "admin_timeout", time.Minute, "Timeout of each request to the admin API") + RaftCmd.AddCommand(bootstrapCmd, addCmd, removeCmd, listCmd, nodeCmd) +} + +// namedGroup is a Raft group of the DSS node, managed through its admin API. +type namedGroup struct { + name string + *client.Group +} + +// selectedGroups returns the Raft groups selected with --groups, in processing order. +func selectedGroups() ([]namedGroup, error) { + for _, name := range groups { + if !slices.Contains(groupNames, name) { + return nil, stacktrace.NewError("unknown Raft group %q in --groups, expected some of %s", name, strings.Join(groupNames, ",")) + } + } + + var result []namedGroup + for _, name := range groupNames { + if !slices.Contains(groups, name) { + continue + } + group, err := client.NewRaftGroup(adminSocket, name, adminTimeout) + if err != nil { + return nil, stacktrace.Propagate(err, "invalid --admin_socket") + } + result = append(result, namedGroup{name: name, Group: group}) + } + if len(result) == 0 { + return nil, stacktrace.NewError("no Raft group selected with --groups") + } + return result, nil +} + +// forEachGroup calls fn for every group concurrently +// then prints the results of the groups that succeeded +// and returns the errors of the failed ones. +func forEachGroup(groups []namedGroup, fn func(g namedGroup) (string, error)) error { + type result struct { + msg string + err error + } + + results := make([]result, len(groups)) + + var wg sync.WaitGroup + for i, g := range groups { + wg.Go(func() { + msg, err := fn(g) + results[i] = result{msg: msg, err: err} + }) + } + wg.Wait() + + var failures []string + for i, g := range groups { + if err := results[i].err; err != nil { + // %#s formats stacktrace errors on a single line, without the stack. + failures = append(failures, fmt.Sprintf("%s: %#s", g.name, err)) + continue + } + fmt.Printf("%s: %s\n", g.name, results[i].msg) + } + + switch len(failures) { + case 0: + return nil + case 1: + return errors.New(failures[0]) + default: + return fmt.Errorf("%d Raft groups failed:\n %s", len(failures), strings.Join(failures, "\n ")) + } +} diff --git a/cmds/dssctl/raft/remove.go b/cmds/dssctl/raft/remove.go new file mode 100644 index 000000000..7b7bbb53b --- /dev/null +++ b/cmds/dssctl/raft/remove.go @@ -0,0 +1,75 @@ +package raft + +import ( + "context" + "fmt" + + "github.com/interuss/stacktrace" + "github.com/spf13/cobra" +) + +var ( + removeCmd = &cobra.Command{ + Use: "remove", + Short: "Remove a node from the Raft groups of the DSS node", + Long: `Remove a node from the Raft groups of the DSS node, all of them unless selected with --groups. + +A node has a different ID in each group: it is removed by --url from all the selected groups, or by --node_id from a +single group selected with --groups. + +The ID of a removed node is retired permanently: it can never be added again to the same group. +A voter cannot be removed if the group would be left with fewer than 3 voters.`, + RunE: remove, + } + + removeURL string + removeNodeID string +) + +func init() { + removeCmd.Flags().StringVar(&removeURL, "url", "", "URL of the node to remove, i.e. its --raft_url") + removeCmd.Flags().StringVar(&removeNodeID, "node_id", "", "ID of the node to remove, only when a single group is selected with --groups") + removeCmd.MarkFlagsMutuallyExclusive("url", "node_id") + removeCmd.MarkFlagsOneRequired("url", "node_id") +} + +func remove(cmd *cobra.Command, _ []string) error { + groups, err := selectedGroups() + if err != nil { + return err + } + if removeNodeID != "" && len(groups) != 1 { + return stacktrace.NewError("--node_id identifies a node in a single group, select it with --groups") + } + + return forEachGroup(groups, func(g namedGroup) (string, error) { + nodeID := removeNodeID + if removeURL != "" { + var err error + nodeID, err = memberID(cmd.Context(), g, removeURL) + if err != nil { + return "", err + } + } + + if err := g.RemoveNode(cmd.Context(), nodeID); err != nil { + return "", err + } + return fmt.Sprintf("removed node %s", nodeID), nil + }) +} + +// memberID returns the ID of the member of g at url. +func memberID(ctx context.Context, g namedGroup, url string) (string, error) { + membership, err := g.Members(ctx) + if err != nil { + return "", err + } + + for _, node := range membership.Nodes { + if node.State == stateMember && node.Url == url { + return node.Id, nil + } + } + return "", stacktrace.NewError("no member has the URL %q", url) +} diff --git a/pkg/admin/client/client.go b/pkg/admin/client/client.go new file mode 100644 index 000000000..ee912f5ca --- /dev/null +++ b/pkg/admin/client/client.go @@ -0,0 +1,154 @@ +package client + +import ( + "context" + "net" + "net/http" + "time" + + restapi "github.com/interuss/dss/pkg/api/adminv1" + dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/stacktrace" +) + +// statusCodes maps the HTTP status of the error responses of the admin API to the code of the error of the DSS node. +var statusCodes = map[int]stacktrace.ErrorCode{ + http.StatusBadRequest: dsserr.BadRequest, + http.StatusNotFound: dsserr.NotFound, + http.StatusConflict: dsserr.AlreadyExists, + http.StatusServiceUnavailable: dsserr.Unavailable, +} + +type Group struct { + client *restapi.Client + socketPath string + name restapi.RaftGroupName +} + +func NewRaftGroup(socketPath string, name string, timeout time.Duration) (*Group, error) { + if socketPath == "" { + return nil, stacktrace.NewError("the path of the Unix socket of the admin API is required") + } + + // The requests are sent over the Unix socket whatever their URL, whose host is only used in the Host header. + transport := &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "unix", socketPath) + }, + } + return &Group{ + client: &restapi.Client{BaseURL: "http://localhost", HTTPClient: &http.Client{Transport: transport, Timeout: timeout}}, + socketPath: socketPath, + name: restapi.RaftGroupName(name), + }, nil +} + +func (g *Group) Node(ctx context.Context) (*restapi.RaftNode, error) { + resp, err := g.client.GetThisRaftNode(ctx, g.name) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) Bootstrap(ctx context.Context) (*restapi.RaftGroup, error) { + resp, err := g.client.CreateRaftGroup(ctx, &restapi.CreateRaftGroupParameters{GroupName: g.name}) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response400 != nil: + return nil, responseError(http.StatusBadRequest, resp.Response400.Message) + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response409 != nil: + return nil, responseError(http.StatusConflict, resp.Response409.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) Members(ctx context.Context) (*restapi.RaftGroup, error) { + resp, err := g.client.GetRaftGroup(ctx, g.name) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) AddNode(ctx context.Context, peerURL string) (*restapi.RaftNode, error) { + resp, err := g.client.CreateRaftGroupMember(ctx, g.name, &restapi.CreateRaftGroupMemberParameters{Url: peerURL}) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response200 != nil: + return resp.Response200, nil + case resp.Response400 != nil: + return nil, responseError(http.StatusBadRequest, resp.Response400.Message) + case resp.Response404 != nil: + return nil, responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response409 != nil: + return nil, responseError(http.StatusConflict, resp.Response409.Message) + case resp.Response503 != nil: + return nil, responseError(http.StatusServiceUnavailable, resp.Response503.Message) + case resp.Response500 != nil: + return nil, responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return nil, stacktrace.NewError("no response received from the admin API") + } +} + +func (g *Group) RemoveNode(ctx context.Context, nodeID string) error { + resp, err := g.client.DeleteRaftGroupMember(ctx, g.name, nodeID) + if err != nil { + return stacktrace.Propagate(err, "failed to call the admin API at %s", g.socketPath) + } + switch { + case resp.Response204 != nil: + return nil + case resp.Response400 != nil: + return responseError(http.StatusBadRequest, resp.Response400.Message) + case resp.Response404 != nil: + return responseError(http.StatusNotFound, resp.Response404.Message) + case resp.Response500 != nil: + return responseError(http.StatusInternalServerError, &resp.Response500.ErrorMessage) + default: + return stacktrace.NewError("no response received from the admin API") + } +} + +// responseError returns the error described by an error response of the admin API, with the code of statusCodes. +func responseError(status int, message *string) error { + text := http.StatusText(status) + if message != nil && *message != "" { + text += ": " + *message + } + if code, ok := statusCodes[status]; ok { + return stacktrace.NewErrorWithCode(code, "%d %s", status, text) + } + return stacktrace.NewError("%d %s", status, text) +} diff --git a/pkg/admin/server/raft_members_handler.go b/pkg/admin/server/raft_members_handler.go new file mode 100644 index 000000000..62b807c5b --- /dev/null +++ b/pkg/admin/server/raft_members_handler.go @@ -0,0 +1,186 @@ +package server + +import ( + "context" + "maps" + "slices" + "strconv" + + "github.com/interuss/dss/pkg/api" + restapi "github.com/interuss/dss/pkg/api/adminv1" + dsserr "github.com/interuss/dss/pkg/errors" + "github.com/interuss/dss/pkg/raftstore/consensus" + "github.com/interuss/dss/pkg/raftstore/consensus/node" + "github.com/interuss/stacktrace" + "go.uber.org/zap" +) + +func (s *Server) GetThisRaftNode(ctx context.Context, req *restapi.GetThisRaftNodeRequest) restapi.GetThisRaftNodeResponseSet { + group, err := s.raftGroup(req.GroupName) + if err != nil { + return restapi.GetThisRaftNodeResponseSet{Response404: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, err)}} + } + + node, err := group.Node() + if err != nil { + return restapi.GetThisRaftNodeResponseSet{Response500: &api.InternalServerErrorBody{ + ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Could not describe this node in the %s raft group", req.GroupName))}} + } + return restapi.GetThisRaftNodeResponseSet{Response200: toRaftNode(node)} +} + +func (s *Server) CreateRaftGroup(ctx context.Context, req *restapi.CreateRaftGroupRequest) restapi.CreateRaftGroupResponseSet { + if req.BodyParseError != nil { + return restapi.CreateRaftGroupResponseSet{Response400: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(req.BodyParseError, dsserr.BadRequest, "Malformed params"))}} + } + group, err := s.raftGroup(req.Body.GroupName) + if err != nil { + return restapi.CreateRaftGroupResponseSet{Response404: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, err)}} + } + + membership, err := group.Bootstrap(ctx) + if err != nil { + err = stacktrace.Propagate(err, "Could not create the %s raft group", req.Body.GroupName) + switch stacktrace.GetCode(err) { + case dsserr.AlreadyExists: + return restapi.CreateRaftGroupResponseSet{Response409: &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}} + default: + return restapi.CreateRaftGroupResponseSet{Response500: &api.InternalServerErrorBody{ + ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}} + } + } + + s.Logger.Info("created raft group", zap.String("group", string(req.Body.GroupName))) + return restapi.CreateRaftGroupResponseSet{Response200: toRaftGroup(req.Body.GroupName, membership)} +} + +func (s *Server) GetRaftGroup(ctx context.Context, req *restapi.GetRaftGroupRequest) restapi.GetRaftGroupResponseSet { + group, err := s.raftGroup(req.GroupName) + if err != nil { + return restapi.GetRaftGroupResponseSet{Response404: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, err)}} + } + + membership, err := group.Members() + if err != nil { + err = stacktrace.Propagate(err, "Could not get the members of the %s raft group", req.GroupName) + switch stacktrace.GetCode(err) { + // NotFound: this node is still waiting to create or join the group. + case dsserr.NotFound: + return restapi.GetRaftGroupResponseSet{Response404: &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)}} + default: + return restapi.GetRaftGroupResponseSet{Response500: &api.InternalServerErrorBody{ + ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}} + } + } + return restapi.GetRaftGroupResponseSet{Response200: toRaftGroup(req.GroupName, membership)} +} + +func (s *Server) CreateRaftGroupMember(ctx context.Context, req *restapi.CreateRaftGroupMemberRequest) restapi.CreateRaftGroupMemberResponseSet { + group, err := s.raftGroup(req.GroupName) + if err != nil { + return restapi.CreateRaftGroupMemberResponseSet{Response404: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, err)}} + } + if req.BodyParseError != nil { + return restapi.CreateRaftGroupMemberResponseSet{Response400: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, stacktrace.PropagateWithCode(req.BodyParseError, dsserr.BadRequest, "Malformed params"))}} + } + + added, err := group.AddNode(ctx, req.Body.Url) + if err != nil { + err = stacktrace.Propagate(err, "Could not add node %s to the %s raft group", req.Body.Url, req.GroupName) + errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} + switch stacktrace.GetCode(err) { + case dsserr.BadRequest: + return restapi.CreateRaftGroupMemberResponseSet{Response400: errResp} + case dsserr.AlreadyExists: + return restapi.CreateRaftGroupMemberResponseSet{Response409: errResp} + // Unavailable: the node cannot be reached, or this node is not a member of the group. + case dsserr.Unavailable: + return restapi.CreateRaftGroupMemberResponseSet{Response503: errResp} + default: + return restapi.CreateRaftGroupMemberResponseSet{Response500: &api.InternalServerErrorBody{ + ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}} + } + } + + s.Logger.Info("added node to raft group", zap.String("group", string(req.GroupName)), zap.Uint64("nodeID", added.ID), zap.String("url", added.URL)) + return restapi.CreateRaftGroupMemberResponseSet{Response200: toRaftNode(added)} +} + +func (s *Server) DeleteRaftGroupMember(ctx context.Context, req *restapi.DeleteRaftGroupMemberRequest) restapi.DeleteRaftGroupMemberResponseSet { + group, err := s.raftGroup(req.GroupName) + if err != nil { + return restapi.DeleteRaftGroupMemberResponseSet{Response404: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, err)}} + } + nodeID, err := parseNodeID(req.NodeId) + if err != nil { + return restapi.DeleteRaftGroupMemberResponseSet{Response400: &restapi.ErrorResponse{ + Message: dsserr.Handle(ctx, err)}} + } + + err = group.RemoveNode(ctx, nodeID) + if err != nil { + err = stacktrace.Propagate(err, "Could not remove node %d from the %s raft group", nodeID, req.GroupName) + errResp := &restapi.ErrorResponse{Message: dsserr.Handle(ctx, err)} + switch stacktrace.GetCode(err) { + case dsserr.BadRequest, dsserr.Unavailable: + return restapi.DeleteRaftGroupMemberResponseSet{Response400: errResp} + default: + return restapi.DeleteRaftGroupMemberResponseSet{Response500: &api.InternalServerErrorBody{ + ErrorMessage: *dsserr.Handle(ctx, stacktrace.Propagate(err, "Got an unexpected error"))}} + } + } + + s.Logger.Info("removed node from raft group", zap.String("group", string(req.GroupName)), zap.Uint64("nodeID", nodeID)) + return restapi.DeleteRaftGroupMemberResponseSet{Response204: &api.EmptyResponseBody{}} +} + +// raftGroup returns the Raft group or a dsserr.NotFound error. +func (s *Server) raftGroup(name restapi.RaftGroupName) (*consensus.Consensus, error) { + group, ok := s.raftGroups[string(name)] + if !ok { + return nil, stacktrace.NewErrorWithCode(dsserr.NotFound, "Unknown raft group %q, expected one of %v", name, slices.Sorted(maps.Keys(s.raftGroups))) + } + return group, nil +} + +// parseNodeID parses a node ID formatted as a decimal string, or returns a dsserr.BadRequest error. +func parseNodeID(s string) (uint64, error) { + id, err := strconv.ParseUint(s, 10, 64) + if err != nil || id == 0 { + return 0, stacktrace.NewErrorWithCode(dsserr.BadRequest, "Invalid node ID %q, expected a non-zero decimal integer", s) + } + return id, nil +} + +func toRaftNode(info node.Info) *restapi.RaftNode { + return &restapi.RaftNode{Id: strconv.FormatUint(info.ID, 10), Url: info.URL, State: string(info.State)} +} + +// toRaftGroup returns the group with its nodes: its members first, then the nodes removed from it, each by ID. +func toRaftGroup(name restapi.RaftGroupName, membership consensus.Membership) *restapi.RaftGroup { + result := &restapi.RaftGroup{ + GroupName: name, + ClusterId: strconv.FormatUint(membership.ClusterID, 10), + Nodes: []restapi.RaftNode{}, + } + // The leader is unknown while an election is in progress. + if membership.LeaderID != 0 { + result.LeaderId = strconv.FormatUint(membership.LeaderID, 10) + } + ids := slices.Sorted(maps.Keys(membership.Nodes)) + for _, state := range []node.State{node.Member, node.Removed} { + for _, id := range ids { + if info := membership.Nodes[id]; info.State == state { + result.Nodes = append(result.Nodes, *toRaftNode(info)) + } + } + } + return result +} diff --git a/pkg/admin/server/server.go b/pkg/admin/server/server.go new file mode 100644 index 000000000..ceacee907 --- /dev/null +++ b/pkg/admin/server/server.go @@ -0,0 +1,13 @@ +package server + +import ( + "github.com/interuss/dss/pkg/raftstore/consensus" + "go.uber.org/zap" +) + +// Server implements adminv1.Implementation. +type Server struct { + Logger *zap.Logger + // raftGroups are the Raft groups of the DSS node, indexed by name. + raftGroups map[string]*consensus.Consensus +} diff --git a/pkg/admin/server/start.go b/pkg/admin/server/start.go new file mode 100644 index 000000000..c400c524f --- /dev/null +++ b/pkg/admin/server/start.go @@ -0,0 +1,59 @@ +package server + +import ( + "context" + "errors" + "io/fs" + "net" + "net/http" + "os" + "path/filepath" + "time" + + "github.com/interuss/dss/pkg/api" + restapi "github.com/interuss/dss/pkg/api/adminv1" + "github.com/interuss/dss/pkg/raftstore/consensus" + "github.com/interuss/stacktrace" + "go.uber.org/zap" +) + +// Start serves the admin API on the Unix socket at socketPath, with requests failing after timeout. +// The directory of the socket is created if needed. A leftover socket at socketPath is replaced. +func Start(ctx context.Context, logger *zap.Logger, socketPath string, timeout time.Duration, raftGroups map[string]*consensus.Consensus) (*http.Server, error) { + socketDir := filepath.Dir(socketPath) + if err := os.MkdirAll(socketDir, 0o700); err != nil { + return nil, stacktrace.Propagate(err, "failed to create the directory of the admin socket at %s", socketDir) + } + + if err := os.Remove(socketPath); err != nil && !errors.Is(err, fs.ErrNotExist) { + return nil, stacktrace.Propagate(err, "failed to remove the previous admin socket at %s", socketPath) + } + var listenConfig net.ListenConfig + listener, err := listenConfig.Listen(ctx, "unix", socketPath) + if err != nil { + return nil, stacktrace.Propagate(err, "failed to listen on the admin socket at %s", socketPath) + } + if err := os.Chmod(socketPath, 0o600); err != nil { + _ = listener.Close() + return nil, stacktrace.Propagate(err, "failed to restrict the permissions of the admin socket at %s", socketPath) + } + + router := restapi.MakeAPIRouter(&Server{Logger: logger, raftGroups: raftGroups}, nil) + handler := &api.MultiRouter{Routers: []api.PartialRouter{&router}} + + server := &http.Server{ + Handler: http.TimeoutHandler(handler, timeout, "request timeout"), + ReadHeaderTimeout: 15 * time.Second, + ReadTimeout: 15 * time.Second, + } + + go func() { + logger.Info("starting admin server") + // The socket file is removed when the server is shut down, which closes the listener. + if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Error("admin server failed", zap.Error(err)) + } + }() + + return server, nil +} diff --git a/pkg/raftstore/README.md b/pkg/raftstore/README.md index 048a0f4e4..8920fe9d0 100644 --- a/pkg/raftstore/README.md +++ b/pkg/raftstore/README.md @@ -5,7 +5,7 @@ ## Key assumptions - **Single Raft group:** Each service (aux, rid, scd) is backed by exactly one Raft group (one [`Consensus`](consensus/consensus.go) instance). There is no sharding or partitioning of data across multiple groups for now. -- **Single listener:** The Raft groups of a DSS node share one `rafthttp` listener ([consensus/listener.go](consensus/listener.go)), at the node's URL in `--raft_peers`. Each group has its own cluster ID, derived from `--raft_cluster_id` and the group name, which is used to route the requests to the right group. The listener is started by the process before any group is created and lives as long as the process, whether or not groups are registered on it. +- **Single listener:** The Raft groups of a DSS node share one `rafthttp` listener ([consensus/listener.go](consensus/listener.go)), on the port of the node's `--raft_url`. Each group has its own cluster ID, derived from `--raft_cluster_id` and the group name, which is used to route the requests to the right group. The listener is started by the process before any group is created and lives as long as the process, whether or not groups are registered on it. - **Persistent log:** The Raft log and snapshots are durable: each node writes its entries to a [write-ahead log (WAL)](https://pkg.go.dev/go.etcd.io/etcd/server/v3/storage/wal) and periodically persists snapshots to disk ([consensus/storage.go](consensus/storage.go)), so a node can restart and rejoin the cluster by replaying its own WAL. The log consists of a series of Raft proposals (see the data model section below), and snapshots are serialized representations of the in-memory store. - **In-memory data projection:** The application state machine derived from the committed log (i.e. ISAs, subscriptions, operational intents, etc.) lives in-memory. It is not persisted but rather reconstructed by replaying the log / snapshot on startup. - **Data model:** Each service (`rid`, `scd`, `aux_`) defines its own request/response structs. During replication, they are wrapped in a [`Proposal`](consensus/proposal.go), which carries metadata (an ID, the originating node, and a timestamp for determinism) with a `RequestType` string and a request payload as a serialized `Value`. [`pkg/raftstore/store.go`](store.go) replicates the `Proposal` through Raft and, once committed, the service specific raftstore implementation interprets `RequestType`/`Value` and applies the resulting change to its own in-memory store. @@ -23,7 +23,7 @@ Raft nodes authenticate each other with mutual TLS, configured by these three fl - Each participant operates its own CA and issues one certificate per node. The bundle holds the CA of every participant of the pool, including the node's own. - The same certificate is presented to peers and to nodes connecting to this one. - A connecting peer is accepted when its certificate is signed by one of these CAs. The certificate is not matched against the peer's source IP, which differs from the addresses in the certificate behind NAT or a load balancer. Other connections are dropped during the TLS handshake, before any Raft message is read. -- Every peer URL must use `https`. A node with an `http` peer URL, or with only some of the three flags set, does not start. +- `--raft_url` must use `https`, so that the URLs at which the members reach each other do. A node with an `http` `--raft_url`, or with only some of the three flags set, does not start. - A node does not start if its certificate cannot be loaded or has expired, or if every CA of the bundle has expired. Each expired CA is logged as a warning. - Restart the node after replacing its certificate or the CA bundle. - For local development and tests only, you can use `--raft_insecure` to disable TLS. @@ -35,7 +35,7 @@ Most of the concepts defined below come from the Raft consensus algorithm itself ### Actors - **Raft group:** A set of Raft nodes that replicate one Raft log via the Raft consensus protocol (one leader and its followers). -- **Raft Node:** One participant in a Raft group, identified by a unique `NodeID`. One [`Consensus`](consensus/consensus.go) instance wraps one Raft node. +- **Raft Node:** One participant in a Raft group, identified by a unique `NodeID` generated on its first start. One [`Consensus`](consensus/consensus.go) instance wraps one Raft node. - **DSS Node:** One running core-service. Each DSS Node runs 3 Raft nodes (aux, rid, scd). - **Peer:** Another Raft node in the same Raft group, reachable over `rafthttp` at a configured URL. Peers exchange Raft messages (votes, log replication, heartbeats) with each other. - **Client:** An entity sending requests to a DSS node (e.g. via the REST ASTM API). @@ -64,17 +64,20 @@ Most of the concepts defined below come from the Raft consensus algorithm itself ## Membership -The members of a Raft group are bootstrapped from the `--raft_peers` flag on the group's very first start. Afterwards, nodes are added to and removed from a running group one at a time with [`Consensus.AddNode` and `Consensus.RemoveNode`](consensus/consensus.go), which propose a configuration change through the Raft log and block until it is applied. Each Raft group is managed independently: a DSS node is added to or removed from each of its groups with one configuration change per group. +Each Raft group is managed independently: a DSS node bootstraps, joins or leaves each of its groups separately, through the [admin API](../../interfaces/admin/admin.yaml) served by the core-service on the Unix socket at `--admin_socket` using `dssctl raft`, so a change may succeed for some groups and fail for others, in which case it is retried for the failed groups only. Nodes are added to and removed from a running group one at a time. [`Consensus.AddNode` and `Consensus.RemoveNode`](consensus/consensus.go) propose a configuration change through the Raft log and block until it is applied. -- **Joining:** Starting a node that joins a running group is not supported yet, see [#1802](https://github.com/interuss/dss/issues/1802). +- **Identity:** A DSS node is identified by its `--raft_url`, which must be unique within the cluster. On its first start, the node generates a random ID per group and records it in the metadata of the group's WAL. The node keeps this ID across restarts. +- **Startup:** What a node does when it starts only depends on its data directory: a node with Raft state restarts as a member, a node without Raft state waits (`GET /admin/v1/raft/groups/{group_name}/this_node` reports its state) to be added to a running group or to create a new one. A waiting or removed node rejects the requests to the group. +- **Bootstrapping:** `POST /admin/v1/raft/groups` on a waiting node creates a new group with this node as its only member. The other nodes are then added to it. +- **Joining:** A node is started, then added with `POST /admin/v1/raft/groups/{group_name}/members` sent to a member, with the URL of the node. The member reads the ID of the node from the node itself, on its Raft listener, and rejects a node that is unreachable, not waiting, or not running this group in this cluster. The node learns the configuration and the application state of the group from the leader. +- **Replacing:** Replacing a node at the same URL is not supported yet. A node that lost its data directory starts with a new ID, and its old ID still uses its URL. With 4 voters or more, its old ID is removed, then the node is added. A group of 3 voters cannot go below the minimum: the node must be restarted with a different `--raft_url` and added before its old ID is removed. - **Validation:** A configuration change is validated when it is proposed, to fail fast, and again by every node when it is applied. The following changes are rejected: - adding a node that is already a member, with an invalid peer URL, or with the peer URL of another member. - removing a node that is not a member, or a voter if the group would be left with fewer than 3 voters. - adding or removing a node ID that was previously removed from the group. -- **Removal:** A node stops when it applies its own removal. +- **Removal:** A node stops when it applies its own removal, and its state becomes `removed`. - **Persistence:** The peer URLs of the members and of the removed nodes are replicated state: they are rebuilt from the configuration changes of the log and stored in snapshots alongside the application state. ## Future Work -- **Client-initiated config changes are not supported yet:** See the `TODO` in [consensus/consensus.go](consensus/consensus.go). - **No background WAL cleanup:** Old, WAL segment files that are no longer needed are unlocked (`ReleaseLockTo`) but not actively deleted by a cleanup goroutine yet. See the `TODO` in [consensus/storage.go](consensus/storage.go). diff --git a/pkg/raftstore/consensus/consensus.go b/pkg/raftstore/consensus/consensus.go index 109ab2c76..677c34601 100644 --- a/pkg/raftstore/consensus/consensus.go +++ b/pkg/raftstore/consensus/consensus.go @@ -41,7 +41,13 @@ type Consensus struct { nodeID uint64 group string publicURL string - node *node.NodeState + // config is used to start the raft node, either when bootstrapping a group or when joining one. + config *raft.Config + + tickInterval time.Duration + snapshotInterval uint64 + + node *node.NodeState transport *rafthttp.Transport listener *listener @@ -72,42 +78,27 @@ type Consensus struct { } func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params.ConnectParameters, provider func() ([]byte, error), commitC chan<- EntryCommit) (*Consensus, error) { - storage, old, err := newStorage(ctx, logger.With(zap.String("component", "storage")), connectParams.DataDir, connectParams.NodeID, connectParams.SnapshotCatchupEntries) + storage, nodeID, hasState, err := newStorage(ctx, logger.With(zap.String("component", "storage")), connectParams.DataDir, connectParams.SnapshotCatchupEntries) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize storage") } - peers, err := connectParams.PeerMap() + nodeURL, err := url.Parse(connectParams.URL) if err != nil { - return nil, stacktrace.Propagate(err, "failed to parse peer map") - } - - nodeUrl, ok := peers[connectParams.NodeID] - if !ok { - return nil, stacktrace.NewError("node ID %d not found in peers map", connectParams.NodeID) - } - - var raftNode raft.Node - config := connectParams.RaftConfig(storage) - if old { - logger.Info("restarting raft node", zap.String("address", nodeUrl.String())) - raftNode = raft.RestartNode(config) - } else { - logger.Info("starting new raft node", zap.String("address", nodeUrl.String())) - raftPeers, err := peersList(peers) - if err != nil { - return nil, stacktrace.Propagate(err, "failed to build the list of bootstrap peers") - } - raftNode = raft.StartNode(config, raftPeers) + return nil, stacktrace.Propagate(err, "failed to parse the URL of the node") } consensus := &Consensus{ - logger: logging.WithValuesFromContext(ctx, logger), + logger: logging.WithValuesFromContext(ctx, logger).With(zap.Uint64("nodeID", nodeID)), + + nodeID: nodeID, + group: connectParams.Group, + publicURL: nodeURL.String(), + config: connectParams.RaftConfig(nodeID, storage), + tickInterval: connectParams.TickInterval, + snapshotInterval: connectParams.SnapshotIntervalEntries, + node: node.NewNodeState(), - nodeID: connectParams.NodeID, - group: connectParams.Group, - publicURL: nodeUrl.String(), - node: node.NewNodeState(raftNode), storage: storage, getAppSnapshot: provider, commitC: commitC, @@ -119,13 +110,13 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. readTracker: newReadIndexTracker(), membership: membership{ - nodeID: connectParams.NodeID, + nodeID: nodeID, members: map[uint64]member{}, removed: map[uint64]member{}, }, } - err = consensus.initTransport(ctx, connectParams.NodeID, connectParams.GroupClusterID, peers, NewTLSInfo(connectParams, logger)) + err = consensus.initTransport(ctx, nodeID, connectParams.GroupClusterID, nodeURL, NewTLSInfo(connectParams, logger)) if err != nil { return nil, stacktrace.Propagate(err, "failed to initialize transport") } @@ -148,10 +139,6 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. if err != nil { return nil, stacktrace.Propagate(err, "failed to restore membership") } - // A removal in the log is recorded when it is applied again, but not one that is already in the snapshot. - if consensus.membership.isIDRemoved(connectParams.NodeID) { - consensus.node.SetRemoved() - } consensus.commitC <- EntryCommit{SnapshotData: initSnapshot.AppData} } @@ -160,11 +147,30 @@ func NewConsensus(ctx context.Context, logger *zap.Logger, connectParams params. return nil, stacktrace.Propagate(err, "failed to register raft metrics") } - consensus.startRaftUpdatesConsumer(connectParams.TickInterval, connectParams.SnapshotIntervalEntries) + switch { + case consensus.membership.isIDRemoved(nodeID): + consensus.logger.Warn("this node was removed from the Raft group") + consensus.node.SetRemoved() + case hasState: + consensus.logger.Info("restarting raft node", zap.String("address", nodeURL.String())) + consensus.startNode(func() raft.Node { return raft.RestartNode(consensus.config) }) + default: + consensus.logger.Info("waiting to be added to a Raft group or to bootstrap one", zap.String("address", nodeURL.String())) + } return consensus, nil } +// startNode makes this node a member of the group with the raft node returned by newNode and starts the consumer of +// its updates, if this node is waiting. It returns whether the raft node was started, and the state of this node. +func (c *Consensus) startNode(newNode func() raft.Node) (bool, node.State) { + started, state := c.node.Start(newNode) + if started { + c.startRaftUpdatesConsumer(c.tickInterval, c.snapshotInterval) + } + return started, state +} + // TODO: ctx is currently ignored (see issue: https://github.com/interuss/dss/issues/1610) func (c *Consensus) Stop(ctx context.Context) { // TODO: remove once (see issue: https://github.com/interuss/dss/issues/1610) @@ -181,9 +187,11 @@ func (c *Consensus) Stop(ctx context.Context) { c.transport.Stop() c.logger.Info("transport stopped") c.endpoint.close() - raftNode := c.node.RaftNode() - raftNode.Stop() - c.logger.Info("raft node stopped") + // A waiting node has no raft node. + if raftNode := c.node.RaftNode(); raftNode != nil { + raftNode.Stop() + c.logger.Info("raft node stopped") + } }) } @@ -321,31 +329,12 @@ type member struct { URL string `json:"url,omitempty"` } -func peersList(peers map[uint64]*url.URL) ([]raft.Peer, error) { - result := make([]raft.Peer, 0, len(peers)) - for id, peerURL := range peers { - peerCtx, err := json.Marshal(member{URL: peerURL.String()}) - if err != nil { - return nil, stacktrace.Propagate(err, "failed to marshal context of peer %d", id) - } - - // We pass member as the Context so that we can recover the peer URL when applying a config change. - result = append(result, raft.Peer{ID: id, Context: peerCtx}) - } - return result, nil -} - -func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, peers map[uint64]*url.URL, tlsInfo etcdtransport.TLSInfo) error { +func (c *Consensus) initTransport(ctx context.Context, nodeID uint64, clusterID uint64, nodeURL *url.URL, tlsInfo etcdtransport.TLSInfo) error { nodeIDStr := fmt.Sprintf("%d", nodeID) if tlsInfo.Empty() { c.logger.Warn("--raft_insecure is set: Raft nodes do not authenticate each other and connection is not encrypted. Don't use this node in production environments.") } - nodeURL, ok := peers[nodeID] - if !ok { - return stacktrace.NewError("node ID %d not found in peers map", nodeID) - } - transport := &rafthttp.Transport{ Logger: logging.WithValuesFromContext(ctx, c.logger.With(zap.String("component", "transport"))), ID: types.ID(nodeID), @@ -455,7 +444,12 @@ func (c *Consensus) registerMetrics() error { } c.metricsRegistration, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error { - status := c.node.RaftNode().Status() + raftNode := c.node.RaftNode() + if raftNode == nil { + // A waiting node, or one found removed at startup, has no raft node to report on. + return nil + } + status := raftNode.Status() attrs := metric.WithAttributes(c.metricAttributes()...) o.ObserveInt64(gauges["raft_term"], int64(status.GetTerm()), attrs) @@ -792,7 +786,12 @@ func (c *Consensus) RemoveNode(ctx context.Context, nodeID uint64) error { } // Members returns the membership of the Raft group as currently known by this node. +// It returns a dsserr.NotFound error while this node is waiting to bootstrap or join the group. func (c *Consensus) Members() (Membership, error) { + if c.node.Current() == node.Waiting { + return Membership{}, stacktrace.NewErrorWithCode(dsserr.NotFound, "this node is waiting to bootstrap or join the Raft group") + } + c.membership.RLock() nodes := make(map[uint64]node.Info, len(c.membership.members)+len(c.membership.removed)) for id, m := range c.membership.members { @@ -803,9 +802,15 @@ func (c *Consensus) Members() (Membership, error) { } c.membership.RUnlock() + // A node found removed at startup has no raft node, so it knows no leader. + var leaderID uint64 + if raftNode := c.node.RaftNode(); raftNode != nil { + leaderID = raftNode.Status().Lead + } + return Membership{ ClusterID: uint64(c.transport.ClusterID), - LeaderID: c.node.RaftNode().Status().Lead, + LeaderID: leaderID, Nodes: nodes, }, nil } @@ -818,11 +823,40 @@ func (c *Consensus) Node() (node.Info, error) { }, nil } -// Bootstrap starts a new Raft group with this node as its only member. -// TODO: The groups are still formed from --raft_peers: this node is a member of its group from its first start, or -// was removed from it, so a dsserr.AlreadyExists error is always returned. -func (c *Consensus) Bootstrap() (Membership, error) { - return Membership{}, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "the Raft group is formed from --raft_peers, the state of this node is %s", c.node.Current()) +// Bootstrap starts a new Raft group with this node as its only member, and blocks until this node is its leader or +// ctx is cancelled. Only a waiting node can bootstrap a group, otherwise a dsserr.AlreadyExists error is returned. +func (c *Consensus) Bootstrap(ctx context.Context) (Membership, error) { + selfCtx, err := json.Marshal(member{URL: c.publicURL}) + if err != nil { + return Membership{}, stacktrace.Propagate(err, "failed to marshal the context of this node") + } + + // We pass member as the Context so that every node records the URL of this node when applying the config change. + started, state := c.startNode(func() raft.Node { + return raft.StartNode(c.config, []raft.Peer{{ID: c.nodeID, Context: selfCtx}}) + }) + if !started { + return Membership{}, stacktrace.NewErrorWithCode(dsserr.AlreadyExists, "only a waiting node can bootstrap a Raft group, the state of this node is %s", state) + } + c.logger.Info("bootstrapping a new Raft group") + + ticker := time.NewTicker(c.tickInterval) + defer ticker.Stop() + for { + membership, err := c.Members() + if err != nil { + return Membership{}, err + } + if membership.LeaderID == c.nodeID && len(membership.Nodes) > 0 { + return membership, nil + } + + select { + case <-ticker.C: + case <-ctx.Done(): + return Membership{}, ctx.Err() + } + } } // Ready returns whether this node can serve requests, i.e. it is a member of the Raft group, knows its leader and the @@ -1002,10 +1036,34 @@ func decodeSnapshot(data []byte) (snapshot, error) { // Process implements the rafthttp.Raft interface. func (c *Consensus) Process(ctx context.Context, m *raftpb.Message) error { - raftNode := c.node.RaftNode() + if m.GetTo() != c.nodeID { + return stacktrace.NewError("received a message for node %d, this node is %d", m.GetTo(), c.nodeID) + } + raftNode, err := c.nodeForMessage(m.GetFrom()) + if err != nil { + return err + } return raftNode.Step(ctx, m) } +// nodeForMessage returns the raft node that processes a message received from a peer. +// A waiting node only receives messages from the leader of a group it was added to: it then starts its raft node to +// join the group, and learns the configuration and the application state of the group from the leader. +func (c *Consensus) nodeForMessage(from uint64) (raft.Node, error) { + raftNode, state := c.node.Get() + if state == node.Waiting { + if started, _ := c.startNode(func() raft.Node { return raft.RestartNode(c.config) }); started { + c.logger.Info("joining the Raft group", zap.Uint64("from", from)) + } + // The raft node was started either above or concurrently, by another message or by Bootstrap. + raftNode, state = c.node.Get() + } + if state == node.Removed { + return nil, stacktrace.NewError("this node was removed from the Raft group") + } + return raftNode, nil +} + // IsIDRemoved implements the rafthttp.Raft interface. func (c *Consensus) IsIDRemoved(id uint64) bool { return c.membership.isIDRemoved(id) diff --git a/pkg/raftstore/consensus/node/node.go b/pkg/raftstore/consensus/node/node.go index 8dcbb9052..d2f1a4842 100644 --- a/pkg/raftstore/consensus/node/node.go +++ b/pkg/raftstore/consensus/node/node.go @@ -30,42 +30,56 @@ type Info struct { type NodeState struct { mu sync.RWMutex state State - node raft.Node + // node is nil until this node is a member of the group. Once set, it never changes. + node raft.Node } -// NewNodeState returns the NodeState of a node that is a member of the group. -func NewNodeState(node raft.Node) *NodeState { - return &NodeState{state: Member, node: node} +// NewNodeState returns the NodeState of a node that is waiting to bootstrap or join the group. +func NewNodeState() *NodeState { + return &NodeState{state: Waiting} } -// get returns the raft node and the state of this node. -func (t *NodeState) get() (raft.Node, State) { +// Get returns the raft node, nil until this node is a member of the group, and the state of this node. +func (t *NodeState) Get() (raft.Node, State) { t.mu.RLock() defer t.mu.RUnlock() return t.node, t.state } -// RaftNode returns the raft node. +// RaftNode returns the raft node, nil until this node is a member of the group. func (t *NodeState) RaftNode() raft.Node { - node, _ := t.get() + node, _ := t.Get() return node } // Current returns the state of this node. func (t *NodeState) Current() State { - _, state := t.get() + _, state := t.Get() return state } // Member returns the raft node, or a dsserr.Unavailable error if this node is not a member of the group. func (t *NodeState) Member() (raft.Node, error) { - node, state := t.get() + node, state := t.Get() if state != Member { return nil, stacktrace.NewErrorWithCode(dsserr.Unavailable, "this node is not a member of the Raft group, its state is %s", state) } return node, nil } +// Start makes this node a member of the group with the raft node returned by newNode, if this node is waiting. +// It returns whether the raft node was started, and the state of this node. +func (t *NodeState) Start(newNode func() raft.Node) (bool, State) { + t.mu.Lock() + defer t.mu.Unlock() + if t.state != Waiting { + return false, t.state + } + t.node = newNode() + t.state = Member + return true, t.state +} + // SetRemoved records that this node was removed from the group. func (t *NodeState) SetRemoved() { t.mu.Lock() diff --git a/pkg/raftstore/consensus/storage.go b/pkg/raftstore/consensus/storage.go index 698cddb8d..d4036d311 100644 --- a/pkg/raftstore/consensus/storage.go +++ b/pkg/raftstore/consensus/storage.go @@ -2,8 +2,10 @@ package consensus import ( "context" + "crypto/rand" + "encoding/binary" + "encoding/json" "errors" - "fmt" "os" "path" @@ -31,22 +33,29 @@ type storage struct { snapshotCatchUpEntries uint64 } +// identity identifies the node in its Raft group. It is generated and persisted in the metadata of the WAL on the first +// start, so that the node keeps it across restarts. +type identity struct { + NodeID uint64 `json:"node_id"` +} + // newStorage initializes the storage by loading the latest snapshot and wal entries from the disk -// and applies them to the Raft memory storage. -// It returns the initialized storage, a boolean indicating whether the storage was pre-existent or an error. -func newStorage(ctx context.Context, logger *zap.Logger, dataDir string, nodeID uint64, snapshotCatchUpEntries uint64) (*storage, bool, error) { +// and applies them to the Raft memory storage. On the first start, the WAL is created with a newly generated node ID. +// It returns the initialized storage, the ID of the node, a boolean indicating whether the storage holds Raft state or +// an error. +func newStorage(ctx context.Context, logger *zap.Logger, dataDir string, snapshotCatchUpEntries uint64) (*storage, uint64, bool, error) { logger = logging.WithValuesFromContext(ctx, logger) // load the latest snapshot - snapshotPath := path.Join(dataDir, fmt.Sprintf("snapshot_%d", nodeID)) + snapshotPath := path.Join(dataDir, "snap") if !fileutil.Exist(snapshotPath) { err := os.MkdirAll(snapshotPath, 0o750) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to create directory for snapshot storage at: %s", snapshotPath) + return nil, 0, false, stacktrace.Propagate(err, "failed to create directory for snapshot storage at: %s", snapshotPath) } } - walPath := path.Join(dataDir, fmt.Sprintf("wal_%d", nodeID)) + walPath := path.Join(dataDir, "wal") snapper := snap.New(logger, snapshotPath) @@ -55,23 +64,32 @@ func newStorage(ctx context.Context, logger *zap.Logger, dataDir string, nodeID err error snapshot *raftpb.Snapshot ) - ok := wal.Exist(walPath) - if !ok { + if !wal.Exist(walPath) { + nodeID, err := generateNodeID() + if err != nil { + return nil, 0, false, stacktrace.Propagate(err, "failed to generate node ID") + } + metadata, err := json.Marshal(identity{NodeID: nodeID}) + if err != nil { + return nil, 0, false, stacktrace.Propagate(err, "failed to marshal node identity") + } + if err = os.MkdirAll(walPath, 0o750); err != nil { - return nil, false, stacktrace.Propagate(err, "failed to create directory for wal storage at: %s", walPath) + return nil, 0, false, stacktrace.Propagate(err, "failed to create directory for wal storage at: %s", walPath) } - w, err = wal.Create(logger, walPath, nil) + w, err = wal.Create(logger, walPath, metadata) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to create wal at: %s", walPath) + return nil, 0, false, stacktrace.Propagate(err, "failed to create wal at: %s", walPath) } if err = w.Close(); err != nil { - return nil, false, stacktrace.Propagate(err, "failed to close wal at: %s", walPath) + return nil, 0, false, stacktrace.Propagate(err, "failed to close wal at: %s", walPath) } + logger.Info("generated new node ID", zap.Uint64("nodeID", nodeID)) snapshot = &raftpb.Snapshot{} } else { snapshot, err = loadSnapshot(logger, walPath, snapper) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to load snapshot") + return nil, 0, false, stacktrace.Propagate(err, "failed to load snapshot") } } @@ -82,12 +100,17 @@ func newStorage(ctx context.Context, logger *zap.Logger, dataDir string, nodeID ConfState: snapshot.GetMetadata().GetConfState(), }) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to open wal with index %d, term %d and confstate %v at: %s", snapshot.GetMetadata().GetIndex(), snapshot.GetMetadata().GetTerm(), snapshot.GetMetadata().GetConfState(), walPath) + return nil, 0, false, stacktrace.Propagate(err, "failed to open wal with index %d, term %d and confstate %v at: %s", snapshot.GetMetadata().GetIndex(), snapshot.GetMetadata().GetTerm(), snapshot.GetMetadata().GetConfState(), walPath) } - _, state, entries, err := w.ReadAll() + metadata, state, entries, err := w.ReadAll() if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to read wal state and entries") + return nil, 0, false, stacktrace.Propagate(err, "failed to read wal state and entries") + } + + var id identity + if err := json.Unmarshal(metadata, &id); err != nil { + return nil, 0, false, stacktrace.Propagate(err, "failed to read the node identity from the wal at %s", walPath) } // initialize the raft memory storage with the loaded snapshot and wal entries @@ -95,18 +118,18 @@ func newStorage(ctx context.Context, logger *zap.Logger, dataDir string, nodeID if !raft.IsEmptySnap(snapshot) { err = raftMemoryStorage.ApplySnapshot(snapshot) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to apply snapshot to raft memory storage") + return nil, 0, false, stacktrace.Propagate(err, "failed to apply snapshot to raft memory storage") } } err = raftMemoryStorage.SetHardState(state) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to set hard state to raft memory storage") + return nil, 0, false, stacktrace.Propagate(err, "failed to set hard state to raft memory storage") } err = raftMemoryStorage.Append(entries) if err != nil { - return nil, false, stacktrace.Propagate(err, "failed to append entries to raft memory storage") + return nil, 0, false, stacktrace.Propagate(err, "failed to append entries to raft memory storage") } logger.Info("Loaded previous hardstate and entries to Raft memory storage", zap.Any("hard-state", state), zap.Int("entries-number", len(entries))) @@ -121,7 +144,20 @@ func newStorage(ctx context.Context, logger *zap.Logger, dataDir string, nodeID snapper: snapper, snapshotCatchUpEntries: snapshotCatchUpEntries, - }, ok, nil + }, id.NodeID, !raft.IsEmptySnap(snapshot) || !raft.IsEmptyHardState(state) || len(entries) > 0, nil +} + +// generateNodeID returns a random, non-zero node ID. +func generateNodeID() (uint64, error) { + var b [8]byte + for { + if _, err := rand.Read(b[:]); err != nil { + return 0, stacktrace.Propagate(err, "failed to read random bytes") + } + if id := binary.BigEndian.Uint64(b[:]); id != raft.None { + return id, nil + } + } } func loadSnapshot(logger *zap.Logger, walPath string, snapshotter *snap.Snapshotter) (*raftpb.Snapshot, error) { diff --git a/pkg/raftstore/params/params.go b/pkg/raftstore/params/params.go index 85680bb27..7b2c17464 100644 --- a/pkg/raftstore/params/params.go +++ b/pkg/raftstore/params/params.go @@ -6,11 +6,10 @@ import ( "hash/fnv" "net/url" "path/filepath" - "strconv" - "strings" "time" "github.com/interuss/stacktrace" + "go.etcd.io/etcd/client/pkg/v3/types" "go.etcd.io/raft/v3" ) @@ -36,8 +35,7 @@ const ( type ( // ConnectParameters bundles up parameters used for connecting nodes in a raftstore cluster. ConnectParameters struct { - NodeID uint64 - Peers string + URL string // DataDir is the join between the base data directory provided by the user and a subfolder for the specific group (aux, rid, scd). DataDir string @@ -72,55 +70,21 @@ func (c ConnectParameters) TLSEnabled() bool { return !c.Insecure } -// PeerMap parses the Peers string into a map of node ID to peer URL. -func (c ConnectParameters) PeerMap() (map[uint64]*url.URL, error) { - peers := make(map[uint64]*url.URL) - scheme := "http" +// URLScheme is the scheme of the URLs at which the Raft nodes reach each other: https with mutual TLS, http otherwise. +func (c ConnectParameters) URLScheme() string { if c.TLSEnabled() { - scheme = "https" + return "https" } - - for entry := range strings.SplitSeq(c.Peers, ",") { - parts := strings.SplitN(entry, "=", 2) - if len(parts) != 2 { - return nil, stacktrace.NewError("invalid peer entry %s: must be in format nodeID=peerURL", entry) - } - - id, err := strconv.ParseUint(parts[0], 10, 64) - if err != nil { - return nil, stacktrace.Propagate(err, "invalid peer ID %s", parts[0]) - } - - if id == 0 { - return nil, stacktrace.NewError("invalid peer ID 0: peer IDs must be greater than 0") - } - - if _, exists := peers[id]; exists { - return nil, stacktrace.NewError("duplicate peer ID %d", id) - } - - peerURL, err := url.Parse(parts[1]) - if err != nil { - return nil, stacktrace.Propagate(err, "invalid peer URL %s", parts[1]) - } - - if peerURL.Scheme != scheme { - return nil, stacktrace.NewError("invalid peer URL %s: scheme must be %s (https when mutual TLS is enabled, http otherwise)", parts[1], scheme) - } - - peers[id] = peerURL - } - - return peers, nil + return "http" } func (c ConnectParameters) ElectionInterval() time.Duration { return time.Duration(c.ElectionTick) * c.TickInterval } -func (c ConnectParameters) RaftConfig(storage raft.Storage) *raft.Config { +func (c ConnectParameters) RaftConfig(nodeID uint64, storage raft.Storage) *raft.Config { return &raft.Config{ - ID: c.NodeID, + ID: nodeID, ElectionTick: c.ElectionTick, HeartbeatTick: c.HeartbeatTick, MaxSizePerMsg: c.MaxSizePerMsg, @@ -136,10 +100,9 @@ var ( ) func init() { - flag.Uint64Var(&connectParameters.NodeID, "raft_node_id", 0, "Raft node ID for this instance (must be non-zero and unique within the cluster).") - flag.StringVar(&connectParameters.Peers, "raft_peers", "", `Comma-separated "nodeID=peerURL" pairs defining all cluster members including this node, e.g. "1=http://node1:9011,2=http://node2:9011,3=http://node3:9011". The Raft groups (aux, rid, scd) of a node share a single listener at its URL.`) + flag.StringVar(&connectParameters.URL, "raft_url", "", "URL at which the other DSS nodes of the cluster reach this node, unique within the cluster, e.g. http://node1:9011. The Raft groups (aux, rid, scd) of the node share a single listener on its port.") flag.Uint64Var(&connectParameters.ClusterID, "raft_cluster_id", 1, "ID of the cluster, used to isolate different Raft clusters running in the same network (must be the same for all nodes in the cluster).") - flag.StringVar(&connectParameters.DataDir, "raft_datadir", defaultDataDir, "Directory for raft data (WAL segments and snapshots), required for restarts. These should not be deleted while the node is running or across restarts unless the node is being permanently shut down. If the directory is lost, the node must be removed from the cluster and re-added as a new node with a new NodeID.") + flag.StringVar(&connectParameters.DataDir, "raft_datadir", defaultDataDir, "Directory for raft data (node identity, WAL segments and snapshots), required for restarts. These should not be deleted while the node is running or across restarts unless the node is being permanently shut down. A node starting with an empty directory gets a new ID and waits to be added to the cluster: if the directory is lost, the node must be added again and its old ID removed.") flag.Uint64Var(&connectParameters.SnapshotCatchupEntries, "raft_snapshot_catchup_entries", defaultSnapshotCatchupEntries, "Log entries retained after compaction so a slow follower can catch up via replication rather than a full snapshot. Higher values tolerate slower followers but increase disk usage.") @@ -163,11 +126,8 @@ func init() { // GetConnectParameters returns a ConnectParameters instance that gets populated from well-known CLI flags. func GetConnectParameters(subfolder string) (ConnectParameters, error) { - if connectParameters.NodeID == 0 { - return ConnectParameters{}, stacktrace.NewError("--raft_node_id is required and must be non-zero") - } - if connectParameters.Peers == "" { - return ConnectParameters{}, stacktrace.NewError("--raft_peers is required") + if _, err := types.NewURLs([]string{connectParameters.URL}); err != nil { + return ConnectParameters{}, stacktrace.Propagate(err, "--raft_url is required and must be a valid URL with a port and without a path") } p := connectParameters if p.Insecure && (p.TLSCertFile != "" || p.TLSKeyFile != "" || p.TLSCAFile != "") { @@ -176,6 +136,9 @@ func GetConnectParameters(subfolder string) (ConnectParameters, error) { if !p.Insecure && (p.TLSCertFile == "" || p.TLSKeyFile == "" || p.TLSCAFile == "") { return ConnectParameters{}, stacktrace.NewError("--raft_tls_cert_file, --raft_tls_key_file and --raft_tls_ca_file are required unless --raft_insecure is set") } + if nodeURL, err := url.Parse(p.URL); err != nil || nodeURL.Scheme != p.URLScheme() { + return ConnectParameters{}, stacktrace.NewError("invalid --raft_url %s: scheme must be %s (https when mutual TLS is enabled, http otherwise)", p.URL, p.URLScheme()) + } p.DataDir = filepath.Join(connectParameters.DataDir, subfolder) p.Group = subfolder groupID, err := groupClusterID(connectParameters.ClusterID, subfolder) diff --git a/pkg/raftstore/params/params_test.go b/pkg/raftstore/params/params_test.go index 2b8cc0c6b..249c34a64 100644 --- a/pkg/raftstore/params/params_test.go +++ b/pkg/raftstore/params/params_test.go @@ -2,7 +2,6 @@ package params import ( "flag" - "net/url" "strconv" "testing" @@ -10,104 +9,6 @@ import ( "github.com/stretchr/testify/require" ) -func TestPeerMap(t *testing.T) { - tests := []struct { - name string - peers string - mutualTLS bool - want map[uint64]*url.URL - wantError bool - }{ - { - name: "valid single peer", - peers: "1=http://node1:9021", - want: map[uint64]*url.URL{1: mustParseURL("http://node1:9021")}, - }, - { - name: "valid multiple peers", - peers: "1=http://node1:9021,2=http://node2:9021,3=http://node3:9021", - want: map[uint64]*url.URL{ - 1: mustParseURL("http://node1:9021"), - 2: mustParseURL("http://node2:9021"), - 3: mustParseURL("http://node3:9021"), - }, - }, - { - name: "valid URL with equals sign in query string", - peers: "1=http://node1:9021?token=abc123", - want: map[uint64]*url.URL{1: mustParseURL("http://node1:9021?token=abc123")}, - }, - { - name: "invalid empty peers string", - peers: "", - wantError: true, - }, - { - name: "invalid entry format", - peers: "invalidentry", - wantError: true, - }, - { - name: "invalid non-numeric node ID", - peers: "abc=http://node1:9021", - wantError: true, - }, - { - name: "invalid negative node ID", - peers: "-1=http://node1:9021", - wantError: true, - }, - { - name: "mixed valid and invalid entries", - peers: "1=http://node1:9021,badentry", - wantError: true, - }, - { - name: "invalid zero peer ID", - peers: "0=http://node1:9021", - wantError: true, - }, - { - name: "duplicate peer IDs", - peers: "1=http://node1:9021,1=http://node2:9021", - wantError: true, - }, - { - name: "valid https peers with mutual TLS", - peers: "1=https://node1:9021,2=https://node2:9021", - mutualTLS: true, - want: map[uint64]*url.URL{ - 1: mustParseURL("https://node1:9021"), - 2: mustParseURL("https://node2:9021"), - }, - }, - { - name: "invalid http peer with mutual TLS", - peers: "1=https://node1:9021,2=http://node2:9021", - mutualTLS: true, - wantError: true, - }, - { - name: "invalid https peer without mutual TLS", - peers: "1=http://node1:9021,2=https://node2:9021", - wantError: true, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - c := ConnectParameters{Peers: tc.peers, Insecure: !tc.mutualTLS} - got, err := c.PeerMap() - if tc.wantError { - require.Error(t, err) - return - } - require.NoError(t, err) - assert.Equal(t, tc.want, got) - }) - } -} - func TestGetConnectParametersTLSFlags(t *testing.T) { setFlag := func(t *testing.T, name string, value string) { require.NoError(t, flag.Set(name, value)) @@ -117,15 +18,19 @@ func TestGetConnectParametersTLSFlags(t *testing.T) { const ( required = "--raft_tls_cert_file, --raft_tls_key_file and --raft_tls_ca_file are required unless --raft_insecure is set" notWithTLS = "--raft_insecure cannot be used with --raft_tls_cert_file, --raft_tls_key_file or --raft_tls_ca_file" + badScheme = "invalid --raft_url" ) tests := []struct { name string + url string certFile string keyFile string caFile string insecure bool wantError string }{ + {name: "http URL with mutual TLS", url: "http://127.0.0.1:9011", certFile: "node.crt", keyFile: "node.key", caFile: "cas.crt", wantError: badScheme}, + {name: "https URL without mutual TLS", url: "https://127.0.0.1:9011", insecure: true, wantError: badScheme}, {name: "all TLS flags: mutual TLS", certFile: "node.crt", keyFile: "node.key", caFile: "cas.crt"}, {name: "insecure only: plain HTTP", insecure: true}, {name: "nothing set", wantError: required}, @@ -141,8 +46,14 @@ func TestGetConnectParametersTLSFlags(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - setFlag(t, "raft_node_id", "1") - setFlag(t, "raft_peers", "1=http://127.0.0.1:9011") + nodeURL := tc.url + if nodeURL == "" { + nodeURL = "https://127.0.0.1:9011" + if tc.insecure { + nodeURL = "http://127.0.0.1:9011" + } + } + setFlag(t, "raft_url", nodeURL) setFlag(t, "raft_tls_cert_file", tc.certFile) setFlag(t, "raft_tls_key_file", tc.keyFile) setFlag(t, "raft_tls_ca_file", tc.caFile) @@ -174,12 +85,3 @@ func TestGroupClusterID(t *testing.T) { } } } - -func mustParseURL(s string) *url.URL { - u, err := url.Parse(s) - if err != nil { - panic(err) - } - - return u -} diff --git a/pkg/raftstore/store.go b/pkg/raftstore/store.go index 60b22bbd4..fac847287 100644 --- a/pkg/raftstore/store.go +++ b/pkg/raftstore/store.go @@ -3,6 +3,7 @@ package raftstore import ( "context" "fmt" + "net/url" "github.com/interuss/dss/pkg/logging" "github.com/interuss/dss/pkg/memstore" @@ -75,14 +76,9 @@ func StartListener(ctx context.Context, logger *zap.Logger, onError func(error)) return stacktrace.Propagate(err, "failed to get raft parameters") } - peers, err := params.PeerMap() + nodeURL, err := url.Parse(params.URL) if err != nil { - return stacktrace.Propagate(err, "failed to parse peer map") - } - - nodeURL, ok := peers[params.NodeID] - if !ok { - return stacktrace.NewError("node ID %d not found in peers map", params.NodeID) + return stacktrace.Propagate(err, "failed to parse the URL of the node") } err = consensus.StartListener(ctx, logger.With(zap.String("component", "listener")), fmt.Sprintf(":%s", nodeURL.Port()), consensus.NewTLSInfo(params, logger), onError)