[feat:#75][storage:runtime]: storage runtime dependency

Makefile

@@ -46,12 +46,14 @@ clean-build:
 	rm -f bin/lind
 	cd web/ && make web_clean
 
-
-clean:  ## Clean up useless files.
-	$(clean-build)
+clean-tmp: ## clean up tmp and test out files
 	find . -type f -name '*.out' -exec rm -f {} +
 	find . -type f -name '.DS_Store' -exec rm -f {} +
 	find . -type f -name '*.test' -exec rm -f {} +
 	find . -type f -name '*.prof' -exec rm -f {} +
 	find . -type s -name 'localhost:*' -exec rm -f {} +
 	find . -type s -name '127.0.0.1:*' -exec rm -f {} +
+
+clean:  ## Clean up useless files.
+	$(clean-build)
+	$(clean-tmp)

broker/rpc/broker_client.go

@@ -6,8 +6,8 @@ import (
 
 	"google.golang.org/grpc"
 
-	"github.com/eleme/lindb/rpc/pkg/broker"
-	"github.com/eleme/lindb/rpc/pkg/common"
+	"github.com/eleme/lindb/rpc/proto/broker"
+	"github.com/eleme/lindb/rpc/proto/common"
 )
 
 type BrokerClient interface {

broker/rpc/broker_server.go

@@ -9,8 +9,8 @@ import (
 
 	"github.com/eleme/lindb/pkg/logger"
 	"github.com/eleme/lindb/rpc"
-	"github.com/eleme/lindb/rpc/pkg/broker"
-	"github.com/eleme/lindb/rpc/pkg/common"
+	"github.com/eleme/lindb/rpc/proto/broker"
+	"github.com/eleme/lindb/rpc/proto/common"
 )
 
 type BrokerServer interface {

broker/rpc/broker_test.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/eleme/lindb/pkg/logger"
 	"github.com/eleme/lindb/rpc"
-	"github.com/eleme/lindb/rpc/pkg/common"
+	"github.com/eleme/lindb/rpc/proto/common"
 )
 
 const (

broker/runtime.go

@@ -34,23 +34,19 @@ type apiHandler struct {
 	databaseAPI       *admin.DatabaseAPI
 }
 
-// Runtime represents broker runtime dependency
+// runtime represents broker runtime dependency
 type runtime struct {
-	state server.State
-
+	state   server.State
 	cfgPath string
 	config  config.Broker
+	ctx     context.Context
+	cancel  context.CancelFunc
 
 	// init value when runtime
-	repo state.Repository
-
-	srv srv
-
+	repo       state.Repository
+	srv        srv
 	httpServer *http.Server
 
-	ctx    context.Context
-	cancel context.CancelFunc
-
 	log *zap.Logger
 }
 
@@ -93,9 +89,7 @@ func (r *runtime) Run() error {
 	r.buildAPIDependency()
 
 	// start http server
-	go func() {
-		r.startHTTPServer()
-	}()
+	r.startHTTPServer()
 
 	r.state = server.Running
 	return nil
@@ -125,6 +119,7 @@ func (r *runtime) Stop() error {
 		}
 	}
 
+	r.log.Info("broker server stop complete")
 	r.state = server.Terminated
 	return nil
 }
@@ -143,9 +138,12 @@ func (r *runtime) startHTTPServer() {
 		IdleTimeout:  time.Second * 60,
 		Handler:      router,
 	}
-	if err := r.httpServer.ListenAndServe(); err != nil {
-		panic(fmt.Sprintf("start http server error:%s", err))
-	}
+	go func() {
+		if err := r.httpServer.ListenAndServe(); err != http.ErrServerClosed {
+			panic(fmt.Sprintf("start http server error:%s", err))
+		}
+		r.log.Info("http server stop complete")
+	}()
 }
 
 // startStateRepo starts state repository

broker/runtime_test.go

@@ -2,6 +2,7 @@ package broker
 
 import (
 	"testing"
+	"time"
 
 	"gopkg.in/check.v1"
 
@@ -19,7 +20,7 @@ type testBrokerRuntimeSuite struct {
 	mock.RepoTestSuite
 }
 
-func TestDatabaseAPI(t *testing.T) {
+func TestBrokerRuntime(t *testing.T) {
 	check.Suite(&testBrokerRuntimeSuite{})
 	test = t
 	check.TestingT(t)
@@ -51,6 +52,9 @@ func (ts *testBrokerRuntimeSuite) TestBrokerRun(c *check.C) {
 	if err != nil {
 		c.Fatal(err)
 	}
+	// wait run finish
+	time.Sleep(500 * time.Millisecond)
+
 	c.Assert(server.Running, check.Equals, broker.State())
 
 	broker.Stop()

cmd/lind/broker.go

@@ -54,15 +54,15 @@ func serveBroker(cmd *cobra.Command, args []string) error {
 	// start broker server
 	broker := broker.NewBrokerRuntime(brokerCfgPath)
 	if err := broker.Run(); err != nil {
-		return fmt.Errorf("run broker error:%s", err)
+		return fmt.Errorf("run broker server error:%s", err)
 	}
 
 	// waiting system exit signal
 	<-ctx.Done()
 
 	// stop broker server
 	if err := broker.Stop(); err != nil {
-		return fmt.Errorf("stop broker error:%s", err)
+		return fmt.Errorf("stop broker server error:%s", err)
 	}
 
 	return nil

cmd/lind/root.go

@@ -5,9 +5,6 @@ import (
 )
 
 const (
-	// default config file location of LinDB
-	cfgFilePath = "/etc/lindb"
-
 	linDBText = `
     __     _             ____     ____ 
    / /    (_)   ____    / __ \   / __ )

cmd/lind/storage.go

@@ -3,22 +3,10 @@ package lind
 import (
 	"fmt"
 	_ "net/http/pprof" // for profiling
-	"os"
 
-	"github.com/eleme/lindb/config"
-	"github.com/eleme/lindb/pkg/logger"
-	"github.com/eleme/lindb/query"
 	"github.com/eleme/lindb/storage"
-	"github.com/eleme/lindb/tsdb"
 
-	"github.com/BurntSushi/toml"
 	"github.com/spf13/cobra"
-	"go.uber.org/zap"
-)
-
-const (
-	storageCfgName        = "storage.toml"
-	defaultStorageCfgFile = cfgFilePath + "/" + storageCfgName
 )
 
 var (
@@ -34,7 +22,7 @@ func newStorageCmd() *cobra.Command {
 		Short:   "The storage layer of LinDB",
 	}
 	runStorageCmd.PersistentFlags().StringVar(&storageCfgPath, "config", "",
-		fmt.Sprintf("storage config file path, default is %s", defaultStorageCfgFile))
+		fmt.Sprintf("storage config file path, default is %s", storage.DefaultStorageCfgFile))
 	runStorageCmd.PersistentFlags().BoolVar(&storageDebug, "debug", false,
 		"profiling Go programs with pprof")
 
@@ -61,38 +49,22 @@ var initializeStorageConfigCmd = &cobra.Command{
 }
 
 func serveStorage(cmd *cobra.Command, args []string) error {
-	log := logger.GetLogger()
 	ctx := newCtxWithSignals()
 
-	if storageCfgPath == "" {
-		storageCfgPath = defaultStorageCfgFile
+	// start storage server
+	storage := storage.NewStorageRuntime(storageCfgPath)
+	if err := storage.Run(); err != nil {
+		return fmt.Errorf("run storage server error:%s", err)
 	}
-	if _, err := os.Stat(storageCfgPath); err != nil {
-		return fmt.Errorf("config file doesn't exist, see how to initialize the config by `lind storage -h`")
-	}
-	fmt.Printf("load config file: %v successfully\n", storageCfgPath)
 
-	storageConfig := config.StorageConfig{}
-	if _, err := toml.DecodeFile(storageCfgPath, &storageConfig); err != nil {
-		return err
-	}
-	// start the repository server
-	storageServer := storage.New(ctx, &storageConfig)
-	if err := storageServer.Start(); err != nil {
-		log.Error("storage start failed", zap.Error(err))
-		return err
-	}
+	// waiting system exit signal
+	<-ctx.Done()
 
-	// create a new engine
-	engine, err := tsdb.NewEngine(storageConfig.Name, storageConfig.Path)
-	if err != nil {
-		return err
+	// stop storage server
+	if err := storage.Stop(); err != nil {
+		return fmt.Errorf("stop storage server error:%s", err)
 	}
-	// todo: fix this
-	_ = query.NewTSDBExecutor(engine, nil, nil)
-
-	<-ctx.Done()
-	return engine.Close()
+	return nil
 }
 
 // databaseCmd provides the ability to control the database of storage

config/storage.go

@@ -2,22 +2,22 @@ package config
 
 import "github.com/eleme/lindb/pkg/state"
 
-// StorageConfig represents a storage configuration
-type StorageConfig struct {
-	StorageRepositoryConfig `toml:"StorageRepositoryConfig"`
-	Engine                  `toml:"engine"`
-	StoragePort             uint16 `toml:"Port"`
+// Storage represents a storage configuration
+type Storage struct {
+	Coordinator state.Config `toml:"coordinator"`
+	Server      Server       `toml:"server"`
+
+	Engine `toml:"engine"`
+}
+
+// Server represents tcp server config
+type Server struct {
+	Port uint16 `toml:"port"`
+	TTL  int64  `toml:"ttl"`
 }
 
 // Engine represents an engine level configuration
 type Engine struct {
 	Path string `toml:"path"`
 	Name string `toml:"name"`
 }
-
-// RepositoryConfig represents the repository config
-type StorageRepositoryConfig struct {
-	state.Config    `toml:"RepositoryConfig"`
-	HeartBeatTTL    int64  `toml:"HeartBeatTTL"`
-	HeartBeatPrefix string `toml:"HeartBeatPrefix"`
-}

coordinator/discovery/constants.go

@@ -0,0 +1,4 @@
+package discovery
+
+// ActiveNodesPath represents active nodes prefix path for node register
+const ActiveNodesPath = "/active/nodes"