diff --git a/Dockerfile b/Dockerfile index b8b155a..a800e77 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,7 +4,7 @@ WORKDIR /app COPY go.mod go.sum ./ RUN go mod download COPY . . -RUN CGO_ENABLED=0 go build -o zeno main.go +RUN CGO_ENABLED=0 go build -o zeno ./cmd/zeno FROM alpine:3.21 RUN apk add --no-cache redis diff --git a/bench/main.go b/bench/main.go index 5b38839..c369d43 100644 --- a/bench/main.go +++ b/bench/main.go @@ -3,7 +3,7 @@ // at the server, and reports throughput and latency percentiles. // // Because every client command is forwarded to the leader over gRPC (see -// nodes.SendCommand), the target address must be a reachable *leader*. In +// controlplane.SendCommand), the target address must be a reachable *leader*. In // practice that is the a1 node from docker-compose.local.yml, which publishes // 6379/6380 to the host: // diff --git a/main.go b/cmd/zeno/main.go similarity index 85% rename from main.go rename to cmd/zeno/main.go index 2096a67..a61e7fc 100644 --- a/main.go +++ b/cmd/zeno/main.go @@ -10,7 +10,8 @@ import ( "time" "zeno/pb" "zeno/src/aof" - "zeno/src/nodes" + "zeno/src/controlplane" + "zeno/src/node" "zeno/src/resp" "zeno/src/utils" "zeno/src/writer" @@ -44,7 +45,7 @@ func (s *taskServer) SendTask(ctx context.Context, req *pb.SendTaskRequest) (*pb // Client -> TCP Request -> RESP deserialze -> commands hander -> RESP serialze -> Response func main() { - n, err := nodes.MakeNodes() + n, err := controlplane.New() messageQueue := utils.NewQueue[utils.Message]() if err != nil { panic("Failed to make nodes") @@ -59,7 +60,7 @@ func main() { } grpcServer := grpc.NewServer() pb.RegisterTaskServiceServer(grpcServer, &taskServer{}) - pb.RegisterNodeServiceServer(grpcServer, &nodes.NodeServer{}) + pb.RegisterNodeServiceServer(grpcServer, &node.NodeServer{}) reflection.Register(grpcServer) slog.Info("gRPC server running", "port", 6380) if err := grpcServer.Serve(lis); err != nil { @@ -81,7 +82,13 @@ func main() { utils.RunOnInterval(30*time.Second, heartbeatStopChan, n.SendHeartbeat, func(failedNodes []string) { if len(failedNodes) > 0 { slog.Warn("nodes failed heartbeat", "nodes", failedNodes) - // _ = n.RestartNodes(failedNodes) + } + // Diff this round against known state; any node that was down and + // is now responding gets caught up from the replication queue. + for _, node := range n.ReconcileHealth(failedNodes) { + if err := n.RestartNode(node, messageQueue); err != nil { + slog.Error("failed to resync recovered node", "node", node, "error", err) + } } }) } @@ -113,7 +120,7 @@ func main() { } } -func handleConnection(conn net.Conn, _ *aof.Aof, n *nodes.Nodes, q *utils.Queue[utils.Message]) { +func handleConnection(conn net.Conn, _ *aof.Aof, n *controlplane.Cluster, q *utils.Queue[utils.Message]) { defer conn.Close() for { response := resp.NewResp(conn) diff --git a/makefile b/makefile index 80d6e47..4579ab1 100644 --- a/makefile +++ b/makefile @@ -7,9 +7,9 @@ proto: protoc --go_out=. --go_opt=module=zeno --go-grpc_out=. --go-grpc_opt=module=zeno proto/*.proto build: - GOARCH=amd64 GOOS=darwin go build -o ${APP_EXECUTABLE}-darwin main.go - GOARCH=amd64 GOOS=linux go build -o ${APP_EXECUTABLE}-linux main.go - GOARCH=amd64 GOOS=windows go build -o ${APP_EXECUTABLE}-windows main.go + GOARCH=amd64 GOOS=darwin go build -o ${APP_EXECUTABLE}-darwin ./cmd/zeno + GOARCH=amd64 GOOS=linux go build -o ${APP_EXECUTABLE}-linux ./cmd/zeno + GOARCH=amd64 GOOS=windows go build -o ${APP_EXECUTABLE}-windows ./cmd/zeno run: build ./${APP_EXECUTABLE}-darwin diff --git a/src/nodes/client.go b/src/controlplane/client.go similarity index 98% rename from src/nodes/client.go rename to src/controlplane/client.go index b73a4f9..42092e1 100644 --- a/src/nodes/client.go +++ b/src/controlplane/client.go @@ -1,4 +1,4 @@ -package nodes +package controlplane import ( "context" diff --git a/src/controlplane/cluster.go b/src/controlplane/cluster.go new file mode 100644 index 0000000..b436a40 --- /dev/null +++ b/src/controlplane/cluster.go @@ -0,0 +1,259 @@ +package controlplane + +import ( + "errors" + "log/slog" + "net" + "sort" + "sync" + "zeno/src/resp" + "zeno/src/utils" + + "zeno/pb" +) + +// Cluster holds cluster membership and coordination policy: who the leader +// is, which peers exist, and whether *this* process should be acting as +// coordinator. This is the control-plane view of the cluster. Outbound RPC +// mechanics live in peerClient (client.go); the data-plane server that +// applies commands lives in the node package (node/server.go). + +const ( + NoReadOrWriteState = 0 + ReadState = 1 + WriteState = 2 + ReadAndWriteState = 3 + DownState = 4 +) + +type Cluster struct { + leader string // ip + nodes []string // list of node ips (excludes the leader) + ipToContainer map[string]string + nodeState map[string]int // ip: node state (0 - no read or write, 1 - read, 2 - write, 3 - read and write (normal state), 4 - node is down) + rrCounter uint64 // round-robin cursor for read routing + mu sync.Mutex +} + +// hard coded for now but set a1 to leader +func New() (*Cluster, error) { + nodes := &Cluster{ + leader: "10.10.1.10", + nodes: []string{"10.10.1.11", "10.10.2.10", "10.10.2.11"}, + ipToContainer: map[string]string{ + "10.10.1.10": "zeno-a1", + "10.10.1.11": "zeno-a2", + "10.10.2.10": "zeno-b1", + "10.10.2.11": "zeno-b2", + }, + nodeState: map[string]int{ + "10.10.1.10": ReadAndWriteState, + "10.10.1.11": ReadAndWriteState, + "10.10.2.10": WriteState, + "10.10.2.11": ReadState, + }, + mu: sync.Mutex{}, + } + + return nodes, nil +} + +// IsLeader reports whether this process is the leader, by checking whether +// the leader ip is bound to one of our local interfaces. Only the leader +// should drive heartbeats/coordination — otherwise every node pings every +// other node and heartbeats multiply by the cluster size. +func (n *Cluster) IsLeader() bool { + addrs, err := net.InterfaceAddrs() + if err != nil { + slog.Error("failed to read local interface addresses", "error", err) + return false + } + for _, a := range addrs { + if ipnet, ok := a.(*net.IPNet); ok && ipnet.IP.String() == n.leader { + return true + } + } + return false +} + +// ReconcileHealth diffs this round's heartbeat failures against the last +// known state and returns the nodes that just recovered (were down, now +// responding). Newly failed nodes are recorded as down. This is what turns a +// stateless per-round heartbeat into up/down *transition* detection. +func (n *Cluster) ReconcileHealth(failedNodes []string) (recovered []string) { + n.mu.Lock() + defer n.mu.Unlock() + + failedSet := make(map[string]bool, len(failedNodes)) + for _, ip := range failedNodes { + failedSet[ip] = true + if n.nodeState[ip] != DownState { + slog.Warn("node went down", "node", ip) + n.nodeState[ip] = DownState + } + } + + for ip := range n.nodeState { + if !failedSet[ip] { // was down, responded this round + slog.Info("node recovered", "node", ip) + recovered = append(recovered, ip) + n.nodeState[ip] = ReadAndWriteState + } + } + return recovered +} + +// RestartNodes takes a list of node ips (strings) and will wait for these nodes to come back to life. +// If node B goes down RestartNode is responsible for making sure node B is up to date when node B comes +// back up +func (n *Cluster) RestartNode(node string, queuedMessages *utils.Queue[utils.Message]) error { + // make sure node is in readonly + if n.nodeState[node] != 3 { + // send command to update node state to other + slog.Info("node not in correct state", "nodeState", n.nodeState[node]) + } + + slog.Info("Starting restart node process", "queue_len", queuedMessages.QueueLength()) + for !queuedMessages.IsEmpty() { + item, _ := queuedMessages.Dequeue() + + res, err := newPeerClient(node).forwardCommand(item.Command, item.Args) + if err != nil { + slog.Error("failed to get response", "error", err) + return err + } + slog.Info("res from forwardcommand RestartNode()", "res", res) + } + + // TODO: we still need to sync from the main items, how do we ensure that the node stays up to date now. + // example if node cannot be written to by the main args, since its being sync'd do we need to queue the live feed again? + return nil + +} + +func (n *Cluster) SendHeartbeat() ([]string, error) { + // check leader node first + // TODO: handle leader failure later, do we vote a new leader? + result, err := newPeerClient(n.leader).heartbeat() + if err != nil { + return nil, err + } + if result.Response != "OK" { + slog.Info("leader node responded with failure", "error", result.Response) + return nil, errors.New("leader failed health check") + } + + var ( + mu sync.Mutex + wg sync.WaitGroup + failedNodes = []string{} + ) + for _, node := range n.nodes { + wg.Add(1) + go func(node string) { + defer wg.Done() + resp, err := newPeerClient(node).heartbeat() + slog.Info("heartbeat resp", "resp", resp) + if err != nil || resp.Response != "OK" { + mu.Lock() + failedNodes = append(failedNodes, node) + mu.Unlock() + } + }(node) + } + wg.Wait() + return failedNodes, nil +} + +// writeCommands classifies which verbs mutate state. Everything else is +// treated as a read and can be served by any read-capable replica. +var writeCommands = map[string]bool{ + "SET": true, + "HSET": true, +} + +func isWrite(command string) bool { + return writeCommands[command] +} + +// SendCommand is the single client-facing entry point: the client talks to +// the control plane and we decide where the command goes. Writes go to the +// leader (source of truth) and replicate to healthy followers; reads are +// served by any healthy read-capable replica. +func (n *Cluster) SendCommand(command string, args []resp.Value) (*pb.ForwardCommandResponse, error) { + if isWrite(command) { + return n.routeWrite(command, args) + } + return n.routeRead(command, args) +} + +// routeWrite applies the write on the leader, then asynchronously replicates +// it to every follower that is currently write-capable. The leader's result +// is what the client sees; replication is fire-and-forget. +func (n *Cluster) routeWrite(command string, args []resp.Value) (*pb.ForwardCommandResponse, error) { + result, err := newPeerClient(n.leader).forwardCommand(command, args) + if err != nil { + return nil, err + } + + for _, node := range n.nodes { + // skip leader node. + if node == n.leader { + continue + } + n.mu.Lock() + state := n.nodeState[node] + n.mu.Unlock() + + if state != ReadAndWriteState && state != WriteState { + slog.Info("skipping replication due to state", "node", n.ipToContainer[node], "state", state) + continue + } + + go func(node string) { + if _, err := newPeerClient(node).forwardCommand(command, args); err != nil { + slog.Error("replication failed", "node", node, "error", err) + } + }(node) + } + + return result, nil +} + +// routeRead picks one healthy read-capable node and serves the read from it. +func (n *Cluster) routeRead(command string, args []resp.Value) (*pb.ForwardCommandResponse, error) { + target := n.pickReadNode() + if target == "" { + return nil, errors.New("no read-capable node available") + } + return newPeerClient(target).forwardCommand(command, args) +} + +// pickReadNode round-robins across all read-capable nodes (leader included). +// Sorting gives a stable order so the round-robin counter is meaningful even +// though nodeState iteration order is random. +func (n *Cluster) pickReadNode() string { + n.mu.Lock() + defer n.mu.Unlock() + + candidates := make([]string, 0, len(n.nodeState)) + for ip, state := range n.nodeState { + if state == ReadAndWriteState || state == ReadState { + candidates = append(candidates, ip) + } + } + if len(candidates) == 0 { + return "" + } + sort.Strings(candidates) + + target := candidates[n.rrCounter%uint64(len(candidates))] + n.rrCounter++ + return target +} + +func (n *Cluster) printNodes() { + for _, v := range n.nodes { + slog.Info("node", "address", v) + } +} diff --git a/src/handler/handler_bench_test.go b/src/handler/handler_bench_test.go index feeda1c..f054b6e 100644 --- a/src/handler/handler_bench_test.go +++ b/src/handler/handler_bench_test.go @@ -152,7 +152,7 @@ func BenchmarkHGetAll(b *testing.B) { // BenchmarkCommandRoundTrip measures the full per-command CPU cost the server // pays for one request: parse the RESP array, dispatch to the handler, and // marshal the reply. It deliberately excludes the network and the gRPC -// leader-forwarding hop (see nodes.SendCommand) so it isolates command +// leader-forwarding hop (see controlplane.SendCommand) so it isolates command // processing. Use the load generator in ./bench for end-to-end numbers. func BenchmarkCommandRoundTrip(b *testing.B) { SETs = map[string]string{"foo": "bar"} diff --git a/src/nodes/node_server.go b/src/node/server.go similarity index 98% rename from src/nodes/node_server.go rename to src/node/server.go index 8ef9431..5627a9e 100644 --- a/src/nodes/node_server.go +++ b/src/node/server.go @@ -1,4 +1,4 @@ -package nodes +package node import ( "context" diff --git a/src/nodes/cluster.go b/src/nodes/cluster.go deleted file mode 100644 index 739d77c..0000000 --- a/src/nodes/cluster.go +++ /dev/null @@ -1,136 +0,0 @@ -package nodes - -import ( - "errors" - "log/slog" - "net" - "sync" - "zeno/src/resp" - - "zeno/pb" -) - -// Nodes holds cluster membership and coordination policy: who the leader -// is, which peers exist, and whether *this* process should be acting as -// coordinator. Outbound RPC mechanics live in peerClient (client.go); -// inbound RPC handling lives in NodeServer (node_server.go). -type Nodes struct { - leader string // ip - nodes []string // list of node ips (excludes the leader) - ipToContainer map[string]string -} - -// hard coded for now but set a1 to leader -func MakeNodes() (*Nodes, error) { - nodes := &Nodes{ - leader: "10.10.1.10", - nodes: []string{"10.10.1.11", "10.10.2.10", "10.10.2.11"}, - ipToContainer: map[string]string{ - "10.10.1.10": "zeno-a1", - "10.10.1.11": "zeno-a2", - "10.10.2.10": "zeno-b1", - "10.10.2.11": "zeno-b2", - }, - } - - return nodes, nil -} - -// IsLeader reports whether this process is the leader, by checking whether -// the leader ip is bound to one of our local interfaces. Only the leader -// should drive heartbeats/coordination — otherwise every node pings every -// other node and heartbeats multiply by the cluster size. -func (n *Nodes) IsLeader() bool { - addrs, err := net.InterfaceAddrs() - if err != nil { - slog.Error("failed to read local interface addresses", "error", err) - return false - } - for _, a := range addrs { - if ipnet, ok := a.(*net.IPNet); ok && ipnet.IP.String() == n.leader { - return true - } - } - return false -} - -// RestartNodes takes a list of node ips (strings) and will wait for these nodes to come back to life. -// If node B goes down RestartNode is responsible for making sure node B is up to date when node B comes -// back up -// func (n *Nodes) RestartNodes(nodes []string) []string { -// deadNodes := []string{} - -// for i := range len(nodes) { -// containerName, exists := n.ipToContainer[nodes[i]] -// if !exists { -// slog.Error("ip does not exist in container", "ip", nodes[i]) -// continue -// } - -// cmd := exec.Command("docker", "restart", containerName) -// _, err := cmd.Output() -// if err != nil { -// slog.Error("failed to restart container", "container", containerName, "error", err) -// deadNodes = append(deadNodes, nodes[i]) -// } -// } -// return deadNodes - -// } - -func (n *Nodes) SendHeartbeat() ([]string, error) { - // check leader node first - // TODO: handle leader failure later, do we vote a new leader? - result, err := newPeerClient(n.leader).heartbeat() - if err != nil { - return nil, err - } - if result.Response != "OK" { - slog.Info("leader node responded with failure", "error", result.Response) - return nil, errors.New("leader failed health check") - } - - var ( - mu sync.Mutex - wg sync.WaitGroup - failedNodes = []string{} - ) - for _, node := range n.nodes { - wg.Add(1) - go func(node string) { - defer wg.Done() - resp, err := newPeerClient(node).heartbeat() - slog.Info("heartbeat resp", "resp", resp) - if err != nil || resp.Response != "OK" { - mu.Lock() - failedNodes = append(failedNodes, node) - mu.Unlock() - } - }(node) - } - wg.Wait() - return failedNodes, nil -} - -func (n *Nodes) SendCommand(command string, args []resp.Value) (*pb.ForwardCommandResponse, error) { - // send to leader. - result, err := newPeerClient(n.leader).forwardCommand(command, args) - if err != nil { - return nil, err - } - - // fan out to other nodes - for _, node := range n.nodes { - go func(node string) { - newPeerClient(node).forwardCommand(command, args) - }(node) - } - - return result, nil -} - -func (n *Nodes) printNodes() { - for _, v := range n.nodes { - slog.Info("node", "address", v) - } -} diff --git a/src/utils/queue.go b/src/utils/queue.go index 54ea6a3..e3fe0bd 100644 --- a/src/utils/queue.go +++ b/src/utils/queue.go @@ -3,8 +3,8 @@ package utils import "zeno/src/resp" type Message struct { - command string - arg []resp.Value + Command string + Args []resp.Value } type Queue[T any] struct { @@ -13,8 +13,8 @@ type Queue[T any] struct { func NewMessage(command string, arg []resp.Value) *Message { return &Message{ - command: command, - arg: arg, + Command: command, + Args: arg, } }