Openstack 实现技术分解 (4) 通用技术 — TaskFlow
目录 目录 前文列表 扩展阅读 简介 基本概念 实现样例 最后 前文列表 Openstack 实现技术分解 (1) 开发环境 — Devstack 部署案例详解 Openstack 实现技术分解 (2) 虚拟机初始化工具 — Cloud-Init & metadata & userdata Openstack 实现技术分解 (3) 开发工具 — VIM & dotfiles 扩展阅读 TaskFlow 代码库 TaskFlow 文档 简介 TaskFlow is a Python library that helps to make task execution easy, consistent and reliable. A library to do [jobs, tasks, flows] in a highly available, easy to understand and declarative manner (and more!) to be used with OpenStack and other projects. 简而言之, TaskFlow 能够控制应用程序中的长流程业务逻辑任务的暂停、重启、恢复以及回滚, 主要用于保证长流程任务执行的可靠性和一致性。 主要应用场景有如 Cinder 的 create volume 这般复杂、冗长、容易失败, 却又要求保持数据与环境一致的业务逻辑. 从 create volume 流程图看, Cinder 在 create_volume.py(cinder/volume/flows/manager/create_volume.py) 模块中定义了大量的 Tasks class 来组成 TaskFlow: OnFailureRescheduleTask ExtractVolumeRefTask ExtractVolumeSpecTask NotifyVolumeActionTask CreateVolumeFromSpecTask CreateVolumeOnFinishTask 如果在执行任务流的过程中失败了, TaskFlow 的回滚机制能够让程序流和执行环境回滚到初始状态, 并且可以重新开始执行. 总的来说, TaskFlow 非常适合于 面向任务 的应用场景. 基本概念 Atom: An atom is the smallest unit in TaskFlow which acts as the base for other classes Atom: Atom 是 TaskFlow 的最小单位, 其他的所有类, 包括 Task 类都是 Atom 类的子类. Task: A task (derived from an atom) is a unit of work that can have an execute & rollback sequence associated with it (they are nearly analogous to functions). Task: task 是拥有执行和回滚功能额最小工作单元. 在 Task 类中开发者能够自定义 execute(执行) 和 revert(回滚) method. Flow: Linear/Unordered/Graph Flow: 在 TaskFlow 中使用 flow(流) 来关联各个 Task, 并且规定这些 Task 之间的执行和回滚顺序. flow 中不仅能够包含 task 还能够嵌套 flow. 常见类型有以下几种: Linear(linear_flow.Flow): 线性流, 该类型 flow 中的 task/flow 按照加入的顺序来依次执行, 按照加入的倒序依次回滚. Unordered(unordered_flow.Flow): 无序流, 该类型 flow 中的 task/flow 可能按照任意的顺序来执行和回滚. Graph(graph_flow.Flow): 图流, 该类型 flow 中的 task/flow 按照显式指定的依赖关系或通过其间的 provides/requires 属性的隐含依赖关系来执行和回滚. Retry: A retry (derived from an atom) is a special unit of work that handles errors, controls flow execution and can (for example) retry other atoms with other parameters if needed. Retry: Retry 是一个控制当错误发生时, 如何进行重试的特殊工作单元, 而且当你需要的时候还能够以其他参数来重试执行别的 Atom 子类. 常见类型: AlwaysRevert: 错误发生时, 回滚子流 AlwaysRevertAll: 错误发生时, 回滚所有流 Times: 错误发生时, 重试子流 ForEach: 错误发生时, 为子流中的 Atom 提供一个新的值, 然后重试, 直到成功或 retry 中定义的值用完为止. ParameterizedForEach: 错误发生时, 从后台存储(由 store 参数提供)中获取重试的值, 然后重试, 直到成功或后台存储中的值用完为止. Engine: Engines are what really runs your atoms. Engine: Engines 才是真正运行 Atoms 的对象, 用于 load(载入) 一个 flow, 然后驱动这个 flow 中的 task/flow 开始运行. 我们可以通过 engine_conf 参数来指定不同的 engine 类型. 常见的 engine 类型如下: serial: 所有的 task 都在调用了 engine.run 的线程中运行. parallel: task 可以被调度到不同的线程中运行. worker-based: task 可以被调度到不同的 woker 中运行. 实现样例 源码请浏览 Github #!/usr/bin/env python #filename: tasks.py import taskflow.engines from taskflow.patterns import linear_flow as lt from taskflow import task from taskflow.types import failure as task_failed class CallJim(task.Task): default_provides = set(['jim_new_number']) def execute(self, jim_number, *args, **kwargs): print "Calling Jim %s." % jim_number print '=' * 10 jim_new_number = jim_number + 'new' return {'jim_new_number': jim_new_number} def revert(self, result, *args, **kwargs): if isinstance(result, task_failed.Failure): print "jim result" return None jim_new_number = result['jim_new_number'] print "Calling jim %s and apologizing." % jim_new_number class CallJoe(task.Task): default_provides = set(['joe_new_number', 'jim_new_number']) def execute(self, joe_number, jim_new_number, *args, **kwargs): print "Calling jim %s." % jim_new_number print "Calling Joe %s." % joe_number print '=' * 10 joe_new_number = joe_number + 'new' return {'jim_new_number': jim_new_number, 'joe_new_number': joe_new_number} def revert(self, result, *args, **kwargs): if isinstance(result, task_failed.Failure): print "joe result" return None jim_new_number = result['jim_new_number'] joe_new_number = result['joe_new_number'] print "Calling joe %s and apologizing." % joe_new_number class CallJmilkFan(task.Task): default_provides = set(['new_numbers']) def execute(self, jim_new_number, joe_new_number, jmilkfan_number, *args, **kwargs): print "Calling jim %s" % jim_new_number print "Calling joe %s" % joe_new_number print "Calling jmilkfan %s" % jmilkfan_number print '=' * 10 jmilkfan_new_number = jmilkfan_number + 'new' raise ValueError('Error') new_numbers = {'jim_new_number': jim_new_number, 'joe_new_number': joe_new_number, 'jmilkfan_new_number': jmilkfan_new_number} return {'new_numbers': new_numbers} def revert(self, result, *args, **kwargs): if isinstance(result, task_failed.Failure): print "jmilkfan result" return None jim_new_number = result['jim_new_number'] joe_new_number = result['joe_new_number'] jmilkfan_new_number = result['jmilkfan_new_number'] print "Calling jmilkfan %s and apologizing." % jmilkfan_new_number def get_flow(flow, numbers): flow_name = flow flow_api = lt.Flow(flow_name) flow_api.add(CallJim(), CallJoe(), CallJmilkFan()) return taskflow.engines.load(flow_api, engine_conf={'engine': 'serial'}, store=numbers) def main(): numbers = {'jim_number': '1'*6, 'joe_number': '2'*6, 'jmilkfan_number': '3'*6} try: flow_engine = get_flow(flow='taskflow-demo', numbers=numbers) flow_engine.run() except Exception: print "TaskFlow Failed!" raise new_numbers = flow_engine.storage.fetch('new_numbers') if __name__ == '__main__': main() Output: fanguiju@fanguiju:~/project/my-code-repertory/TaskFlow-demo$ python tasks.py Calling Jim 111111. ========== Calling jim 111111new. Calling Joe 222222. ========== Calling jim 111111new Calling joe 222222new Calling jmilkfan 333333 ========== jmilkfan result Calling joe 222222new and apologizing. Calling jim 111111new and apologizing. TaskFlow Failed! Traceback (most recent call last): File "tasks.py", line 114, in <module> main() File "tasks.py", line 105, in main flow_engine.run() File "/usr/local/lib/python2.7/dist-packages/taskflow/engines/action_engine/engine.py", line 159, in run for _state in self.run_iter(): File "/usr/local/lib/python2.7/dist-packages/taskflow/engines/action_engine/engine.py", line 223, in run_iter failure.Failure.reraise_if_any(it) File "/usr/local/lib/python2.7/dist-packages/taskflow/types/failure.py", line 292, in reraise_if_any failures[0].reraise() File "/usr/local/lib/python2.7/dist-packages/taskflow/types/failure.py", line 299, in reraise six.reraise(*self._exc_info) File "/usr/local/lib/python2.7/dist-packages/taskflow/engines/action_engine/executor.py", line 82, in _execute_task result = task.execute(**arguments) File "tasks.py", line 66, in execute raise ValueError('Error') ValueError: Error NOTE 1: 在 function get_flow 中使用 linear_flow.Flow 生成一个 TaskFlow(线性任务流) 对象 flow_api , 再通过flow_api.add method 添加要 顺序执行且倒序回滚 的 Task class(CallJim/CallJom/CallJmilkFan). NOTE 2: 使用 taskflow.engines.load method 来加载 TaskFlow(flow_api)对象/后台存储数据(store)/ engine配置 等信息并生成 Task Engine 对象. NOTE 3: 最后调用 Task Engine 对象的 flow_engine.run method 来开始执行该任务流. NOTE 4: 后台存储 store 的数据在该任务流中被所有 Task class 共享, 并且以 Task class 中的 execute method 的形参作为对接入口. e.g. 上述实现的 store 后台存储中含有 {jim_number: '1'*6}, 那么 CallJim 的 execute method 就可以通过形参 jim_number 来获取 '1'*6 的值. NOTE 5: Task class 的属性 default_provides 用于声明在执行过程中新添到后台存储的元素的名称, 其相应的值会自动的从 execute method 返回值中匹配获取, 最终存储后台存储. e.g. CallJim 的属性 default_provides = set(['jim_new_number']) 其中 jim_new_number 的值会从 execute method 的返回 return {'jim_new_number': jim_new_number} 中获取. NOTE 6: provides 的实现能够有效的帮助传递 Task class 之间在执行时产生的新属性对象. 将上一个 Task 的结果传递给后一个 Task 使用. 最后 当实现的 TaskFlow 中包含了多个 Task(的确可能存在只有一个 Task 的 TaskFlow) 时, 有两点是需要注意的: 在使用线性流类型的 TaskFlow 时, Task class revert method 回滚的应该是上一个 Task class execute method 的业务. e.g. BTask.revert 应该回滚 ATask.execute, 因为只有在 ATask.execute 成功执行的前提之下才有 revert 的价值. 所以在 revert method 的定义中需要实现语句 if isinstance(result, task_failed.Failure): return None. 当一个 Task 的 execute method 执行失败时, 那么 revert method 接收的 result 实参就是 taskflow.types.failure.Failure 的实例对象. 尽量让每个 Task class 都仅处理一件事情, 这是为了让每一次回滚都足够精准. e.g. 尽管创建虚拟机和开启虚拟机都同属于对虚拟机的操作, 但是我们仍然应该将两者各自定义一个 Task class. 假如启动虚拟机失败时, 我们只需再次重试启动虚拟机, 而无须再次重复创建虚拟机.
)
