abstract whole agent

agent.go

@@ -1,26 +1,25 @@
 package main
 
 import (
-	"bufio"
-	"bytes"
 	"fmt"
+	"math/rand"
 	"os"
-
-	_ "go.uber.org/automaxprocs"
+	"os/signal"
+	"syscall"
+	"time"
 
 	"github.com/projecteru2/agent/api"
-	"github.com/projecteru2/agent/engine"
+	"github.com/projecteru2/agent/manager/node"
+	"github.com/projecteru2/agent/manager/workload"
 	"github.com/projecteru2/agent/selfmon"
 	"github.com/projecteru2/agent/types"
 	"github.com/projecteru2/agent/utils"
 	"github.com/projecteru2/agent/version"
-	"github.com/projecteru2/agent/watcher"
 
 	"github.com/jinzhu/configor"
-	"github.com/sethvargo/go-signalcontext"
 	log "github.com/sirupsen/logrus"
 	cli "github.com/urfave/cli/v2"
-	"gopkg.in/yaml.v3"
+	_ "go.uber.org/automaxprocs"
 )
 
 func setupLogLevel(l string) error {
@@ -40,26 +39,14 @@ func initConfig(c *cli.Context) *types.Config {
 		log.Fatalf("[main] load config failed %v", err)
 	}
 
-	config.PrepareConfig(c)
-	printConfig(config)
+	config.Prepare(c)
+	config.Print()
 	return config
 }
 
