Tidb cloud

.github/workflows/go.yml

@@ -27,6 +27,20 @@ jobs:
         mongodb-version: 4.2
         mongodb-username: root
         mongodb-password: pwd
+        mongodb-port: 27017
+        mongodb-db: fastflow
+
+    - name: Shutdown Ubuntu MySQL (SUDO)
+      run: sudo service mysql stop # Shutdown the Default MySQL, "sudo" is necessary, please not remove it
+
+    - name: Set up MySQL
+      uses: mirromutth/[email protected]
+      with:
+        host port: 55000 # Optional, default value is 3306. The port of host
+        container port: 55000 # Optional, default value is 3306. The port of container
+        mysql version: '8.0' # Optional, default value is "latest". The version of the MySQL
+        mysql database: 'fastflow' # Optional, default value is "test". The specified database which will be create
+        mysql root password: mysqlpw # Required if "mysql user" is empty, default is empty. The root superuser password
 
     - name: Test
       run: make g-test
@@ -35,3 +49,11 @@ jobs:
       uses: codecov/codecov-action@v2
       with:
         file: ./coverage.out
+
+    - name: Integration Test
+      run: go test -race -coverprofile=integration-coverage.out ./... -tags=integration
+
+    - name: Upload Coverage report to CodeCov
+      uses: codecov/codecov-action@v2
+      with:
+        file: ./integration-coverage.out

Makefile

@@ -26,4 +26,10 @@ g-test:
 mock:
 	for file in `find . -type d \( -path ./.git -o -path ./.github \) -prune -o -name '*.go' -print | xargs grep --files-with-matches -e '//go:generate mockgen'`; do \
 		go generate $$file; \
-	done
+	done
+
+.PHONY: build
+GO     := GO111MODULE=on go
+GOBUILD := CGO_ENABLED=0 $(GO) build
+build:
+	GOARCH=amd64 GOOS=linux $(GOBUILD) -gcflags "all=-N -l" -o dist/fastflow examples/mysql/main.go

examples/mysql/main.go

