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         errs = ""
 410         for process in self.job.tasks:
 411             if isinstance(process, asyncio.subprocess.Process):
 412                 rc = yield from process.wait()
 413                 err = yield from process.stderr.read()
 414
 415             else:
 416                 # Task instance
 417                 rc = yield from process
 418                 err = ''
 419
 420             self.log.debug("Process {} returned rc: {}, err: {}".
 421                            format(process, rc, err))
 422
 423             if len(err):
 424                 errs += "<error>{}</error>".format(err)
 425             result |= rc
 426
 427         if result == 0:
 428             self.job.job_status = "success"
 429         else:
 430             self.job.job_status = "failure"
 431
 432         if len(errs):
 433             self.job.job.job_status_details = errs
 434
 435         registration_handle.update_element(self.job.xpath, self.job.job)
 436
 437     def get_error_details(self):
 438         '''Get the error details from failed primitives'''
 439         errs = ''
 440         for vnfr in self.job.job.vnfr:
 441             if vnfr.vnf_job_status != "failure":
 442                 continue
 443
 444             for primitive in vnfr.primitive:
 445                 if primitive.execution_status == "failure":
 446                     errs += '<error>'
 447                     if primitive.execution_error_details:
 448                         errs += primitive.execution_error_details
 449                     else:
 450                         errs += '{}: Unknown error'.format(primitive.name)
 451                     errs += "</error>"
 452
 453         return errs
 454
 455     @asyncio.coroutine
 456     def publish_action_status(self):
 457         """
 458         Starts publishing the status for jobs/primitives
 459         """
 460         registration_handle = yield from self.dts.register(
 461                 xpath=self.job.xpath,
 462                 handler=rift.tasklets.DTS.RegistrationHandler(),
 463                 flags=(rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ),
 464                 )
 465
 466         self.log.debug('preparing to publish job status for {}'.format(self.job.xpath))
 467
 468         try:
 469             registration_handle.create_element(self.job.xpath, self.job.job)
 470
 471             # If the config is done via a user defined script
 472             if self.job.tasks is not None:
 473                 yield from self._monitor_processes(registration_handle)
 474                 return
 475
 476             prev = time.time()
 477             # Run until pending moves to either failure/success
 478             while self.job.job_status == "pending":
 479                 curr = time.time()
 480
 481                 if curr - prev < self.polling_period:
 482                     pause = self.polling_period - (curr - prev)
 483                     yield from asyncio.sleep(pause, loop=self.loop)
 484
 485                 prev = time.time()
 486
 487                 tasks = []
 488                 for vnfr in self.job.job.vnfr:
 489                     task = self.loop.create_task(self.get_vnfr_status(vnfr))
 490                     tasks.append(task)
 491
 492                 # Exit, if no tasks are found
 493                 if not tasks:
 494                     break
 495
 496                 yield from asyncio.wait(tasks, loop=self.loop)
 497
 498                 job_status = [task.result() for task in tasks]
 499
 500                 if "failure" in job_status:
 501                     self.job.job_status = "failure"
 502                     errs = self.get_error_details()
 503                     if len(errs):
 504                         self.job.job.job_status_details = errs
 505                 elif "pending" in job_status:
 506                     self.job.job_status = "pending"
 507                 else:
 508                     self.job.job_status = "success"
 509
 510                 # self.log.debug("Publishing job status: {} at {} for nsr id: {}".format(
 511                 #     self.job.job_status,
 512                 #     self.job.xpath,
 513                 #     self.job.nsr_id))
 514
 515                 registration_handle.update_element(self.job.xpath, self.job.job)
 516
 517
 518         except Exception as e:
 519             self.log.exception(e)
 520             raise
 521
 522
 523     @asyncio.coroutine
 524     def get_vnfr_status(self, vnfr):
 525         """Schedules tasks for all containing primitives and updates it's own
 526         status.
 527
 528         Args:
 529             vnfr : Vnfr job record containing primitives.
 530
 531         Returns:
 532             (str): "success|failure|pending"
 533         """
 534         tasks = []
 535         job_status = []
 536
 537         for primitive in vnfr.primitive:
 538             if primitive.execution_status != 'pending':
 539                 continue
 540
 541             if primitive.execution_id == "":
 542                 # Actions which failed to queue can have empty id
 543                 job_status.append(primitive.execution_status)
 544                 continue
 545
 546             elif primitive.execution_id == "config":
 547                 # Config job. Check if service is active
 548                 task = self.loop.create_task(self.get_service_status(vnfr.id, primitive))
 549
 550             else:
 551                 task = self.loop.create_task(self.get_primitive_status(primitive))
 552
 553             tasks.append(task)
 554
 555         if tasks:
 556             yield from asyncio.wait(tasks, loop=self.loop)
 557
 558         job_status.extend([task.result() for task in tasks])
 559         if "failure" in job_status:
 560             vnfr.vnf_job_status = "failure"
 561             return "failure"
 562
 563         elif "pending" in job_status:
 564             vnfr.vnf_job_status = "pending"
 565             return "pending"
 566
 567         else:
 568             vnfr.vnf_job_status = "success"
 569             return "success"
 570
 571     @asyncio.coroutine
 572     def get_service_status(self, vnfr_id, primitive):
 573         try:
 574             status = yield from self.loop.run_in_executor(
 575                 self.executor,
 576                 self.config_plugin.get_service_status,
 577                 vnfr_id
 578             )
 579
 580             self.log.debug("Service status: {}".format(status))
 581             if status in ['error', 'blocked']:
 582                 self.log.warning("Execution of config {} failed: {}".
 583                                  format(primitive.execution_id, status))
 584                 primitive.execution_error_details = 'Config failed'
 585                 status = 'failure'
 586             elif status in ['active']:
 587                 status = 'success'
 588             elif status is None:
 589                 status = 'failure'
 590             else:
 591                 status = 'pending'
 592
 593         except Exception as e:
 594             self.log.exception(e)
 595             status = "failed"
 596
 597         primitive.execution_status = status
 598         return primitive.execution_status
 599
 600     @asyncio.coroutine
 601     def get_primitive_status(self, primitive):
 602         """
 603         Queries the juju api and gets the status of the execution id.
 604
 605         Args:
 606             primitive : Primitive containing the execution ID.
 607         """
 608
 609         try:
 610             resp = yield from self.loop.run_in_executor(
 611                     self.executor,
 612                     self.config_plugin.get_action_status,
 613                     primitive.execution_id
 614                     )
 615
 616             self.log.debug("Action status: {}".format(resp))
 617             status = resp['status']
 618             if status == 'failed':
 619                 self.log.warning("Execution of action {} failed: {}".
 620                                  format(primitive.execution_id, resp))
 621                 primitive.execution_error_details = resp['message']
 622
 623         except Exception as e:
 624             self.log.exception(e)
 625             status = "failed"
 626
 627         # Handle case status is None
 628         if status:
 629             primitive.execution_status = ConfigAgentJob.STATUS_MAP[status]
 630         else:
 631             primitive.execution_status = "failure"
 632
 633         return primitive.execution_status
 634
 635
 636 class CfgAgentJobDtsHandler(object):
 637     """Dts Handler for CfgAgent"""
 638     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr/nsr:config-agent-job"
 639
 640     def __init__(self, dts, log, loop, nsm, cfgm):
 641         """
 642         Args:
 643             dts  : Dts Handle.
 644             log  : Log handle.
 645             loop : Event loop.
 646             nsm  : NsmManager.
 647             cfgm : ConfigManager.
 648         """
 649         self._dts = dts
 650         self._log = log
 651         self._loop = loop
 652         self._cfgm = cfgm
 653         self._nsm = nsm
 654
 655         self._regh = None
 656
 657     @property
 658     def regh(self):
 659         """ Return registration handle """
 660         return self._regh
 661
 662     @property
 663     def nsm(self):
 664         """ Return the NSManager manager instance """
 665         return self._nsm
 666
 667     @property
 668     def cfgm(self):
 669         """ Return the ConfigManager manager instance """
 670         return self._cfgm
 671
 672     @staticmethod
 673     def cfg_job_xpath(nsr_id, job_id):
 674         return ("D,/nsr:ns-instance-opdata" +
 675                 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']" +
 676                 "/nsr:config-agent-job[nsr:job-id='{}']").format(nsr_id, job_id)
 677
 678     @asyncio.coroutine
 679     def register(self):
 680         """ Register for NS monitoring read from dts """
 681
 682         @asyncio.coroutine
 683         def on_prepare(xact_info, action, ks_path, msg):
 684             """ prepare callback from dts """
 685             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
 686             if action == rwdts.QueryAction.READ:
 687                 schema = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr.schema()
 688                 path_entry = schema.keyspec_to_entry(ks_path)
 689                 try:
 690                     nsr_id = path_entry.key00.ns_instance_config_ref
 691
 692                     #print("###>>> self.nsm.nsrs:", self.nsm.nsrs)
 693                     nsr_ids = []
 694                     if nsr_id is None or nsr_id == "":
 695                         nsrs = list(self.nsm.nsrs.values())
 696                         nsr_ids = [nsr.id for nsr in nsrs if nsr is not None]
 697                     else:
 698                         nsr_ids = [nsr_id]
 699
 700                     for nsr_id in nsr_ids:
 701                         job = self.cfgm.get_job(nsr_id)
 702
 703                         # If no jobs are queued for the NSR
 704                         if job is None:
 705                             continue
 706
 707                         xact_info.respond_xpath(
 708                             rwdts.XactRspCode.MORE,
 709                             CfgAgentJobDtsHandler.cfg_job_xpath(nsr_id, job.job_id),
 710                             job)
 711
 712                 except Exception as e:
 713                     self._log.exception("Caught exception:%s", str(e))
 714                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 715
 716             else:
 717                 xact_info.respond_xpath(rwdts.XactRspCode.NA)
 718
 719         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
 720         with self._dts.group_create() as group:
 721             self._regh = group.register(xpath=CfgAgentJobDtsHandler.XPATH,
 722                                         handler=hdl,
 723                                         flags=rwdts.Flag.PUBLISHER,
 724                                         )
 725
 726
 727 class ConfigAgentJobManager(object):
 728     """A central class that manager all the Config Agent related data,
 729     Including updating the status
 730
 731     TODO: Needs to support multiple config agents.
 732     """
 733     def __init__(self, dts, log, loop, nsm):
 734         """
 735         Args:
 736             dts  : Dts handle
 737             log  : Log handler
 738             loop : Event loop
 739             nsm  : NsmTasklet instance
 740         """
 741         self.jobs = {}
 742         self.dts = dts
 743         self.log = log
 744         self.loop = loop
 745         self.nsm = nsm
 746         self.handler = CfgAgentJobDtsHandler(dts, log, loop, nsm, self)
 747         self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
 748
 749     def add_job(self, rpc_output, tasks=None):
 750         """Once an RPC is trigger add a now job
 751
 752         Args:
 753             rpc_output (YangOutput_Nsr_ExecNsConfigPrimitive): Rpc output
 754             rpc_input (YangInput_Nsr_ExecNsConfigPrimitive): Rpc input
 755             tasks(list) A list of asyncio.Tasks
 756
 757         """
 758         nsr_id = rpc_output.nsr_id_ref
 759
 760         self.jobs[nsr_id] = ConfigAgentJob.convert_rpc_input_to_job(nsr_id, rpc_output, tasks)
 761
 762         self.log.debug("Creating a job monitor for Job id: {}".format(
 763                 rpc_output.job_id))
 764
 765         # If the tasks are none, assume juju actions
 766         # TBD: This logic need to be revisited
 767         ca = self.nsm.config_agent_plugins[0]
 768         if tasks is None:
 769             for agent in self.nsm.config_agent_plugins:
 770                 if agent.agent_type == 'juju':
 771                     ca = agent
 772                     break
 773
 774         # For every Job we will schedule a new monitoring process.
 775         job_monitor = ConfigAgentJobMonitor(
 776             self.dts,
 777             self.log,
 778             self.jobs[nsr_id],
 779             self.executor,
 780             self.loop,
 781             ca
 782             )
 783         task = self.loop.create_task(job_monitor.publish_action_status())
 784
 785     def get_job(self, nsr_id):
 786         """Get the job associated with the NSR Id, if present."""
 787         try:
 788             return self.jobs[nsr_id].job
 789         except KeyError:
 790             return None
 791
 792     @asyncio.coroutine
 793     def register(self):
 794         yield from self.handler.register()