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

.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"`
 }