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         self._regh = None
 289
 290     @property
 291     def id(self):
 292         """Job id"""
 293         return self._job.job_id
 294
 295     @property
 296     def name(self):
 297         """Job name"""
 298         return self._job.job_name
 299
 300     @property
 301     def job_status(self):
 302         """Status of the job (success|pending|failure)"""
 303         return self._job.job_status
 304
 305     @job_status.setter
 306     def job_status(self, value):
 307         """Setter for job status"""
 308         self._job.job_status = value
 309
 310     @property
 311     def job(self):
 312         """Gi object"""
 313         return self._job
 314
 315     @property
 316     def xpath(self):
 317         """Xpath of the job"""
 318         return ("D,/nsr:ns-instance-opdata" +
 319                 "/nsr:nsr[nsr:ns-instance-config-ref='{}']" +
 320                 "/nsr:config-agent-job[nsr:job-id='{}']"
 321                 ).format(self.nsr_id, self.id)
 322
 323     @property
 324     def regh(self):
 325         """Registration handle for the job"""
 326         return self._regh
 327
 328     @regh.setter
 329     def regh(self, hdl):
 330         """Setter for registration handle"""
 331         self._regh = hdl
 332
 333     @staticmethod
 334     def convert_rpc_input_to_job(nsr_id, rpc_output, tasks):
 335         """A helper function to convert the YangOutput_Nsr_ExecNsConfigPrimitive
 336         to YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob (NsrYang)
 337
 338         Args:
 339             nsr_id (uuid): NSR ID
 340             rpc_output (YangOutput_Nsr_ExecNsConfigPrimitive): RPC output
 341             tasks (list): A list of asyncio.Tasks
 342
 343         Returns:
 344             ConfigAgentJob
 345         """
 346         # Shortcuts to prevent the HUUGE names.
 347         CfgAgentJob = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob
 348         CfgAgentVnfr = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob_Vnfr
 349         CfgAgentPrimitive = NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob_Vnfr_Primitive
 350         CfgAgentPrimitiveParam =  NsrYang.YangData_Nsr_NsInstanceOpdata_Nsr_ConfigAgentJob_Vnfr_Primitive_Parameter
 351
 352         job = CfgAgentJob.from_dict({
 353                 "job_id": rpc_output.job_id,
 354                 "job_name" : rpc_output.name,
 355                 "job_status": "pending",
 356                 "triggered_by": rpc_output.triggered_by,
 357                 "create_time": rpc_output.create_time,
 358                 "job_status_details": rpc_output.job_status_details if rpc_output.job_status_details is not None else None,
 359                 "parameter": [param.as_dict() for param in rpc_output.parameter],
 360                 "parameter_group": [pg.as_dict() for pg in rpc_output.parameter_group]
 361             })
 362
 363         for vnfr in rpc_output.vnf_out_list:
 364             vnfr_job = CfgAgentVnfr.from_dict({
 365                     "id": vnfr.vnfr_id_ref,
 366                     "vnf_job_status": "pending",
 367                     })
 368
 369             for primitive in vnfr.vnf_out_primitive:
 370                 vnf_primitive = CfgAgentPrimitive.from_dict({
 371                         "name": primitive.name,
 372                         "execution_status": ConfigAgentJob.STATUS_MAP[primitive.execution_status],
 373                         "execution_id": primitive.execution_id
 374                     })
 375
 376                 # Copy over the input param
 377                 for param in primitive.parameter:
 378                     vnf_primitive.parameter.append(
 379                             CfgAgentPrimitiveParam.from_dict({
 380                                     "name": param.name,
 381                                     "value": param.value
 382                             }))
 383
 384                 vnfr_job.primitive.append(vnf_primitive)
 385
 386             job.vnfr.append(vnfr_job)
 387
 388         return ConfigAgentJob(nsr_id, job, tasks)
 389
 390
 391 class ConfigAgentJobMonitor(object):
 392     """Job monitor: Polls the Juju controller and get the status.
 393     Rules:
 394         If all Primitive are success, then vnf & nsr status will be "success"
 395         If any one Primitive reaches a failed state then both vnf and nsr will fail.
 396     """
 397     POLLING_PERIOD = 2
 398
 399     def __init__(self, dts, log, job, executor, loop, config_plugin):
 400         """
 401         Args:
 402             dts : DTS handle
 403             log : log handle
 404             job (ConfigAgentJob): ConfigAgentJob instance
 405             executor (concurrent.futures): Executor for juju status api calls
 406             loop (eventloop): Current event loop instance
 407             config_plugin : Config plugin to be used.
 408         """
 409         self.job = job
 410         self.log = log
 411         self.loop = loop
 412         self.executor = executor
 413         self.polling_period = ConfigAgentJobMonitor.POLLING_PERIOD
 414         self.config_plugin = config_plugin
 415         self.dts = dts
 416
 417     @asyncio.coroutine
 418     def _monitor_processes(self, registration_handle):
 419         result = 0
 420         errs = ""
 421         for process in self.job.tasks:
 422             if isinstance(process, asyncio.subprocess.Process):
 423                 rc = yield from process.wait()
 424                 err = yield from process.stderr.read()
 425
 426             else:
 427                 # Task instance
 428                 rc = yield from process
 429                 err = ''
 430
 431             self.log.debug("Process {} returned rc: {}, err: {}".
 432                            format(process, rc, err))
 433
 434             if len(err):
 435                 errs += "<error>{}</error>".format(err)
 436             result |= rc
 437
 438         if result == 0:
 439             self.job.job_status = "success"
 440         else:
 441             self.job.job_status = "failure"
 442
 443         if len(errs):
 444             self.job.job.job_status_details = errs
 445
 446         registration_handle.update_element(self.job.xpath, self.job.job)
 447
 448     def get_error_details(self):
 449         '''Get the error details from failed primitives'''
 450         errs = ''
 451         for vnfr in self.job.job.vnfr:
 452             if vnfr.vnf_job_status != "failure":
 453                 continue
 454
 455             for primitive in vnfr.primitive:
 456                 if primitive.execution_status == "failure":
 457                     errs += '<error>'
 458                     if primitive.execution_error_details:
 459                         errs += primitive.execution_error_details
 460                     else:
 461                         errs += '{}: Unknown error'.format(primitive.name)
 462                     errs += "</error>"
 463
 464         return errs
 465
 466     @asyncio.coroutine
 467     def publish_action_status(self):
 468         """
 469         Starts publishing the status for jobs/primitives
 470         """
 471         registration_handle = yield from self.dts.register(
 472                 xpath=self.job.xpath,
 473                 handler=rift.tasklets.DTS.RegistrationHandler(),
 474                 flags=(rwdts.Flag.PUBLISHER | rwdts.Flag.NO_PREP_READ),
 475                 )
 476
 477         self.log.debug('preparing to publish job status for {}'.format(self.job.xpath))
 478         self.job.regh = registration_handle
 479
 480         try:
 481             registration_handle.create_element(self.job.xpath, self.job.job)
 482
 483             # If the config is done via a user defined script
 484             if self.job.tasks is not None:
 485                 yield from self._monitor_processes(registration_handle)
 486                 return
 487
 488             prev = time.time()
 489             # Run until pending moves to either failure/success
 490             while self.job.job_status == "pending":
 491                 curr = time.time()
 492
 493                 if curr - prev < self.polling_period:
 494                     pause = self.polling_period - (curr - prev)
 495                     yield from asyncio.sleep(pause, loop=self.loop)
 496
 497                 prev = time.time()
 498
 499                 tasks = []
 500                 for vnfr in self.job.job.vnfr:
 501                     task = self.loop.create_task(self.get_vnfr_status(vnfr))
 502                     tasks.append(task)
 503
 504                 # Exit, if no tasks are found
 505                 if not tasks:
 506                     break
 507
 508                 yield from asyncio.wait(tasks, loop=self.loop)
 509
 510                 job_status = [task.result() for task in tasks]
 511
 512                 if "failure" in job_status:
 513                     self.job.job_status = "failure"
 514                     errs = self.get_error_details()
 515                     if len(errs):
 516                         self.job.job.job_status_details = errs
 517                 elif "pending" in job_status:
 518                     self.job.job_status = "pending"
 519                 else:
 520                     self.job.job_status = "success"
 521
 522                 # self.log.debug("Publishing job status: {} at {} for nsr id: {}".format(
 523                 #     self.job.job_status,
 524                 #     self.job.xpath,
 525                 #     self.job.nsr_id))
 526
 527                 registration_handle.update_element(self.job.xpath, self.job.job)
 528
 529
 530         except Exception as e:
 531             self.log.exception(e)
 532             raise
 533
 534
 535     @asyncio.coroutine
 536     def get_vnfr_status(self, vnfr):
 537         """Schedules tasks for all containing primitives and updates it's own
 538         status.
 539
 540         Args:
 541             vnfr : Vnfr job record containing primitives.
 542
 543         Returns:
 544             (str): "success|failure|pending"
 545         """
 546         tasks = []
 547         job_status = []
 548
 549         for primitive in vnfr.primitive:
 550             if primitive.execution_status != 'pending':
 551                 continue
 552
 553             if primitive.execution_id == "":
 554                 # Actions which failed to queue can have empty id
 555                 job_status.append(primitive.execution_status)
 556                 continue
 557
 558             elif primitive.execution_id == "config":
 559                 # Config job. Check if service is active
 560                 task = self.loop.create_task(self.get_service_status(vnfr.id, primitive))
 561
 562             else:
 563                 task = self.loop.create_task(self.get_primitive_status(primitive))
 564
 565             tasks.append(task)
 566
 567         if tasks:
 568             yield from asyncio.wait(tasks, loop=self.loop)
 569
 570         job_status.extend([task.result() for task in tasks])
 571         if "failure" in job_status:
 572             vnfr.vnf_job_status = "failure"
 573             return "failure"
 574
 575         elif "pending" in job_status:
 576             vnfr.vnf_job_status = "pending"
 577             return "pending"
 578
 579         else:
 580             vnfr.vnf_job_status = "success"
 581             return "success"
 582
 583     @asyncio.coroutine
 584     def get_service_status(self, vnfr_id, primitive):
 585         try:
 586             status = yield from self.loop.run_in_executor(
 587                 self.executor,
 588                 self.config_plugin.get_service_status,
 589                 vnfr_id
 590             )
 591
 592             self.log.debug("Service status: {}".format(status))
 593             if status in ['error', 'blocked']:
 594                 self.log.warning("Execution of config {} failed: {}".
 595                                  format(primitive.execution_id, status))
 596                 primitive.execution_error_details = 'Config failed'
 597                 status = 'failure'
 598             elif status in ['active']:
 599                 status = 'success'
 600             elif status is None:
 601                 status = 'failure'
 602             else:
 603                 status = 'pending'
 604
 605         except Exception as e:
 606             self.log.exception(e)
 607             status = "failed"
 608
 609         primitive.execution_status = status
 610         return primitive.execution_status
 611
 612     @asyncio.coroutine
 613     def get_primitive_status(self, primitive):
 614         """
 615         Queries the juju api and gets the status of the execution id.
 616
 617         Args:
 618             primitive : Primitive containing the execution ID.
 619         """
 620
 621         try:
 622             resp = yield from self.loop.run_in_executor(
 623                     self.executor,
 624                     self.config_plugin.get_action_status,
 625                     primitive.execution_id
 626                     )
 627
 628             self.log.debug("Action status: {}".format(resp))
 629             status = resp['status']
 630             if status == 'failed':
 631                 self.log.warning("Execution of action {} failed: {}".
 632                                  format(primitive.execution_id, resp))
 633                 primitive.execution_error_details = resp['message']
 634
 635         except Exception as e:
 636             self.log.exception(e)
 637             status = "failed"
 638
 639         # Handle case status is None
 640         if status:
 641             primitive.execution_status = ConfigAgentJob.STATUS_MAP[status]
 642         else:
 643             primitive.execution_status = "failure"
 644
 645         return primitive.execution_status
 646
 647
 648 class CfgAgentJobDtsHandler(object):
 649     """Dts Handler for CfgAgent"""
 650     XPATH = "D,/nsr:ns-instance-opdata/nsr:nsr/nsr:config-agent-job"
 651
 652     def __init__(self, dts, log, loop, nsm, cfgm):
 653         """
 654         Args:
 655             dts  : Dts Handle.
 656             log  : Log handle.
 657             loop : Event loop.
 658             nsm  : NsmManager.
 659             cfgm : ConfigManager.
 660         """
 661         self._dts = dts
 662         self._log = log
 663         self._loop = loop
 664         self._cfgm = cfgm
 665         self._nsm = nsm
 666
 667         self._regh = None
 668         self._nsr_regh = None
 669
 670     @property
 671     def regh(self):
 672         """ Return registration handle """
 673         return self._regh
 674
 675     @property
 676     def nsm(self):
 677         """ Return the NSManager manager instance """
 678         return self._nsm
 679
 680     @property
 681     def cfgm(self):
 682         """ Return the ConfigManager manager instance """
 683         return self._cfgm
 684
 685     @staticmethod
 686     def cfg_job_xpath(nsr_id, job_id):
 687         return ("D,/nsr:ns-instance-opdata" +
 688                 "/nsr:nsr[nsr:ns-instance-config-ref = '{}']" +
 689                 "/nsr:config-agent-job[nsr:job-id='{}']").format(nsr_id, job_id)
 690
 691     @asyncio.coroutine
 692     def register(self):
 693         """ Register for NS monitoring read from dts """
 694
 695         @asyncio.coroutine
 696         def on_prepare(xact_info, action, ks_path, msg):
 697             """ prepare callback from dts """
 698             xpath = ks_path.to_xpath(RwNsrYang.get_schema())
 699             if action == rwdts.QueryAction.READ:
 700                 schema = RwNsrYang.YangData_Nsr_NsInstanceOpdata_Nsr.schema()
 701                 path_entry = schema.keyspec_to_entry(ks_path)
 702                 try:
 703                     nsr_id = path_entry.key00.ns_instance_config_ref
 704
 705                     #print("###>>> self.nsm.nsrs:", self.nsm.nsrs)
 706                     nsr_ids = []
 707                     if nsr_id is None or nsr_id == "":
 708                         nsrs = list(self.nsm.nsrs.values())
 709                         nsr_ids = [nsr.id for nsr in nsrs if nsr is not None]
 710                     else:
 711                         nsr_ids = [nsr_id]
 712
 713                     for nsr_id in nsr_ids:
 714                         jobs = self.cfgm.get_job(nsr_id)
 715
 716                         for job in jobs:
 717                             xact_info.respond_xpath(
 718                                 rwdts.XactRspCode.MORE,
 719                                 CfgAgentJobDtsHandler.cfg_job_xpath(nsr_id, job.id),
 720                                 job.job)
 721
 722                 except Exception as e:
 723                     self._log.exception("Caught exception:%s", str(e))
 724                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 725
 726             else:
 727                 xact_info.respond_xpath(rwdts.XactRspCode.NA)
 728
 729         hdl = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare,)
 730         with self._dts.group_create() as group:
 731             self._regh = group.register(xpath=CfgAgentJobDtsHandler.XPATH,
 732                                         handler=hdl,
 733                                         flags=rwdts.Flag.PUBLISHER,
 734                                         )
 735
 736     @asyncio.coroutine
 737     def _terminate_nsr(self, nsr_id):
 738         self._log.debug("NSR {} being terminated".format(nsr_id))
 739         jobs = self.cfgm.get_job(nsr_id)
 740         for job in jobs:
 741             path = CfgAgentJobDtsHandler.cfg_job_xpath(nsr_id, job.id)
 742             with self._dts.transaction() as xact:
 743                 self._log.debug("Deleting job: {}".format(path))
 744                 job.regh.delete_element(path)
 745                 self._log.debug("Deleted job: {}".format(path))
 746
 747         # Remove the NSR id in manager
 748         self.cfgm.del_nsr(nsr_id)
 749
 750     @property
 751     def nsr_xpath(self):
 752         return "D,/nsr:ns-instance-opdata/nsr:nsr"
 753
 754     @asyncio.coroutine
 755     def register_for_nsr(self):
 756         """ Register for NSR changes """
 757
 758         @asyncio.coroutine
 759         def on_prepare(xact_info, query_action, ks_path, msg):
 760             """ This NSR is created """
 761             self._log.debug("Received NSR instantiate on_prepare (%s:%s:%s)",
 762                             query_action,
 763                             ks_path,
 764                             msg)
 765
 766             if (query_action == rwdts.QueryAction.UPDATE or
 767                 query_action == rwdts.QueryAction.CREATE):
 768                 pass
 769             elif query_action == rwdts.QueryAction.DELETE:
 770                 nsr_id = msg.ns_instance_config_ref
 771                 asyncio.ensure_future(self._terminate_nsr(nsr_id), loop=self._loop)
 772             else:
 773                 raise NotImplementedError(
 774                     "%s action on cm-state not supported",
 775                     query_action)
 776
 777             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 778
 779         try:
 780             handler = rift.tasklets.DTS.RegistrationHandler(on_prepare=on_prepare)
 781             self._nsr_regh = yield from self._dts.register(self.nsr_xpath,
 782                                                           flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY,
 783                                                           handler=handler)
 784         except Exception as e:
 785             self._log.error("Failed to register for NSR changes as %s", str(e))
 786
 787
 788 class ConfigAgentJobManager(object):
 789     """A central class that manager all the Config Agent related data,
 790     Including updating the status
 791
 792     TODO: Needs to support multiple config agents.
 793     """
 794     def __init__(self, dts, log, loop, nsm):
 795         """
 796         Args:
 797             dts  : Dts handle
 798             log  : Log handler
 799             loop : Event loop
 800             nsm  : NsmTasklet instance
 801         """
 802         self.jobs = {}
 803         self.dts = dts
 804         self.log = log
 805         self.loop = loop
 806         self.nsm = nsm
 807         self.handler = CfgAgentJobDtsHandler(dts, log, loop, nsm, self)
 808         self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
 809
 810     def add_job(self, rpc_output, tasks=None):
 811         """Once an RPC is trigger add a now job
 812
 813         Args:
 814             rpc_output (YangOutput_Nsr_ExecNsConfigPrimitive): Rpc output
 815             rpc_input (YangInput_Nsr_ExecNsConfigPrimitive): Rpc input
 816             tasks(list) A list of asyncio.Tasks
 817
 818         """
 819         nsr_id = rpc_output.nsr_id_ref
 820
 821         job = ConfigAgentJob.convert_rpc_input_to_job(nsr_id, rpc_output, tasks)
 822
 823         self.log.debug("Creating a job monitor for Job id: {}".format(
 824                 rpc_output.job_id))
 825
 826         if nsr_id not in self.jobs:
 827             self.jobs[nsr_id] = [job]
 828         else:
 829             self.jobs[nsr_id].append(job)
 830
 831         # If the tasks are none, assume juju actions
 832         # TBD: This logic need to be revisited
 833         ca = self.nsm.config_agent_plugins[0]
 834         if tasks is None:
 835             for agent in self.nsm.config_agent_plugins:
 836                 if agent.agent_type == 'juju':
 837                     ca = agent
 838                     break
 839
 840         # For every Job we will schedule a new monitoring process.
 841         job_monitor = ConfigAgentJobMonitor(
 842             self.dts,
 843             self.log,
 844             job,
 845             self.executor,
 846             self.loop,
 847             ca
 848             )
 849         task = self.loop.create_task(job_monitor.publish_action_status())
 850
 851     def get_job(self, nsr_id):
 852         """Get the job associated with the NSR Id, if present."""
 853         try:
 854             return self.jobs[nsr_id]
 855         except KeyError:
 856             return []
 857
 858     def del_nsr(self, nsr_id):
 859         """Delete a NSR id from the jobs list"""
 860         if nsr_id in self.jobs:
 861             self.jobs.pop(nsr_id)
 862
 863     @asyncio.coroutine
 864     def register(self):
 865         yield from self.handler.register()
 866         yield from self.handler.register_for_nsr()