Merge pull request #1 from van-sprundel/heartbeat

feat: add heartbeat to node, update in kubelet

Author: van-sprundel

.github/workflows/ci.yml

@@ -30,48 +30,8 @@ jobs:
   benchmark:
     name: run benchmarks
     runs-on: ubuntu-latest
+    if: github.event_name == 'push'
     steps:
       - uses: actions/checkout@v6
       - uses: actions/setup-go@v6
-
-      - name: Install benchstat
-        run: go install golang.org/x/perf/cmd/benchstat@latest
-
-      - name: Run benchmarks on PR branch
-        if: github.event_name == 'pull_request'
-        run: go test -run=^$ -bench=. -benchmem -benchtime=100ms -count=3 ./... 2>&1 | tee benchmark-pr.txt
-
-      - name: Run benchmarks on master
-        if: github.event_name == 'push'
-        run: go test -run=^$ -bench=. -benchmem -benchtime=100ms -count=3 ./...
-
-      - name: Checkout base branch
-        if: github.event_name == 'pull_request'
-        uses: actions/checkout@v4
-        with:
-          ref: ${{ github.base_ref }}
-          clean: false
-
-      - name: Run benchmarks on base branch
-        if: github.event_name == 'pull_request'
-        run: go test -run=^$ -bench=. -benchmem -benchtime=100ms -count=3 ./... 2>&1 | tee benchmark-base.txt
-
-      - name: Compare benchmarks
-        if: github.event_name == 'pull_request'
-        run: |
-          echo "## 📊 Benchmark Results" > benchmark_results.md
-          echo "" >> benchmark_results.md
-          echo "Comparing \`${{ github.head_ref }}\` (PR) vs \`${{ github.base_ref }}\` (base)" >> benchmark_results.md
-          echo "" >> benchmark_results.md
-          echo '```' >> benchmark_results.md
-          benchstat benchmark-base.txt benchmark-pr.txt >> benchmark_results.md 2>&1 || echo "No comparable benchmarks found" >> benchmark_results.md
-          echo '```' >> benchmark_results.md
-          echo "" >> benchmark_results.md
-          echo "<sub>📏 Using Go's benchstat for statistical comparison. Lower is better for time/op and B/op.</sub>" >> benchmark_results.md
-
-      - name: Comment PR with results
-        if: github.event_name == 'pull_request'
-        uses: thollander/actions-comment-pull-request@v2
-        with:
-          filePath: benchmark_results.md
-          comment_tag: benchmark_results
+      - run: go test -run=^$ -bench=. -benchmem ./...

.pre-commit-config.yaml

@@ -0,0 +1,9 @@
+repos:
+  - repo: local
+    hooks:
+      - id: golangci-lint
+        name: golangci-lint
+        entry: golangci-lint run --enable errcheck
+        types: [go]
+        language: system
+        pass_filenames: false

README.md

@@ -49,7 +49,6 @@ Very nice!
 
 # TODOS
 
-- [ ] Mark nodes as `NotReady` if kubelet stops responding (will require healthchecks)
 - [ ] Round-robin for scheduler
 - [ ] Pod deletion cleanup (stop containers if the pods are deleted via API)
 - [ ] etcd (k/v store for cluster state so restarts persist)
@@ -70,6 +69,17 @@ TODO
 | Running    | exited          | Update pod to Failed                         |
 | Running    | doesn't exist   | Weird state - maybe re-create or mark Failed |
 
+```
+           (heartbeat)
+  Kubelet ─────────────>  NodeStore
+                              |
+                              V
+                          NodeController ("is heartbeat stale?")
+                              |
+                              V
+                          NotReady if stale
+```
+
 # Disclaimer
 
 This project is purely for my own education. That means **no** LLM's, which also means it's not going to be production-ready code. ~~The Pod spec is purposefully simple (name, img, state) because I do not need anything else for my goals.~~ Turns out that was a lie

cmd/miniku/main.go

