Openstack Periodic Task

周期性任务在各个模块的manager.py（computer，scheduler，cell，network）中添加。

添加方法：在模块manager类实现中添加方法，并用装饰器periodic_task.periodic_task装饰。

实现机制

以ComputerManager类为例。

模块mananger类添加方法

在资源上报方法为例，装饰器指定该方法为周期性任务，并通过spacing参数指定任务执行间隔，不指定spacing参数则默认为60.0s。

nova.computer.manager.ComputerManager

@periodic_task.periodic_task(spacing=CONF.update_resources_interval)

      　　def update_available_resource

       　　 　　…

periodic_task装饰器

periodic_task装饰器指明方法是周期性任务，并给周期性方法增加一些属性，包括周期型任务名称，任务执行间隔等。

oslo_service.periodic_task.periodic_task

def periodic_task(*args, **kwargs):

    ...

Manager类继承关系

nova.computer.manager.ComputerManager

nova.manager.Manager

nova.manager.PeriodicTask

oslo_service.periodic_task.PeriodicTask

PeriodicTask类对象通过元类PeriodicTasksMeta创建。

@six.add_metaclass(_PeriodicTasksMeta)

class PeriodicTasks(object):

    　　...

PeriodicTasksMeta元类

PeriodicTasksMeta元类收集模块manager类中定义的周期性任务，并保存在_periodic_tasks列表中。

oslo_service.PeriodicTasksMeta

DEFAULT_INTERVAL = 60.0

class _PeriodicTasksMeta(type):

    def _add_periodic_task(cls, task):

       　　 """Add a periodic task to the list of periodic tasks.

       　　 The task should already be decorated by @periodic_task.

        　　:return: whether task was actually enabled

        　　"""

        　　name = task._periodic_name

        　　…

	　　if task._periodic_spacing == 0:

	　　　　task._periodic_spacing = DEFAULT_INTERVAL

        　　cls._periodic_tasks.append((name, task))

        　　cls._periodic_spacing[name] = task._periodic_spacing

       　　return True

def __init__(cls, names, bases, dict_):

   　　 """Metaclass that allows us to collect decorated periodic tasks."""

	　　…

         　　for value in cls.__dict__.values():

            　　　　if getattr(value, '_periodic_task', False):

                　　　　　　cls._add_periodic_task(value)

触发执行周期性任务

Computer进程启动时调用service的start方法，start方法中创建定时器，定时执行periodic_tasks方法。

1、nova.service.Service.

class Service(service.Service):

    """Service object for binaries running on hosts.

    def start(self):

        　　…

        　　if self.periodic_enable:

            　　　　if self.periodic_fuzzy_delay:

                　　　　　　initial_delay = random.randint(0, self.periodic_fuzzy_delay)

           　　　　 else:

                　　　　　　initial_delay = None

            　　　　self.tg.add_dynamic_timer(self.periodic_tasks,

                                     　　　　　　　　initial_delay=initial_delay,

                                     　　　　　　　　periodic_interval_max=

                                        　　　　　　　　self.periodic_interval_max)

def periodic_tasks(self, raise_on_error=False):

    　　"""Tasks to be run at a periodic interval."""

    　　ctxt = context.get_admin_context()

    　　return self.manager.periodic_tasks(ctxt, raise_on_error=raise_on_error)

2、nova.manager.Manger

class Manager(base.Base, PeriodicTasks):

    　　def periodic_tasks(self, context, raise_on_error=False):

        　　　　"""Tasks to be run at a periodic interval."""

        　　　　return self.run_periodic_tasks(context, raise_on_error=raise_on_error)

3、oslo_service.periodic_task.PeriodicTask

run_periodic_tasks方法遍历执行_periodic_tasks中收集的周期性任务。

class PeriodicTasks(object):

    　　def run_periodic_tasks(self, context, raise_on_error=False):

        　　　　"""Tasks to be run at a periodic interval."""

        　　　　idle_for = DEFAULT_INTERVAL

        　　　　for task_name, task in self._periodic_tasks:

            　　　　　　if (task._periodic_external_ok and not

               　　　　　　self.conf.run_external_periodic_tasks):

                　　　　　　continue

            　　　　cls_name = reflection.get_class_name(self, fully_qualified=False)

            　　　　full_task_name = '.'.join([cls_name, task_name])

            　　　　spacing = self._periodic_spacing[task_name]

            　　　　last_run = self._periodic_last_run[task_name]

            　　　　# Check if due, if not skip

            　　　　idle_for = min(idle_for, spacing)

            　　　　if last_run is not None:

                　　　　　　delta = last_run + spacing - now()

                　　　　　　if delta > 0:

                    　　　　　　　　idle_for = min(idle_for, delta)

                    　　　　　　　　continue

            　　　　LOG.debug("Running periodic task %(full_task_name)s",

                      　　　　　　　　{"full_task_name": full_task_name})

            　　　　self._periodic_last_run[task_name] = _nearest_boundary(

                　　　　　　　　last_run, spacing)

            　　　　try:

                　　　　　　task(self, context)

            　　　　except Exception:

                　　　　　　　　if raise_on_error:

                    　　　　　　　　　　raise

                　　　　　　　　LOG.exception(_LE("Error during %(full_task_name)s"),

                              　　　　　　　　　　{"full_task_name": full_task_name})

            　　　　time.sleep(0)

        　　　　return idle_for

巴特西

Openstack Periodic Task

Openstack Periodic Task

实现机制

模块mananger类添加方法

periodic_task装饰器

Manager类继承关系

PeriodicTasksMeta元类

触发执行周期性任务

最新文章

热门文章