Merge pull request #252 from ByteStorage/develop

refactor component for store and region

cluster/meta/meta.go

@@ -1,6 +1,7 @@
 package meta
 
 import (
+	"fmt"
 	"github.com/ByteStorage/FlyDB/cluster/region"
 	"github.com/ByteStorage/FlyDB/cluster/store"
 	"github.com/ByteStorage/FlyDB/config"
@@ -24,6 +25,8 @@ type MetadataManager interface {
 	GetRegionByID(id uint64) (*region.Region, error)
 	// GetStoreByID gets a store by id.
 	GetStoreByID(id uint64) (*store.Store, error)
+	// ApplyConfig applies a new config to the cluster.
+	ApplyConfig(config *config.Config) error
 }
 
 // meta stores the metadata of the cluster.
@@ -40,3 +43,82 @@ type meta struct {
 	scheduler     *Scheduler                // scheduler, to schedule the cluster.
 	raft          *raft.Raft                // raft, to store the raft group.
 }
+
+// GetStore gets a store by address.
+func (m *meta) GetStore(addr string) (*store.Store, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+// AddStore adds a new store to the cluster.
+func (m *meta) AddStore(addr string) error {
+	//TODO implement me
+	panic("implement me")
+}
+
+// RemoveStore removes a store from the cluster.
+func (m *meta) RemoveStore(addr string) error {
+	//TODO implement me
+	panic("implement me")
+}
+
+// GetAllStores gets all stores in the cluster.
+func (m *meta) GetAllStores() []*store.Store {
+	//TODO implement me
+	panic("implement me")
+}
+
+// GetRegionByID gets a region by id.
+func (m *meta) GetRegionByID(id uint64) (*region.Region, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+// GetStoreByID gets a store by id.
+func (m *meta) GetStoreByID(id uint64) (*store.Store, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
+// ApplyConfig applies a new config to the cluster.
+func (m *meta) ApplyConfig(config *config.Config) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m != nil {
+		err := m.stop()
+		if err != nil {
+			// if err = ErrNotStarted, it means the meta node has not started yet.
+		}
+		err = m.start()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (m *meta) start() error {
+	for _, metaNode := range m.clusterConfig.MetaNodes {
+		// ssh to the meta node
+		// start the meta node
+		fmt.Println(metaNode)
+		panic("implement me")
+	}
+	return nil
+}
+
+func (m *meta) stop() error {
+	//TODO implement me
+	panic("implement me")
+}
+
+// NewMeta creates a new meta.
+func NewMeta(conf config.Config) MetadataManager {
+	return &meta{
+		clusterConfig: &conf,
+		heartbeat:     make(map[string]time.Time),
+		dirTree:       dirtree.NewDirTree(),
+		stores:        make(map[string]*store.Store),
+		regions:       make(map[uint64]*region.Region),
+	}
+}

cluster/region/region.go

@@ -2,6 +2,7 @@ package region
 
 import (
 	"errors"
+	"github.com/ByteStorage/FlyDB/config"
 	"github.com/ByteStorage/FlyDB/engine"
 	"github.com/hashicorp/raft"
 	"sync"
@@ -14,7 +15,7 @@ import (
 // region and replicas are a raft group, and the one region is the leader of the raft group.
 // region stores the data of a region.
 type region struct {
-	id         uint64                // region id
+	id         int64                 // region id
 	startKey   []byte                // start key
 	endKey     []byte                // end key
 	db         *engine.DB            // db, to store the data.
@@ -24,6 +25,7 @@ type region struct {
 	peers      []string              // peers
 	size       int64                 // size
 	mu         sync.RWMutex          // mutex, to protect the region.
+	conf       config.Config         // config
 }
 
 // Region is the interface of region.
@@ -50,6 +52,66 @@ type Region interface {
 	RemovePeer(peer string) error
 	// GetSize gets the total size of the region.
 	GetSize() int64
+	// GetID gets the id of the region.
+	GetID() int64
+}
+
+func NewRegion(conf config.RegionConfig) (Region, error) {
+	db, err := engine.NewDB(conf.Options)
+	if err != nil {
+		return nil, errors.New("new db failed")
+	}
+	raftNode, err := newRaftNode(conf.Config)
+	if err != nil {
+		return nil, errors.New("new raft node failed")
+	}
+	return &region{
+		startKey:   conf.Start,
+		endKey:     conf.End,
+		raftGroups: make(map[uint64]*raft.Raft),
+		db:         db,
+		mu:         sync.RWMutex{},
+		conf:       conf.Config,
+		raft:       raftNode,
+	}, nil
+}
+
+// newRaftNode creates a new raft node for the store.
+func newRaftNode(conf config.Config) (*raft.Raft, error) {
+	// All new methods below can add other return values as needed, such as err
+
+	// create default config for raft
+	raftConfig := newDefaultConfig()
+
+	// setup Raft communication
+	t := newTransport()
+
+	// create the snapshot store. This allows the Raft to truncate the log.
+	snapshots, err := newSnapshotStore(conf)
+	if err != nil {
+		return nil, err
+	}
+
+	// create the log store and stable store
+	logStore, err := newRaftLog(conf)
+	if err != nil {
+		return nil, err
+	}
+	stableStore, err := newStableLog(conf)
+	if err != nil {
+		return nil, err
+	}
+
+	// create a new finite state machine
+	f := newFSM()
+
+	// instantiate the Raft system
+	r, err := raft.NewRaft(raftConfig, f, logStore, stableStore, snapshots, t)
+	if err != nil {
+		return nil, err
+	}
+
+	return r, nil
 }
 
 func (r *region) Put(key []byte, value []byte) error {
@@ -145,3 +207,9 @@ func (r *region) GetSize() int64 {
 	defer r.mu.RUnlock()
 	return r.size
 }
+
+func (r *region) GetID() int64 {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return r.id
+}

cluster/store/store_config.go → cluster/region/region_config.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"github.com/hashicorp/raft"
@@ -7,7 +7,7 @@ import (
 )
 
 // newDefaultConfig returns a new default raft config
-func (s *store) newDefaultConfig() *raft.Config {
+func newDefaultConfig() *raft.Config {
 	return &raft.Config{
 		ProtocolVersion:          raft.ProtocolVersionMax, // using latest protocol version
 		HeartbeatTimeout:         1000 * time.Millisecond,

cluster/store/store_fsm.go → cluster/region/region_fsm.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"github.com/hashicorp/raft"

cluster/store/store_log.go → cluster/region/region_log.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"fmt"

cluster/store/store_log_test.go → cluster/region/region_log_test.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"github.com/ByteStorage/FlyDB/config"

cluster/store/store_snapshot.go → cluster/region/region_snapshot.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"errors"

cluster/store/store_snapshot_test.go → cluster/region/region_snapshot_test.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"github.com/ByteStorage/FlyDB/config"

cluster/store/store_stable.go → cluster/region/region_stable.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"github.com/ByteStorage/FlyDB/config"

cluster/store/store_transport.go → cluster/region/region_transport.go

@@ -1,4 +1,4 @@
-package store
+package region
 
 import (
 	"github.com/hashicorp/raft"