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