运筹决策 BaiCe(百策)
运筹决策 BaiCe(百策)
1 系统设计
+-----------------------------------------------------------+
| +source-------+ |
| | +---------+ | |
| | | bianque | | +-----------+ +------------+ |
| | +---------+ | event | | action | | |
| | +------>| baice +------->| xingqiao | |
| | +---------+ | | | | | |
| | | xxx | | +-----+-----+ +------------+ |
| | +---------+ | | |
| +-------------+ | |
| +-----V-----+ |
| | DB | |
| +-----------+ |
+-----------------------------------------------------------+1.1 存储 cloudevent
将 cloudevent 存储
1.2 发布事件
发布 cloudevent create 事件
发布 cloudevent 数据变化事件
1.3 执行决策(插件)
执行相应插件，发起 action(可以创建星桥 workflow，也可以直接执行某个任务)
局部决策:
监听 source create 事件, 进行决策
监听 source 数据变化事件, 进行决策(IFTTT: if this than that)
全局决策
定时任务
可以设置冷却期，冷却期间不再发 action 任务
2 上下游
| 外部输入
+----------------------------------------------------------|---------------------------------------+
| +-------------------------V------------------------------+ |
| +ruqi-----------------------+ | +baichuan-----------+ +callback----------------+ | |
| | target + cron | | | | | | | |
| +------------+--------------+ | +---------+---------+ +------------+-----------+ | |
| | +------------+----------------------------+--------------+ |
| | | | |
| |1a |1b |1c |
| V V V |
| +----------------------------------------------------------------------------------------------+ |
| | queue | |
| +------------------^------------------------------^-------------------------------^------------+ |
| |2 |3 |4 |
| +bianque_agent-----+--------------+ +baice-------+--------------+ +xingqiao-----+------------+ |
| | +plugin-----------------------+ | |+-------++-------++-------+| |+plugin------------------+| | 故障自愈
| | | detection + trigger | | || baice || baice || baice || ||unit_migrate/delete/copy|| |
| | +-----------------------------+ | |+-------++-------++-------+| ||hotkey/bigkey || |
| +---------------------------------+ +------------+--------------| ||... || |
| | |+------------------------+| |
| +DB----------V--------------+ +--------------------------+ |
| | | |
| +---------------------------+ |
| |
| <sense + trigger> <save + decision> <execute> |
+--------------------------------------------------------------------------------------------------+
+--------------------------------------------------------------------------------------------------+
| +------------+ +------------+ +------------+ +------------+ +------------+ |
| | huoyan | | ruqi | | wuxing | | baichuan | | xingqiao | | 通用服务
| +------------+ +------------+ +------------+ +------------+ +------------+ |
+--------------------------------------------------------------------------------------------------+
〖扁鹊〗产生一个 event(cloudevent)，将其 push 到消息队列中
〖百策〗将 event 进行存储，并根据 rule 规则生成任务
存储 cloudevent
发布 cloudevent create 事件
发布 cloudevent 数据变化事件
〖星桥〗工作流进行执行具体任务
3 实践
3.1 volume_resize_demo
import logging
from xlib import db
from xlib.util import http_util
from handlers.baice import base_rule
baichuan_cache = db.my_caches["baichuan"]
# 300s 内同一个 event_target 仅能触发一次操作
rate_limit = baichuan_cache.rate_limit('volume_resize', limit=1, per=300)
logger = logging.getLogger("plugin")
class Rule(base_rule.BaseRule):
"""
Rule
"""
def __init__(self):
self.endpoint = "http://<ip>:<port>"
def decision(self, event_obj, event_count):
"""
决策
event_obj.data
{
"data":{
"volume_cursize_in_gb":600
"volume_id":"v-F5RIqyMQ"
}
"mnt_disk_free_value":48.3838726679484
}
Returns:
action_result
"""
if rate_limit.limit(event_obj.event_target):
logger.info("event_target={}, event_count={}, status=skip, msg=rate_limit".format(
event_obj.event_target, event_count))
return {"stat": "ERR_RATE_LIMIE"}
return self._action(event_obj)
def _action(self, event_obj):
"""
执行
"""
data = {
"job_namespace": "scs",
"job_type": "volume_resize",
"job_extra": {
"region": event_obj.event_region,
"host_uuid": event_obj.event_target,
"volume_incrsize_in_gb": 50,
"volume_cursize_in_gb": event_obj.data["data"]["volume_cursize_in_gb"]
}
}
res = http_util.post_json("{endpoint}/xingqiao/create_job".format(endpoint=self.endpoint),
data=data,
check_key="stat",
check_value="OK"
)
if not res.success():
return {"stat": "ERR_CREATE_JOB_FAILED"}
return {"stat": "OK", "job_id": res.output()["job_id"]}
def setup(app):
"""
插件注册函数
"""
app.register_formatter(
# 事件来源
source="scs_volume_free_check",
# 水平触发(level) 和 边缘触发(edge) 方式
watch_method="level",
# 事件状态
watch_status_list=["ERR_VOLUME_FREE_TOO_SMALL"],
formatter=Rule)Last updated