@@ -27,15 +27,19 @@ func main() {
 
 	// reconcile pods -> containers (one per node)
 	rt := &runtime.DockerCLIRuntime{}
-	kubelet1 := kubelet.New(podStore, rt, "node-1")
-	kubelet2 := kubelet.New(podStore, rt, "node-2")
+	kubelet1 := kubelet.New(podStore, nodeStore, rt, "node-1")
+	kubelet2 := kubelet.New(podStore, nodeStore, rt, "node-2")
 	go kubelet1.Run()
 	go kubelet2.Run()
 
 	// reconcile replicasets -> pods
 	rsController := controller.New(podStore, rsStore)
 	go rsController.Run()
 
+	// mark nodes NotReady if heartbeat is stale
+	nodeController := controller.NewNodeController(nodeStore)
+	go nodeController.Run()
+
 	// API Server
 	srv := &api.Server{
 		PodStore: podStore,

pkg/controller/node.go

@@ -0,0 +1,35 @@
+package controller
+
+import (
+	"miniku/pkg/store"
+	"miniku/pkg/types"
+	"time"
+)
+
+const NODE_HEARTBEAT_THRESHOLD = 15 * time.Second
+
+type NodeController struct {
+	nodeStore store.NodeStore
+}
+
+func NewNodeController(nodeStore store.NodeStore) *NodeController {
+	return &NodeController{nodeStore}
+}
+
+func (c *NodeController) Run() {
+	for {
+		for _, node := range c.nodeStore.List() {
+			c.reconcile(node)
+		}
+		time.Sleep(POLL_INTERVAL)
+	}
+}
+
+func (c *NodeController) reconcile(node types.Node) {
+	if time.Since(node.LastHeartbeat) > NODE_HEARTBEAT_THRESHOLD {
+		node.Status = types.NodeStateNotReady
+	} else {
+		node.Status = types.NodeStateReady
+	}
+	c.nodeStore.Put(node.Name, node)
+}

pkg/controller/node_test.go

@@ -0,0 +1,89 @@
+package controller
+
+import (
+	"miniku/pkg/store"
+	"miniku/pkg/types"
+	"testing"
+	"time"
+)
+
+func TestNodeControllerReconcile(t *testing.T) {
+	tests := []struct {
+		name           string
+		node           types.Node
+		expectedStatus types.NodeState
+	}{
+		{
+			name: "fresh heartbeat stays ready",
+			node: types.Node{
+				Name:          "node-1",
+				Status:        types.NodeStateReady,
+				LastHeartbeat: time.Now(),
+			},
+			expectedStatus: types.NodeStateReady,
+		},
+		{
+			name: "stale heartbeat marked notready",
+			node: types.Node{
+				Name:          "node-1",
+				Status:        types.NodeStateReady,
+				LastHeartbeat: time.Now().Add(-30 * time.Second),
+			},
+			expectedStatus: types.NodeStateNotReady,
+		},
+		{
+			name: "zero heartbeat marked notready",
+			node: types.Node{
+				Name:   "node-1",
+				Status: types.NodeStateReady,
+			},
+			expectedStatus: types.NodeStateNotReady,
+		},
+		{
+			name: "recovered node marked ready",
+			node: types.Node{
+				Name:          "node-1",
+				Status:        types.NodeStateNotReady,
+				LastHeartbeat: time.Now(),
+			},
+			expectedStatus: types.NodeStateReady,
+		},
+		{
+			name: "heartbeat exactly at threshold",
+			node: types.Node{
+				Name:          "node-1",
+				Status:        types.NodeStateReady,
+				LastHeartbeat: time.Now().Add(-NODE_HEARTBEAT_THRESHOLD),
+			},
+			expectedStatus: types.NodeStateNotReady,
+		},
+		{
+			name: "heartbeat just before threshold",
+			node: types.Node{
+				Name:          "node-1",
+				Status:        types.NodeStateReady,
+				LastHeartbeat: time.Now().Add(-NODE_HEARTBEAT_THRESHOLD + time.Second),
+			},
+			expectedStatus: types.NodeStateReady,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			nodeStore := store.NewMemStore[types.Node]()
+			nodeStore.Put(tt.node.Name, tt.node)
+
+			controller := NewNodeController(nodeStore)
+			controller.reconcile(tt.node)
+
+			updatedNode, found := nodeStore.Get(tt.node.Name)
+			if !found {
+				t.Fatal("node not found in store")
+			}
+
+			if updatedNode.Status != tt.expectedStatus {
+				t.Errorf("expected status %s, got %s", tt.expectedStatus, updatedNode.Status)
+			}
+		})
+	}
+}

pkg/kubelet/kubelet.go

@@ -17,15 +17,17 @@ const BASE_DELAY time.Duration = 1000
 const MAX_DELAY time.Duration = 60_000
 
 type Kubelet struct {
-	name    string
-	store   store.PodStore
-	runtime runtime.Runtime
+	name      string
+	podStore  store.PodStore
+	nodeStore store.NodeStore
+	runtime   runtime.Runtime
 }
 
-func New(store store.PodStore, runtime runtime.Runtime, name string) Kubelet {
+func New(podStore store.PodStore, nodeStore store.NodeStore, runtime runtime.Runtime, name string) Kubelet {
 	return Kubelet{
 		name,
-		store,
+		podStore,
+		nodeStore,
 		runtime,
 	}
 }
@@ -42,7 +44,7 @@ func (k *Kubelet) Sync() {
 	}
 
 	for _, container := range containers {
-		pod, exists := k.store.Get(container.Name)
+		pod, exists := k.podStore.Get(container.Name)
 		if !exists {
 			log.Printf("sync: removing orphan container %s (%s)", container.Name, container.ID)
 			if err := k.runtime.Stop(container.ID); err != nil {
@@ -63,7 +65,7 @@ func (k *Kubelet) Sync() {
 			log.Printf("sync: linking container %s to pod %s", container.ID, pod.Spec.Name)
 			pod.ContainerID = container.ID
 			pod.Status = types.PodStatusRunning
-			k.store.Put(pod.Spec.Name, pod)
+			k.podStore.Put(pod.Spec.Name, pod)
 		}
 	}
 }
@@ -72,7 +74,7 @@ func (k *Kubelet) Run() {
 	k.Sync()
 
 	for {
-		for _, pod := range k.store.List() {
+		for _, pod := range k.podStore.List() {
 			// only reconcile pods assigned to this node
 			if pod.Spec.NodeName != k.name {
 				continue
@@ -83,6 +85,7 @@ func (k *Kubelet) Run() {
 		}
 
 		// polling
+		k.updateHeartbeat()
 		time.Sleep(POLL_INTERVAL)
 	}
 }
@@ -134,7 +137,7 @@ func (k *Kubelet) reconcilePod(pod types.Pod) error {
 		return fmt.Errorf("unhandled state")
 	}
 
-	k.store.Put(updatedPod.Spec.Name, updatedPod)
+	k.podStore.Put(updatedPod.Spec.Name, updatedPod)
 	return nil
 }
 
@@ -178,3 +181,11 @@ func calculateNextRetry(pod types.Pod) time.Time {
 	delay := min(MAX_DELAY, BASE_DELAY*time.Duration(1<<pod.RetryCount))
 	return time.Now().Add(delay)
 }
+
+func (k *Kubelet) updateHeartbeat() {
+	node, ok := k.nodeStore.Get(k.name)
+	if ok {
+		node.LastHeartbeat = time.Now()
+		k.nodeStore.Put(k.name, node)
+	}
+}