feat: alert message

examples/hello/main.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/LilithGames/agent-go/pkg/agent"
-	"github.com/LilithGames/agent-go/tools/log"
 	"github.com/hasura/go-graphql-client"
 	"github.com/magicsea/behavior3go"
 	"github.com/magicsea/behavior3go/core"
@@ -27,7 +26,8 @@ func main() {
 	behavior.RegisterNode("TestSubscriber", NewSubscriber)
 	behavior.RegisterTreeConfig(helloB3)
 	engine := behavior.BuildEngineFromConfig(task)
-	a := agent.NewAgent(engine)
+	alert := &Echo{}
+	a := agent.NewAgent(engine, nil, agent.WithAlert(alert))
 	a.Start()
 }
 
@@ -55,11 +55,6 @@ func HelloA(tick agent.Ticker) (behavior3go.Status, error) {
 		p = one.(*Player)
 	}
 	tick.Blackboard().SetMem("player", p)
-	fmt.Println("current player id: ", p.ID())
-	log.Debug("test debug log...")
-	log.Info("test info log...")
-	log.Error("test error log...")
-	log.DPanic("test panic panic...")
 	return behavior3go.SUCCESS, nil
 }
 
@@ -92,7 +87,8 @@ func HelloD(tick agent.Ticker) (behavior3go.Status, error) {
 }
 
 func HelloE(tick agent.Ticker) (behavior3go.Status, error) {
-	return behavior3go.SUCCESS, nil
+	return behavior3go.FAILURE, fmt.Errorf("test error essage")
+	// return behavior3go.SUCCESS, nil
 }
 
 func newUser(tick agent.Ticker) (behavior3go.Status, error) {
@@ -135,7 +131,7 @@ func buildTeam(tick agent.Ticker) (behavior3go.Status, error) {
 }
 
 func NewSubscription() core.IBaseNode {
-	subscription := agent.NewGqlSubscription(agent.WithLog(nil))
+	subscription := agent.NewGqlSubscription("", agent.WithLog(nil))
 	return subscription
 }
 
@@ -170,3 +166,11 @@ func NewPlayer(id string) *Player {
 func (p *Player) ID() string {
 	return p.id
 }
+
+type Echo struct {
+}
+
+func (e *Echo) SendMsg(msg *agent.ErrMsg) error {
+	fmt.Println(msg.Name, msg.Intro, msg.Detail, "echo alert...")
+	return nil
+}

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/magicsea/behavior3go v0.0.0-20201106103304-15430dcfecd8
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/rs/xid v1.3.0
+	github.com/spf13/viper v1.9.0
 	go.opentelemetry.io/contrib/instrumentation/runtime v0.27.0
 	go.opentelemetry.io/otel v1.2.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.25.0
@@ -26,27 +27,38 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/emirpasic/gods v1.12.0 // indirect
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/uuid v1.1.2 // indirect
+	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/klauspost/compress v1.10.3 // indirect
+	github.com/magiconair/properties v1.8.5 // indirect
 	github.com/mattn/go-runewidth v0.0.9 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/mitchellh/mapstructure v1.4.2 // indirect
 	github.com/orcaman/concurrent-map v0.0.0-20190107190726-7ed82d9cb717 // indirect
+	github.com/pelletier/go-toml v1.9.4 // indirect
 	github.com/prometheus/client_golang v1.11.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.26.0 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect
+	github.com/spf13/afero v1.6.0 // indirect
+	github.com/spf13/cast v1.4.1 // indirect
+	github.com/spf13/jwalterweatherman v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/subosito/gotenv v1.2.0 // indirect
 	go.opentelemetry.io/otel/internal/metric v0.25.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.2.0 // indirect
 	go.opentelemetry.io/otel/trace v1.2.0 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
-	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 // indirect
-	golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40 // indirect
-	golang.org/x/text v0.3.5 // indirect
-	google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c // indirect
-	gopkg.in/yaml.v2 v2.3.0 // indirect
+	golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420 // indirect
+	golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf // indirect
+	golang.org/x/text v0.3.6 // indirect
+	google.golang.org/genproto v0.0.0-20210828152312-66f60bf46e71 // indirect
+	gopkg.in/ini.v1 v1.63.2 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	nhooyr.io/websocket v1.8.6 // indirect
 )
 

pkg/agent/agent.go

@@ -2,11 +2,12 @@ package agent
 
 import (
 	"context"
-	"github.com/LilithGames/agent-go/tools/metric"
 	"net/http"
-	"os"
 	"strconv"
 
+	"github.com/LilithGames/agent-go/tools/metric"
+	"github.com/spf13/viper"
+
 	"github.com/LilithGames/agent-go/pkg/transfer"
 	"github.com/LilithGames/agent-go/tools/log"
 	"github.com/rs/xid"
@@ -18,22 +19,39 @@ import (
 const masterAddr = "MASTER_ADDR"
 
 type Agent struct {
+	id       string
+	ctx      context.Context
+	cancel   context.CancelFunc
 	engine   *Engine
+	cfg      *viper.Viper
 	endpoint string
-	opt      *AgentOpt
-}
-
-func IsTestMode() bool {
-	endpoint := os.Getenv(masterAddr)
-	return endpoint == ""
+	view     *ViewOpt
+	alert    Alert
+	stream    *proxyStream
 }
 
-func NewAgent(engine *Engine, opts ...*AgentOpt) *Agent {
+func NewAgent(engine *Engine, cfg *viper.Viper, opts ...Option) *Agent {
 	if len(engine.plans) == 0 {
 		log.Panic("absent plans")
 	}
-	endpoint := os.Getenv(masterAddr)
-	return &Agent{engine: engine, endpoint: endpoint, opt: mergeAgentOpt(opts...)}
+	if cfg == nil {
+		cfg = viper.New()
+	}
+	id := cfg.GetString("ID")
+	endpoint := cfg.GetString(masterAddr)
+	ctx, cancel := context.WithCancel(context.Background())
+	at := &Agent{
+		id:id, 
+		ctx: ctx,
+		cancel: cancel,
+		engine: engine, 
+		cfg: cfg, 
+		endpoint: endpoint,
+	}
+	for _, opt := range opts {
+		opt(at)
+	}
+	return at
 }
 
 func (a *Agent) Start() {
@@ -46,8 +64,8 @@ func (a *Agent) Start() {
 }
 
 func (a *Agent) startDefaultAgent() {
-	c := newProxyStream(nil, a.opt.getView())
-	newManager(a.engine, c).startLocalService()
+	a.stream = newProxyFromAgent(a)
+	newManagerFromAgent(a).startLocalService()
 }
 
 func (a *Agent) startClusterAgent() {
@@ -58,8 +76,8 @@ func (a *Agent) startClusterAgent() {
 	if err != nil {
 		log.Panic("request grpc courier error", zap.Error(err))
 	}
-	var c = newProxyStream(client, a.opt.getView())
-	newManager(a.engine, c).startClusterService()
+	a.stream = newProxyFromAgent(a, client)
+	newManagerFromAgent(a).startClusterService()
 }
 
 func (a *Agent) dialMaster() *grpc.ClientConn {
@@ -80,7 +98,7 @@ func (a *Agent) newOutgoingContext() context.Context {
 		}
 	}
 	robotNum := strconv.Itoa(int(rc))
-	data := map[string]string{"agentID": agentID, "ID": os.Getenv("ID"), "robotNum": robotNum}
+	data := map[string]string{"agentID": agentID, "ID": a.id, "robotNum": robotNum}
 	return metadata.NewOutgoingContext(context.Background(), metadata.New(data))
 }
 

pkg/agent/data.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"bytes"
 	"fmt"
-	"os"
 	"sort"
 	"strconv"
 	"strings"
@@ -104,7 +103,7 @@ func printErrorMessage(name string, qs map[string]*transfer.Quantity, opts ...*V
 		es := q.ErrorMap
 		for err, count := range es {
 			row := make([]string, len(header))
-			row[0] = name
+			row[0] = q.Name
 			row[1] = err
 			row[2] = strconv.Itoa(int(count))
 			table.Append(row)
@@ -169,11 +168,9 @@ func pushLocalData(report *transfer.Report) error {
 }
 
 func echoLocalData(planName string, view *ViewOpt) {
-	if os.Getenv("echo") != "ct" {
-		printQuantitySlice(planName+":H", quantities.Handler, view)
-		printQuantitySlice(planName+":E", quantities.Event, view)
-		printErrorMessage(planName+":H", quantities.Handler, view)
-		printErrorMessage(planName+":E", quantities.Event, view)
-	}
+	printQuantitySlice(planName+":H", quantities.Handler, view)
+	printQuantitySlice(planName+":E", quantities.Event, view)
+	printErrorMessage(planName+":H", quantities.Handler, view)
+	printErrorMessage(planName+":E", quantities.Event, view)
 	quantities = newQuantities()
 }

pkg/agent/engine.go

@@ -3,7 +3,6 @@ package agent
 import (
 	"encoding/json"
 	"fmt"
-	"os"
 
 	"github.com/LilithGames/agent-go/pkg/transfer"
 	"github.com/LilithGames/agent-go/tools/log"
@@ -56,16 +55,6 @@ func NewBehavior() *Behavior {
 	}
 }
 
-func (b *Behavior) InitLocalEnvs(envs map[string]string) error {
-	for k, v := range envs {
-		err := os.Setenv(k, v)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func (b *Behavior) RegisterHandler(name string, handler Handler) {
 	creator := func() core.IBaseNode {
 		return &Action{handler: handler}

pkg/agent/manager.go

@@ -1,40 +1,29 @@
 package agent
 
 import (
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"math"
+	"strconv"
+	"sync"
+	"time"
+
 	"github.com/AsynkronIT/protoactor-go/actor"
 	"github.com/LilithGames/agent-go/pkg/transfer"
 	"github.com/LilithGames/agent-go/tools/log"
 	"github.com/magicsea/behavior3go/core"
 	"github.com/magicsea/behavior3go/loader"
 	"go.uber.org/zap"
-	"io"
-	"math"
-	"os"
-	"strconv"
-	"sync"
-	"time"
 )
 
 type manager struct {
-	engine *Engine
-	stream *proxyStream
-	ctx    context.Context
-	cancel context.CancelFunc
+	*Agent
 }
 
-func newManager(engine *Engine, stream *proxyStream) *manager {
-	ctx, cancel := context.WithCancel(context.Background())
-	stream.withContext(ctx)
-	return &manager{
-		engine: engine,
-		stream: stream,
-		ctx:    ctx,
-		cancel: cancel,
-	}
+func newManagerFromAgent(agent *Agent) *manager {
+	return &manager{agent}
 }
 
 func (m *manager) startClusterService() {
@@ -94,7 +83,7 @@ func (m *manager) startAgentEngine(content []byte, circle bool) error {
 
 func (m *manager) startLocalService() {
 	var circle bool
-	if os.Getenv("mode") == "circle" {
+	if m.cfg.GetString("mode") == "circle" {
 		circle = true
 	}
 	m.stream.setPlanCount(len(m.engine.plans), circle)
@@ -103,7 +92,7 @@ func (m *manager) startLocalService() {
 
 func (m *manager) setEnv(envs map[string]string) {
 	for k, v := range envs {
-		os.Setenv(k, v)
+		m.cfg.Set(k, v)
 		if k == "logLevel" && v != "" {
 			log.ResetLogLevel(v)
 		}
@@ -164,7 +153,8 @@ func (m *manager) startExecutor(executor *executor, market *Market) {
 			withWaitGroup(wg).
 			withStatPID(actuaryID).
 			withBeTree(executor.treeCreator()).
-			withMarket(market)
+			withMarket(market).
+			withAlert(m.alert)
 		if i%parallel == 0 && i/parallel > 0 {
 			<-ticker.C
 		}
@@ -180,7 +170,7 @@ func (m *manager) startExecutor(executor *executor, market *Market) {
 
 func (m *manager) getParallel() int {
 	var rs int
-	pe := os.Getenv("parallel")
+	pe := m.cfg.GetString("parallel")
 	if pe != "" {
 		rs, _ = strconv.Atoi(pe)
 		return rs
@@ -190,5 +180,5 @@ func (m *manager) getParallel() int {
 			rs = int(plan.Parallel)
 		}
 	}
-	return math.MaxInt16
+	return math.MaxInt8
 }