fbloc / bloc-server Goto Github PK

View Code? Open in Web Editor NEW

4.0 0.0 2.0 590 KB

License: MIT License

Go 100.00%

bloc-server's Introduction

bloc-server

中文

In a global view, bloc's code repo has below relationship:

about bloc-server

responsibility

api server for bloc-frontend
scheduler for trigger flow、function's run, also include schedule retry、handle intercept、timeout...
store data

relied on:

mongoDB: Database. Use to store functions、flow、run_record...'s infomation. Tested version: 5.0.5
rabbitMQ: MQ. Use to delivery trigger flow/function run msg. Tested version: 3.9.11
minio: Object storage. Used to store function run's output data. Tested version: RELEASE.2021-11-24T23-19-33Z
influxDB: Time series database. Used to store logs. Tested version: 2.1.1

how to run bloc-server

if you just want a local bloc environment（include both upper requirements、bloc-server、bloc-frontend）which can be used to receive your function's register and provide frontend ui. Just follow this tutorial.

after deployed above requirement, you can start up bloc-server by:

$ go run cmd/server/main.go --app_name="$your_app_name" --rabbitMQ_connection_str="$rabbit_user:$rabbit_password@$rabbitMQ_server_address" --minio_connection_str="$minio_user:$minio_password@$mioio_server_address" --mongo_connection_str="$mongodb_user:$mongodb_password@$mongodb_address" --influxdb_connection_str="$influxDB_user:$influxDB_password@$influxDB_address/?token=$influxDB_token&organization=bloc

bloc-server's People

Contributors

Stargazers

Forkers

richardweiu shaoxyz

bloc-server's Issues

入口的RunConsumer需要运行heartbeat watcher，用于重新发布运行失败的任务

采用配置式文件来启动项目，这样有default的值，更为清晰明了

具体有哪些字段我先梳理下后再启动开发

提交draft flow时候的CheckValid方法，缺失了检测当参数输入类型是连接时候的参数数据类型是否一致的检测

对应代码逻辑在aggregate/flow下的CheckValid方法中。

拆分「注册中心」和「function provider client」

正确的处理一种运行情况：function是正常运行成功了的，但依赖其输入或特定条件，此function此次决定拦截此flow的继续运行

flow的运行控制参数AllowParallelRun发挥实际的作用

Flow的运行控制参数AllowParallelRun（是否允许正在运行时再次运行）没有实际在代码中起作用（目前前端已经能有效设置）。需要在后端代码中实际实现此功能：当设置为False的时候，如果一个flow正在运行，将不允许再次运行

测试代码

权限相关实现说明

user

user 权限相关字段有 super，为真表示拥有全局的全部权限 + 能够进入 admin 界面

function

function 权限相关字段有 read 、 execute，assign_permission。具体含义如下：

read：为真的话，能够在 flow 下面查看此 function 的运行历史
execute：为真的话，用户创建新的 flow 的时候，可选择/使用此 function
assign_permission：能够分配其他用户对此 function 的 read/execute/assign_permission 权限

特别说明一个情况，如果 read 为真、execute 为假：表示用户在创建自己的 flow 的时候不能使用此 function，但是能够查看已有此 function 的 flow 中的此 function 及其运行记录

flow

flow 权限相关字段有 read 、 write 、 execute 、 delete 、 assign_permission。具体含义如下：

read：可读此 flow（当前用户获取到的 flow 列表有此项、能够查看此 flow 具体及其运行历史）
write ：可以更改此 flow（可以改此 flow 下的 function 构成/关系，以及每个 function 的参数输入）
execute ：可以在前端触发运行此 flow & 可以修改此 flow 的执行控制参数(crontab、trigger_key...)
delete：能够删除此 flow
assign_permission：能够分配其他用户对此 flow 的 read/write/execute/delete/assign_permission 权限

调整 api/v1/flow/get_latestonline_by_origin_id/${originId} 接口

删除返回体中的flowFunctionID_map_status字段

返回体中新增flowFunctionID_map_info字段，值如下：

{
  ...
  flowFunctionID_map_info: {
    /**
     * 运行结果
     */
    status: 6,
    /**
     * 对应的详情ID
     */
    record_id: "xxxx"
  }
}

完善返回体中的latest_run值，使得和/api/v1/flow_run_record?${flowId}接口返回的值保持一致。如：

{
  ...
  latest_run: {
    arrangement_id: "",
    flow_id: "xxx",
    flow_origin_id: "xxx",
    trigger_type: 1,
    trigger_key: "",
    trigger_source: 2,
    trigger_user_name: "bloc",
    trigger_time: "xxx",
    start_time: "xxx",
    end_time: "xxx",
    status: 6,
    error_msg: "",
    retried_amount: 0,
    timeout_canceled: false,
    canceled: false,
    cancel_user_name: ""
  }
}

// 目前结构
{
  status_code: 200,
  status_message: "",
  data: []
}

// 希望改为如下结构
{
  status_code: 200,
  status_message: "",
  data: {
    total: 100,
    data: []
  }
}

crontab watcher doc

关联commit
提交前已本地测试过

代码相关

主要代码：./crontab_watcher.go
调用主入口：./consumer.go

背景

发布配置了crontab出发周期的flow的脚本可能（也应该）在多个机器上同时运行

所以存在可能重复发布的并发问题

当前如何解决

因为目前存储只用了mongo，故这里只先说在mongo下的解决原理：

下面是在mongo shell用一个测试库测试的：

> db.tmp.findOneAndUpdate({"field": "the unique flag"}, {"$setOnInsert": { foo: "bar" }}, {"returnOriginal": true, "upsert": true})
< null
> db.tmp.findOneAndUpdate({"field": "the unique flag"}, {"$setOnInsert": { foo: "bar" }}, {"returnOriginal": true, "upsert": true})
< { _id: ObjectId("618f71c4162a046430631516"),
  field: 'the unique flag',
  foo: 'bar' }

可见，在没有对应文档时，第一次运行返回老文档就是null，第二次运行，返回了老文档。可利用此原理实现原子性，实现同crontab精度下只发布一次的目的

findOneAndUpdate三个参数说明：

第一个参数是filter、用于查找是否有满足此条件的文档
第二个参数利用$setOnInsert，当查找到没有满足条件的文档时，就设置$setOnInsert后面的值
第三个参数这里需要设置两个参数：
1. "returnOriginal": true：返回老的文档（此时如果是新插入的话就会返回空）
2. "upsert": true，不存在就插入