feat(raft): Implement Raft-based Consistent Hash State Management (#636)

* feat: Add Raft consensus for consistent hashing

This commit introduces Raft consensus to maintain consistency of hash-to-proxy
mappings across multiple GatewayD instances. Key changes include:

- Add new Raft package implementing consensus protocol using HashiCorp's Raft
- Integrate Raft with consistent hashing load balancer
- Store proxy mappings in distributed state machine
- Add configuration options for Raft cluster setup
- Implement leadership monitoring and peer management
- Add FSM snapshot and restore capabilities

The implementation ensures that hash-to-proxy mappings remain consistent across
cluster nodes, improving reliability for consistent hash-based load balancing.

* refactor: update consistent hash implementation with block-based proxy mapping

- Replace proxy ID with block name for consistent hash mapping
- Remove direct raft node dependency from ConsistentHash struct
- Add ProxyByBlock map to Server for block-based proxy lookups
- Include group name in hash key generation for better distribution
- Add proxy initialization during server startup
- Update FSM to use consistent naming for hash map storage

This change improves the consistent hashing mechanism by using block names
instead of proxy IDs, making it more aligned with the block-based
architecture while maintaining backwards compatibility with the original
load balancing strategy.

* refactor: remove proxy ID and related functionality

- Remove unused UUID-based ID field from Proxy struct
- Remove GetID() method from IProxy interface and Proxy implementation
- Remove GetProxyByID() method from Server struct
- Remove uuid package dependency

The proxy ID was not being used meaningfully in the codebase, so removing it
simplifies the proxy implementation.

* feat(raft): Add Raft integration tests and consistent hash improvements

This commit introduces comprehensive Raft testing infrastructure and enhances
the consistent hash implementation with distributed state management.

Key changes:
- Add new test cases for Raft leadership, follower behavior, and FSM operations
- Integrate Raft with consistent hash load balancer for distributed state
- Add TestRaftHelper utility for simplified Raft testing setup
- Update consistent hash tests to use Raft for state persistence
- Add GetState method to RaftNode for state inspection
- Improve test coverage for concurrent operations

The changes ensure that proxy mappings are consistently maintained across
the cluster using Raft consensus, making the load balancer more reliable
in distributed environments.

* feat(raft): add configurable directory and improve test stability

- Add Directory field to Raft config to make raft storage location configurable
- Use t.TempDir() in tests to ensure proper cleanup of test directories
- Rename HashMapCommand to ConsistentHashCommand for better clarity
- Update command type constants and map names to be more descriptive
- Fix test flakiness by using unique node IDs and random available ports
- Remove manual directory cleanup in favor of t.TempDir() cleanup
- Update configuration files with raft directory settings

This change improves test stability and makes the raft storage location
configurable while cleaning up naming conventions throughout the raft package.

* feat(config): add default Raft configuration values

Add default configuration values for Raft consensus implementation:
- RaftAddress: 127.0.0.1:2223
- RaftNodeID: node1
- RaftLeaderID: node1
- RaftDirectory: raft

This change initializes the default Raft configuration in the config loader.

* refactor(raft): improve error handling and code organization

- Enhance error handling with wrapped errors and detailed messages
- Add meaningful constants for timeouts and configuration values
- Rename RaftNode to Node for better clarity
- Fix JSON field names to match Raft convention (nodeId, leaderId)
- Add missing error checks in critical paths
- Improve documentation and code comments
- Update golangci linter settings to include raft package

* Add temporary directory for Raft in Test_pluginScaffoldCmd

- Introduced a temporary directory for Raft using t.TempDir() in the Test_pluginScaffoldCmd test case.
- Set the GATEWAYD_RAFT_DIRECTORY environment variable to the new temporary directory.
- This change ensures that Raft operations during testing are isolated and do not interfere with other tests or system directories.

* feat(config): add JSON parsing for raft peers env variable

- Replace loadEnvVars with loadEnvVarsWithTransform to handle complex env values
- Add special handling for raft.peers to parse JSON array into RaftPeer structs
- Update GlobalKoanf and PluginKoanf to use new transformer function

This change allows proper parsing of list-type environment variables,
specifically for raft peer configurations.

* Add GRPC to raft

Add gRPC support to the Raft implementation to enable proper request forwarding between nodes. Changes include:

- Add protobuf definitions for Raft service with ForwardApply RPC
- Add gRPC server and client implementations for Raft nodes
- Update Raft configuration to include gRPC addresses
- Implement request forwarding logic for non-leader nodes
- Update node configuration to handle gRPC connections
- Add proper cleanup of gRPC resources during shutdown

The changes enable proper forwarding of apply requests from follower nodes to the leader, improving the distributed consensus mechanism.

* feat: add Docker Compose configuration for Raft cluster setup

Add docker-compose-raft.yaml that configures a 3-node GatewayD cluster using Raft consensus protocol. The setup includes:
- 3 GatewayD nodes with Raft configuration
- Separate read/write PostgreSQL instances
- Redis for caching
- Observability stack (Prometheus, Tempo, Grafana)
- Plugin installation service

This configuration enables high availability and leader election through Raft consensus.

* refactor(raft): improve error handling and code clarity

- Improve variable naming in loadEnvVarsWithTransform for better readability
- Clean up error handling in forwardToLeader and ForwardApply
- Add proper error propagation in RPC responses
- Fix string type conversions for peer IDs and addresses
- Organize imports and add missing error package
- Remove unused convertPeers function
- Add clarifying comments for Apply methods

This commit focuses on code quality improvements and better error handling
in the Raft implementation without changing core functionality.

* Add unit tests for Raft RPC server and client

- Implement `TestRPCServer_ForwardApply` to test the `ForwardApply` method of the RPC server, ensuring correct handling of apply requests with various configurations.
- Implement `TestRPCClient` to verify the creation and management of RPC clients, including client retrieval and connection closure.
- Utilize `setupGRPCServer` to create a gRPC server for testing purposes.
- Ensure proper setup and teardown of test nodes and gRPC connections to maintain test isolation and reliability.

* Update Raft configuration in gatewayd.yaml

- Change `nodeId` and `leaderId` from `node2` to `node1`.
- Add `grpcAddress` with value `127.0.0.1:50051`.
- Update `peers` to an empty list instead of an empty dictionary.

These changes adjust the Raft configuration to reflect the new node setup and include a gRPC address for communication.

* Convert RaftPeer slice to string for environment variable compatibility

The function `v1.NewStruct(args)` only accepts `NewValue`, which requires converting certain types to strings. This change adds support for converting a slice of `config.RaftPeer` to a comma-separated string format. Each peer is formatted as "ID:Address:GRPCAddress". This conversion is necessary to overwrite the peers as an environment variable.

* Update checksum in gatewayd_plugins.yaml

- Updated the checksum value for the plugin configuration to ensure integrity and consistency with the latest changes.

* Refactor Raft configuration to use `IsBootstrap` flag

- Replaced `LeaderID` with `IsBootstrap` in Raft configuration across multiple files.
- Updated YAML configuration files (`gatewayd.yaml`, `docker-compose-raft.yaml`) to reflect the new `IsBootstrap` flag.
- Modified Go source files (`config.go`, `constants.go`, `types.go`, `raft.go`) to use `IsBootstrap` instead of `LeaderID`.
- Adjusted test cases in `raft_test.go`, `rpc_test.go`, and `raft_helpers.go` to accommodate the new `IsBootstrap` flag.
- Ensured that the `IsBootstrap` flag is correctly set for nodes intended to bootstrap the Raft cluster.

* Increase the sleep time to pass the test case on the local machine.

* fix: resolve lint issues in rpc_test.go

- Added `t.Helper()` to `setupGRPCServer` and `setupNodes` functions to improve test helper identification.
- Corrected variable naming in `TestRPCServer_ForwardApply` for clarity and consistency.
- Ensured comments end with a period for consistency.
- Updated assertions to use `GetSuccess()` method for better readability.

* feat: Improve code readability with comments and updates

- Updated Docker image references in `docker-compose-raft.yaml` to use `gatewaydio/gatewayd:latest` and added `pull_policy: always` for consistent image updates.
- Changed server and API addresses in `gatewayd.yaml` for better port management.
- Enhanced logging in `raft.go` by switching from `Info` to `Debug` for certain messages to reduce verbosity.
- Added detailed comments in `raft.go` and `rpc.go` to explain the purpose and functionality of key methods, improving code readability and maintainability.
- Introduced new helper functions with comments to clarify their roles in the Raft and RPC processes.

* Improve Redis container setup and async test handling

- Updated `createTestRedis` in `act_helpers_test.go` to use `wait.ForAll` for better reliability by ensuring both log readiness and port listening.
- Enhanced `Test_Run_Async_Redis` in `registry_test.go` by adding a context with a timeout to the consumer subscription for improved test robustness.
- Simplified the sleep duration in `Test_Run_Async_Redis` to reduce unnecessary wait time.

* Handle Fatal Error on Raft Node Initialization Failure

- Added error handling to record and log errors when Raft node initialization fails.
- Ensured the application exits with a specific error code if the Raft node cannot be started.
- Updated tests to set environment variables for Raft node configuration.
- Added a new error code for Raft node startup failure in the error definitions.

This change ensures that if the Raft node cannot be configured and started, the application will terminate gracefully, preventing further execution with an invalid state.

* Update test configuration in gatewayd.yaml

- Changed the raft address from 127.0.0.1:2223 to 127.0.0.1:2222.
- Updated the nodeID from node2 to node1.

These updates are made to the test data configuration to align with the current test case requirements.

* Update comment to accurately describe Raft configuration constants

The comment above the constants was misleading, suggesting they were only command types. Updated the comment to reflect that these constants are related to Raft operations.

* Simplify leader check in monitorLeadership function

- Removed the unnecessary `isLeader` variable in the `monitorLeadership` function.
- Directly checked the node's state against `raft.Leader` in the if condition.

* Fix: Gracefully handle ErrRaftShutdown during Node shutdown

Updated the `Shutdown` method in `raft.go` to gracefully handle the `ErrRaftShutdown` error. This change ensures that if the Raft node is already shut down, the error is ignored, preventing unnecessary error handling.

Author: sinadarbouy

.golangci.yaml

@@ -70,6 +70,7 @@ linters-settings:
           - "github.com/testcontainers/testcontainers-go"
           - "github.com/stretchr/testify/require"
           - "github.com/docker/go-connections/nat"
+          - "github.com/hashicorp/raft"
       test:
         files:
           - $test
@@ -92,6 +93,7 @@ linters-settings:
           - "github.com/redis/go-redis/v9"
           - "github.com/docker/go-connections/nat"
           - "github.com/codingsince1985/checksum"
+          - "github.com/hashicorp/raft"
   tagalign:
     align: false
     sort: false

act/act_helpers_test.go

@@ -62,7 +62,10 @@ func createTestRedis(t *testing.T) string {
 	req := testcontainers.ContainerRequest{
 		Image:        "redis:6",
 		ExposedPorts: []string{"6379/tcp"},
-		WaitingFor:   wait.ForLog("Ready to accept connections"),
+		WaitingFor: wait.ForAll(
+			wait.ForLog("Ready to accept connections"),
+			wait.ForListeningPort("6379/tcp"),
+		),
 	}
 	redisContainer, err := testcontainers.GenericContainer(
 		ctx, testcontainers.GenericContainerRequest{

act/registry_test.go

@@ -747,10 +747,12 @@ func Test_Run_Async_Redis(t *testing.T) {
 	consumer, err := sdkAct.NewConsumer(hclogger, rdb, 5, "test-async-chan")
 	require.NoError(t, err)
 
-	require.NoError(t, consumer.Subscribe(context.Background(), func(ctx context.Context, task []byte) error {
-		err := actRegistry.runAsyncActionFn(ctx, task)
-		waitGroup.Done()
-		return err
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	require.NoError(t, consumer.Subscribe(ctx, func(ctx context.Context, task []byte) error {
+		defer waitGroup.Done()
+		return actRegistry.runAsyncActionFn(ctx, task)
 	}))
 
 	outputs := actRegistry.Apply([]sdkAct.Signal{

cmd/plugin_scaffold_test.go

@@ -28,6 +28,9 @@ func Test_pluginScaffoldCmd(t *testing.T) {
 	postgresAddress2 := postgresHostIP2 + ":" + postgresMappedPort2.Port()
 	t.Setenv("GATEWAYD_CLIENTS_TEST_WRITE_ADDRESS", postgresAddress2)
 
+	raftTempDir := t.TempDir()
+	t.Setenv("GATEWAYD_RAFT_DIRECTORY", raftTempDir)
+
 	globalTestConfigFile := filepath.Join("testdata", "gatewayd.yaml")
 	plugin.IsPluginTemplateEmbedded()
 	pluginTestScaffoldInputFile := "./testdata/scaffold_input.yaml"

cmd/run.go

@@ -29,6 +29,7 @@ import (
 	"github.com/gatewayd-io/gatewayd/network"
 	"github.com/gatewayd-io/gatewayd/plugin"
 	"github.com/gatewayd-io/gatewayd/pool"
+	"github.com/gatewayd-io/gatewayd/raft"
 	"github.com/gatewayd-io/gatewayd/tracing"
 	usage "github.com/gatewayd-io/gatewayd/usagereport/v1"
 	"github.com/getsentry/sentry-go"
@@ -910,6 +911,17 @@ var runCmd = &cobra.Command{
 
 		span.End()
 
+		_, span = otel.Tracer(config.TracerName).Start(runCtx, "Create Raft Node")
+		defer span.End()
+
+		raftNode, originalErr := raft.NewRaftNode(logger, conf.Global.Raft)
+		if originalErr != nil {
+			logger.Error().Err(originalErr).Msg("Failed to start raft node")
+			span.RecordError(originalErr)
+			pluginRegistry.Shutdown()
+			os.Exit(gerr.FailedToStartRaftNode)
+		}
+
 		_, span = otel.Tracer(config.TracerName).Start(runCtx, "Create servers")
 		// Create and initialize servers.
 		for name, cfg := range conf.Global.Servers {
@@ -946,6 +958,7 @@ var runCmd = &cobra.Command{
 					LoadbalancerStrategyName:   cfg.LoadBalancer.Strategy,
 					LoadbalancerRules:          cfg.LoadBalancer.LoadBalancingRules,
 					LoadbalancerConsistentHash: cfg.LoadBalancer.ConsistentHash,
+					RaftNode:                   raftNode,
 				},
 			)
 

cmd/run_test.go

@@ -23,6 +23,11 @@ func Test_runCmd(t *testing.T) {
 	postgresAddress := postgresHostIP + ":" + postgresMappedPort.Port()
 	t.Setenv("GATEWAYD_CLIENTS_DEFAULT_WRITES_ADDRESS", postgresAddress)
 
+	tempDir := t.TempDir()
+	t.Setenv("GATEWAYD_RAFT_DIRECTORY", tempDir)
+	t.Setenv("GATEWAYD_RAFT_ADDRESS", "127.0.0.1:0")
+	t.Setenv("GATEWAYD_RAFT_GRPCADDRESS", "127.0.0.1:0")
+
 	globalTestConfigFile := "./test_global_runCmd.yaml"
 	pluginTestConfigFile := "./test_plugins_runCmd.yaml"
 	// Create a test plugins config file.
@@ -87,6 +92,11 @@ func Test_runCmdWithTLS(t *testing.T) {
 	postgresAddress := postgresHostIP + ":" + postgresMappedPort.Port()
 	t.Setenv("GATEWAYD_CLIENTS_DEFAULT_WRITES_ADDRESS", postgresAddress)
 
+	tempDir := t.TempDir()
+	t.Setenv("GATEWAYD_RAFT_DIRECTORY", tempDir)
+	t.Setenv("GATEWAYD_RAFT_ADDRESS", "127.0.0.1:0")
+	t.Setenv("GATEWAYD_RAFT_GRPCADDRESS", "127.0.0.1:0")
+
 	globalTLSTestConfigFile := "./testdata/gatewayd_tls.yaml"
 	pluginTestConfigFile := "./test_plugins_runCmdWithTLS.yaml"
 	// Create a test plugins config file.
@@ -150,6 +160,11 @@ func Test_runCmdWithMultiTenancy(t *testing.T) {
 	postgresAddress2 := postgresHostIP2 + ":" + postgresMappedPort2.Port()
 	t.Setenv("GATEWAYD_CLIENTS_TEST_WRITE_ADDRESS", postgresAddress2)
 
+	tempDir := t.TempDir()
+	t.Setenv("GATEWAYD_RAFT_DIRECTORY", tempDir)
+	t.Setenv("GATEWAYD_RAFT_ADDRESS", "127.0.0.1:0")
+	t.Setenv("GATEWAYD_RAFT_GRPCADDRESS", "127.0.0.1:0")
+
 	globalTestConfigFile := "./testdata/gatewayd.yaml"
 	pluginTestConfigFile := "./test_plugins_runCmdWithMultiTenancy.yaml"
 	// Create a test plugins config file.
@@ -211,6 +226,11 @@ func Test_runCmdWithCachePlugin(t *testing.T) {
 	postgresAddress := postgresHostIP + ":" + postgresMappedPort.Port()
 	t.Setenv("GATEWAYD_CLIENTS_DEFAULT_WRITES_ADDRESS", postgresAddress)
 
+	tempDir := t.TempDir()
+	t.Setenv("GATEWAYD_RAFT_DIRECTORY", tempDir)
+	t.Setenv("GATEWAYD_RAFT_ADDRESS", "127.0.0.1:0")
+	t.Setenv("GATEWAYD_RAFT_GRPCADDRESS", "127.0.0.1:0")
+
 	globalTestConfigFile := "./test_global_runCmdWithCachePlugin.yaml"
 	pluginTestConfigFile := "./test_plugins_runCmdWithCachePlugin.yaml"
 	// TODO: Remove this once these global variables are removed from cmd/run.go.

cmd/testdata/gatewayd.yaml

@@ -73,3 +73,9 @@ servers:
 
 api:
   enabled: True
+
+raft:
+  address: 127.0.0.1:2222
+  nodeID: node1
+  isBootstrap: true
+  peers: {}

config/config.go

@@ -180,6 +180,13 @@ func (c *Config) LoadDefaults(ctx context.Context) *gerr.GatewayDError {
 			GRPCNetwork: DefaultGRPCAPINetwork,
 			GRPCAddress: DefaultGRPCAPIAddress,
 		},
+		Raft: Raft{
+			Address:     DefaultRaftAddress,
+			NodeID:      DefaultRaftNodeID,
+			IsBootstrap: DefaultRaftIsBootstrap,
+			Directory:   DefaultRaftDirectory,
+			GRPCAddress: DefaultRaftGRPCAddress,
+		},
 	}
 
 	//nolint:nestif
@@ -201,7 +208,7 @@ func (c *Config) LoadDefaults(ctx context.Context) *gerr.GatewayDError {
 				return gerr.ErrConfigParseError.Wrap(err)
 			}
 
-			if configObject == "api" {
+			if configObject == "api" || configObject == "raft" {
 				// Handle API configuration separately
 				// TODO: Add support for multiple API config groups.
 				continue
@@ -309,7 +316,7 @@ func (c *Config) LoadDefaults(ctx context.Context) *gerr.GatewayDError {
 func (c *Config) LoadGlobalEnvVars(ctx context.Context) *gerr.GatewayDError {
 	_, span := otel.Tracer(TracerName).Start(ctx, "Load global environment variables")
 
-	if err := c.GlobalKoanf.Load(loadEnvVars(), nil); err != nil {
+	if err := c.GlobalKoanf.Load(loadEnvVarsWithTransform(), nil); err != nil {
 		span.RecordError(err)
 		span.End()
 		return gerr.ErrConfigParseError.Wrap(
@@ -326,7 +333,7 @@ func (c *Config) LoadGlobalEnvVars(ctx context.Context) *gerr.GatewayDError {
 func (c *Config) LoadPluginEnvVars(ctx context.Context) *gerr.GatewayDError {
 	_, span := otel.Tracer(TracerName).Start(ctx, "Load plugin environment variables")
 
-	if err := c.PluginKoanf.Load(loadEnvVars(), nil); err != nil {
+	if err := c.PluginKoanf.Load(loadEnvVarsWithTransform(), nil); err != nil {
 		span.RecordError(err)
 		span.End()
 		return gerr.ErrConfigParseError.Wrap(
@@ -338,41 +345,52 @@ func (c *Config) LoadPluginEnvVars(ctx context.Context) *gerr.GatewayDError {
 	return nil
 }
 
-func loadEnvVars() *env.Env {
-	return env.Provider(EnvPrefix, ".", transformEnvVariable)
-}
-
-// transformEnvVariable transforms the environment variable name to a format based on JSON tags.
-func transformEnvVariable(envVar string) string {
-	structs := []any{
-		&API{},
-		&Logger{},
-		&Pool{},
-		&Proxy{},
-		&Server{},
-		&Metrics{},
-		&PluginConfig{},
-	}
-	tagMapping := make(map[string]string)
-	generateTagMapping(structs, tagMapping)
+func loadEnvVarsWithTransform() *env.Env {
+	// Use ProviderWithValue to transform both key and value
+	return env.ProviderWithValue(EnvPrefix, ".", func(envKey string, value string) (string, interface{}) {
+		// Transform the key
+		key := strings.ToLower(strings.TrimPrefix(envKey, EnvPrefix))
+
+		structs := []any{
+			&API{},
+			&Logger{},
+			&Pool{},
+			&Proxy{},
+			&Server{},
+			&Metrics{},
+			&PluginConfig{},
+			&Raft{},
+		}
+		tagMapping := make(map[string]string)
+		generateTagMapping(structs, tagMapping)
 
-	lowerEnvVar := strings.ToLower(strings.TrimPrefix(envVar, EnvPrefix))
-	parts := strings.Split(lowerEnvVar, "_")
+		parts := strings.Split(key, "_")
 
-	var transformedParts strings.Builder
+		var transformedParts strings.Builder
 
-	for i, part := range parts {
-		if i > 0 {
-			transformedParts.WriteString(".")
+		for i, part := range parts {
+			if i > 0 {
+				transformedParts.WriteString(".")
+			}
+			if mappedValue, exists := tagMapping[part]; exists {
+				transformedParts.WriteString(mappedValue)
+			} else {
+				transformedParts.WriteString(part)
+			}
 		}
-		if mappedValue, exists := tagMapping[part]; exists {
-			transformedParts.WriteString(mappedValue)
-		} else {
-			transformedParts.WriteString(part)
+
+		// Check if the key is "peers" and transform the value using JSON unmarshal
+		if transformedParts.String() == "raft.peers" {
+			var raftPeers []RaftPeer
+			if err := json.Unmarshal([]byte(value), &raftPeers); err != nil {
+				return transformedParts.String(), fmt.Errorf("failed to unmarshal peers: %w", err)
+			}
+			return transformedParts.String(), raftPeers
 		}
-	}
 
-	return transformedParts.String()
+		// Return the key and value as is if no transformation is needed
+		return transformedParts.String(), value
+	})
 }
 
 // LoadGlobalConfigFile loads the plugin configuration file.

config/constants.go

@@ -126,6 +126,13 @@ const (
 	DefaultActionRedisEnabled = false
 	DefaultRedisAddress       = "localhost:6379"
 	DefaultRedisChannel       = "gatewayd-actions"
+
+	// Raft constants.
+	DefaultRaftAddress     = "127.0.0.1:2223"
+	DefaultRaftNodeID      = "node1"
+	DefaultRaftIsBootstrap = true
+	DefaultRaftDirectory   = "raft"
+	DefaultRaftGRPCAddress = "127.0.0.1:50051"
 )
 
 // Load balancing strategies.

config/types.go

@@ -138,6 +138,21 @@ type API struct {
 	GRPCNetwork string `json:"grpcNetwork" jsonschema:"enum=tcp,enum=udp,enum=unix"`
 }
 
+type Raft struct {
+	Address     string     `json:"address"`
+	NodeID      string     `json:"nodeId"`
+	IsBootstrap bool       `json:"isBootstrap"`
+	Peers       []RaftPeer `json:"peers"`
+	Directory   string     `json:"directory" jsonschema:"default=raft"`
+	GRPCAddress string     `json:"grpcAddress"`
+}
+
+type RaftPeer struct {
+	ID          string `json:"id"`
+	Address     string `json:"address"`
+	GRPCAddress string `json:"grpcAddress"`
+}
+
 type GlobalConfig struct {
 	API     API                           `json:"api"`
 	Loggers map[string]*Logger            `json:"loggers"`
@@ -146,4 +161,5 @@ type GlobalConfig struct {
 	Proxies map[string]map[string]*Proxy  `json:"proxies"`
 	Servers map[string]*Server            `json:"servers"`
 	Metrics map[string]*Metrics           `json:"metrics"`
+	Raft    Raft                          `json:"raft"`
 }