Add an ability to run kvgod in a distributed way on multiple servers at the same time using RAFT consensus algorithm

Author: kgantsov

Gopkg.lock

@@ -1,19 +1,27 @@
 package main
 
 import (
+	"context"
 	"flag"
-	"fmt"
+	"path/filepath"
+	"time"
 
+	pb "github.com/kgantsov/kvgo/pkg/server"
 	server "github.com/kgantsov/kvgo/pkg/server"
 	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
 )
 
 const dbPath = "./data.db"
 const indexPath = "./indexes.idx"
 
 func main() {
-	port := flag.String("port", "56379", "DB port")
-	rpcPort := flag.String("rpc_port", "50051", "RPC DB port")
+	// addr := flag.String("addr", ":56379", "Redis bind address")
+	rpcAddr := flag.String("rpc_addr", ":50051", "RPC bind address")
+	raftDir := flag.String("raft_dir", "", "RPC DB port")
+	raftAddr := flag.String("raft_addr", ":12000", "Raft bind address")
+	joinAddr := flag.String("join_addr", "", "Join address")
+	nodeID := flag.String("node_id", "", "Node ID")
 	logLevel := flag.String("log_level", "info", "Log level")
 	flag.Parse()
 
@@ -26,9 +34,42 @@ func main() {
 	}
 	log.SetLevel(level)
 
+	if *raftDir == "" {
+		log.Fatal("No Raft storage directory specified\n")
+	}
+	if *nodeID == "" {
+		log.Fatal("No nodeID storage directory specified\n")
+	}
+
 	log.Info("Creating storage...")
-	store := server.NewStore(dbPath, indexPath, 1000, 10000)
+	store := server.NewStore(
+		filepath.Join(*raftDir, dbPath), filepath.Join(*raftDir, indexPath), 1000, 10000,
+	)
+	store.RaftDir = *raftDir
+	store.RaftBind = *raftAddr
+
+	if err := store.Open(*joinAddr == "", *nodeID); err != nil {
+		log.Fatalf("failed to open store: %s", err.Error())
+	}
+
+	if *joinAddr != "" {
+		conn, err := grpc.Dial(*joinAddr, grpc.WithInsecure())
+		if err != nil {
+			log.Fatalf("did not connect: %v", err)
+		}
+		defer conn.Close()
+		c := pb.NewKVClient(conn)
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		log.Info("TRYING TO CONNECT OT REMOTE NODE remote node %s at %s", *nodeID, *raftAddr)
+		_, err = c.Join(ctx, &pb.JoinRequest{Addr: *raftAddr, NodeID: *nodeID})
+		if err != nil {
+			log.Fatalf("could not greet: %v", err)
+		}
+	}
 
-	server.ListenAndServ(fmt.Sprintf(":%s", *port), store)
-	server.ListenAndServGrpc(fmt.Sprintf(":%s", *rpcPort), store)
+	server.ListenAndServGrpc(*rpcAddr, store)
+	// server.ListenAndServ(*addr, store)
 }

cmd/kvgod/main.go

@@ -52,3 +52,8 @@ func (s *server) Del(ctx context.Context, in *DelRequest) (*DelResponse, error)
 	s.store.Delete(in.Key)
 	return &DelResponse{Exist: false}, nil
 }
+
+func (s *server) Join(ctx context.Context, in *JoinRequest) (*JoinResponse, error) {
+	s.store.Join(in.NodeID, in.Addr)
+	return &JoinResponse{Joined: true}, nil
+}

pkg/server/grpc.go

@@ -28,9 +28,19 @@ message DelResponse {
   bool exist = 1;
 }
 
+message JoinRequest {
+  string addr = 1;
+  string nodeID = 2;
+}
+
+message JoinResponse {
+  bool joined = 1;
+}
+
 
 service KV {
   rpc Set (SetRequest) returns (SetResponse) {}
   rpc Get (GetRequest) returns (GetResponse) {}
   rpc Del (DelRequest) returns (DelResponse) {}
+  rpc Join (JoinRequest) returns (JoinResponse) {}
 }