-func printConfig(c *types.Config) {
-	bs, err := yaml.Marshal(c)
-	if err != nil {
-		log.Fatalf("[main] print config failed %v", err)
-	}
-
-	log.Info("---- current config ----")
-	scanner := bufio.NewScanner(bytes.NewBuffer(bs))
-	for scanner.Scan() {
-		log.Info(scanner.Text())
-	}
-	log.Info("------------------------")
-}
-
 func serve(c *cli.Context) error {
+	rand.Seed(time.Now().UnixNano())
+
 	if err := setupLogLevel(c.String("log-level")); err != nil {
 		log.Fatal(err)
 	}
@@ -69,22 +56,45 @@ func serve(c *cli.Context) error {
 	defer os.Remove(config.PidFile)
 
 	if c.Bool("selfmon") {
-		return selfmon.Monitor(config)
+		mon, err := selfmon.New(c.Context, config)
+		if err != nil {
+			return err
+		}
+		return mon.Run(c.Context)
 	}
 
-	ctx, cancel := signalcontext.OnInterrupt()
+	ctx, cancel := signal.NotifyContext(c.Context, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
 	defer cancel()
 
-	watcher.InitMonitor()
-	go watcher.LogMonitor.Serve(ctx)
+	errChan := make(chan error, 1)
+
+	workloadManager, err := workload.NewManager(ctx, config)
+	if err != nil {
+		return err
+	}
+	go func() {
+		errChan <- workloadManager.Run(ctx)
+	}()
 
-	agent, err := engine.NewEngine(c.Context, config)
+	nodeManager, err := node.NewManager(ctx, config)
 	if err != nil {
 		return err
 	}
+	go func() {
+		errChan <- nodeManager.Run(ctx)
+	}()
+
+	apiHandler := api.NewHandler(config, workloadManager)
+	go apiHandler.Serve()
 
-	go api.Serve(config.API.Addr)
-	return agent.Run(ctx)
+	select {
+	case err := <-errChan:
+		log.Debugf("[agent] err: %v", err)
+		return err
+	case <-ctx.Done():
+		log.Info("[agent] Agent caught system signal, exiting")
+		return nil
+	}
 }
 
 func main() {
@@ -109,6 +119,12 @@ func main() {
 				Usage:   "set log level",
 				EnvVars: []string{"ERU_AGENT_LOG_LEVEL"},
 			},
+			&cli.StringFlag{
+				Name:    "store",
+				Value:   "",
+				Usage:   "store type",
+				EnvVars: []string{"ERU_AGENT_STORE"},
+			},
 			&cli.StringFlag{
 				Name:    "core-endpoint",
 				Value:   "",
@@ -127,6 +143,12 @@ func main() {
 				Usage:   "core password",
 				EnvVars: []string{"ERU_AGENT_CORE_PASSWORD"},
 			},
+			&cli.StringFlag{
+				Name:    "runtime",
+				Value:   "",
+				Usage:   "runtime type",
+				EnvVars: []string{"ERU_AGENT_RUNTIME"},
+			},
 			&cli.StringFlag{
 				Name:    "docker-endpoint",
 				Value:   "",
@@ -148,7 +170,7 @@ func main() {
 			&cli.StringFlag{
 				Name:    "api-addr",
 				Value:   "",
-				Usage:   "agent API serving address",
+				Usage:   "agent api serving address",
 				EnvVars: []string{"ERU_AGENT_API_ADDR"},
 			},
 			&cli.StringSliceFlag{
@@ -200,6 +222,12 @@ func main() {
 				Value: false,
 				Usage: "run this agent as a selfmon daemon",
 			},
+			&cli.StringFlag{
+				Name:    "kv",
+				Value:   "",
+				Usage:   "kv type",
+				EnvVars: []string{"ERU_AGENT_KV"},
+			},
 			&cli.BoolFlag{
 				Name:  "check-only-mine",
 				Value: false,

agent.yaml.sample

@@ -4,6 +4,18 @@
 # This option is not required as the default value is "/tmp/agent.pid".
 pid: /tmp/agent.pid
 
+# store defines the type of core service.
+# This option is not required as the default value is "grpc".
+store: grpc
+
+# runtime defines the type of runtime.
+# This option is not required as the default value is "docker".
+runtime: docker
+
+# kv defines the type of kv store.
+# This option is not required as the default value is "etcd".
+kv: etcd
+
 # core defines the address of eru-core component.
 # This option is not required as the default value is "127.0.0.1:5001".
 core:
@@ -111,6 +123,11 @@ healthcheck:
 # The default value is "5s", note that "s" in the end.
 global_connection_timeout: 15s
 
+# ha_keepalive_interval defines the time interval for sending heartbeat
+# when selfmon maintains its own active state.
+# The default value is "16s", note that "s" in the end.
+ha_keepalive_interval: 16s
+
 # etcd defines the etcd configuration.
 # This option is required and has no default value.
 # If you don't plan to run this eru-agent in selfmon mode,

api/http.go

@@ -3,26 +3,26 @@ package api
 import (
 	"encoding/json"
 	"net/http"
-	"runtime/pprof"
-
+	"runtime/pprof" // nolint
 	// enable profile
 	_ "net/http/pprof" // nolint
 
+	"github.com/projecteru2/agent/manager/workload"
 	"github.com/projecteru2/agent/types"
 	"github.com/projecteru2/agent/version"
-	"github.com/projecteru2/agent/watcher"
-	coreutils "github.com/projecteru2/core/utils"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-	log "github.com/sirupsen/logrus"
 
 	"github.com/bmizerany/pat"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	log "github.com/sirupsen/logrus"
 )
 
 // JSON define a json
 type JSON map[string]interface{}
 
 // Handler define handler
 type Handler struct {
+	config          *types.Config
+	workloadManager *workload.Manager
 }
 
 // URL /version/
@@ -58,24 +58,27 @@ func (h *Handler) log(w http.ResponseWriter, req *http.Request) {
 			log.Errorf("[apiLog] connect failed %v", err)
 			return
 		}
-		logConsumer := &types.LogConsumer{
-			ID:  coreutils.RandomString(8),
-			App: app, Conn: conn, Buf: buf,
-		}
-		watcher.LogMonitor.ConsumerC <- logConsumer
-		log.Infof("[apiLog] %s %s log attached", app, logConsumer.ID)
+		defer conn.Close()
+		h.workloadManager.Subscribe(req.Context(), app, buf)
+	}
+}
+
+// NewHandler new api http handler
+func NewHandler(config *types.Config, workloadManager *workload.Manager) *Handler {
+	return &Handler{
+		config:          config,
+		workloadManager: workloadManager,
 	}
 }
 
 // Serve start a api service
 // blocks by http.ListenAndServe
 // run this in a separated goroutine
-func Serve(addr string) {
-	if addr == "" {
+func (h *Handler) Serve() {
+	if h.config.API.Addr == "" {
 		return
 	}
 
-	h := &Handler{}
 	restfulAPIServer := pat.New()
 	handlers := map[string]map[string]func(http.ResponseWriter, *http.Request){
 		"GET": {
@@ -93,9 +96,9 @@ func Serve(addr string) {
 
 	http.Handle("/", restfulAPIServer)
 	http.Handle("/metrics", promhttp.Handler())
-	log.Infof("[apiServe] http api started %s", addr)
+	log.Infof("[apiServe] http api started %s", h.config.API.Addr)
 
-	err := http.ListenAndServe(addr, nil)
+	err := http.ListenAndServe(h.config.API.Addr, nil)
 	if err != nil {
 		log.Panicf("http api failed %s", err)
 	}

common/common.go

@@ -17,4 +17,19 @@ const (
 
 	// LocalIP .
 	LocalIP = "127.0.0.1"
+
+	// DockerRuntime use docker as runtime
+	DockerRuntime = "docker"
+	// MocksRuntime use the mock runtime
+	MocksRuntime = "mocks"
+
+	// GRPCStore use gRPC as store
+	GRPCStore = "grpc"
+	// MocksStore use the mock store
+	MocksStore = "mocks"
+
+	// ETCDKV use ETCD as KV
+	ETCDKV = "etcd"
+	// MocksKV use the mock KV
+	MocksKV = "mocks"
 )