15 changed files with 558 additions and 91 deletions
-
+2 -1.gitignore
-
+3 -1README.md
-
+49 -5cmd/main.go
-
+2 -0go.mod
-
+18 -0go.sum
-
+94 -0internal/client/client_node.go
-
+42 -30internal/nodes/init.go
-
+16 -0internal/nodes/log.go
-
+43 -49internal/nodes/node.go
-
+42 -0internal/nodes/server_node.go
-
+1 -1scripts/build.sh
-
+4 -4scripts/run.sh
-
+45 -0test/common.go
-
+112 -0test/restart_follower_test.go
-
+85 -0test/server_client_test.go
+ 2
- 1
.gitignore
View File
+ 3
- 1
README.md
View File
+ 49
- 5
cmd/main.go
View File
+ 2
- 0
go.mod
View File
+ 18
- 0
go.sum
View File
+ 94
- 0
internal/client/client_node.go
View File
| @ -0,0 +1,94 @@ | |||||
| package clientPkg | |||||
| import ( | |||||
| "net/rpc" | |||||
| "simple-kv-store/internal/logprovider" | |||||
| "simple-kv-store/internal/nodes" | |||||
| "go.uber.org/zap" | |||||
| ) | |||||
| var log, _ = logprovider.CreateDefaultZapLogger(zap.InfoLevel) | |||||
| type Client struct { | |||||
| // 连接的server端节点（node1） | |||||
| ServerId string | |||||
| Address string | |||||
| } | |||||
| type Status = uint8 | |||||
| const ( | |||||
| Ok Status = iota + 1 | |||||
| NotFound | |||||
| Fail | |||||
| ) | |||||
| func (client *Client) Write(kvCall nodes.LogEntryCall) Status { | |||||
| log.Info("client write request key :" + kvCall.LogE.Key) | |||||
| c, err := rpc.DialHTTP("tcp", client.Address) | |||||
| if err != nil { | |||||
| log.Error("dialing: ", zap.Error(err)) | |||||
| return Fail | |||||
| } | |||||
| defer func(server *rpc.Client) { | |||||
| err := c.Close() | |||||
| if err != nil { | |||||
| log.Error("client close err: ", zap.Error(err)) | |||||
| } | |||||
| }(c) | |||||
| var reply nodes.ServerReply | |||||
| callErr := c.Call("Node.WriteKV", kvCall, &reply) // RPC | |||||
| if callErr != nil { | |||||
| log.Error("dialing: ", zap.Error(callErr)) | |||||
| return Fail | |||||
| } | |||||
| if reply.Isconnect { // 发送成功 | |||||
| return Ok | |||||
| } else { // 失败 | |||||
| return Fail | |||||
| } | |||||
| } | |||||
| func (client *Client) Read(key string, value *string) Status { // 查不到value为空 | |||||
| log.Info("client read request key :" + key) | |||||
| if value == nil { | |||||
| return Fail | |||||
| } | |||||
| c, err := rpc.DialHTTP("tcp", client.Address) | |||||
| if err != nil { | |||||
| log.Error("dialing: ", zap.Error(err)) | |||||
| return Fail | |||||
| } | |||||
| defer func(server *rpc.Client) { | |||||
| err := c.Close() | |||||
| if err != nil { | |||||
| log.Error("client close err: ", zap.Error(err)) | |||||
| } | |||||
| }(c) | |||||
| var reply nodes.ServerReply | |||||
| callErr := c.Call("Node.ReadKey", key, &reply) // RPC | |||||
| if callErr != nil { | |||||
| log.Error("dialing: ", zap.Error(callErr)) | |||||
| return Fail | |||||
| } | |||||
| if reply.Isconnect { // 发送成功 | |||||
| if reply.HaveValue { | |||||
| *value = reply.Value | |||||
| return Ok | |||||
| } else { | |||||
| return NotFound | |||||
| } | |||||
| } else { // 失败 | |||||
| return Fail | |||||
| } | |||||
| } | |||||
+ 42
- 30
internal/nodes/init.go
View File
+ 16
- 0
internal/nodes/log.go
View File
| @ -1,10 +1,26 @@ | |||||
| package nodes | package nodes | ||||
| const ( | |||||
| Normal State = iota + 1 | |||||
| Delay | |||||
| Fail | |||||
| ) | |||||
| type LogEntry struct { | type LogEntry struct { | ||||
| Key string | Key string | ||||
| Value string | Value string | ||||
| } | } | ||||
| type LogEntryCall struct { | |||||
| LogE LogEntry | |||||
| CallState State | |||||
| } | |||||
| type KVReply struct { | type KVReply struct { | ||||
| Reply bool | Reply bool | ||||
| } | } | ||||
| type LogIdAndEntry struct { | |||||
| LogId int | |||||
| Entry LogEntry | |||||
| } | |||||
+ 43
- 49
internal/nodes/node.go
View File
+ 42
- 0
internal/nodes/server_node.go
View File
| @ -0,0 +1,42 @@ | |||||
| package nodes | |||||
| import ( | |||||
| "strconv" | |||||
| "github.com/syndtr/goleveldb/leveldb" | |||||
| ) | |||||
| // leader node作为server为client注册的方法 | |||||
| type ServerReply struct{ | |||||
| Isconnect bool | |||||
| HaveValue bool | |||||
| Value string | |||||
| } | |||||
| // RPC call | |||||
| func (node *Node) WriteKV(kvCall LogEntryCall, reply *ServerReply) error { | |||||
| logId := node.maxLogId | |||||
| node.maxLogId++ | |||||
| node.log[logId] = kvCall.LogE | |||||
| node.db.Put([]byte(kvCall.LogE.Key), []byte(kvCall.LogE.Value), nil) | |||||
| log.Info("server write : logId = " + strconv.Itoa(logId) + ", key = " + kvCall.LogE.Key) | |||||
| // 广播给其它节点 | |||||
| node.BroadCastKV(logId, kvCall) | |||||
| reply.Isconnect = true | |||||
| return nil | |||||
| } | |||||
| // RPC call | |||||
| func (node *Node) ReadKey(key string, reply *ServerReply) error { | |||||
| log.Info("server read : " + key) | |||||
| // 先只读leader自己 | |||||
| value, err := node.db.Get([]byte(key), nil) | |||||
| if err == leveldb.ErrNotFound { | |||||
| reply.HaveValue = false | |||||
| } else { | |||||
| reply.HaveValue = true | |||||
| reply.Value = string(value) | |||||
| } | |||||
| reply.Isconnect = true | |||||
| return nil | |||||
| } | |||||
+ 1
- 1
scripts/build.sh
View File
| @ -1 +1 @@ | |||||
| go build -o raftnode ./cmd/main.go | |||||
| go build -o main ./cmd/main.go | |||||
+ 4
- 4
scripts/run.sh
View File
+ 45
- 0
test/common.go
View File
| @ -0,0 +1,45 @@ | |||||
| package test | |||||
| import ( | |||||
| "fmt" | |||||
| "os" | |||||
| "os/exec" | |||||
| "strconv" | |||||
| "strings" | |||||
| ) | |||||
| func ExecuteNodeI(i int, isLeader bool, isNewDb bool, clusters []string) *exec.Cmd { | |||||
| tmpClusters := append(clusters[:i], clusters[i+1:]...) | |||||
| port := fmt.Sprintf(":%d", uint16(9090)+uint16(i)) | |||||
| var isleader string | |||||
| if isLeader { | |||||
| isleader = "true" | |||||
| } else { | |||||
| isleader = "false" | |||||
| } | |||||
| var isnewdb string | |||||
| if isNewDb { | |||||
| isnewdb = "true" | |||||
| } else { | |||||
| isnewdb = "false" | |||||
| } | |||||
| cmd := exec.Command( | |||||
| "../main", | |||||
| "-id", strconv.Itoa(i + 1), | |||||
| "-port", port, | |||||
| "-cluster", strings.Join(tmpClusters, ","), | |||||
| "-isleader=" + isleader, | |||||
| "-isNewDb=" + isnewdb, | |||||
| ) | |||||
| cmd.Stdout = os.Stdout | |||||
| cmd.Stderr = os.Stderr | |||||
| // 执行命令 | |||||
| err := cmd.Start() | |||||
| if err != nil { | |||||
| fmt.Println("启动进程出错:", err) | |||||
| return nil | |||||
| } | |||||
| return cmd | |||||
| } | |||||
+ 112
- 0
test/restart_follower_test.go
View File
| @ -0,0 +1,112 @@ | |||||
| package test | |||||
| import ( | |||||
| "fmt" | |||||
| "os/exec" | |||||
| "simple-kv-store/internal/client" | |||||
| "simple-kv-store/internal/nodes" | |||||
| "strconv" | |||||
| "syscall" | |||||
| "testing" | |||||
| "time" | |||||
| ) | |||||
| func TestFollowerRestart(t *testing.T) { | |||||
| // 登记结点信息 | |||||
| n := 3 | |||||
| var clusters []string | |||||
| for i := 0; i < n; i++ { | |||||
| port := fmt.Sprintf("%d", uint16(9090)+uint16(i)) | |||||
| addr := "127.0.0.1:" + port | |||||
| clusters = append(clusters, addr) | |||||
| } | |||||
| // 结点启动 | |||||
| var cmds []*exec.Cmd | |||||
| for i := 0; i < n; i++ { | |||||
| var cmd *exec.Cmd | |||||
| if i == 0 { | |||||
| cmd = ExecuteNodeI(i, true, true, clusters) | |||||
| } else { | |||||
| cmd = ExecuteNodeI(i, false, true, clusters) | |||||
| } | |||||
| if cmd == nil { | |||||
| return | |||||
| } else { | |||||
| cmds = append(cmds, cmd) | |||||
| } | |||||
| } | |||||
| time.Sleep(time.Second) // 等待启动完毕 | |||||
| // client启动, 连接leader | |||||
| cWrite := clientPkg.Client{Address: clusters[0], ServerId: "1"} | |||||
| // 写入 | |||||
| var s clientPkg.Status | |||||
| for i := 0; i < 5; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||||
| s := cWrite.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||||
| if s != clientPkg.Ok { | |||||
| t.Errorf("write test fail") | |||||
| } | |||||
| } | |||||
| time.Sleep(time.Second) // 等待写入完毕 | |||||
| // 模拟最后一个结点崩溃 | |||||
| err := cmds[n - 1].Process.Signal(syscall.SIGTERM) | |||||
| if err != nil { | |||||
| fmt.Println("Error sending signal:", err) | |||||
| return | |||||
| } | |||||
| // 继续写入 | |||||
| for i := 5; i < 10; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||||
| s := cWrite.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||||
| if s != clientPkg.Ok { | |||||
| t.Errorf("write test fail") | |||||
| } | |||||
| } | |||||
| // 恢复结点 | |||||
| cmd := ExecuteNodeI(n - 1, false, false, clusters) | |||||
| if cmd == nil { | |||||
| t.Errorf("recover test1 fail") | |||||
| return | |||||
| } else { | |||||
| cmds[n - 1] = cmd | |||||
| } | |||||
| time.Sleep(time.Second) // 等待启动完毕 | |||||
| // client启动, 连接节点n-1(去读它的数据) | |||||
| cRead := clientPkg.Client{Address: clusters[n - 1], ServerId: "n"} | |||||
| // 读崩溃前写入数据 | |||||
| for i := 0; i < 5; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| var value string | |||||
| s = cRead.Read(key, &value) | |||||
| if s != clientPkg.Ok { | |||||
| t.Errorf("Read test1 fail") | |||||
| } | |||||
| } | |||||
| // 读未写入数据 | |||||
| for i := 5; i < 15; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| var value string | |||||
| s = cRead.Read(key, &value) | |||||
| if s != clientPkg.NotFound { | |||||
| t.Errorf("Read test2 fail") | |||||
| } | |||||
| } | |||||
| // 通知进程结束 | |||||
| for _, cmd := range cmds { | |||||
| err := cmd.Process.Signal(syscall.SIGTERM) | |||||
| if err != nil { | |||||
| fmt.Println("Error sending signal:", err) | |||||
| return | |||||
| } | |||||
| } | |||||
| } | |||||
+ 85
- 0
test/server_client_test.go
View File
| @ -0,0 +1,85 @@ | |||||
| package test | |||||
| import ( | |||||
| "fmt" | |||||
| "os/exec" | |||||
| "simple-kv-store/internal/client" | |||||
| "simple-kv-store/internal/nodes" | |||||
| "strconv" | |||||
| "syscall" | |||||
| "testing" | |||||
| "time" | |||||
| ) | |||||
| func TestServerClient(t *testing.T) { | |||||
| // 登记结点信息 | |||||
| n := 5 | |||||
| var clusters []string | |||||
| for i := 0; i < n; i++ { | |||||
| port := fmt.Sprintf("%d", uint16(9090)+uint16(i)) | |||||
| addr := "127.0.0.1:" + port | |||||
| clusters = append(clusters, addr) | |||||
| } | |||||
| // 结点启动 | |||||
| var cmds []*exec.Cmd | |||||
| for i := 0; i < n; i++ { | |||||
| var cmd *exec.Cmd | |||||
| if i == 0 { | |||||
| cmd = ExecuteNodeI(i, true, true, clusters) | |||||
| } else { | |||||
| cmd = ExecuteNodeI(i, false, true, clusters) | |||||
| } | |||||
| if cmd == nil { | |||||
| return | |||||
| } else { | |||||
| cmds = append(cmds, cmd) | |||||
| } | |||||
| } | |||||
| time.Sleep(time.Second) // 等待启动完毕 | |||||
| // client启动 | |||||
| c := clientPkg.Client{Address: "127.0.0.1:9090", ServerId: "1"} | |||||
| // 写入 | |||||
| var s clientPkg.Status | |||||
| for i := 0; i < 10; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||||
| s := c.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||||
| if s != clientPkg.Ok { | |||||
| t.Errorf("write test fail") | |||||
| } | |||||
| } | |||||
| // 读写入数据 | |||||
| for i := 0; i < 10; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| var value string | |||||
| s = c.Read(key, &value) | |||||
| if s != clientPkg.Ok && value != "hello" + key { | |||||
| t.Errorf("Read test1 fail") | |||||
| } | |||||
| } | |||||
| // 读未写入数据 | |||||
| for i := 10; i < 15; i++ { | |||||
| key := strconv.Itoa(i) | |||||
| var value string | |||||
| s = c.Read(key, &value) | |||||
| if s != clientPkg.NotFound { | |||||
| t.Errorf("Read test2 fail") | |||||
| } | |||||
| } | |||||
| // 通知进程结束 | |||||
| for _, cmd := range cmds { | |||||
| err := cmd.Process.Signal(syscall.SIGTERM) | |||||
| if err != nil { | |||||
| fmt.Println("Error sending signal:", err) | |||||
| return | |||||
| } | |||||
| } | |||||
| } | |||||