[raft/config] Control config changes

.github/workflows/ci.yml

@@ -127,6 +127,41 @@ jobs:
       - name: Bring down local DSS instance
         run: make down-locally
 
+  dss-tests-with-raft:
+    name: DSS tests with Raft
+    runs-on: ubuntu-latest
+    env:
+      COMPOSE_PROFILES: with-raft
+    steps:
+      - name: Job information
+        run: |
+          echo "Job information"
+          echo "Trigger: ${{ github.event_name }}"
+          echo "Host: ${{ runner.os }}"
+          echo "Repository: ${{ github.repository }}"
+          echo "Branch: ${{ github.ref }}"
+          docker images
+          go env
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          submodules: true
+      - name: Build dss image
+        run: make build-dss
+      - name: Tear down any pre-existing local DSS instance
+        run: make down-locally
+      - name: Start local DSS instance
+        run: make start-locally
+      - name: Probe local DSS instance
+        run: make probe-locally
+      - name: Run Qualifier against local DSS instance
+        run: make qualify-locally
+      # Todo: Re add evict tests here once evict is implemented.
+      - name: Run security tests against local DSS instance
+        run: make security-locally
+      - name: Bring down local DSS instance
+        run: make down-locally
+
   certificates-management-tests:
     name: Certificate management tests
     runs-on: ubuntu-latest

build/dev/docker-compose_dss.yaml

@@ -111,6 +111,7 @@ services:
       - $PWD/../test-certs:/var/test-certs:ro
       - $PWD/startup/core_service.sh:/startup/core_service.sh:ro
       - $PWD/startup/coverdata:/startup/coverdata:rw  # we will save coverage info here
+      - raftdata:/raftdata
     environment:
       COMPOSE_PROFILES: ${COMPOSE_PROFILES}
       # Note: requires the Dockerfile to have been built with "-cover" in the EXTRA_GO_INSTALL_FLAGS var
@@ -142,7 +143,7 @@ services:
       interval: 3m
       start_period: 30s
       start_interval: 5s
-    profiles: ["", "with-yugabyte"]
+    profiles: ["", "with-yugabyte", "with-raft"]
 
   local-dss-dummy-oauth:
     build:
@@ -166,3 +167,4 @@ networks:
 
 volumes:
   local-dss-data:
+  raftdata:

build/dev/startup/core_service.sh

@@ -10,8 +10,13 @@ if [ "${COMPOSE_PROFILES#*"with-yugabyte"}" != "${COMPOSE_PROFILES}" ]; then
   echo "Using Yugabyte"
   DATASTORE_CONNECTION="-datastore_host local-dss-ybdb -datastore_user yugabyte --datastore_port 5433"
 else
-  echo "Using CockroachDB"
-  DATASTORE_CONNECTION="-datastore_host local-dss-crdb"
+    if [ "${COMPOSE_PROFILES#*"with-raft"}" != "${COMPOSE_PROFILES}" ]; then
+      echo "Using raft"
+      DATASTORE_CONNECTION="-store_type raft -raft_node_id=1 -rid_raft_peers=1=http://127.0.0.1:9011 -scd_raft_peers=1=http://127.0.0.1:9021 -aux_raft_peers=1=http://127.0.0.1:9031 -raft_datadir /raftdata"
+    else
+      echo "Using CockroachDB"
+      DATASTORE_CONNECTION="-datastore_host local-dss-crdb"
+    fi
 fi
 
 if [ "$DEBUG_ON" = "1" ]; then

cmds/core-service/main.go

