diff --git a/LEARNING_GUIDE.md b/LEARNING_GUIDE.md index b75f495..6ca1450 100644 --- a/LEARNING_GUIDE.md +++ b/LEARNING_GUIDE.md @@ -1,73 +1,103 @@ -# πŸ“˜ KVStore Learning Guide & Implementation Checklist +# πŸ“˜ KVStore Learning Guide & Implementation Workbook -So far, you've done an incredible job laying the foundation. We have a pure memory storage engine, a binary TCP protocol, a persistence layer, an HTTP API, and a beautiful Next.js frontend! πŸš€ +Welcome to your learning guide for the KVStore project! The primary goal of this project is not just to have a working database, but to **make you a better systems programmer**. -Right now, you are tackling **Phase 7: Raft Consensus**. Raft is one of the most rewarding algorithms to build, but also one of the easiest places to introduce subtle bugs. - -Based on our analysis, here is the exact state of your project, what you left out, and how to approach finishing itβ€”**without me writing the code for you**. +This document serves as your interactive workbook. Instead of copy-pasting code, use this guide to understand *what* to build, *why* it matters, and *how* to approach it. --- ## 🎯 Current Project State -- **Phase 1-6:** βœ… Largely Complete. The core store, TCP networking, and API wrappers are well-defined. -- **Phase 7 (Raft):** βœ… Complete. You have implemented robust leader election, replication, and proxying. -- **Phase 8 (Observability):** 🚧 In Progress. This is your next focus area! +- **Phases 1-6:** βœ… **Complete.** Core storage, TCP, persistence, API, and UI are built! +- **Phase 7:** βœ… **Complete.** Raft consensus (leader election, replication) is working. +- **Phase 8:** 🚧 **In Progress.** Observability and Benchmarks. This is your current focus! --- -## βœ… The "Left Out Stuff" Checklist - -Here is exactly what remains for you to build to finish this system: +## πŸš€ Phase 8: Observability and Benchmarks (Your Next Challenge) -### πŸ› οΈ Raft Consensus (Weeks 10-13) -- [x] **Election Timer Loop:** A background goroutine that constantly ticks and triggers an election if a heartbeat hasn't been received in 150-300ms. -- [x] **RPC Structures:** Define `RequestVoteArgs`, `RequestVoteReply` and `AppendEntriesArgs`, `AppendEntriesReply`. -- [x] **Candidate Logic (`startElection`):** Transitioning to candidate, incrementing term, voting for self, and sending `RequestVote` RPCs in parallel. -- [x] **Follower Logic (`RequestVote` Handler):** Evaluating inbound votes. Granting a vote if the candidate's term is newer and their log is at least as up-to-date. -- [x] **Leader Logic (`AppendEntries` Heartbeat):** Sending empty log entries every 50ms so followers don't trigger new elections. -- [x] **Log Replication:** Leader accepts commands from clients, appends to its own log, and broadcasts them via `AppendEntries`. -- [x] **Commit Index Management:** Advancing the `commitIndex` once a majority of nodes acknowledge a log entry. -- [x] **State Machine Application:** A background loop (`applyCommitted()`) that reads from the `commitIndex` and physically executes those commands on the KV `Store`. -- [x] **Leader Write Proxy:** Ensuring followers who receive arbitrary writes proxy them safely over to the active leader. +### The Goal +A production-grade database is flying blind without metrics. In this phase, you will write a Prometheus integration to track performance and a benchmarking tool to prove your system can handle the load. -### πŸ“Š Observability (Week 14) -- [ ] **Prometheus Metrics:** Tracking command counts, latencies, etc. -- [ ] **Load Testing / Benchmarking (`bench/bench.go`):** Validating the >100K ops/sec throughput target. +**Rule of Thumb:** Do not copy code blindly. Read the hint provided, try implementing it in the file yourself, and then test. --- -## 🧭 Your Path Forward (Teacher's Guide) - -Since my goal is to help you become a better programmer, I won’t write this code for you. Instead, let me act as your pair-programming instructor. - -### First Step: The Election Timer -You are currently inside `internal/raft/election.go` and `node.go`. Your very first job is to ensure a node doesn't stay a `Follower` forever. - -**Stop and think about these questions:** -1. *In `node.go`, we have `electionTimeout` and `electionResetAt`. If a node boots up as a Follower, how does it know when to become a Candidate?* -2. *Raft demands "randomized election timeouts" (e.g., 150ms to 300ms). Why? What catastrophe would happen if all 3 nodes had an exact 150ms timeout?* - -**Implementation Direction:** -Create a method `runElectionTimer()` that runs inside a goroutine as soon as `New(...)` is called. It should wake up every few milliseconds, check the current time against `node.electionResetAt` + `node.electionTimeout`. If elapsed, call a `startElection()` method! - -### Second Step: The Vote Request -Once your timer fires, your node is a candidate. It must ask for votes. - -**Questions:** -1. *Before you ask for votes from peers, what two state properties must you change locally? (Hint: See section 5.2 of the Raft paper).* -2. *If a node receives a `RequestVote`, under what exact scenarios must it reject the vote AND under what scenarios must it accept?* +### Step 1: Prepare Your Workspace +Before you write logic, set up the standard library tools used by the industry. +- **Task:** Add the Prometheus Go client to your project. +- **Command to run in `server/`:** + ```bash + go get github.com/prometheus/client_golang/prometheus + go get github.com/prometheus/client_golang/prometheus/promhttp + ``` + +### Step 2: Define the Metrics (`server/internal/metrics/metrics.go`) +You need a dedicated package where all your metrics are strongly typed and globally accessible. + +- **Task:** Create `metrics.go` inside the `metrics` package and define three variables using the `prometheus` module. + - **1. CommandsTotal (CounterVec):** Name it `kvstore_commands_total`. Include a label named `"command"`. + - **2. KeysTotal (Gauge):** Name it `kvstore_keys_total`. No labels required. + - **3. CommandDurationSeconds (HistogramVec):** Name it `kvstore_command_duration_seconds`. Use buckets: `.0001, .0005, .001, .005, .01, .05, .1`. Include a label named `"command"`. +- **Task:** Write a single `func Register()` in this file that calls `prometheus.MustRegister` with all three of your metrics. +- πŸ’‘ **Pedagogical Hint:** Why use a *Vector* for counters and histograms? Because it allows us to group latencies and counts dynamically by command type (e.g., SET vs GET), giving us rich insights in Grafana! + +### Step 3: Instrument the TCP Server (`server/internal/server/handler.go`) +Your database needs to record metrics exactly when it does work. + +- **Task:** Open `executeCommand`. Record the start time at the very top using `time.Now()`. +- **Task:** Use an anonymous `defer` function right after the start time. Inside that defer: + 1. Record duration in the histogram using `time.Since(...)`. + 2. Increment the counter. +- πŸ’‘ **Pedagogical Hint:** Why use `defer` here instead of writing it at the bottom of the function? Notice how many `return` statements exist in your big `switch` statement. A `defer` guarantees execution no matter which path the code takes to return, preventing dropped metrics! + +### Step 4: Instrument the Raft State Machine (`server/internal/raft/node.go`) +Your TCP server handles single-node direct connections, but what about Raft-replicated operations across the cluster? + +- **Task:** Navigate to `applyCommitted()`. This is where all committed Raft logs actually hit the store. +- **Task:** Apply the exact same pattern as Step 3 for any command processed here (incrementing the counter). +- **Task:** Every time a batch of commands is applied in `applyCommitted`, update your `KeysTotal` gauge with the current total count from the store. +- πŸ’‘ **Pedagogical Hint:** Why measure here? Because followers never execute logic in `executeCommand`β€”they only get data via Raft. If you only instrumented the TCP layer, you'd be missing metrics on all your Raft followers! + +### Step 5: Start the Metrics Engine +Now that metrics are being tracked in memory, expose them so a Prometheus scraper can fetch them. + +- **Task:** In `api/server.go` (where your HTTP routes are mounted), mount the prometheus handler: `r.Handle("/metrics", promhttp.Handler())`. +- **Task:** Open `cmd/server/main.go`. Before your server starts running, call your `metrics.Register()` function exactly once to wire up your custom metrics to the registry. + +### Step 6: Expose Pprof (Go Profiling) +To find out why a Go program is slow, we use `pprof`. +- **Task:** In `cmd/server/main.go`, add this specific, blank import: + ```go + import _ "net/http/pprof" + ``` +- πŸ’‘ **Pedagogical Hint:** The `_` (blank identifier) runs the package's `init()` function without you needing to call its methods. For `pprof`, its `init()` automatically injects CPU/Memory profiling endpoints into the default HTTP mux. + +### Step 7: Build the Benchmark Tool (`bench/bench.go`) +You claim this database is fast. Prove it. + +- **Task:** Create a standalone Go program (with its own `main` package) in a new `bench/bench.go` file. +- **Task:** Setup command line flags: `--connections` (default 50), `--duration` (default 10s), etc. +- **Task:** Spawn N goroutines. Each one should establish *one* continuous TCP connection. In a `for` loop, each goroutine should pick a random string, send a `SET` command, then `GET` the same key. +- **Task:** Keep a global atomic counter of operations. Wait until `--duration` ends, then divide the counter by the seconds to print `ops/sec`. +- πŸ’‘ **Pedagogical Hint:** In benchmarks, locks (Mutexes) can be a bottleneck. Track your total operations using `sync/atomic` (like `atomic.AddInt64`). This uses lock-free hardware instructions and ensures your benchmark tool isn't slowing down your test! Also make sure your client uses one persistent connection rather than reconnecting per operation, otherwise you end up benchmarking the OS's handshake speed rather than your KV-store logic. --- -## πŸ“š Essential Reading +### Step 8: Visualize with Grafana & Prometheus (Bonus Challenge) +Raw text metrics at `/metrics` are great, but industry standards use dashboards for real-time monitoring. -Do not write code for Raft without having the answers immediately available. Distribute systems rely purely on getting these edge cases right. - -1. **[The Secret Lives of Data (Raft Visualization)](https://thesecretlivesofdata.com/raft/)**: Watch this animation completely. It's the best summary of everything you're about to write. -2. **[The Raft Paper](https://raft.github.io/raft.pdf)**: Read **Sections 5.1 through 5.4**. You do not need to read the whole paper, but these 4 pages are your "source of truth". They literally define the `if/else` statements you will write in Go. -3. Your local `docs/RAFT.md` file. It breaks down the architecture specifically mapped to *this* KV Store project's types! +- **Task:** Update your `docker-compose.yml` to include the official `prom/prometheus` and `grafana/grafana` images. +- **Task:** Create a `prometheus.yml` config file and mount it into your Prometheus container. Configure it to scrape all 3 of your KVStore Raft nodes (`node1:8080`, `node2:8080`, `node3:8080`). +- **Task:** Boot up Grafana on port `3001` (to avoid conflicting with Next.js), connect Prometheus as a data source, and build a beautiful dashboard showing your `kvstore_command_duration_seconds` heatmap and `kvstore_commands_total` rates! +- πŸ’‘ **Pedagogical Hint:** Using rate functions in Grafana like `rate(kvstore_commands_total[1m])` converts your raw, ever-growing counter into a clean "Operations per Second" graph! --- -Whenever you're ready, take a stab at creating the `runElectionTimer` mechanism or setting up the RPC structures. Let me know what you try, and I will be right here to review it and guide your logic! +### βœ… Phase 8 Definition of Done Check +- Can you `curl localhost:8080/metrics` and see raw Prometheus text output? +- Are the counters accumulating correctly when you run operations via the CLI? +- Does your `bench.go` run successfully and output a >100,000 throughput rate? +- **(Bonus)** Do you have a vibrant Grafana dashboard rendering your Prometheus data in real-time? + +Whenever you get stuck on an implementation detail for these steps, think about the Go primitives you've learned. Good luck, and start writing code! diff --git a/server/bench/bench.go b/server/bench/bench.go new file mode 100644 index 0000000..7278281 --- /dev/null +++ b/server/bench/bench.go @@ -0,0 +1,87 @@ +package main + +import ( + "flag" + "fmt" + "math/rand" + "net" + "sync" + "sync/atomic" + "time" + + "github.com/ARCoder181105/kvstore/internal/protocol" +) + +func main() { + conns := flag.Int("connections", 50, "Number of concurrent connections") + duration := flag.Duration("duration", 10*time.Second, "Duration of the benchmark") + host := flag.String("host", "localhost:6379", "TCP server address") + flag.Parse() + + fmt.Printf("Starting benchmark: %d connections for %v on %s\n", *conns, *duration, *host) + + var opsCount atomic.Uint64 + + startTime := time.Now() + endTime := startTime.Add(*duration) + + var wg sync.WaitGroup + + for i := 0; i < *conns; i++ { + wg.Add(1) + go func(workerID int) { + defer wg.Done() + + conn, err := net.Dial("tcp", *host) + if err != nil { + fmt.Printf("Worker %d failed to connect: %v\n", workerID, err) + return + } + defer conn.Close() + + val := []byte("benchmark_value") + + for time.Now().Before(endTime) { + // Randomize key so we stress different parts of the map + key := fmt.Sprintf("bench:%d", rand.Intn(10000)) + + // SET + if err := protocol.WriteCommand(conn, &protocol.Command{ + ID: protocol.CmdSet, + Key: key, + Value: val, + }); err != nil { + break // connection broken β€” exit this worker + } + if _, err := protocol.ReadResponse(conn); err != nil { + break + } + opsCount.Add(1) + + // GET + if err := protocol.WriteCommand(conn, &protocol.Command{ + ID: protocol.CmdGet, + Key: key, + }); err != nil { + break + } + if _, err := protocol.ReadResponse(conn); err != nil { + break + } + opsCount.Add(1) + } + }(i) + } + + wg.Wait() + + actualDuration := time.Since(startTime) + totalOps := opsCount.Load() + opsPerSec := float64(totalOps) / actualDuration.Seconds() + + fmt.Println("=====================================") + fmt.Printf("Total Operations : %d\n", totalOps) + fmt.Printf("Time Elapsed : %.2f seconds\n", actualDuration.Seconds()) + fmt.Printf("Throughput : %.0f ops/sec\n", opsPerSec) + fmt.Println("=====================================") +} diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 9df26c7..7bb1825 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -10,6 +10,7 @@ import ( "time" "github.com/ARCoder181105/kvstore/internal/api" + "github.com/ARCoder181105/kvstore/internal/metrics" aof "github.com/ARCoder181105/kvstore/internal/persistence" "github.com/ARCoder181105/kvstore/internal/raft" "github.com/ARCoder181105/kvstore/internal/server" @@ -24,6 +25,7 @@ func getEnv(key, fallback string) string { } func main() { + metrics.Register() nodeID := getEnv("NODE_ID", "node1") httpAddr := getEnv("HTTP_ADDR", ":8080") tcpAddr := getEnv("TCP_ADDR", ":6379") diff --git a/server/go.mod b/server/go.mod index aac6de2..4496bca 100644 --- a/server/go.mod +++ b/server/go.mod @@ -1,6 +1,6 @@ module github.com/ARCoder181105/kvstore -go 1.22.2 +go 1.23.0 require ( github.com/chzyer/readline v1.5.1 @@ -11,7 +11,16 @@ require ( ) require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_golang v1.23.2 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect github.com/spf13/pflag v1.0.10 // indirect - golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect + golang.org/x/sys v0.35.0 // indirect + google.golang.org/protobuf v1.36.8 // indirect ) diff --git a/server/go.sum b/server/go.sum index 5ce3d60..4019e50 100644 --- a/server/go.sum +++ b/server/go.sum @@ -1,3 +1,7 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/chzyer/logex v1.2.1 h1:XHDu3E6q+gdHgsdTPH6ImJMIp436vR6MPtH8gP05QzM= github.com/chzyer/logex v1.2.1/go.mod h1:JLbx6lG2kDbNRFnfkgvh4eRJRPX1QCoOIWomwysCBrQ= github.com/chzyer/readline v1.5.1 h1:upd/6fQk4src78LMRzh5vItIt361/o4uq553V8B5sGI= @@ -13,13 +17,29 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5 h1:y/woIyUBFbpQGKS0u1aHF/40WUDnek3fPOyD08H5Vng= golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/server/internal/api/server.go b/server/internal/api/server.go index 346832f..4d0e9b0 100644 --- a/server/internal/api/server.go +++ b/server/internal/api/server.go @@ -4,12 +4,14 @@ import ( "context" "fmt" "net/http" + "net/http/pprof" "time" "github.com/ARCoder181105/kvstore/internal/raft" "github.com/ARCoder181105/kvstore/internal/store" "github.com/go-chi/chi/v5" "github.com/go-chi/cors" + "github.com/prometheus/client_golang/prometheus/promhttp" ) type APIServer struct { @@ -42,6 +44,15 @@ func (s *APIServer) setupRoutes() { r.Post("/raft/requestvote", s.handleRequestVote) r.Post("/raft/appendentries", s.handleAppendEntries) + // pprof endpoints + r.HandleFunc("/debug/pprof/", pprof.Index) + r.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) + r.HandleFunc("/debug/pprof/profile", pprof.Profile) + r.HandleFunc("/debug/pprof/symbol", pprof.Symbol) + r.HandleFunc("/debug/pprof/trace", pprof.Trace) + + r.Handle("/metrics", promhttp.Handler()) + s.router = r } diff --git a/server/internal/metrics/metrics.go b/server/internal/metrics/metrics.go new file mode 100644 index 0000000..15bb504 --- /dev/null +++ b/server/internal/metrics/metrics.go @@ -0,0 +1,33 @@ +package metrics + +import "github.com/prometheus/client_golang/prometheus" + +var ( + CommandsTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "kvstore_commands_total", + Help: "Total number of commands processed.", + }, + []string{"command"}, + ) + + KeysTotal = prometheus.NewGauge( + prometheus.GaugeOpts{ + Name: "kvstore_keys_total", + Help: "Current number of keys in store.", + }, + ) + + CommandDurationSeconds = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "kvstore_command_duration_seconds", + Help: "Latency of each command in seconds.", + Buckets: []float64{0.0001, 0.0005, 0.001, 0.005, 0.01, 0.05, 0.1}, + }, + []string{"command"}, + ) +) + +func Register() { + prometheus.MustRegister(CommandsTotal, KeysTotal, CommandDurationSeconds) +} diff --git a/server/internal/raft/election.go b/server/internal/raft/election.go index f9ea93b..cb4b55d 100644 --- a/server/internal/raft/election.go +++ b/server/internal/raft/election.go @@ -27,6 +27,7 @@ func (r *RaftNode) startElection() { r.currentTerm += 1 r.state = Candidate r.votedFor = r.id + r.electionResetAt = time.Now() term := r.currentTerm lastIndex := r.getLastIndex() @@ -110,6 +111,12 @@ func (r *RaftNode) advanceCommitIndex() { } if matches > (len(r.peers)+1)/2 { r.commitIndex = n + + select { + case r.commitReady <- struct{}{}: + default: + } + return } } @@ -133,7 +140,7 @@ func (r *RaftNode) runHeartbeatLoop() { for peerID, peerURL := range r.peers { nextIdx := r.nextIndex[peerID] prevLogIndex := nextIdx - 1 - + // getEntry handles out of bounds safely var prevLogTerm uint64 if prevLogIndex < uint64(len(r.log)) { @@ -190,7 +197,7 @@ func (r *RaftNode) runHeartbeatLoop() { } }(peerID, peerURL, nextIdx, prevLogIndex, prevLogTerm, entries) } - + // Advance commit index for the leader itself. // This is critical for single-node deployments where the peer loop // above never executes, so commitIndex would never advance otherwise. diff --git a/server/internal/raft/node.go b/server/internal/raft/node.go index e7dbea0..a0d4e79 100644 --- a/server/internal/raft/node.go +++ b/server/internal/raft/node.go @@ -6,6 +6,7 @@ import ( "sync" "time" + "github.com/ARCoder181105/kvstore/internal/metrics" aof "github.com/ARCoder181105/kvstore/internal/persistence" "github.com/ARCoder181105/kvstore/internal/protocol" "github.com/ARCoder181105/kvstore/internal/store" @@ -67,8 +68,9 @@ type RaftNode struct { // pending client requests (index -> response channel) pending map[uint64]chan interface{} - store *store.Store - aofWriter *aof.AOFWriter + store *store.Store + aofWriter *aof.AOFWriter + commitReady chan struct{} } func New(id NodeID, peers map[NodeID]string, store *store.Store, aofWriter *aof.AOFWriter) *RaftNode { @@ -86,6 +88,7 @@ func New(id NodeID, peers map[NodeID]string, store *store.Store, aofWriter *aof. aofWriter: aofWriter, electionResetAt: time.Now(), electionTimeout: time.Duration(150+rand.Intn(150)) * time.Millisecond, + commitReady: make(chan struct{}, 1), } go r.runElectionTimer() @@ -112,7 +115,7 @@ func (r *RaftNode) Submit(cmd protocol.Command) (interface{}, error) { index := uint64(len(r.log)) term := r.currentTerm - + waitCh := make(chan interface{}, 1) r.pendingMu.Lock() r.pending[index] = waitCh @@ -129,15 +132,24 @@ func (r *RaftNode) Submit(cmd protocol.Command) (interface{}, error) { case res := <-waitCh: return res, nil case <-time.After(2 * time.Second): + r.pendingMu.Lock() + delete(r.pending, index) + r.pendingMu.Unlock() return nil, fmt.Errorf("raft submit timeout") } } func (r *RaftNode) applyCommitted() { for { - time.Sleep(10 * time.Millisecond) - r.mu.Lock() + + // If there is nothing to apply, unlock and wait for a signal + for r.commitIndex <= r.lastApplied { + r.mu.Unlock() + <-r.commitReady // Go scheduler puts this goroutine to sleep here! + r.mu.Lock() + } + var commandsToApply []LogEntry for r.commitIndex > r.lastApplied { r.lastApplied++ @@ -148,8 +160,10 @@ func (r *RaftNode) applyCommitted() { for _, entry := range commandsToApply { cmd := entry.Command - if cmd.ID == protocol.CmdSet { + switch cmd.ID { + case protocol.CmdSet: r.store.Set(cmd.Key, cmd.Value, cmd.TTL) + metrics.CommandsTotal.WithLabelValues(raftCommandName(cmd.ID)).Inc() if r.aofWriter != nil { var expiresAt int64 if cmd.TTL > 0 { @@ -163,8 +177,10 @@ func (r *RaftNode) applyCommitted() { ExpiresAt: expiresAt, }) } - } else if cmd.ID == protocol.CmdDel { + + case protocol.CmdDel: r.store.Delete(cmd.Key) + metrics.CommandsTotal.WithLabelValues(raftCommandName(cmd.ID)).Inc() if r.aofWriter != nil { r.aofWriter.Append(aof.AOFEntry{ Timestamp: time.Now().UnixNano(), @@ -172,8 +188,10 @@ func (r *RaftNode) applyCommitted() { Key: cmd.Key, }) } - } else if cmd.ID == protocol.CmdExpire { + + case protocol.CmdExpire: r.store.Expire(cmd.Key, cmd.TTL) + metrics.CommandsTotal.WithLabelValues(raftCommandName(cmd.ID)).Inc() if r.aofWriter != nil { var expiresAt int64 if cmd.TTL > 0 { @@ -196,6 +214,11 @@ func (r *RaftNode) applyCommitted() { } r.pendingMu.Unlock() } + + // Update key count gauge once after the whole batch + if r.store != nil { + metrics.KeysTotal.Set(float64(r.store.Count())) + } } } @@ -206,3 +229,15 @@ func (r *RaftNode) GetPeerURL(id NodeID) (string, bool) { return url, ok } +func raftCommandName(id byte) string { + switch id { + case protocol.CmdSet: + return "set" + case protocol.CmdDel: + return "del" + case protocol.CmdExpire: + return "expire" + default: + return "unknown" + } +} diff --git a/server/internal/raft/rpc.go b/server/internal/raft/rpc.go index 9dc29ac..e56fcec 100644 --- a/server/internal/raft/rpc.go +++ b/server/internal/raft/rpc.go @@ -144,6 +144,11 @@ func (r *RaftNode) AppendEntries(args AppendEntriesArgs) AppendEntriesReply { } else { r.commitIndex = lastIndex } + + select { + case r.commitReady <- struct{}{}: + default: + } } return AppendEntriesReply{ diff --git a/server/internal/server/handler.go b/server/internal/server/handler.go index 7403798..0055e6b 100644 --- a/server/internal/server/handler.go +++ b/server/internal/server/handler.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/ARCoder181105/kvstore/internal/metrics" aof "github.com/ARCoder181105/kvstore/internal/persistence" "github.com/ARCoder181105/kvstore/internal/protocol" ) @@ -33,6 +34,41 @@ func (s *Server) handleConn(conn net.Conn) { } func (s *Server) executeCommand(cmd *protocol.Command) *protocol.Response { + start := time.Now() + commandName := func(id byte) string { + switch id { + case protocol.CmdSet: + return "set" + case protocol.CmdGet: + return "get" + case protocol.CmdDel: + return "del" + case protocol.CmdExpire: + return "expire" + case protocol.CmdTTL: + return "ttl" + case protocol.CmdKeys: + return "keys" + case protocol.CmdIncr: + return "incr" + case protocol.CmdMSet: + return "mset" + case protocol.CmdMGet: + return "mget" + case protocol.CmdPing: + return "ping" + default: + return "unknown" + } + } + + cmdLabel := commandName(cmd.ID) + + defer func() { + metrics.CommandDurationSeconds.WithLabelValues(cmdLabel).Observe(time.Since(start).Seconds()) + metrics.CommandsTotal.WithLabelValues(cmdLabel).Inc() + }() + switch cmd.ID { case protocol.CmdSet: diff --git a/server/internal/store/store.go b/server/internal/store/store.go index d421fc1..bf8ee99 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -396,6 +396,10 @@ func (s *Store) publish(event Event) { select { case s.subscribers[i] <- event: default: + // LOAD SHEDDING: If a subscriber (like a WebSocket client) is reading + // too slowly and their buffer fills up, we drop the event. + // This prevents a single slow client from blocking the database's + // write-path, as publish() is called synchronously during Set/Del. } }