common/python/rift/mano/config_agent/operdata.py

   1 #
   2 #   Copyright 2016 RIFT.IO Inc
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16
  17 import asyncio
  18 import concurrent.futures
  19 import time
  20
  21 from gi.repository import (
  22     NsrYang,
  23     RwTypes,
  24     RwcalYang,
  25     RwNsrYang,
  26     RwConfigAgentYang,
  27     RwDts as rwdts)
  28
  29 import rift.tasklets
  30
  31 import rift.mano.utils.juju_api as juju
  32
  33
  34 class ConfigAgentAccountNotFound(Exception):
  35     pass
  36
  37 class JujuClient(object):
  38     def __init__(self, log, ip, port, user, passwd):
  39         self._log = log
  40         self._ip = ip
  41         self._port = port
  42         self._user = user
  43         self._passwd = passwd
  44
  45         self._api = juju.JujuApi(log=log,
  46                                  server=ip, port=port,
  47                                  user=user, secret=passwd)
  48
  49
  50     def validate_account_creds(self):
  51         status = RwcalYang.CloudConnectionStatus()
  52         try:
  53             env = self._api._get_env()
  54         except juju.JujuEnvError as e:
  55             msg = "JujuClient: Invalid account credentials: %s", str(e)
  56             self._log.error(msg)
  57             raise Exception(msg)
  58         except ConnectionRefusedError as e:
  59             msg = "JujuClient: Wrong IP or Port: %s", str(e)
  60             self._log.error(msg)
  61             raise Exception(msg)
  62         except Exception as e:
  63             msg = "JujuClient: Connection Failed: %s", str(e)
  64             self._log.error(msg)
  65             raise Exception(msg)
  66         else:
  67             status.status = "success"
  68             status.details = "Connection was successful"
  69             self._log.info("JujuClient: Connection Successful")
  70
  71         return status
  72
  73
  74 class ConfigAgentAccount(object):
  75     def __init__(self, log, account_msg):
  76         self._log = log
  77         self._account_msg = account_msg.deep_copy()
  78
  79         if account_msg.account_type == "juju":
  80             self._cfg_agent_client_plugin = JujuClient(
  81                     log,
  82                     account_msg.juju.ip_address,
  83                     account_msg.juju.port,
  84                     account_msg.juju.user,
  85                     account_msg.juju.secret)
  86         else:
  87             self._cfg_agent_client_plugin = None
  88
  89         self._status = RwConfigAgentYang.ConfigAgentAccount_ConnectionStatus(
  90                 status="unknown",
  91                 details="Connection status lookup not started"
  92                 )
  93
  94         self._validate_task = None
  95
  96     @property
  97     def name(self):
  98         return self._account_msg.name
  99
 100     @property
 101     def account_msg(self):
 102         return self._account_msg
 103
 104     @property
 105     def account_type(self):
 106         return self._account_msg.account_type
 107
 108     @property
 109     def connection_status(self):
 110         return self._status
 111
 112     def update_from_cfg(self, cfg):
 113         self._log.debug("Updating parent ConfigAgentAccount to %s", cfg)
 114         raise NotImplementedError("Update config agent account not yet supported")
 115
 116     @asyncio.coroutine
 117     def validate_cfg_agent_account_credentials(self, loop):
 118         self._log.debug("Validating Config Agent Account %s, credential status %s", self._account_msg, self._status)
 119
 120         self._status = RwConfigAgentYang.ConfigAgentAccount_ConnectionStatus(
 121                 status="validating",
 122                 details="Config Agent account connection validation in progress"
 123                 )
 124
 125         if self._cfg_agent_client_plugin is None:
 126             self._status = RwConfigAgentYang.ConfigAgentAccount_ConnectionStatus(
 127                     status="unknown",
 128                     details="Config Agent account does not support validation of account creds"
 129                     )
 130         else:
 131             try:
 132                 status = yield from loop.run_in_executor(
 133                     None,
 134                     self._cfg_agent_client_plugin.validate_account_creds
 135                     )
 136                 self._status = RwConfigAgentYang.ConfigAgentAccount_ConnectionStatus.from_dict(status.as_dict())
 137             except Exception as e:
 138                 self._status = RwConfigAgentYang.ConfigAgentAccount_ConnectionStatus(
 139                     status="failure",
 140                     details="Error - " + str(e)
 141                     )
 142
 143         self._log.info("Got config agent account validation response: %s", self._status)
 144
 145     def start_validate_credentials(self, loop):
 146         if self._validate_task is not None:
 147             self._validate_task.cancel()
 148             self._validate_task = None
 149
 150         self._validate_task = asyncio.ensure_future(
 151                 self.validate_cfg_agent_account_credentials(loop),
 152                 loop=loop
 153                 )
 154
 155 class CfgAgentDtsOperdataHandler(object):
 156     def __init__(self, dts, log, loop):
 157         self._dts = dts
 158         self._log = log
 159         self._loop = loop
 160
 161         self.cfg_agent_accounts = {}
 162
 163     def add_cfg_agent_account(self, account_msg):
 164         account = ConfigAgentAccount(self._log, account_msg)
 165         self.cfg_agent_accounts[account.name] = account
 166         self._log.info("ConfigAgent Operdata Handler added. Starting account validation")
 167
 168         account.start_validate_credentials(self._loop)
 169
 170     def delete_cfg_agent_account(self, account_name):
 171         del self.cfg_agent_accounts[account_name]
 172         self._log.info("ConfigAgent Operdata Handler deleted.")
 173
 174     def get_saved_cfg_agent_accounts(self, cfg_agent_account_name):
 175         ''' Get Config Agent Account corresponding to passed name, or all saved accounts if name is None'''
 176         saved_cfg_agent_accounts = []
 177
 178         if cfg_agent_account_name is None or cfg_agent_account_name == "":
 179             cfg_agent_accounts = list(self.cfg_agent_accounts.values())
 180             saved_cfg_agent_accounts.extend(cfg_agent_accounts)
 181         elif cfg_agent_account_name in self.cfg_agent_accounts:
 182             account = self.cfg_agent_accounts[cfg_agent_account_name]
 183             saved_cfg_agent_accounts.append(account)
 184         else:
 185             errstr = "Config Agent account {} does not exist".format(cfg_agent_account_name)
 186             raise KeyError(errstr)
 187
 188         return saved_cfg_agent_accounts
 189
 190
 191     def _register_show_status(self):
 192         def get_xpath(cfg_agent_name=None):
 193             return "D,/rw-config-agent:config-agent/account{}/connection-status".format(
 194                     "[name='%s']" % cfg_agent_name if cfg_agent_name is not None else ''
 195                     )
 196
 197         @asyncio.coroutine
 198         def on_prepare(xact_info, action, ks_path, msg):
 199             path_entry = RwConfigAgentYang.ConfigAgentAccount.schema().keyspec_to_entry(ks_path)
 200             cfg_agent_account_name = path_entry.key00.name
 201             self._log.debug("Got show cfg_agent connection status request: %s", ks_path.create_string())
 202
 203             try:
 204                 saved_accounts = self.get_saved_cfg_agent_accounts(cfg_agent_account_name)
 205                 for account in saved_accounts:
 206                     connection_status = account.connection_status
 207                     self._log.debug("Responding to config agent connection status request: %s", connection_status)
 208                     xact_info.respond_xpath(
 209                             rwdts.XactRspCode.MORE,
 210                             xpath=get_xpath(account.name),
 211                             msg=account.connection_status,
 212                             )
 213             except KeyError as e:
 214                 self._log.warning(str(e))
 215                 xact_info.respond_xpath(rwdts.XactRspCode.NA)
 216                 return
 217
 218             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 219
 220         yield from self._dts.register(
 221                 xpath=get_xpath(),
 222                 handler=rift.tasklets.DTS.RegistrationHandler(
 223                     on_prepare=on_prepare),
 224                 flags=rwdts.Flag.PUBLISHER,
 225                 )
 226
 227     def _register_validate_rpc(self):
 228         def get_xpath():
 229             return "/rw-config-agent:update-cfg-agent-status"
 230
 231         @asyncio.coroutine
 232         def on_prepare(xact_info, action, ks_path, msg):
 233             if not msg.has_field("cfg_agent_account"):
 234                 raise ConfigAgentAccountNotFound("Config Agent account name not provided")
 235
 236             cfg_agent_account_name = msg.cfg_agent_account
 237             try:
 238                 account = self.cfg_agent_accounts[cfg_agent_account_name]
 239             except KeyError:
 240                 raise ConfigAgentAccountNotFound("Config Agent account name %s not found" % cfg_agent_account_name)
 241
 242             account.start_validate_credentials(self._loop)
 243
 244             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 245
 246         yield from self._dts.register(
 247                 xpath=get_xpath(),
 248                 handler=rift.tasklets.DTS.RegistrationHandler(
 249                     on_prepare=on_prepare
 250                     ),
 251                 flags=rwdts.Flag.PUBLISHER,
 252                 )
 253
 254     @asyncio.coroutine
 255     def register(self):
 256         yield from self._register_show_status()
 257         yield from self._register_validate_rpc()
 258
 259 class ConfigAgentJob(object):
 260     """A wrapper over the config agent job object, providing some
 261     convenience functions.
 262
 263     YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob contains
 264     ||
 265      ==> VNFRS
 266           ||
 267            ==> Primitives
 268
 269     """
 270     # The normalizes the state terms from Juju to our yang models
 271     # Juju : Yang model
 272     STATUS_MAP = {"completed": "success",
 273                   "pending"  : "pending",
 274                   "running"  : "pending",
 275                   "failed"   : "failure"}
 276
 277     def __init__(self, nsr_id, job, tasks=None):
 278         """
 279         Args:
 280             nsr_id (uuid): ID of NSR record
 281             job (YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob): Gi object
 282             tasks: List of asyncio.tasks. If provided the job monitor will
 283                 use it to monitor the tasks instead of the execution IDs
 284         """
 285         self._job = job
 286         self.nsr_id = nsr_id
 287         self.tasks = tasks
 288
 289     @property
 290     def id(self):
 291         """Job id"""
 292         return self._job.job_id
 293
 294     @property
 295     def name(self):
 296         """Job name"""
 297         return self._job.job_name
 298
 299     @property
 300     def job_status(self):
 301         """Status of the job (success|pending|failure)"""
 302         return self._job.job_status
 303
 304     @job_status.setter
 305     def job_status(self, value):
 306         """Setter for job status"""
 307         self._job.job_status = value
 308
 309     @property
 310     def job(self):
 311         """Gi object"""
 312         return self._job
 313
 314     @property
 315     def xpath(self):
 316         """Xpath of the job"""
 317         return ("D,/nsr:ns-instance-opdata" +
 318                 "/nsr:nsr[nsr:ns-instance-config-ref='{}']" +
 319                 "/nsr:config-agent-job[nsr:job-id='{}']"
 320                 ).format(self.nsr_id, self.id)
 321
 322     @staticmethod
 323     def convert_rpc_input_to_job(nsr_id, rpc_output, tasks):
 324         """A helper function to convert the YangOutput_Nsr_ExecNsConfigPrimitive
 325         to YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob (NsrYang)
 326
 327         Args:
 328             nsr_id (uuid): NSR ID
 329             rpc_output (YangOutput_Nsr_ExecNsConfigPrimitive): RPC output
 330             tasks (list): A list of asyncio.Tasks
 331
 332         Returns:
 333             ConfigAgentJob
 334         """
 335         # Shortcuts to prevent the HUUGE names.
 336         CfgAgentJob = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob
 337         CfgAgentVnfr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob_Vnfr
 338         CfgAgentPrimitive = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob_Vnfr_Primitive
 339         CfgAgentPrimitiveParam =  NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob_Vnfr_Primitive_Parameter
 340
 341         job = CfgAgentJob.from_dict({
 342                 "job_id": rpc_output.job_id,
 343                 "job_name" : rpc_output.name,
 344                 "job_status": "pending",
 345                 "triggered_by": rpc_output.triggered_by,
 346                 "create_time": rpc_output.create_time,
 347                 "job_status_details": rpc_output.job_status_details if rpc_output.job_status_details is not None else None,
 348                 "parameter": [param.as_dict() for param in rpc_output.parameter],
 349                 "parameter_group": [pg.as_dict() for pg in rpc_output.parameter_group]
 350             })
 351
 352         for vnfr in rpc_output.vnf_out_list:
 353             vnfr_job = CfgAgentVnfr.from_dict({
 354                     "id": vnfr.vnfr_id_ref,
 355                     "vnf_job_status": "pending",
 356                     })
 357
 358             for primitive in vnfr.vnf_out_primitive:
 359                 vnf_primitive = CfgAgentPrimitive.from_dict({
 360                         "name": primitive.name,
 361                         "execution_status": ConfigAgentJob.STATUS_MAP[primitive.execution_status],
 362                         "execution_id": primitive.execution_id
 363                     })
 364
 365                 # Copy over the input param
 366                 for param in primitive.parameter:
 367                     vnf_primitive.parameter.append(
 368                             CfgAgentPrimitiveParam.from_dict({
 369                                     "name": param.name,
 370                                     "value": param.value
 371                             }))
 372
 373                 vnfr_job.primitive.append(vnf_primitive)
 374
 375             job.vnfr.append(vnfr_job)
 376
 377         return ConfigAgentJob(nsr_id, job, tasks)
 378
 379
 380 class ConfigAgentJobMonitor(object):
 381     """Job monitor: Polls the Juju controller and get the status.
 382     Rules:
 383         If all Primitive are success, then vnf & nsr status will be "success"
 384         If any one Primitive reaches a failed state then both vnf and nsr will fail.
 385     """
 386     POLLING_PERIOD = 2
 387
 388     def __init__(self, dts, log, job, executor, loop, config_plugin):
 389         """
 390         Args:
 391             dts : DTS handle
 392             log : log handle
 393             job (ConfigAgentJob): ConfigAgentJob instance
 394             executor (concurrent.futures): Executor for juju status api calls
 395             loop (eventloop): Current event loop instance
 396             config_plugin : Config plugin to be used.
 397         """
 398         self.job = job
 399         self.log = log
 400         self.loop = loop
 401         self.executor = executor
 402         self.polling_period = ConfigAgentJobMonitor.POLLING_PERIOD
 403         self.config_plugin = config_plugin
 404         self.dts = dts
 405
 406     @asyncio.coroutine
 407     def _monitor_processes(self, registration_handle):
 408         result = 0
 409         for process in self.job.tasks:
 410             rc = yield from process
 411             self.log.debug("Process {} returned rc: {}".format(process, rc))
 412             result |= rc
 413
 414         if result == 0:
 415             self.job.job_status = "success"
 416         else:
 417             self.job.job_status = "failure"
 418
 419         registration_handle.update_element(self.job.xpath, self.job.job)
 420
 421     def get_error_details(self):
 422         '''Get the error details from failed primitives'''
 423         errs = ''
 424         for vnfr in self.job.job.vnfr:
 425             if vnfr.vnf_job_status != "failure":
 426                 continue
 427
 428             for primitive in vnfr.primitive:
 429                 if primitive.execution_status == "failure":
 430                     errs += '<error>'
 431                     errs += primitive.execution_error_details
 432                     errs += "</error>"
 433
 434         return errs
 435
 436     @asyncio.coroutine
 437     def publish_action_status(self):
 438         """
 439         Starts publishing the status for jobs/primitives
 440         """
 441         registration_handle = yield from self.dts.register(
 442                 xpath=self.job.xpath,
 443                 handler=rift.tasklets.DTS.RegistrationHandler(),
 444                 flags=(rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ),
 445                 )
 446
 447         self.log.debug('preparing to publish job status for {}'.format(self.job.xpath))
 448
 449         try:
 450             registration_handle.create_element(self.job.xpath, self.job.job)
 451
 452             # If the config is done via a user defined script
 453             if self.job.tasks is not None:
 454                 yield from self._monitor_processes(registration_handle)
 455                 return
 456
 457             prev = time.time()
 458             # Run until pending moves to either failure/success
 459             while self.job.job_status == "pending":
 460                 curr = time.time()
 461
 462                 if curr - prev < self.polling_period:
 463                     pause = self.polling_period - (curr - prev)
 464                     yield from asyncio.sleep(pause, loop=self.loop)
 465
 466                 prev = time.time()
 467
 468                 tasks = []
 469                 for vnfr in self.job.job.vnfr:
 470                     task = self.loop.create_task(self.get_vnfr_status(vnfr))
 471                     tasks.append(task)
 472
 473                 # Exit, if no tasks are found
 474                 if not tasks:
 475                     break
 476
 477                 yield from asyncio.wait(tasks, loop=self.loop)
 478
 479                 job_status = [task.result() for task in tasks]
 480
 481                 if "failure" in job_status:
 482                     self.job.job_status = "failure"
 483                     errs = self.get_error_details()
 484                     if len(errs):
 485                         self.job.job.job_status_details = errs
 486                 elif "pending" in job_status:
 487                     self.job.job_status = "pending"
 488                 else:
 489                     self.job.job_status = "success"
 490
 491                 # self.log.debug("Publishing job status: {} at {} for nsr id: {}".format(
 492                 #     self.job.job_status,
 493                 #     self.job.xpath,
 494                 #     self.job.nsr_id))
 495
 496                 registration_handle.update_element(self.job.xpath, self.job.job)
 497
 498
 499         except Exception as e:
 500             self.log.exception(e)
 501             raise
 502
 503
 504     @asyncio.coroutine
 505     def get_vnfr_status(self, vnfr):
 506         """Schedules tasks for all containing primitives and updates it's own
 507         status.
 508
 509         Args:
 510             vnfr : Vnfr job record containing primitives.
 511
 512         Returns:
 513             (str): "success|failure|pending"
 514         """
 515         tasks = []
 516         job_status = []
 517
 518         for primitive in vnfr.primitive:
 519             if primitive.execution_id == "":
 520                 # TODO: For some config data, the id will be empty, check if
 521                 # mapping is needed.
 522                 job_status.append(primitive.execution_status)
 523                 continue
 524
 525             task = self.loop.create_task(self.get_primitive_status(primitive))
 526             tasks.append(task)
 527
 528         if tasks:
 529             yield from asyncio.wait(tasks, loop=self.loop)
 530
 531         job_status.extend([task.result() for task in tasks])
 532         if "failure" in job_status:
 533             vnfr.vnf_job_status = "failure"
 534             return "failure"
 535
 536         elif "pending" in job_status:
 537             vnfr.vnf_job_status = "pending"
 538             return "pending"
 539
 540         else:
 541             vnfr.vnf_job_status = "success"
 542             return "success"
 543
 544     @asyncio.coroutine
 545     def get_primitive_status(self, primitive):
 546         """
 547         Queries the juju api and gets the status of the execution id.
 548
 549         Args:
 550             primitive : Primitive containing the execution ID.
 551         """
 552
 553         try:
 554             resp = yield from self.loop.run_in_executor(
 555                     self.executor,
 556                     self.config_plugin.get_action_status,
 557                     primitive.execution_id
 558                     )
 559
 560             status = resp['status']
 561             if status == 'failed':
 562                 self.log.warning("Execution of action {} failed: {}".
 563                                  format(primitive.execution_id, resp))
 564                 primitive.execution_error_details = resp['message']
 565
 566         except Exception as e:
 567             self.log.exception(e)
 568             status = "failed"
 569
 570         # Handle case status is None
 571         if status:
 572             primitive.execution_status = ConfigAgentJob.STATUS_MAP[status]
 573         else:
 574             primitive.execution_status = "failure"
 575
 576         return primitive.execution_status
 577
 578
 579 class CfgAgentJobDtsHandler(object):
 580     """Dts Handler for CfgAgent"""
 581     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr/nsr:config-agent-job"
 582
 583     def __init__(self, dts, log, loop, nsm, cfgm):
 584         """
 585         Args:
 586             dts  : Dts Handle.
 587             log  : Log handle.
 588             loop : Event loop.
 589             nsm  : NsmManager.
 590             cfgm : ConfigManager.
 591         """
 592         self._dts = dts
 593         self._log = log
 594         self._loop = loop
 595         self._cfgm = cfgm
 596         self._nsm = nsm
 597
 598         self._regh = None
 599
 600     @property
 601     def regh(self):
 602         """ Return registration handle """
 603         return self._regh
 604
 605     @property
 606     def nsm(self):
 607         """ Return the NSManager manager instance """
 608         return self._nsm
 609
 610     @property
 611     def cfgm(self):
 612         """ Return the ConfigManager manager instance """
 613         return self._cfgm
 614
 615     @staticmethod
 616     def cfg_job_xpath(nsr_id, job_id):
 617         return ("D,/nsr:ns-instance-opdata" +
 618                 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']" +
 619                 "/nsr:config-agent-job[nsr:job-id='{}']").format(nsr_id, job_id)
 620
 621     @asyncio.coroutine
 622     def register(self):
 623         """ Register for NS monitoring read from dts """
 624
 625         @asyncio.coroutine
 626         def on_prepare(xact_info, action, ks_path, msg):
 627             """ prepare callback from dts """
 628             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
 629             if action == rwdts.QueryAction.READ:
 630                 schema = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr.schema()
 631                 path_entry = schema.keyspec_to_entry(ks_path)
 632                 try:
 633                     nsr_id = path_entry.key00.ns_instance_config_ref
 634
 635                     #print("###>>> self.nsm.nsrs:", self.nsm.nsrs)
 636                     nsr_ids = []
 637                     if nsr_id is None or nsr_id == "":
 638                         nsrs = list(self.nsm.nsrs.values())
 639                         nsr_ids = [nsr.id for nsr in nsrs if nsr is not None]
 640                     else:
 641                         nsr_ids = [nsr_id]
 642
 643                     for nsr_id in nsr_ids:
 644                         job = self.cfgm.get_job(nsr_id)
 645
 646                         # If no jobs are queued for the NSR
 647                         if job is None:
 648                             continue
 649
 650                         xact_info.respond_xpath(
 651                             rwdts.XactRspCode.MORE,
 652                             CfgAgentJobDtsHandler.cfg_job_xpath(nsr_id, job.job_id),
 653                             job)
 654
 655                 except Exception as e:
 656                     self._log.exception("Caught exception:%s", str(e))
 657                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 658
 659             else:
 660                 xact_info.respond_xpath(rwdts.XactRspCode.NA)
 661
 662         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
 663         with self._dts.group_create() as group:
 664             self._regh = group.register(xpath=CfgAgentJobDtsHandler.XPATH,
 665                                         handler=hdl,
 666                                         flags=rwdts.Flag.PUBLISHER,
 667                                         )
 668
 669
 670 class ConfigAgentJobManager(object):
 671     """A central class that manager all the Config Agent related data,
 672     Including updating the status
 673
 674     TODO: Needs to support multiple config agents.
 675     """
 676     def __init__(self, dts, log, loop, nsm):
 677         """
 678         Args:
 679             dts  : Dts handle
 680             log  : Log handler
 681             loop : Event loop
 682             nsm  : NsmTasklet instance
 683         """
 684         self.jobs = {}
 685         self.dts = dts
 686         self.log = log
 687         self.loop = loop
 688         self.nsm = nsm
 689         self.handler = CfgAgentJobDtsHandler(dts, log, loop, nsm, self)
 690         self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
 691
 692     def add_job(self, rpc_output, tasks=None):
 693         """Once an RPC is trigger add a now job
 694
 695         Args:
 696             rpc_output (YangOutput_Nsr_ExecNsConfigPrimitive): Rpc output
 697             rpc_input (YangInput_Nsr_ExecNsConfigPrimitive): Rpc input
 698             tasks(list) A list of asyncio.Tasks
 699
 700         """
 701         nsr_id = rpc_output.nsr_id_ref
 702
 703         self.jobs[nsr_id] = ConfigAgentJob.convert_rpc_input_to_job(nsr_id, rpc_output, tasks)
 704
 705         self.log.debug("Creating a job monitor for Job id: {}".format(
 706                 rpc_output.job_id))
 707
 708         # For every Job we will schedule a new monitoring process.
 709         job_monitor = ConfigAgentJobMonitor(
 710             self.dts,
 711             self.log,
 712             self.jobs[nsr_id],
 713             self.executor,
 714             self.loop,
 715             self.nsm.config_agent_plugins[0]  # Hack
 716             )
 717         task = self.loop.create_task(job_monitor.publish_action_status())
 718
 719     def get_job(self, nsr_id):
 720         """Get the job associated with the NSR Id, if present."""
 721         try:
 722             return self.jobs[nsr_id].job
 723         except KeyError:
 724             return None
 725
 726     @asyncio.coroutine
 727     def register(self):
 728         yield from self.handler.register()