@@ -0,0 +1,197 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+	"github.com/shiningrush/fastflow"
+	mysqlKeeper "github.com/shiningrush/fastflow/keeper/mysql"
+	"github.com/shiningrush/fastflow/pkg/entity"
+	"github.com/shiningrush/fastflow/pkg/entity/run"
+	"github.com/shiningrush/fastflow/pkg/exporter"
+	"github.com/shiningrush/fastflow/pkg/mod"
+	"github.com/shiningrush/fastflow/pkg/utils/data"
+	mysqlStore "github.com/shiningrush/fastflow/store/mysql"
+)
+
+type ActionParam struct {
+	Name string
+	Desc string
+}
+
+type ActionA struct {
+	code string
+}
+
+func (a *ActionA) Name() string {
+	return fmt.Sprintf("Action-%s", a.code)
+}
+func (a *ActionA) RunBefore(ctx run.ExecuteContext, params interface{}) error {
+	input := params.(*ActionParam)
+	log.Println(fmt.Sprintf("%s run before, p.Name: %s, p.Desc: %s", a.Name(), input.Name, input.Desc))
+	time.Sleep(time.Second)
+	if a.code != "B" && a.code != "C" {
+		ctx.ShareData().Set(fmt.Sprintf("%s-key", a.code), fmt.Sprintf("%s value", a.code))
+	}
+	return nil
+}
+func (a *ActionA) Run(ctx run.ExecuteContext, params interface{}) error {
+	input := params.(*ActionParam)
+	log.Println(fmt.Sprintf("%s run, p.Name: %s, p.Desc: %s", a.Name(), input.Name, input.Desc))
+	ctx.Trace("run start", run.TraceOpPersistAfterAction)
+	time.Sleep(2 * time.Second)
+	ctx.Trace("run end")
+	return nil
+}
+func (a *ActionA) RunAfter(ctx run.ExecuteContext, params interface{}) error {
+	input := params.(*ActionParam)
+	log.Println(fmt.Sprintf("%s run after, p.Name: %s, p.Desc: %s", a.Name(), input.Name, input.Desc))
+	time.Sleep(time.Second)
+	return nil
+}
+func (a *ActionA) ParameterNew() interface{} {
+	return &ActionParam{}
+}
+
+func ensureDagCreated() error {
+	dag := &entity.Dag{
+		BaseInfo: entity.BaseInfo{
+			ID: "test-dag",
+		},
+		Name: "test",
+		Vars: entity.DagVars{
+			"var": {DefaultValue: "default-var"},
+		},
+		Status: entity.DagStatusNormal,
+		Tasks: []entity.Task{
+			{ID: "task1", ActionName: "Action-A", Params: map[string]interface{}{
+				"Name": "task-p1",
+				"Desc": "{{var}}",
+			}, TimeoutSecs: 5},
+			{ID: "task2", ActionName: "Action-B", DependOn: []string{"task1"}, Params: map[string]interface{}{
+				"Name": "task-p1",
+				"Desc": "{{var}}",
+			}},
+			{ID: "task3", ActionName: "Action-C", DependOn: []string{"task1"}, Params: map[string]interface{}{
+				"Name": "task-p1",
+				"Desc": "{{var}}",
+			}},
+			{ID: "task4", ActionName: "Action-D", DependOn: []string{"task2", "task3"}, Params: map[string]interface{}{
+				"Name": "task-p1",
+				"Desc": "{{var}}",
+			}},
+		},
+	}
+	oldDag, err := mod.GetStore().GetDag(dag.ID)
+	if errors.Is(err, data.ErrDataNotFound) {
+		if err := mod.GetStore().CreateDag(dag); err != nil {
+			return err
+		}
+	}
+	if oldDag != nil {
+		if err := mod.GetStore().UpdateDag(dag); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func main() {
+	// init action
+	fastflow.RegisterAction([]run.Action{
+		&ActionA{code: "A"},
+		&ActionA{code: "B"},
+		&ActionA{code: "C"},
+		&ActionA{code: "D"},
+	})
+	// init keeper
+	keeper := mysqlKeeper.NewKeeper(&mysqlKeeper.KeeperOption{
+		Key: "worker-1",
+		MySQLConfig: &mysql.Config{
+			Addr:   "127.0.0.1:55000",
+			User:   "root",
+			Passwd: "mysqlpw",
+			DBName: "fastflow",
+		},
+		MigrationSwitch: true,
+	})
+	if err := keeper.Init(); err != nil {
+		log.Fatal(fmt.Errorf("init keeper failed: %w", err))
+		return
+	}
+
+	// init store
+	st := mysqlStore.NewStore(&mysqlStore.StoreOption{
+		MySQLConfig: &mysql.Config{
+			Addr:   "127.0.0.1:55000",
+			User:   "root",
+			Passwd: "mysqlpw",
+			DBName: "fastflow",
+		},
+		MigrationSwitch: true,
+	})
+	if err := st.Init(); err != nil {
+		log.Fatal(fmt.Errorf("init store failed: %w", err))
+		return
+	}
+
+	// init fastflow
+	if err := fastflow.Init(&fastflow.InitialOption{
+		Keeper:            keeper,
+		Store:             st,
+		ParserWorkersCnt:  10,
+		ExecutorWorkerCnt: 50,
+	}); err != nil {
+		panic(fmt.Sprintf("init fastflow failed: %s", err))
+	}
+
+	// create a dag as template
+	if err := ensureDagCreated(); err != nil {
+		log.Fatalf(err.Error())
+		return
+	}
+	// run dag interval
+	go runInstance()
+
+	// listen a http endpoint to serve metrics
+	if err := http.ListenAndServe(":9090", exporter.HttpHandler()); err != nil {
+		panic(fmt.Sprintf("metrics serve failed: %s", err))
+	}
+}
+
+func runInstance() {
+	// wait init completed
+	time.Sleep(2 * time.Second)
+	dag, err := mod.GetStore().GetDag("test-dag")
+	if err != nil {
+		panic(err)
+	}
+
+	count := uint64(0)
+	for {
+		runVar := map[string]string{
+			"var": "run-var",
+		}
+		if count%2 == 0 {
+			runVar = nil
+		}
+		dagIns, err := dag.Run(entity.TriggerManually, runVar)
+		if err != nil {
+			panic(err)
+		}
+
+		dagIns.Tags = entity.NewDagInstanceTags(map[string]string{"testKey": "testValue", "testKey2": "testValue2", "testKey3": "testValue3"})
+
+		err = mod.GetStore().CreateDagIns(dagIns)
+		if err != nil {
+			panic(err)
+		}
+
+		count++
+		time.Sleep(1 * time.Second)
+	}
+}

examples/programming/main.go

@@ -76,7 +76,8 @@ func createDagAndInstance() {
 		BaseInfo: entity.BaseInfo{
 			ID: "test-dag",
 		},
-		Name: "test",
+		Name:   "test",
+		Status: entity.DagStatusNormal,
 		Tasks: []entity.Task{
 			{ID: "task1", ActionName: "PrintAction"},
 			{ID: "task2", ActionName: "PrintAction", DependOn: []string{"task1"}},

fastflow_test.go

@@ -87,7 +87,7 @@ func Test_readDagFromDir(t *testing.T) {
 			givePathDagMap: map[string][]byte{
 				"dag1": []byte(`tasks: 123`),
 			},
-			wantErr: fmt.Errorf("unmarshal dag1 failed: %w", &yaml.TypeError{Errors: []string{"line 1: cannot unmarshal !!int `123` into []entity.Task"}}),
+			wantErr: fmt.Errorf("unmarshal dag1 failed: %w", &yaml.TypeError{Errors: []string{"line 1: cannot unmarshal !!int `123` into entity.DagTasks"}}),
 		},
 		{
 			caseDesc:  "normal",

go.mod

@@ -1,16 +1,50 @@
 module github.com/shiningrush/fastflow
 
-go 1.14
+go 1.20
 
 require (
+	github.com/go-sql-driver/mysql v1.7.0
 	github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e
 	github.com/golang/mock v1.6.0
 	github.com/mitchellh/mapstructure v1.1.2
 	github.com/prometheus/client_golang v1.14.0
 	github.com/shiningrush/goevent v0.1.0
+	github.com/shiningrush/goext v0.2.4-0.20230805045150-8b8c5748342b
 	github.com/sony/sonyflake v1.0.0
 	github.com/spaolacci/murmur3 v1.1.0
-	github.com/stretchr/testify v1.6.1
+	github.com/stretchr/testify v1.7.0
 	go.mongodb.org/mongo-driver v1.5.4
 	gopkg.in/yaml.v3 v3.0.0
+	gorm.io/driver/mysql v1.5.0
+	gorm.io/gorm v1.25.1
+)
+
+require (
+	github.com/aws/aws-sdk-go v1.34.28 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-stack/stack v1.8.0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/snappy v0.0.1 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/klauspost/compress v1.9.5 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.3.0 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/stretchr/objx v0.1.1 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.0.2 // indirect
+	github.com/xdg-go/stringprep v1.0.2 // indirect
+	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
+	golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 // indirect
+	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f // indirect
+	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
 )

go.sum

@@ -74,6 +74,8 @@ github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
 github.com/go-logfmt/logfmt v0.5.1/go.mod h1:WYhtIu8zTZfxdn5+rREduYbwxfcBr/Vr6KEVveWlfTs=
 github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
+github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc=
+github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI=
 github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gobuffalo/attrs v0.0.0-20190224210810-a9411de4debd/go.mod h1:4duuawTqi2wkkpB4ePgWMaai6/Kc6WEz83bhFwpHzj0=
@@ -147,7 +149,6 @@ github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
-github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
@@ -165,6 +166,10 @@ github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
+github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
+github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
+github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
+github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
 github.com/jmespath/go-jmespath v0.4.0 h1:BEgLn5cpjn8UN1mAw4NjwDrS35OdebyEtFe+9YPoQUg=
 github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
 github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8=
@@ -245,6 +250,8 @@ github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/shiningrush/goevent v0.1.0 h1:084IrgoL3KbudRtYSEVgnGUNNEVwG5aCvzCjAPP1G/g=
 github.com/shiningrush/goevent v0.1.0/go.mod h1:c242Xdp8/ot6idcZ2xdUVSe0I82aobcOfO9yel3PZxU=
+github.com/shiningrush/goext v0.2.4-0.20230805045150-8b8c5748342b h1:qvObgZt8h6Tgeg46fn5+INxDClWZIip5jDOB8VLrDkQ=
+github.com/shiningrush/goext v0.2.4-0.20230805045150-8b8c5748342b/go.mod h1:XAD+HxmZjdrVdmQCVmejcdG5LZRzahJT2J8koaGVBCU=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMBDgk/93Q=
@@ -262,8 +269,9 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
-github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
@@ -356,7 +364,6 @@ golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81R
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
-golang.org/x/net v0.0.0-20220225172249-27dd8689420f h1:oA4XRj0qtSt8Yo1Zms0CUlsT3KG69V2UGQWPBxujDmc=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -580,6 +587,11 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA=
 gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gorm.io/driver/mysql v1.5.0 h1:6hSAT5QcyIaty0jfnff0z0CLDjyRgZ8mlMHLqSt7uXM=
+gorm.io/driver/mysql v1.5.0/go.mod h1:FFla/fJuCvyTi7rJQd27qlNX2v3L6deTR1GgTjSOLPo=
+gorm.io/gorm v1.24.7-0.20230306060331-85eaf9eeda11/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=
+gorm.io/gorm v1.25.1 h1:nsSALe5Pr+cM3V1qwwQ7rOkw+6UeLrX5O4v3llhHa64=
+gorm.io/gorm v1.25.1/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=