@@ -32,6 +32,7 @@ import (
 	scds "github.com/interuss/dss/pkg/scd/store"
 	"github.com/interuss/dss/pkg/store"
 	"github.com/interuss/dss/pkg/store/params"
+	"github.com/interuss/dss/pkg/timestamp"
 	"github.com/interuss/dss/pkg/version"
 	"github.com/interuss/dss/pkg/versioning"
 	"github.com/interuss/stacktrace"
@@ -340,6 +341,7 @@ func RunHTTPServer(ctx context.Context, ctxCanceler func(), address, locality st
 	handler = authorizer.TokenMiddleware(handler)
 	handler = http.TimeoutHandler(handler, *timeout, "request timeout")
 	handler = logging.HTTPMiddleware(logger, *dumpRequests, handler)
+	handler = timestamp.Middleware(handler)
 
 	if *enableMetrics || *enableTracing {
 		// We use the default settings; the APIRouter handler will override the span value accordingly, as it has more information.

cmds/db-manager/cleanup/evict.go

@@ -100,7 +100,7 @@ func evict(cmd *cobra.Command, _ []string) error {
 		}
 		return nil
 	}
-	if err = scdStore.Transact(ctx, scdAction); err != nil {
+	if _, err = scdStore.Transact(ctx, "", nil, scdAction); err != nil {
 		return fmt.Errorf("failed to execute SCD transaction: %w", err)
 	}
 
@@ -145,7 +145,7 @@ func evict(cmd *cobra.Command, _ []string) error {
 
 		return nil
 	}
-	if err = ridStore.Transact(ctx, ridAction); err != nil {
+	if _, err = ridStore.Transact(ctx, "", nil, ridAction); err != nil {
 		return fmt.Errorf("failed to execute RID transaction: %w", err)
 	}
 

cmds/raftctl/main.go

@@ -0,0 +1,20 @@
+package main
+
+import (
+	"github.com/interuss/dss/cmds/raftctl/node"
+	"github.com/spf13/cobra"
+)
+
+var rootCmd = &cobra.Command{
+	Use:   "raftctl",
+	Short: "Manage a DSS instance's raftstore clusters",
+}
+
+func init() {
+	rootCmd.AddCommand(node.NodeCmd)
+}
+
+func main() {
+	err := rootCmd.Execute()
+	cobra.CheckErr(err)
+}

cmds/raftctl/node/add.go

@@ -0,0 +1,22 @@
+package node
+
+import (
+	"github.com/spf13/cobra"
+	"go.etcd.io/raft/v3/raftpb"
+)
+
+var addAddresses map[string]string
+
+var addCmd = &cobra.Command{
+	Use:   "add",
+	Short: "Add a node to the raftstore",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return changeMembership(addr, raftpb.ConfChangeAddNode, nodeID, addAddresses)
+	},
+}
+
+func init() {
+	addCmd.Flags().StringToStringVar(&addAddresses, "addresses", nil, "host:port the new node's Raft transport will listen on for each store, e.g. \"rid=node4:8081,scd=node4:8082,aux=node4:8083\" (required)")
+	err := addCmd.MarkFlagRequired("addresses")
+	cobra.CheckErr(err)
+}

cmds/raftctl/node/client.go

@@ -0,0 +1,77 @@
+// Package node implements the `raftctl node` subcommands, which change membership of a DSS
+// instance's raftstore clusters by talking to its admin endpoint (see pkg/raftstore/admin).
+package node
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"sort"
+
+	"go.etcd.io/raft/v3/raftpb"
+)
+
+const membersPath = "/admin/members"
+
+type memberChangeRequest struct {
+	Action    raftpb.ConfChangeType `json:"action"`
+	NodeID    uint64                `json:"node_id"`
+	Addresses map[string]string     `json:"addresses,omitempty"`
+}
+
+type storeResult struct {
+	ConfState any    `json:"conf_state,omitempty"`
+	Error     string `json:"error,omitempty"`
+}
+
+type memberChangeResponse struct {
+	Results map[string]storeResult `json:"results"`
+}
+
+func changeMembership(addr string, action raftpb.ConfChangeType, nodeID uint64, addresses map[string]string) error {
+	body, err := json.Marshal(memberChangeRequest{Action: action, NodeID: nodeID, Addresses: addresses})
+	if err != nil {
+		return fmt.Errorf("failed to marshal request: %w", err)
+	}
+
+	resp, err := http.Post(addr+membersPath, "application/json", bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("failed to send request to %s: %w", addr, err)
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read response: %w", err)
+	}
+
+	var result memberChangeResponse
+	if err := json.Unmarshal(respBody, &result); err != nil {
+		return fmt.Errorf("failed to decode response (status %s): %s", resp.Status, respBody)
+	}
+
+	names := make([]string, 0, len(result.Results))
+	for name := range result.Results {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	failed := false
+	for _, name := range names {
+		r := result.Results[name]
+		if r.Error != "" {
+			failed = true
+			fmt.Printf("%s: error: %s\n", name, r.Error)
+			continue
+		}
+		fmt.Printf("%s: ok, conf state: %v\n", name, r.ConfState)
+	}
+
+	if failed {
+		return fmt.Errorf("one or more stores failed to apply the membership change")
+	}
+
+	return nil
+}

cmds/raftctl/node/node.go

@@ -0,0 +1,29 @@
+package node
+
+import "github.com/spf13/cobra"
+
+const (
+	addrFlag = "addr"
+	idFlag   = "id"
+)
+
+var (
+	addr   string
+	nodeID uint64
+
+	NodeCmd = &cobra.Command{
+		Use:   "node",
+		Short: "Request configuration changes for raftstore",
+	}
+)
+
+func init() {
+	NodeCmd.PersistentFlags().StringVar(&addr, addrFlag, "", "address of the DSS instance's raft admin endpoint, e.g. http://localhost:8082 (required)")
+	NodeCmd.PersistentFlags().Uint64Var(&nodeID, idFlag, 0, "ID of the node to add/update/remove (required)")
+	err := NodeCmd.MarkPersistentFlagRequired(addrFlag)
+	cobra.CheckErr(err)
+	err = NodeCmd.MarkPersistentFlagRequired(idFlag)
+	cobra.CheckErr(err)
+
+	NodeCmd.AddCommand(addCmd, updateCmd, removeCmd)
+}

cmds/raftctl/node/remove.go

@@ -0,0 +1,14 @@
+package node
+
+import (
+	"github.com/spf13/cobra"
+	"go.etcd.io/raft/v3/raftpb"
+)
+
+var removeCmd = &cobra.Command{
+	Use:   "remove",
+	Short: "Remove a node from every raftstore cluster known to the target DSS instance",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return changeMembership(addr, raftpb.ConfChangeRemoveNode, nodeID, nil)
+	},
+}

cmds/raftctl/node/update.go

@@ -0,0 +1,22 @@
+package node
+
+import (
+	"github.com/spf13/cobra"
+	"go.etcd.io/raft/v3/raftpb"
+)
+
+var updateAddresses map[string]string
+
+var updateCmd = &cobra.Command{
+	Use:   "update",
+	Short: "Update a node's address across every raftstore cluster known to the target DSS instance",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return changeMembership(addr, raftpb.ConfChangeUpdateNode, nodeID, updateAddresses)
+	},
+}
+
+func init() {
+	updateCmd.Flags().StringToStringVar(&updateAddresses, "addresses", nil, "new host:port the node's Raft transport will listen on for each store, e.g. \"rid=node4-new:8081,scd=node4-new:8082,aux=node4-new:8083\" (required)")
+	err := updateCmd.MarkFlagRequired("addresses")
+	cobra.CheckErr(err)
+}

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/go-jose/go-jose/v4 v4.1.4
 	github.com/golang-jwt/jwt/v4 v4.5.2
 	github.com/golang/geo v0.0.0-20230421003525-6adc56603217
+	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/interuss/stacktrace v1.0.0
 	github.com/jackc/pgx/v5 v5.9.2
@@ -27,11 +28,13 @@ require (
 	go.opentelemetry.io/otel v1.43.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.39.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.65.0
+	go.opentelemetry.io/otel/metric v1.43.0
 	go.opentelemetry.io/otel/sdk v1.43.0
 	go.opentelemetry.io/otel/sdk/metric v1.43.0
 	go.opentelemetry.io/otel/trace v1.43.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
+	golang.org/x/sync v0.20.0
 )
 
 require (
@@ -71,13 +74,11 @@ require (
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect
-	go.opentelemetry.io/otel/metric v1.43.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	golang.org/x/crypto v0.49.0 // indirect
 	golang.org/x/net v0.52.0 // indirect
 	golang.org/x/oauth2 v0.34.0 // indirect
-	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.36.0 // indirect
 	golang.org/x/time v0.14.0 // indirect

pkg/aux_/store/memstore/doc.go

@@ -0,0 +1,3 @@
+// Package aux_.store.memstore provides a full implementation of store.Store[aux_.repos.Repository]
+// storing data in memory. It is meant to be used by raftstore.
+package memstore