feat: impl bbolt store

.gitignore

@@ -0,0 +1 @@
+miniku.db

cmd/miniku/main.go

@@ -4,6 +4,8 @@ import (
 	"log"
 	"net/http"
 
+	bolt "go.etcd.io/bbolt"
+
 	"miniku/pkg/api"
 	"miniku/pkg/controller"
 	"miniku/pkg/kubelet"
@@ -14,9 +16,19 @@ import (
 )
 
 func main() {
-	podStore := store.NewMemStore[types.Pod]()
-	rsStore := store.NewMemStore[types.ReplicaSet]()
-	nodeStore := store.NewMemStore[types.Node]()
+	db, err := bolt.Open("miniku.db", 0600, nil)
+	if err != nil {
+		log.Fatalf("failed to open database: %v", err)
+	}
+	defer func() {
+		if err := db.Close(); err != nil {
+			log.Printf("failed to close database: %v", err)
+		}
+	}()
+
+	podStore := store.NewBoltStore[types.Pod](db, "pods")
+	rsStore := store.NewBoltStore[types.ReplicaSet](db, "replicasets")
+	nodeStore := store.NewBoltStore[types.Node](db, "nodes")
 
 	nodeStore.Put("node-1", types.Node{Name: "node-1", Status: types.NodeStateReady})
 	nodeStore.Put("node-2", types.Node{Name: "node-2", Status: types.NodeStateReady})

go.mod

@@ -1,3 +1,8 @@
 module miniku
 
 go 1.24.12
+
+require (
+	go.etcd.io/bbolt v1.4.3 // indirect
+	golang.org/x/sys v0.29.0 // indirect
+)

go.sum

@@ -0,0 +1,4 @@
+go.etcd.io/bbolt v1.4.3 h1:dEadXpI6G79deX5prL3QRNP6JB8UxVkqo4UPnHaNXJo=
+go.etcd.io/bbolt v1.4.3/go.mod h1:tKQlpPaYCVFctUIgFKFnAlvbmB3tpy1vkTnDWohtc0E=
+golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
+golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

pkg/controller/node.go

@@ -9,19 +9,23 @@ import (
 const NODE_HEARTBEAT_THRESHOLD = 15 * time.Second
 
 type NodeController struct {
-	nodeStore store.NodeStore
+	nodeStore    store.NodeStore
+	PollInterval time.Duration
 }
 
 func NewNodeController(nodeStore store.NodeStore) *NodeController {
-	return &NodeController{nodeStore}
+	return &NodeController{
+		nodeStore:    nodeStore,
+		PollInterval: 5 * time.Second,
+	}
 }
 
 func (c *NodeController) Run() {
 	for {
 		for _, node := range c.nodeStore.List() {
 			c.reconcile(node)
 		}
-		time.Sleep(POLL_INTERVAL)
+		time.Sleep(c.PollInterval)
 	}
 }
 

pkg/controller/replicaset.go

@@ -17,17 +17,17 @@ import (
 	"time"
 )
 
-const POLL_INTERVAL = time.Millisecond * 5000
-
 type ReplicaSetController struct {
-	podStore store.PodStore
-	rsStore  store.ReplicaSetStore
+	podStore     store.PodStore
+	rsStore      store.ReplicaSetStore
+	PollInterval time.Duration
 }
 
 func New(podStore store.PodStore, rsStore store.ReplicaSetStore) *ReplicaSetController {
 	return &ReplicaSetController{
-		podStore,
-		rsStore,
+		podStore:     podStore,
+		rsStore:      rsStore,
+		PollInterval: 5 * time.Second,
 	}
 }
 
@@ -39,7 +39,7 @@ func (c *ReplicaSetController) Run() {
 			}
 		}
 
-		time.Sleep(POLL_INTERVAL)
+		time.Sleep(c.PollInterval)
 	}
 }
 func (c *ReplicaSetController) reconcile(rs types.ReplicaSet) error {
@@ -73,6 +73,9 @@ func (c *ReplicaSetController) reconcile(rs types.ReplicaSet) error {
 func (c *ReplicaSetController) getMatchingPods(rs types.ReplicaSet) []types.Pod {
 	var result []types.Pod
 	for _, pod := range c.podStore.List() {
+		if pod.Status == types.PodStatusFailed {
+			continue
+		}
 		if matchesSelector(pod, rs.Selector) {
 			result = append(result, pod)
 		}

pkg/kubelet/kubelet.go

@@ -9,26 +9,26 @@ import (
 	"time"
 )
 
-const POLL_INTERVAL = time.Millisecond * 5000
-
 const MAX_RETRY_COUNT = 3
 
 const BASE_DELAY time.Duration = 1000
 const MAX_DELAY time.Duration = 60_000
 
 type Kubelet struct {
-	name      string
-	podStore  store.PodStore
-	nodeStore store.NodeStore
-	runtime   runtime.Runtime
+	name         string
+	podStore     store.PodStore
+	nodeStore    store.NodeStore
+	runtime      runtime.Runtime
+	PollInterval time.Duration
 }
 
 func New(podStore store.PodStore, nodeStore store.NodeStore, runtime runtime.Runtime, name string) Kubelet {
 	return Kubelet{
-		name,
-		podStore,
-		nodeStore,
-		runtime,
+		name:         name,
+		podStore:     podStore,
+		nodeStore:    nodeStore,
+		runtime:      runtime,
+		PollInterval: 5 * time.Second,
 	}
 }
 
@@ -108,7 +108,7 @@ func (k *Kubelet) Run() {
 
 		// polling
 		k.updateHeartbeat()
-		time.Sleep(POLL_INTERVAL)
+		time.Sleep(k.PollInterval)
 	}
 }
 
@@ -125,6 +125,11 @@ func (k *Kubelet) reconcilePod(pod types.Pod) error {
 		}
 	}
 
+	// terminal state, nothing to do
+	if podStatus == types.PodStatusFailed {
+		return nil
+	}
+
 	var updatedPod types.Pod
 	var err error
 	switch {

pkg/scheduler/scheduler.go

@@ -9,18 +9,17 @@ import (
 )
 
 type Scheduler struct {
-	podStore  store.PodStore
-	nodeStore store.NodeStore
-	nextIndex uint
+	podStore     store.PodStore
+	nodeStore    store.NodeStore
+	nextIndex    uint
+	PollInterval time.Duration
 }
 
-const POLL_INTERVAL = time.Millisecond * 5000
-
 func New(podStore store.PodStore, nodeStore store.NodeStore) *Scheduler {
 	return &Scheduler{
-		podStore,
-		nodeStore,
-		0,
+		podStore:     podStore,
+		nodeStore:    nodeStore,
+		PollInterval: 5 * time.Second,
 	}
 }
 
@@ -37,7 +36,7 @@ func (s *Scheduler) Run() {
 			}
 		}
 
-		time.Sleep(POLL_INTERVAL)
+		time.Sleep(s.PollInterval)
 	}
 }
 

