再次咨询，action not found问题

[chenxushao]

情况是这样的，fastflow部署了一共2台机器。运行的时候，经常报action not found
报这个错误的代码在这里：
execturor.go
func (e *DefExecutor) runAction(taskIns *entity.TaskInstance) error {
act := ActionMap[taskIns.ActionName]
if act == nil {
return fmt.Errorf("action not found: %s", taskIns.ActionName)
}

再翻看源代码，ActionMap 是定义的一个全局变量。
var (
ActionMap = map[string]run.Action{}

defExc Executor
defStore Store
defKeeper Keeper
defParser Parser
defCommander Commander
)

Action 的注册，是将 Action 放到这个 Map 中，相当于数据是存储在‹内存›中，并没有持久化。
在程序中，我是这么注册Action 的。

// 注册 action
fastflow.RegisterAction(actions)

我看到官方文档上是这样描述：
当你开始运行一个 Dag 后，则会为本次执行生成一个执行记录，它被称为 DagInstance，当它生成以后，会由 Leader 实例将其分发到一个健康的 Worker，再由其解析、执行。

假设fastflow.RegisterAction(actions)在2台机器中的A执行，那么是不有可能由另一台机器去执行，是不是因为这个造成的报action not found呢？

[chenxushao]

另外，部署的代码是同一套。

[chenxushao]

Dag和注册Action我这边均是由前端发起请求，动态注册的，假如是A机器收到请求，那么这个action肯定是在A机器了。
但是真正的执行Action根据文档描述，可能是由另一台worker机器B执行，这是不是出现了action not found的根因呢

[chenxushao]

诚盼回复。

[philhuan]

这里 action 的注册用法是在服务启动的时候注册，不是通过接口注册的。

[philhuan]

要注册得有一个实现了接口的结构体吧，通过接口注册怎么关联这个结构体，新增action 得改代码吧？

[chenxushao]

动态不支持吗，比如新增一个编排任务，这时候会新创建dag和新注册action，应该是比较合理的需求。

[chenxushao]

要注册得有一个实现了接口的结构体吧，通过接口注册怎么关联这个结构体，新增action 得改代码吧？

您好，是有两个固化的结构体，只是每次注册name不一样，因为处理的逻辑不一样。

[ShiningRush]

action not found的唯一原因就是它没注册上，看你描述，可能导致这个现象有几个点：

1. 你的注册逻辑可能晚于fastflow的初始化
2. 注册的actionname不对，或者注册失败

[chenxushao]

action not found的唯一原因就是它没注册上，看你描述，可能导致这个现象有几个点：

1. 你的注册逻辑可能晚于fastflow的初始化
2. 注册的actionname不对，或者注册失败

你好。 fasttflow的初始化，只需要也只能够执行一次对吗？

目前fastflow的初始化，是在go main方法中进行的。
之后，通过controller（使用的gin框架）来触发提交的任务。

在main语文教学中初始化fastflow的代码如下：

func InitFlow() {
fastFlowConf := configs.Get().FastFlow
connStr := fastFlowConf.ConnStr
dataBase := fastFlowConf.Database
prefix := fastFlowConf.Prefix

workKeyNum := BuildWorkNum()
logrus.Infof("InitFlow. workKeyNum:%d", workKeyNum)

// 初始化选举组件
Keeper = mongoKeeper.NewKeeper(&mongoKeeper.KeeperOption{
	Key: "worker-" + util.StringUtils.AsString(workKeyNum),
	// if your mongo does not set user/pwd, you should remove it
	ConnStr:  connStr,
	Database: dataBase,
	Prefix:   prefix,
})

logrus.Infof("init keeper begin. keeper:%v", Keeper)
if err := Keeper.Init(); err != nil {
	errMsg := fmt.Sprintf("init keeper failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}

logrus.Infof("init keeper finish. workKey:%s", Keeper.WorkerKey())

isLeader := Keeper.IsLeader()
if isLeader {
	notice.RobotNotice(fmt.Sprintf("keeper leader ip:%s，workKey:%s", runtimes.AppInfo.Ip, Keeper.WorkerKey()))
}

// 初始化存储组件
st := mongoStore.NewStore(&mongoStore.StoreOption{
	// if your mongo does not set user/pwd, you should remove it
	ConnStr:  connStr,
	Database: dataBase,
	Prefix:   prefix,
})

logrus.Infof("init store begin. st:%v", st)
if err := st.Init(); err != nil {
	errMsg := fmt.Sprintf("init store failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}
logrus.Info("init store finish")

// 启动fastflow
logrus.Info("start flow begin")
if err := fastflow.Start(&fastflow.InitialOption{
	Keeper:             Keeper,
	Store:              st,
	DagScheduleTimeout: 300 * time.Second,
	ExecutorTimeout:    150 * time.Second,
}); err != nil {
	errMsg := fmt.Sprintf("init fastflow failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}
logrus.Info("start flow finish")

}

之后，通过http请求来构建Dag并触发执行。代码形如下：

dagVars, dagVarMap := BuildDagVars(globalVarItems)
tasks, actions := BuildTagElement(rawDagBO.TaskNodes)

// 注册 action
fastflow.RegisterAction(actions)
logrus.Infof("RegisterAction finish.dagExplainId:%d", dagExplainId)

//构建dag模型
dag := initialize.CreateDag(util.StringUtils.AsString(dagId), rawDagBO.Name, rawDagBO.Descr, tasks, dagVars)

ch := make(chan string)
//创建dag并运行
go initialize.CreateDagAndInstance(dag, dagVarMap, ch)

这种使用方式是可行的吗？

[chenxushao]

通过阅读源码， fastflow.RegisterAction(actions) 只会在接收到请求的那台机器的内存中（是一个map)才会有数据。我想的是，之所以报action not found，是否是由于具体的执行被调度到另一台机器上了。 另外，还有一个问题，一个Dag上的N外节点，只会被同一台机器执行吧？因为一个dag节点还存在数据共享和传递。

[chenxushao]

action not found的唯一原因就是它没注册上，看你描述，可能导致这个现象有几个点：

1. 你的注册逻辑可能晚于fastflow的初始化
2. 注册的actionname不对，或者注册失败

第1点有一个疑问：
fastflow.RegisterAction
必须在
fastflow.start之前执行是吗？

[ShiningRush]

action not found的唯一原因就是它没注册上，看你描述，可能导致这个现象有几个点：

1. 你的注册逻辑可能晚于fastflow的初始化
2. 注册的actionname不对，或者注册失败

第1点有一个疑问： fastflow.RegisterAction 必须在 fastflow.start之前执行是吗？

最好是，最晚必须在工作流实例被执行前，否则就会在action集合中找不到

[ShiningRush]

通过阅读源码， fastflow.RegisterAction(actions) 只会在接收到请求的那台机器的内存中（是一个map)才会有数据。我想的是，之所以报action not found，是否是由于具体的执行被调度到另一台机器上了。 另外，还有一个问题，一个Dag上的N外节点，只会被同一台机器执行吧？因为一个dag节点还存在数据共享和传递。

你的理解是对的，同一个DagInstance只会被同一个worker执行

[chenxushao]

action not found的唯一原因就是它没注册上，看你描述，可能导致这个现象有几个点：

1. 你的注册逻辑可能晚于fastflow的初始化
2. 注册的actionname不对，或者注册失败

你好。 fasttflow的初始化，只需要也只能够执行一次对吗？

目前fastflow的初始化，是在go main方法中进行的。 之后，通过controller（使用的gin框架）来触发提交的任务。

在main语文教学中初始化fastflow的代码如下：

func InitFlow() { fastFlowConf := configs.Get().FastFlow connStr := fastFlowConf.ConnStr dataBase := fastFlowConf.Database prefix := fastFlowConf.Prefix

workKeyNum := BuildWorkNum()
logrus.Infof("InitFlow. workKeyNum:%d", workKeyNum)

// 初始化选举组件
Keeper = mongoKeeper.NewKeeper(&mongoKeeper.KeeperOption{
	Key: "worker-" + util.StringUtils.AsString(workKeyNum),
	// if your mongo does not set user/pwd, you should remove it
	ConnStr:  connStr,
	Database: dataBase,
	Prefix:   prefix,
})

logrus.Infof("init keeper begin. keeper:%v", Keeper)
if err := Keeper.Init(); err != nil {
	errMsg := fmt.Sprintf("init keeper failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}

logrus.Infof("init keeper finish. workKey:%s", Keeper.WorkerKey())

isLeader := Keeper.IsLeader()
if isLeader {
	notice.RobotNotice(fmt.Sprintf("keeper leader ip:%s，workKey:%s", runtimes.AppInfo.Ip, Keeper.WorkerKey()))
}

// 初始化存储组件
st := mongoStore.NewStore(&mongoStore.StoreOption{
	// if your mongo does not set user/pwd, you should remove it
	ConnStr:  connStr,
	Database: dataBase,
	Prefix:   prefix,
})

logrus.Infof("init store begin. st:%v", st)
if err := st.Init(); err != nil {
	errMsg := fmt.Sprintf("init store failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}
logrus.Info("init store finish")

// 启动fastflow
logrus.Info("start flow begin")
if err := fastflow.Start(&fastflow.InitialOption{
	Keeper:             Keeper,
	Store:              st,
	DagScheduleTimeout: 300 * time.Second,
	ExecutorTimeout:    150 * time.Second,
}); err != nil {
	errMsg := fmt.Sprintf("init fastflow failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}
logrus.Info("start flow finish")

}

之后，通过http请求来构建Dag并触发执行。代码形如下：

dagVars, dagVarMap := BuildDagVars(globalVarItems)
tasks, actions := BuildTagElement(rawDagBO.TaskNodes)

// 注册 action
fastflow.RegisterAction(actions)
logrus.Infof("RegisterAction finish.dagExplainId:%d", dagExplainId)

//构建dag模型
dag := initialize.CreateDag(util.StringUtils.AsString(dagId), rawDagBO.Name, rawDagBO.Descr, tasks, dagVars)

ch := make(chan string)
//创建dag并运行
go initialize.CreateDagAndInstance(dag, dagVarMap, ch)

这种使用方式是可行的吗？

这种方式可行吗？

[chenxushao]

action not found的唯一原因就是它没注册上，看你描述，可能导致这个现象有几个点：

1. 你的注册逻辑可能晚于fastflow的初始化
2. 注册的actionname不对，或者注册失败

你好。 fasttflow的初始化，只需要也只能够执行一次对吗？
目前fastflow的初始化，是在go main方法中进行的。 之后，通过controller（使用的gin框架）来触发提交的任务。
在main语文教学中初始化fastflow的代码如下：
func InitFlow() { fastFlowConf := configs.Get().FastFlow connStr := fastFlowConf.ConnStr dataBase := fastFlowConf.Database prefix := fastFlowConf.Prefix

workKeyNum := BuildWorkNum()
logrus.Infof("InitFlow. workKeyNum:%d", workKeyNum)

// 初始化选举组件
Keeper = mongoKeeper.NewKeeper(&mongoKeeper.KeeperOption{
	Key: "worker-" + util.StringUtils.AsString(workKeyNum),
	// if your mongo does not set user/pwd, you should remove it
	ConnStr:  connStr,
	Database: dataBase,
	Prefix:   prefix,
})

logrus.Infof("init keeper begin. keeper:%v", Keeper)
if err := Keeper.Init(); err != nil {
	errMsg := fmt.Sprintf("init keeper failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}

logrus.Infof("init keeper finish. workKey:%s", Keeper.WorkerKey())

isLeader := Keeper.IsLeader()
if isLeader {
	notice.RobotNotice(fmt.Sprintf("keeper leader ip:%s，workKey:%s", runtimes.AppInfo.Ip, Keeper.WorkerKey()))
}

// 初始化存储组件
st := mongoStore.NewStore(&mongoStore.StoreOption{
	// if your mongo does not set user/pwd, you should remove it
	ConnStr:  connStr,
	Database: dataBase,
	Prefix:   prefix,
})

logrus.Infof("init store begin. st:%v", st)
if err := st.Init(); err != nil {
	errMsg := fmt.Sprintf("init store failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}
logrus.Info("init store finish")

// 启动fastflow
logrus.Info("start flow begin")
if err := fastflow.Start(&fastflow.InitialOption{
	Keeper:             Keeper,
	Store:              st,
	DagScheduleTimeout: 300 * time.Second,
	ExecutorTimeout:    150 * time.Second,
}); err != nil {
	errMsg := fmt.Sprintf("init fastflow failed: %s", err.Error())
	logrus.Error(errMsg)
	notice.RobotNotice(errMsg)
	return
}
logrus.Info("start flow finish")

}
之后，通过http请求来构建Dag并触发执行。代码形如下：

dagVars, dagVarMap := BuildDagVars(globalVarItems)
tasks, actions := BuildTagElement(rawDagBO.TaskNodes)

// 注册 action
fastflow.RegisterAction(actions)
logrus.Infof("RegisterAction finish.dagExplainId:%d", dagExplainId)

//构建dag模型
dag := initialize.CreateDag(util.StringUtils.AsString(dagId), rawDagBO.Name, rawDagBO.Descr, tasks, dagVars)

ch := make(chan string)
//创建dag并运行
go initialize.CreateDagAndInstance(dag, dagVarMap, ch)

这种使用方式是可行的吗？

这种方式可行吗？

目前这样，就是遇到 了action not found异常。 我部署了2台机器