pkg/store/bolt.go

@@ -0,0 +1,101 @@
+package store
+
+import (
+	"encoding/json"
+	"log"
+	"sync"
+
+	bolt "go.etcd.io/bbolt"
+)
+
+type BoltStore[T any] struct {
+	mu     sync.RWMutex
+	db     *bolt.DB
+	bucket []byte
+}
+
+func NewBoltStore[T any](db *bolt.DB, bucket string) *BoltStore[T] {
+	if err := db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists([]byte(bucket))
+		return err
+	}); err != nil {
+		log.Fatalf("bolt: failed to create bucket %q: %v", bucket, err)
+	}
+
+	return &BoltStore[T]{
+		db:     db,
+		bucket: []byte(bucket),
+	}
+}
+
+func (s *BoltStore[T]) List() []T {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	var out []T
+	if err := s.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucket)
+		return b.ForEach(func(k, v []byte) error {
+			var item T
+			if err := json.Unmarshal(v, &item); err != nil {
+				return err
+			}
+			out = append(out, item)
+			return nil
+		})
+	}); err != nil {
+		log.Printf("bolt: list %q: %v", s.bucket, err)
+	}
+	return out
+}
+
+func (s *BoltStore[T]) Get(name string) (T, bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	var item T
+	var found bool
+	if err := s.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucket)
+		v := b.Get([]byte(name))
+		if v == nil {
+			return nil
+		}
+		if err := json.Unmarshal(v, &item); err != nil {
+			return err
+		}
+		found = true
+		return nil
+	}); err != nil {
+		log.Printf("bolt: get %q/%s: %v", s.bucket, name, err)
+	}
+	return item, found
+}
+
+func (s *BoltStore[T]) Put(name string, t T) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if err := s.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucket)
+		data, err := json.Marshal(t)
+		if err != nil {
+			return err
+		}
+		return b.Put([]byte(name), data)
+	}); err != nil {
+		log.Printf("bolt: put %q/%s: %v", s.bucket, name, err)
+	}
+}
+
+func (s *BoltStore[T]) Delete(name string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if err := s.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(s.bucket)
+		return b.Delete([]byte(name))
+	}); err != nil {
+		log.Printf("bolt: delete %q/%s: %v", s.bucket, name, err)
+	}
+}