From 766ea1b5248b56a8e489e7b77c455b1080891072 Mon Sep 17 00:00:00 2001 From: Abhishek-121 Date: Tue, 22 Oct 2024 18:16:50 +0530 Subject: [PATCH 1/4] Add the new feature of deploying the template to the devices using devices specific details or site specific with filtering parameters as device family, role, tag etc. and write the playbook, examples and update the documentation for the same, fix the issue of deleting the project/template from the progress state. Also add some common helper functions in dnac.py --- playbooks/template_workflow_manager.yml | 30 +- plugins/module_utils/dnac.py | 237 +++++ plugins/modules/inventory_workflow_manager.py | 108 --- plugins/modules/template_workflow_manager.py | 852 +++++++++++++++++- 4 files changed, 1092 insertions(+), 135 deletions(-) diff --git a/playbooks/template_workflow_manager.yml b/playbooks/template_workflow_manager.yml index be296ff2e9..cd1478897d 100644 --- a/playbooks/template_workflow_manager.yml +++ b/playbooks/template_workflow_manager.yml @@ -13,11 +13,11 @@ dnac_password: "{{ dnac_password }}" dnac_verify: "{{ dnac_verify }}" dnac_debug: "{{ dnac_debug }}" - dnac_log: True + dnac_log: true dnac_log_level: DEBUG - dnac_log_append: True + dnac_log_append: true dnac_log_file_path: "{{ dnac_log_file_path }}" - validate_response_schema: False + validate_response_schema: false state: "merged" config_verify: true #ignore_errors: true #Enable this to continue execution even the task fails @@ -41,6 +41,30 @@ import: project: "{{ item.import_project }}" template: "{{ item.import_template }}" + + deploy_template: + project_name: "{{ item.proj_name }}" + template_name: "{{ item.temp_name }}" + force_push_template: "{{ item.force_push_template }}" + device_template_params: + - param_name: "{{ item.device_template_params.param_name }}" + param_value: "{{ item.device_template_params.param_value }}" + - param_name: "{{ item.device_template_params.param_name }}" + param_value: "{{ item.device_template_params.param_value }}" + device_specific_details: + # Provide any of the one device_specific details either device_ips_list, device_hostnames_list + # serial_number_list, mac_address_list to deploy template to the devices + # device_ips_list: "{{ item.device_specific_details.device_ips_list }}" + device_hostnames_list: "{{ item.device_specific_details.device_hostnames_list }}" + # serial_number_list: "{{ item.device_specific_details.serial_number_list }}" + # mac_address_list: "{{ item.device_specific_details.mac_address_list }}" + site_associated_provisioning: + # Provide the site name and other parameters are optional to narrow down the results + - site_name: "{{ item.site_associated_provisioning.site_name }}" + device_family: "{{ item.site_associated_provisioning.device_family }}" + device_role: "{{ item.site_associated_provisioning.device_role }}" + device_tag: "{{ item.site_associated_provisioning.device_tag }}" + register: template_result with_items: '{{ template_details }}' tags: diff --git a/plugins/module_utils/dnac.py b/plugins/module_utils/dnac.py index 28c79fbfd9..73a4afcc9f 100644 --- a/plugins/module_utils/dnac.py +++ b/plugins/module_utils/dnac.py @@ -1034,6 +1034,243 @@ def update_site_type_key(self, config): return new_config + def get_device_ips_from_hostname(self, hostname_list): + """ + Get the list of unique device IPs for list of specified hostnames of devices in Cisco Catalyst Center. + Parameters: + self (object): An instance of a class used for interacting with Cisco Catalyst Center. + hostname_list (list): The hostnames of devices for which you want to retrieve the device IPs. + Returns: + list: The list of unique device IPs for the specified devices hostname list. + Description: + Queries Cisco Catalyst Center to retrieve the unique device IP's associated with a device having the specified + list of hostnames. If a device is not found in Cisco Catalyst Center, an error log message is printed. + """ + + device_ips = [] + for hostname in hostname_list: + try: + response = self.dnac._exec( + family="devices", + function='get_device_list', + op_modifies=True, + params={"hostname": hostname} + ) + if response: + self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + response = response.get("response") + if response: + device_ip = response[0]["managementIpAddress"] + if device_ip: + device_ips.append(device_ip) + except Exception as e: + error_message = "Exception occurred while fetching device from Cisco Catalyst Center: {0}".format(str(e)) + self.log(error_message, "ERROR") + + return device_ips + + def get_device_ips_from_serial_number(self, serial_number_list): + """ + Get the list of unique device IPs for a specified list of serial numbers in Cisco Catalyst Center. + Parameters: + self (object): An instance of a class used for interacting with Cisco Catalyst Center. + serial_number_list (list): The list of serial number of devices for which you want to retrieve the device IPs. + Returns: + list: The list of unique device IPs for the specified devices with serial numbers. + Description: + Queries Cisco Catalyst Center to retrieve the unique device IPs associated with a device having the specified + serial numbers.If a device is not found in Cisco Catalyst Center, an error log message is printed. + """ + + device_ips = [] + for serial_number in serial_number_list: + try: + response = self.dnac._exec( + family="devices", + function='get_device_list', + op_modifies=True, + params={"serialNumber": serial_number} + ) + if response: + self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + response = response.get("response") + if response: + device_ip = response[0]["managementIpAddress"] + if device_ip: + device_ips.append(device_ip) + except Exception as e: + error_message = "Exception occurred while fetching device from Cisco Catalyst Center - {0}".format(str(e)) + self.log(error_message, "ERROR") + + return device_ips + + def get_device_ips_from_mac_address(self, mac_address_list): + """ + Get the list of unique device IPs for list of specified mac address of devices in Cisco Catalyst Center. + Parameters: + self (object): An instance of a class used for interacting with Cisco Catalyst Center. + mac_address_list (list): The list of mac address of devices for which you want to retrieve the device IPs. + Returns: + list: The list of unique device IPs for the specified devices. + Description: + Queries Cisco Catalyst Center to retrieve the unique device IPs associated with a device having the specified + mac addresses. If a device is not found in Cisco Catalyst Center, an error log message is printed. + """ + + device_ips = [] + for mac_address in mac_address_list: + try: + response = self.dnac._exec( + family="devices", + function='get_device_list', + op_modifies=True, + params={"macAddress": mac_address} + ) + if response: + self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + response = response.get("response") + if response: + device_ip = response[0]["managementIpAddress"] + if device_ip: + device_ips.append(device_ip) + except Exception as e: + self.status = "failed" + self.msg = "Exception occurred while fetching device from Cisco Catalyst Center - {0}".format(str(e)) + self.result['response'] = self.msg + self.log(self.msg, "ERROR") + self.check_return_status() + + return device_ips + + def get_device_ids_from_device_ips(self, device_ips): + """ + Get the list of unique device IPs for list of specified hostnames of devices in Cisco Catalyst Center. + Parameters: + self (object): An instance of a class used for interacting with Cisco Catalyst Center. + hostname_list (list): The hostnames of devices for which you want to retrieve the device IPs. + Returns: + list: The list of unique device IPs for the specified devices hostname list. + Description: + Queries Cisco Catalyst Center to retrieve the unique device IP's associated with a device having the specified + list of hostnames. If a device is not found in Cisco Catalyst Center, an error log message is printed. + """ + + device_ids = [] + for device_ip in device_ips: + try: + response = self.dnac._exec( + family="devices", + function='get_device_list', + op_modifies=False, + params={"management_ip_address": device_ip} + ) + response = response.get("response") + if not response: + self.log("Unable to fetch the device id for the device '{0}' due to absence of device.".format(device_ip), "WARNING") + continue + + self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + device_id = response[0]["id"] + if device_id: + self.log("Received the device id '{0}' for the device {1}".format(device_id, device_ip), "DEBUG") + device_ids.append(device_id) + except Exception as e: + error_message = ( + "Exception occurred while fetching device id for the device '{0} 'from " + "Cisco Catalyst Center: {1}" + ).format(device_ip, str(e)) + self.log(error_message, "ERROR") + + return device_ids + + def get_device_ips_from_device_ids(self, device_ids): + """ + Retrieves the management IP addresses of devices based on the provided device IDs. + + Args: + device_ids (list): A list of device IDs for which the management IP addresses need to be fetched. + Returns: + device_ips (list): A list of management IP addresses corresponding to the provided device IDs. If a device ID + doesn't have an associated IP or there is an error, the corresponding IP is not included in the list. + Description: + This function iterates over a list of device IDs, makes an API call to Cisco Catalyst Center to fetch + the management IP addresses of the devices, and returns a list of these IPs. If a device is not found + or an exception occurs, it logs the error or warning and continues to the next device ID. + """ + + device_ips = [] + + for device_id in device_ids: + try: + response = self.dnac._exec( + family="devices", + function='get_device_list', + op_modifies=False, + params={"id": device_id} + ) + response = response.get("response") + if not response: + self.log("Unable to fetch the device ip for the device '{0}' due to absence of device.".format(device_id), "WARNING") + continue + + self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + device_ip = response[0]["managementIpAddress"] + if device_ip: + self.log("Received the device ip '{0}' for the device having id {1}".format(device_ip, device_id), "DEBUG") + device_ips.append(device_ip) + + except Exception as e: + error_message = ( + "Exception occurred while fetching device ip with device id'{0} 'from " + "Cisco Catalyst Center: {1}" + ).format(device_id, str(e)) + self.log(error_message, "ERROR") + + return device_ips + + def get_network_device_tag_id(self, tag_name): + """ + Retrieves the ID of a network device tag from the Cisco Catalyst Center based on the tag name. + + Args: + self (object): An instance of the class used for interacting with Cisco Catalyst Center. + tag_name (str): The name of the tag whose ID is to be retrieved. + Returns: + str or None: The tag ID if found, or `None` if the tag is not available or an error occurs. + Description: + This function queries the Cisco Catalyst Center API to retrieve the ID of a tag by its name. + It sends a request to the 'get_tag' API endpoint with the specified `tag_name`. If the tag is found, + the function extracts and returns its `id`. If no tag is found or an error occurs during the API call, + it logs appropriate messages and returns `None`. + """ + + device_tag_id = None + + try: + response = self.dnac._exec( + family="tag", + function='get_tag', + op_modifies=False, + params={"name": tag_name} + ) + response = response.get("response") + if not response: + self.log("Unable to fetch the tag details for the tag '{0}'.".format(tag_name), "WARNING") + return device_tag_id + + self.log("Received API response from 'get_tag': {0}".format(str(response)), "DEBUG") + device_tag_id = response[0]["id"] + self.log("Received the tag id '{0}' for the tag: {1}".format(device_tag_id, tag_name), "INFO") + + except Exception as e: + self.msg = ( + "Exception occurred while fetching tag id for the tag '{0} 'from " + "Cisco Catalyst Center: {1}" + ).format(tag_name, str(e)) + self.set_operation_result("failed", False, self.msg, "INFO").check_return_status() + + return device_tag_id + def is_valid_ipv4(self, ip_address): """ Validates an IPv4 address. diff --git a/plugins/modules/inventory_workflow_manager.py b/plugins/modules/inventory_workflow_manager.py index 7c4754a2e3..9e8fafabe6 100644 --- a/plugins/modules/inventory_workflow_manager.py +++ b/plugins/modules/inventory_workflow_manager.py @@ -2398,114 +2398,6 @@ def get_device_ids(self, device_ips): return device_ids - def get_device_ips_from_hostname(self, hostname_list): - """ - Get the list of unique device IPs for list of specified hostnames of devices in Cisco Catalyst Center. - Parameters: - self (object): An instance of a class used for interacting with Cisco Catalyst Center. - hostname_list (list): The hostnames of devices for which you want to retrieve the device IPs. - Returns: - list: The list of unique device IPs for the specified devices hostname list. - Description: - Queries Cisco Catalyst Center to retrieve the unique device IP's associated with a device having the specified - list of hostnames. If a device is not found in Cisco Catalyst Center, an error log message is printed. - """ - - device_ips = [] - for hostname in hostname_list: - try: - response = self.dnac._exec( - family="devices", - function='get_device_list', - op_modifies=True, - params={"hostname": hostname} - ) - if response: - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") - response = response.get("response") - if response: - device_ip = response[0]["managementIpAddress"] - if device_ip: - device_ips.append(device_ip) - except Exception as e: - error_message = "Exception occurred while fetching device from Cisco Catalyst Center: {0}".format(str(e)) - self.log(error_message, "ERROR") - - return device_ips - - def get_device_ips_from_serial_number(self, serial_number_list): - """ - Get the list of unique device IPs for a specified list of serial numbers in Cisco Catalyst Center. - Parameters: - self (object): An instance of a class used for interacting with Cisco Catalyst Center. - serial_number_list (list): The list of serial number of devices for which you want to retrieve the device IPs. - Returns: - list: The list of unique device IPs for the specified devices with serial numbers. - Description: - Queries Cisco Catalyst Center to retrieve the unique device IPs associated with a device having the specified - serial numbers.If a device is not found in Cisco Catalyst Center, an error log message is printed. - """ - - device_ips = [] - for serial_number in serial_number_list: - try: - response = self.dnac._exec( - family="devices", - function='get_device_list', - op_modifies=True, - params={"serialNumber": serial_number} - ) - if response: - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") - response = response.get("response") - if response: - device_ip = response[0]["managementIpAddress"] - if device_ip: - device_ips.append(device_ip) - except Exception as e: - error_message = "Exception occurred while fetching device from Cisco Catalyst Center - {0}".format(str(e)) - self.log(error_message, "ERROR") - - return device_ips - - def get_device_ips_from_mac_address(self, mac_address_list): - """ - Get the list of unique device IPs for list of specified mac address of devices in Cisco Catalyst Center. - Parameters: - self (object): An instance of a class used for interacting with Cisco Catalyst Center. - mac_address_list (list): The list of mac address of devices for which you want to retrieve the device IPs. - Returns: - list: The list of unique device IPs for the specified devices. - Description: - Queries Cisco Catalyst Center to retrieve the unique device IPs associated with a device having the specified - mac addresses. If a device is not found in Cisco Catalyst Center, an error log message is printed. - """ - - device_ips = [] - for mac_address in mac_address_list: - try: - response = self.dnac._exec( - family="devices", - function='get_device_list', - op_modifies=True, - params={"macAddress": mac_address} - ) - if response: - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") - response = response.get("response") - if response: - device_ip = response[0]["managementIpAddress"] - if device_ip: - device_ips.append(device_ip) - except Exception as e: - self.status = "failed" - self.msg = "Exception occurred while fetching device from Cisco Catalyst Center - {0}".format(str(e)) - self.result['response'] = self.msg - self.log(self.msg, "ERROR") - self.check_return_status() - - return device_ips - def get_interface_from_id_and_name(self, device_id, interface_name): """ Retrieve the interface ID for a device in Cisco Catalyst Center based on device id and interface name. diff --git a/plugins/modules/template_workflow_manager.py b/plugins/modules/template_workflow_manager.py index 5bc116a126..98b8f93bac 100644 --- a/plugins/modules/template_workflow_manager.py +++ b/plugins/modules/template_workflow_manager.py @@ -7,7 +7,7 @@ from __future__ import absolute_import, division, print_function __metaclass__ = type -__author__ = ['Madhan Sankaranarayanan, Rishita Chowdhary, Akash Bhaskaran, Muthu Rakesh'] +__author__ = ['Madhan Sankaranarayanan, Rishita Chowdhary, Akash Bhaskaran, Muthu Rakesh, Abhishek Maheshwari'] DOCUMENTATION = r""" --- @@ -29,6 +29,7 @@ Rishita Chowdhary (@rishitachowdhary) Akash Bhaskaran (@akabhask) Muthu Rakesh (@MUTHU-RAKESH-27) + Abhishek Maheshwari (@abmahesh) options: config_verify: description: Set to True to verify the Cisco Catalyst Center after applying the playbook config. @@ -809,6 +810,79 @@ description: ProjectName path parameter. Project name to create template under the project. type: str + deploy_template: + description: To deploy the template to the devices based on either list of site provisionig details with further filtering + criteria like device family, device role, device tag or by providing the device specific details which includes device_ips_list, + device_hostnames_list, serial_number_list or mac_address_list. + type: dict + suboptions: + project_name: + description: Provide the name of project under which template is available. + type: str + template_name: + description: Name of the template to be deployed. + type: str + force_push_template: + description: Boolean flag to indicate whether the template should be forcefully pushed to the devices, overriding any existing + configuration. + type: bool + is_composite: + description: Boolean flag indicating whether the template is composite, which means the template is built using multiple smaller + templates. + type: bool + device_template_params: + description: A list of parameter name-value pairs used for customizing the template with specific values for each device. + type: list + elements: dict + suboptions: + param_name: + description: Name of the parameter in the template that needs to be replaced with a specific value. + type: str + param_value: + description: Value assigned to the parameter for deployment to devices. + type: str + device_specific_details: + description: Details specific to devices where the template will be deployed, including lists of device IPs, hostnames, + serial numbers, or MAC addresses. + type: list + elements: dict + suboptions: + device_ips_list: + description: A list of IP addresses of the devices where the template will be deployed. + type: list + elements: str + device_hostnames_list: + description: A list of hostnames of the devices where the template will be deployed. + type: list + elements: str + serial_number_list: + description: A list of serial numbers of the devices where the template will be deployed. + type: list + elements: str + mac_address_list: + description: A list of MAC addresses of the devices where the template will be deployed. + type: list + elements: str + site_associated_provisioning: + description: Parameters related to site-based provisioning, allowing the deployment of templates to devices associated with specific sites, with + optional filtering by device family, role, or tag. + type: list + elements: dict + suboptions: + site_name: + description: Name of the site where the devices are associated for provisioning. + type: list + elements: str + device_family: + description: Family of the devices (e.g., switches, routers) used to filter devices for template deployment. + type: str + device_role: + description: Role of the devices (e.g., access, core, edge) used to filter devices for template deployment. + type: str + device_tag: + description: Specific device tag used to filter devices for template deployment. + type: str + requirements: - dnacentersdk >= 2.7.2 @@ -953,6 +1027,81 @@ project_name: string template_file: string +- name: Deploy the given template to the devices based on site specific details and other filtering mode + cisco.dnac.template_workflow_manager: + dnac_host: "{{dnac_host}}" + dnac_username: "{{dnac_username}}" + dnac_password: "{{dnac_password}}" + dnac_verify: "{{dnac_verify}}" + dnac_port: "{{dnac_port}}" + dnac_version: "{{dnac_version}}" + dnac_debug: "{{dnac_debug}}" + dnac_log: True + dnac_log_level: "{{dnac_log_level}}" + state: merged + config_verify: True + config: + deploy_template: + project_name: "Sample_Project" + template_name: "Sample Template" + force_push_template: true + device_template_params: + - param_name: "vlan_id" + param_value: "1431" + - param_name: "vlan_name" + param_value: "testvlan31" + site_associated_provisioning: + - site_name: "Global/Bangalore/Building14/Floor1" + device_family: "Switches and Hubs" + +- name: Deploy the given template to the devices based on device specific details + cisco.dnac.template_workflow_manager: + dnac_host: "{{dnac_host}}" + dnac_username: "{{dnac_username}}" + dnac_password: "{{dnac_password}}" + dnac_verify: "{{dnac_verify}}" + dnac_port: "{{dnac_port}}" + dnac_version: "{{dnac_version}}" + dnac_debug: "{{dnac_debug}}" + dnac_log: True + dnac_log_level: "{{dnac_log_level}}" + state: merged + config_verify: True + config: + deploy_template: + project_name: "Sample_Project" + template_name: "Sample Template" + force_push_template: true + device_template_params: + - param_name: "vlan_id" + param_value: "1431" + - param_name: "vlan_name" + param_value: "testvlan31" + device_specific_details: + - device_ips_list: ["10.1.2.1", "10.2.3.4"] + +- name: Delete the given project or template from the Cisco Catalyst Center + cisco.dnac.template_workflow_manager: + dnac_host: "{{dnac_host}}" + dnac_username: "{{dnac_username}}" + dnac_password: "{{dnac_password}}" + dnac_verify: "{{dnac_verify}}" + dnac_port: "{{dnac_port}}" + dnac_version: "{{dnac_version}}" + dnac_debug: "{{dnac_debug}}" + dnac_log: True + dnac_log_level: "{{dnac_log_level}}" + state: deleted + config_verify: True + config: + configuration_templates: + project_name: "Sample_Project" + template_name: "Sample Template" + language: "velocity" + software_type: "IOS-XE" + device_types: + - product_family: "Switches and Hubs" + """ RETURN = r""" @@ -1027,6 +1176,7 @@ import copy import json +import time from ansible.module_utils.basic import AnsibleModule from ansible_collections.cisco.dnac.plugins.module_utils.dnac import ( DnacBase, @@ -1046,6 +1196,7 @@ def __init__(self, module): self.supported_states = ["merged", "deleted"] self.accepted_languages = ["JINJA", "VELOCITY"] self.export_template = [] + self.max_timeout = self.params.get('dnac_api_task_timeout') self.result['response'] = [ {"configurationTemplate": {"response": {}, "msg": {}}}, {"export": {"response": {}}}, @@ -1106,6 +1257,35 @@ def validate_input(self): 'template_name': {'type': 'str'}, 'version': {'type': 'str'} }, + 'deploy_template': { + 'type': 'dict', + 'project_name': {'type': 'str'}, + 'template_name': {'type': 'str'}, + 'force_push_template': {'type': 'bool'}, + 'is_composite': {'type': 'bool'}, + 'device_template_params': { + 'type': 'list', + 'elements': 'dict', + 'param_name': {'type': 'str'}, + 'param_value': {'type': 'str'}, + }, + 'device_specific_details': { + 'type': 'list', + 'elements': 'dict', + 'device_ips_list': {'type': 'list', 'elements': 'str'}, + 'device_hostnames_list': {'type': 'list', 'elements': 'str'}, + 'serial_number_list': {'type': 'list', 'elements': 'str'}, + 'mac_address_list': {'type': 'list', 'elements': 'str'}, + }, + 'site_associated_provisioning': { + 'type': 'list', + 'elements': 'dict', + 'site_name': {'type': 'str'}, + 'device_family': {'type': 'str'}, + 'device_role': {'type': 'str'}, + 'device_tag': {'type': 'str'}, + } + }, 'export': { 'type': 'dict', 'project': {'type': 'list', 'elements': 'str'}, @@ -1120,11 +1300,13 @@ def validate_input(self): 'type': 'dict', 'project': { 'type': 'dict', + 'project_file': {'type': 'str'}, 'do_version': {'type': 'str', 'default': 'False'}, }, 'template': { 'type': 'dict', 'do_version': {'type': 'str', 'default': 'False'}, + 'template_file': {'type': 'str'}, 'payload': { 'type': 'list', 'elements': 'dict', @@ -1443,6 +1625,33 @@ def get_templates_details(self, name): self.log("Received API response from 'get_templates_details': {0}".format(items), "DEBUG") return result + def get_project_defined_template_details(self, project_name, template_name): + """ + Get the template details from the template name provided in the playbook. + Parameters: + project_name (str) - Name of the project under which templates are associated. + template_name (str) - Name of the template provided in the playbook. + Returns: + result (dict) - Template details for the given template name. + """ + + result = None + items = self.dnac_apply['exec']( + family="configuration_templates", + function="get_templates_details", + op_modifies=True, + params={ + "project_name": project_name, + "name": template_name + } + ) + if items: + result = items + + self.log("Received API response from 'get_templates_details': {0}".format(items), "DEBUG") + + return result + def get_containing_templates(self, containing_templates): """ Store tags from the playbook for template processing in Cisco Catalyst Center. @@ -1656,6 +1865,99 @@ def get_template(self, config): self.result['response'][0].get("configurationTemplate").update({"items": items}) return result + def get_uncommitted_template_id(self, project_name, template_name): + """ + Retrieves the ID of an uncommitted template from a specified project in the Cisco Catalyst Center. + + Args: + self (object): An instance of the class used for interacting with Cisco Catalyst Center. + project_name (str): The name of the project under which the template is located. + template_name (str): The name of the template whose uncommitted ID is to be retrieved. + Returns: + str or None: The template ID if found, otherwise `None` if the template is not available or uncommitted. + Description: + This function queries the Cisco Catalyst Center for uncommitted templates within a specified project. + It checks if the template list contains the specified `template_name` and if found, returns the associated + `templateId`. If the template is not found, the function logs a warning message and returns `None`. + The function is useful for identifying templates that are not yet committed, which can then be versioned + or deployed. If the template is unavailable, an appropriate log message is recorded and the function + exits early with `None`. + """ + + template_id = None + template_list = self.dnac_apply['exec']( + family="configuration_templates", + function="gets_the_templates_available", + op_modifies=False, + params={ + "projectNames": project_name, + "un_committed": True + }, + ) + msg = ( + "Given template '{0}' is not available under the project '{1}' " + "so cannot commit or deploy the template in device(s)." + ).format(template_name, project_name) + + if not template_list: + self.log(msg, "WARNING") + self.msg = msg + return template_id + + for template in template_list: + if template.get("name") == template_name: + template_id = template.get("templateId") + return template_id + + return template_id + + def versioned_given_template(self, project_name, template_name, template_id): + """ + Versions (commits) a specified template in the Cisco Catalyst Center. + + Args: + self (object): An instance of the class used for interacting with Cisco Catalyst Center. + project_name (str): The name of the project under which the template resides. + template_name (str): The name of the template to be versioned. + template_id (str): The unique identifier of the template to be versioned. + Returns: + self (object): The instance of the class itself, with the operation result (success/failure) set accordingly. + Description: + This function handles the process of versioning or committing a template in the Cisco Catalyst Center. + It constructs a request payload with versioning comments and template ID, and then calls the API to + initiate the versioning task. + The function returns the class instance for further chaining of operations. + """ + + try: + comments = ( + "Given template '{0}' under the project '{1}' versioned successfully." + ).format(template_name, project_name) + + version_params = { + "comments": comments, + "templateId": template_id + } + task_name = "version_template" + task_id = self.get_taskid_post_api_call("configuration_templates", task_name, version_params) + + if not task_id: + self.msg = "Unable to retrieve the task_id for the task '{0}'.".format(task_name) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + success_msg = "Given template '{0}' versioned/committed successfully in the Cisco Catalyst Center.".format(template_name) + self.get_task_status_from_tasks_by_id(task_id, task_name, success_msg) + + except Exception as e: + self.msg = ( + "An exception occured while versioning the template '{0}' in the Cisco Catalyst " + "Center: {1}" + ).format(template_name, str(e)) + self.set_operation_result("failed", False, self.msg, "ERROR") + + return self + def get_have_project(self, config): """ Get the current project related information from Cisco Catalyst Center. @@ -1769,6 +2071,7 @@ def get_have(self, config): Returns: self """ + have = {} configuration_templates = config.get("configuration_templates") if configuration_templates: if not configuration_templates.get("project_name"): @@ -1779,6 +2082,22 @@ def get_have(self, config): if template_available: self.get_have_template(config, template_available) + deploy_temp_details = config.get("deploy_template") + if deploy_temp_details: + template_name = deploy_temp_details.get("template_name") + project_name = deploy_temp_details.get("project_name") + temp_details = self.get_project_defined_template_details(project_name, template_name).get("response") + + if temp_details: + self.log("Given template '{0}' is already committed in the Catalyst Center.".format(template_name), "INFO") + have["temp_id"] = temp_details[0].get("id") + + self.log("Successfully collect the details for the template '{0}' from the " + "Cisco Catalyst Center.".format(template_name), "INFO" + ) + + self.have = have + self.msg = "Successfully collected all project and template \ parameters from Cisco Catalyst Center for comparison" self.status = "success" @@ -1830,6 +2149,39 @@ def get_want(self, config): want["project_params"] = project_params want["comments"] = version_comments + deploy_temp_details = config.get("deploy_template") + if deploy_temp_details: + project_name = deploy_temp_details.get("project_name") + if not project_name: + self.msg = ( + "To Deploy the template in the devices, parameter 'project_name' " + "must be given in the playboook." + ) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + template_name = deploy_temp_details.get("template_name") + if not template_name: + self.msg = ( + "To Deploy the template in the devices, parameter 'template_name' " + "must be given in the playboook." + ) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + device_specific_details = deploy_temp_details.get("device_specific_details") + site_associated_provisioning = deploy_temp_details.get("site_associated_provisioning") + + if not (device_specific_details or site_associated_provisioning): + self.msg = ( + "Either give the parameter 'device_specific_details' or 'site_associated_provisioning' " + "in the playbook to fetch the device ids and proceed for the deployment of template {0}." + ).format(template_name) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + want["deploy_tempate"] = deploy_temp_details + self.want = want self.msg = "Successfully collected all parameters from playbook " + \ "for comparison" @@ -2184,7 +2536,8 @@ def update_configuration_templates(self, configuration_templates): else: self.msg = str(task_details.get("progress")) self.status = "failed" - return self + + return self def handle_export(self, export): """ @@ -2397,11 +2750,331 @@ def handle_import(self, _import): return self + def filter_devices_with_family_role(self, site_assign_device_ids, device_family=None, device_role=None): + """ + Filters devices based on their family and role from a list of site-assigned device IDs. + + Args: + self (object): An instance of the class interacting with Cisco Catalyst Center. + site_assign_device_ids (list): A list of device IDs (strings) assigned to a site that need to be filtered. + device_family (str, optional): The family of devices to filter by (e.g., 'Switches and Hubs'). If None, + this filter is not applied. Defaults to None. + device_role (str, optional): The role of the devices to filter by (e.g., 'ACCESS', 'CORE'). If None, + this filter is not applied. Defaults to None. + Returns: + list (str): A list of filtered device IDs (strings) that belong to the specified device family and role. + If no matching devices are found, the list will be empty. + Description: + This function filters a list of device IDs based on the specified `device_family` and `device_role` by querying + the Cisco Catalyst Center API. It iterates over each device ID, checking if the device belongs to the specified + family and has the desired role. Devices that match the criteria are added to the `filtered_device_list`. + If a device does not match the criteria or no response is received from the API, the function logs an + informational message and skips that device. In the event of an error during the API call, it logs the error + message and continues processing the remaining devices. + The function returns the list of devices that meet the filtering criteria. + """ + + filtered_device_list = [] + + for device_id in site_assign_device_ids: + try: + response = self.dnac._exec( + family="devices", + function='get_device_list', + op_modifies=True, + params={ + "family": device_family, + "id": device_id, + "role": device_role + } + ) + response = response.get('response') + + if not response: + self.log( + "Device with id '{0}' does not belong to the device family '{1}' or not having the " + " device role as {2}".format(device_id, device_family, device_role), "INFO" + ) + continue + + filtered_device_list.append(device_id) + + except Exception as e: + error_message = "Error while getting the response of device from Cisco Catalyst Center: {0}".format(str(e)) + self.log(error_message, "CRITICAL") + continue + + return filtered_device_list + + def get_latest_template_version_id(self, template_id, template_name): + version_temp_id = None + + try: + response = self.dnac._exec( + family="configuration_templates", + function='get_template_versions', + op_modifies=True, + params={ + "template_id": template_id, + } + ) + + if not response: + self.log( + "There is no versioning present for the template {0} in the Cisco " + "Catalyst Center.".format(template_name), "INFO" + ) + response = response[0].get("versionsInfo") + self.log( + "Received API response from 'get_tempget_template_versionslate_version' for template " + "{0}: {1}".format(str(response), template_name), "DEBUG" + ) + latest_version = max(response, key=lambda x: x["versionTime"]) + version_temp_id = latest_version.get("id") + + except Exception as e: + error_message = "Error while getting the latest version id for the template {0}: {1}".format(template_name, str(e)) + self.log(error_message, "CRITICAL") + + return version_temp_id + + def create_payload_for_template_deploy(self, deploy_temp_details, device_ids): + """ + Creates a payload for deploying a template to specified devices in the Cisco Catalyst Center. + + Args: + self (object): An instance of the class interacting with Cisco Catalyst Center. + deploy_temp_details (dict): A dictionary containing details about the template to be deployed. + device_ids (list): A list of device UUIDs to which the template should be deployed. + Returns: + dict: A dictionary representing the payload required to deploy the template. + Description: + This function generates the necessary payload for deploying a template to devices in the Cisco Catalyst Center. + It first checks if the given template is already committed. If not, it fetches its uncommitted version, commits it, + and uses its template ID for deployment. The payload includes information about target devices and their respective + template parameters. + The function logs appropriate messages during the process, including if a template is already committed, if + parameters are updated, and when the payload is successfully collected. + """ + + project_name = deploy_temp_details.get("project_name") + template_name = deploy_temp_details.get("template_name") + # Check if the template is available but not yet committed + if self.have.get("temp_id"): + self.log("Given template '{0}' is already committed in the Cisco Catalyst Center".format(template_name), "INFO") + template_id = self.have.get("temp_id") + else: + template_id = self.get_uncommitted_template_id(project_name, template_name) + + if not template_id: + self.msg = ( + "Unable to fetch the details for the template '{0}' from the Cisco " + "Catalyst Center." + ).format(template_name) + self.set_operation_result("failed", False, self.msg, "ERROR").check_return_status() + + self.log("Given template '{0}' is available and is not committed yet.".format(template_name), "INFO") + + # Commit or versioned the given template in the Catalyst Center + self.versioned_given_template(project_name, template_name, template_id).check_return_status() + + deploy_payload = { + "forcePushTemplate": deploy_temp_details.get("force_push_template", False), + "isComposite": deploy_temp_details.get("is_composite", False), + "templateId": template_id, + } + target_info_list = [] + template_dict = {} + device_template_params = deploy_temp_details.get("device_template_params") + if not device_template_params: + self.msg = "Template parameters is not given in the playbook so cannot deploy {0} to the devices.".format(template_name) + self.set_operation_result("failed", False, self.msg, "ERROR").check_return_status() + + for param in device_template_params: + name = param["param_name"] + value = param["param_value"] + self.log("Update the template placeholder for the name '{0}' with value {1}".format(name, value), "DEBUG") + template_dict[name] = value + + # Get the latest version template ID + version_template_id = self.get_latest_template_version_id(template_id, template_name) + if not version_template_id: + self.log("No versioning found for the template: {0}".format(template_name), "INFO") + version_template_id = template_id + + for device_id in device_ids: + target_device_dict = { + "id": device_id, + "type": "MANAGED_DEVICE_UUID", + "versionedTemplateId": version_template_id, + "params": template_dict, + } + target_info_list.append(target_device_dict) + del target_device_dict + + deploy_payload["targetInfo"] = target_info_list + self.log("Successfully collected the payload for the deploy of template '{0}'.".format(template_name), "INFO") + + return deploy_payload + + def deploy_template_to_devices(self, deploy_temp_payload, template_name, device_ips): + """ + Deploys a specified template to devices associated with a site in the Cisco Catalyst Center. + + Args: + self (object): An instance of the class used for interacting with Cisco Catalyst Center. + deploy_temp_payload (dict): The payload containing the details required to deploy the template. + This includes the template ID, device details, and template parameters. + template_name (str): The name of the template to be deployed. + device_ips (list): The management ip address of the devices to which template will be deployed. + Returns: + self (object): The instance of the class itself, with the operation result (success or failure) + set accordingly. + Description: + This function handles the deployment of a template to a set of devices managed in the Cisco Catalyst Center. + It sends a POST request with the deployment payload and retrieves the task ID associated with the deployment task. + It then monitors the status of the task using the task ID and logs the result. + If the task ID is not retrieved or an exception occurs during deployment, the function logs an error message, + sets the operation result to "failed," and returns the instance. + The success message indicates that the template has been successfully deployed to all the devices in the specified + site, while any exceptions are caught and logged with appropriate details. + """ + + try: + self.log("Deploying the given template {0} to the device(s) {1}.".format(template_name, device_ips)) + payload = {"payload": deploy_temp_payload} + task_name = "deploy_template_v2" + task_id = self.get_taskid_post_api_call("configuration_templates", task_name, payload) + + if not task_id: + self.msg = "Unable to retrive the task_id for the task '{0}'.".format(task_name) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + while True: + task_details = self.get_task_details_by_id(task_id) + progress = task_details.get("progress") + self.log("Task details for the API {0}: {1}".format(task_name, progress), "DEBUG") + + if "not deploying" in progress: + self.log("Deployment of the template {0} gets failed because of: {1}".format(template_name, progress), "WARNING") + self.msg = progress + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + if "ApplicableTargets" in progress: + self.msg = ( + "Given template '{0}' deployed successfully to all the device(s) '{1}' " + " in the Cisco Catalyst Center." + ).format(template_name, device_ips) + self.set_operation_result("success", True, self.msg, "INFO") + return self + + time.sleep(self.params.get('dnac_task_poll_interval')) + + except Exception as e: + self.msg = ( + "An exception occured while deploying the template '{0}' to the device(s) {1} " + " in the Cisco Catalyst Center: {2}." + ).format(template_name, device_ips, str(e)) + self.set_operation_result("failed", False, self.msg, "ERROR") + + return self + + def get_device_ips_from_config_priority(self, device_specific_details): + """ + Retrieve device IPs based on the configuration. + Parameters: + - self (object): An instance of a class used for interacting with Cisco Cisco Catalyst Center. + Returns: + list: A list containing device IPs. + Description: + This method retrieves device IPs based on the priority order specified in the configuration. + It first checks if device IPs are available. If not, it checks hostnames, serial numbers, + and MAC addresses in order and retrieves IPs based on availability. + If none of the information is available, an empty list is returned. + """ + # Retrieve device IPs from the configuration + device_ips = device_specific_details.get("device_ips_list") + + if device_ips: + return device_ips + + # If device IPs are not available, check hostnames + device_hostnames = device_specific_details.get("device_hostnames_list") + if device_hostnames: + return self.get_device_ips_from_hostname(device_hostnames) + + # If hostnames are not available, check serial numbers + device_serial_numbers = device_specific_details.get("serial_number_list") + if device_serial_numbers: + return self.get_device_ips_from_serial_number(device_serial_numbers) + + # If serial numbers are not available, check MAC addresses + device_mac_addresses = device_specific_details.get("mac_address_list") + if device_mac_addresses: + return self.get_device_ips_from_mac_address(device_mac_addresses) + + # If no information is available, return an empty list + return [] + + def get_device_ids_from_tag(self, tag_id, tag_name): + """ + Retrieves the device IDs associated with a specific tag from the Cisco Catalyst Center. + + Args: + self (object): An instance of the class used for interacting with Cisco Catalyst Center. + tag_id (str): The unique identifier of the tag from which to retrieve associated device IDs. + tag_name (str): The name of the tag, used for logging purposes. + Returns: + list (str): A list of device IDs (strings) associated with the specified tag. If no devices are found or + an error occurs, the function returns an empty list. + Description: + This function queries the Cisco Catalyst Center API to retrieve a list of devices associated with a given tag. + It calls the `get_tag_members_by_id` function using the tag's ID, specifying that the tag members should be of + type "networkdevice". If the API response contains device data, the function extracts and returns the device IDs. + The function logs whether the tag has associated devices and details about the API response. In the event of an + exception, it logs an error message, sets the operation result to "failed," and returns an empty list. + """ + + device_ids = [] + + try: + response = self.dnac._exec( + family="tag", + function='get_tag_members_by_id', + op_modifies=False, + params={ + "id": tag_id, + "member_type": "networkdevice", + } + ) + response = response.get("response") + if not response: + self.log("No device(s) are associated with the tag '{0}'.".format(tag_name), "WARNING") + return device_ids + + self.log("Received API response from 'get_tag_members_by_id' for the tag {0}: {1}".format(tag_name, response), "DEBUG") + for tag in response: + device_id = tag.get("id") + device_ids.append(device_id) + + except Exception as e: + self.msg = ( + "Exception occurred while fetching tag id for the tag '{0} 'from " + "Cisco Catalyst Center: {1}" + ).format(tag_name, str(e)) + self.set_operation_result("failed", False, self.msg, "INFO").check_return_status() + + return device_ids + def get_diff_merged(self, config): """ Update/Create templates and projects in CCC with fields provided in Cisco Catalyst Center. Export the tempaltes and projects. Import the templates and projects. + Deploy the template to the devices based on device specific details or by fetching the device + details from site using other filtering parameters like device tag, device family, device role. Check using check_return_status(). Parameters: @@ -2413,24 +3086,126 @@ def get_diff_merged(self, config): configuration_templates = config.get("configuration_templates") if configuration_templates: - self.update_configuration_templates(configuration_templates) - if self.status == "failed": - return self + self.update_configuration_templates(configuration_templates).check_return_status() + + _import = config.get("import") + if _import: + self.handle_import(_import).check_return_status() export = config.get("export") if export: - self.handle_export(export) - if self.status == "failed": - return self + self.handle_export(export).check_return_status() + + deploy_temp_details = config.get("deploy_template") + if deploy_temp_details: + template_name = deploy_temp_details.get("template_name") + device_specific_details = deploy_temp_details.get("device_specific_details") + site_specific_details = deploy_temp_details.get("site_associated_provisioning") + + if device_specific_details: + device_ips = self.get_device_ips_from_config_priority(device_specific_details) + if not device_ips: + self.msg = ( + "There is no matched device management ip addresss found for the " + "deployment of template '{0}'" + ).format(template_name) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self - _import = config.get("import") - if _import: - self.handle_import(_import) - if self.status == "failed": + device_ids = self.get_device_ids_from_device_ips(device_ips) + device_missing_msg = ( + "There are no device id found for the device(s) '{0}' in the " + "Cisco Catalyst Center so cannot deploy the given template '{1}'." + ).format(device_ips, template_name) + elif site_specific_details: + device_ids, site_name_list = [], [] + + for site in site_specific_details: + site_name = site.get("site_name") + site_exists, site_id = self.get_site_id(site_name) + if not site_exists: + self.msg = ( + "To Deploy the template in the devices, given site '{0}' must be " + "present in the Cisco Catalyst Center and it's not there currently." + ).format(site_name) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + site_response, site_assign_device_ids = self.get_device_ids_from_site(site_id) + site_name_list.append(site_name) + device_missing_msg = ( + "There is no device currently associated with the site '{0}' in the " + "Cisco Catalyst Center so cannot deploy the given template '{1}'." + ).format(site_name, template_name) + + if not site_assign_device_ids: + self.msg = device_missing_msg + self.log(device_missing_msg, "WARNING") + continue + + device_family = site.get("device_family") + device_role = site.get("device_role") + + # Filter devices based on the device family or device role + if device_family or device_role: + self.log("Filtering devices based on the given family/role for the site {0}.".format(site_name), "INFO") + site_assign_device_ids = self.filter_devices_with_family_role(site_assign_device_ids, device_family, device_role) + + # Filter devices based on the device tag given to the devices + tag_name = site.get("device_tag") + tag_device_ids = None + if tag_name: + self.log("Filtering out the devices based on the given device tag: {0}".format(tag_name), "INFO") + tag_id = self.get_network_device_tag_id(tag_name) + self.log("Successfully collected the tag id {0} for the tag {1}".format(tag_id, tag_name), "INFO") + # Get the device ids associated with the given tag for given site + tag_device_ids = self.get_device_ids_from_tag(tag_id, tag_name) + self.log("Successfully collected the device ids {0} associated with the tag {1}".format(tag_device_ids, tag_name), "INFO") + + self.log("Getting the device ids based on device assoicated with tag or site or both.", "DEBUG") + + if tag_device_ids and site_assign_device_ids: + self.log("Getting the common device ids based on devices fetched from site and with tag.", "DEBUG") + common_device_ids = list(set(tag_device_ids).intersection(set(site_assign_device_ids))) + device_ids.extend(common_device_ids) + elif site_assign_device_ids and not tag_device_ids: + self.log("Getting the device ids based on devices fetched from site.", "DEBUG") + device_ids.extend(site_assign_device_ids) + elif tag_device_ids and not site_assign_device_ids: + self.log("Getting the device ids based on devices fetched with the tag {0}.".format(tag_name), "DEBUG") + device_ids.extend(tag_device_ids) + else: + self.log( + "There is no matching device ids found for the deployment of template {0} " + "for the given site {1}".format(template_name, site_name), "WARNING" + ) + continue + + device_missing_msg = ( + "There is no device id found for the given site(s) '{0}' in the " + "Cisco Catalyst Center so cannot deploy the template '{1}'." + ).format(site_name_list, template_name) + else: + self.msg = ( + "Unable to provision the template '{0}' as device related details are " + "not given in the playboook. Please provide it either via the parameter " + "device_specific_details or with site_associated_provisioning." + ).format(self.msg) + self.set_operation_result("failed", False, self.msg, "INFO").check_return_status() + + if not device_ids: + self.msg = device_missing_msg + self.set_operation_result("failed", False, self.msg, "INFO") return self + device_ips = self.get_device_ips_from_device_ids(device_ids) + self.log("Successfully collect the device ips {0} for the device ids {1}.".format(device_ips, device_ids), "INFO") + deploy_temp_payload = self.create_payload_for_template_deploy(deploy_temp_details, device_ids) + self.deploy_template_to_devices(deploy_temp_payload, template_name, device_ips).check_return_status() + self.msg = "Successfully completed merged state execution" self.status = "success" + return self def delete_project_or_template(self, config, is_delete_project=False): @@ -2462,21 +3237,41 @@ def delete_project_or_template(self, config, is_delete_project=False): params=params_key, ) task_id = response.get("response").get("taskId") - if task_id: - task_details = self.get_task_details(task_id) - self.result['changed'] = True - self.result['response'][0].get("configurationTemplate")['msg'] = task_details.get('progress') - self.result['response'][0].get("configurationTemplate")['diff'] = config.get("configuration_templates") + if not task_id: + self.msg = "Unable to retrive the task_id for the task '{0}'.".format(deletion_value) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self - self.log("Task details for '{0}': {1}".format(deletion_value, task_details), "DEBUG") - self.result['response'][0].get("configurationTemplate")['response'] = task_details if task_details else response - if not self.result['response'][0].get("configurationTemplate")['msg']: - self.result['response'][0].get("configurationTemplate")['msg'] = "Error while deleting {name} : " - self.status = "failed" + while True: + task_details = self.get_task_details_by_id(task_id) + self.log("Printing task details: {0}".format(task_details), "DEBUG") + if not task_details: + self.msg = "Unable to delete {0} as task details is empty.".format(deletion_value) + self.set_operation_result("failed", False, self.msg, "ERROR") return self - self.msg = "Successfully deleted {0} ".format(name) - self.status = "success" + progress = task_details.get("progress") + self.log("Task details for the API {0}: {1}".format(deletion_value, progress), "DEBUG") + + if "deleted" in progress: + self.log("Successfully perform the operation of {0} for {1}".format(deletion_value, name), "INFO") + self.msg = "Successfully deleted {0} ".format(name) + self.set_operation_result("success", True, self.msg, "INFO") + break + + if task_details.get("isError"): + failure_reason = task_details.get("failureReason") + if failure_reason: + self.msg = ( + "Failed to perform the operation of {0} for {1} because of: {2}" + ).format(deletion_value, name, failure_reason) + else: + self.msg = "Failed to perform the operation of {0} for {1}.".format(deletion_value, name) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + time.sleep(self.params.get('dnac_task_poll_interval')) + return self def get_diff_deleted(self, config): @@ -2525,6 +3320,15 @@ def get_diff_deleted(self, config): self.status = "failed" return self + deploy_temp_details = config.get("deploy_template") + if deploy_temp_details: + template_name = deploy_temp_details.get("template_name") + self.msg = ( + "Deleting/removing the device configuration using deployment of template is not supported " + "for the template {0} in the Cisco Catalyst Center." + ).format(template_name) + self.set_operation_result("failed", False, self.msg, "ERROR").check_return_status() + self.msg = "Successfully completed delete state execution" self.status = "success" return self From 96b27629de7e6acfb2363ce214750b67a7d20080 Mon Sep 17 00:00:00 2001 From: Abhishek-121 Date: Thu, 24 Oct 2024 12:36:27 +0530 Subject: [PATCH 2/4] Address review comments, fix the issue of get_device_ids_from_site having site_name as argument in all the modules including site, accesspoint, device credentials, inventory by adding the site_name in argument. --- playbooks/template_workflow_manager.yml | 39 +- plugins/module_utils/dnac.py | 252 +++++++--- .../modules/accesspoint_workflow_manager.py | 7 +- .../device_credential_workflow_manager.py | 2 +- plugins/modules/inventory_intent.py | 17 +- plugins/modules/inventory_workflow_manager.py | 9 +- plugins/modules/site_workflow_manager.py | 2 +- plugins/modules/template_workflow_manager.py | 435 ++++++++++++------ 8 files changed, 510 insertions(+), 253 deletions(-) diff --git a/playbooks/template_workflow_manager.yml b/playbooks/template_workflow_manager.yml index cd1478897d..17d3127d4a 100644 --- a/playbooks/template_workflow_manager.yml +++ b/playbooks/template_workflow_manager.yml @@ -12,11 +12,12 @@ dnac_username: "{{ dnac_username }}" dnac_password: "{{ dnac_password }}" dnac_verify: "{{ dnac_verify }}" + dnac_version: "{{dnac_version}}" dnac_debug: "{{ dnac_debug }}" dnac_log: true dnac_log_level: DEBUG dnac_log_append: true - dnac_log_file_path: "{{ dnac_log_file_path }}" + # dnac_log_file_path: "{{ dnac_log_file_path }}" validate_response_schema: false state: "merged" config_verify: true @@ -45,25 +46,25 @@ deploy_template: project_name: "{{ item.proj_name }}" template_name: "{{ item.temp_name }}" - force_push_template: "{{ item.force_push_template }}" - device_template_params: - - param_name: "{{ item.device_template_params.param_name }}" - param_value: "{{ item.device_template_params.param_value }}" - - param_name: "{{ item.device_template_params.param_name }}" - param_value: "{{ item.device_template_params.param_value }}" - device_specific_details: - # Provide any of the one device_specific details either device_ips_list, device_hostnames_list - # serial_number_list, mac_address_list to deploy template to the devices - # device_ips_list: "{{ item.device_specific_details.device_ips_list }}" - device_hostnames_list: "{{ item.device_specific_details.device_hostnames_list }}" - # serial_number_list: "{{ item.device_specific_details.serial_number_list }}" - # mac_address_list: "{{ item.device_specific_details.mac_address_list }}" - site_associated_provisioning: + force_push: "{{ item.force_push }}" + template_parameters: + - param_name: "{{ item.template_parameters.param_name }}" + param_value: "{{ item.template_parameters.param_value }}" + - param_name: "{{ item.template_parameters.param_name }}" + param_value: "{{ item.template_parameters.param_value }}" + device_details: + # Provide any of the one device_specific details either device_ips, device_hostnames + # serial_numbers, mac_addresses to deploy template to the devices + # device_ips: "{{ item.device_details.device_ips }}" + device_hostnames: "{{ item.device_details.device_hostnames }}" + # serial_numbers: "{{ item.device_details.serial_numbers }}" + # mac_addresses: "{{ item.device_details.mac_addresses }}" + site_provisioning_details: # Provide the site name and other parameters are optional to narrow down the results - - site_name: "{{ item.site_associated_provisioning.site_name }}" - device_family: "{{ item.site_associated_provisioning.device_family }}" - device_role: "{{ item.site_associated_provisioning.device_role }}" - device_tag: "{{ item.site_associated_provisioning.device_tag }}" + - site_name: "{{ item.site_provisioning_details.site_name }}" + device_family: "{{ item.site_provisioning_details.device_family }}" + device_role: "{{ item.site_provisioning_details.device_role }}" + device_tag: "{{ item.site_provisioning_details.device_tag }}" register: template_result with_items: '{{ template_details }}' diff --git a/plugins/module_utils/dnac.py b/plugins/module_utils/dnac.py index 73a4afcc9f..ab2b550b51 100644 --- a/plugins/module_utils/dnac.py +++ b/plugins/module_utils/dnac.py @@ -506,11 +506,11 @@ def reset_values(self): self.have.clear() self.want.clear() - def get_execution_details(self, execid): + def get_execution_details(self, exec_id): """ Get the execution details of an API Args: - execid (str) - Id for API execution + exec_id (str) - Id for API execution Returns: response (dict) - Status for API execution """ @@ -519,10 +519,10 @@ def get_execution_details(self, execid): response = self.dnac._exec( family="task", function='get_business_api_execution_details', - params={"execution_id": execid} + params={"execution_id": exec_id} ) self.log("Successfully retrieved execution details by the API 'get_business_api_execution_details' for execution ID: {0}, Response: {1}" - .format(execid, response), "DEBUG") + .format(exec_id, response), "DEBUG") except Exception as e: # Log an error message and fail if an exception occurs self.log_traceback() @@ -1034,21 +1034,23 @@ def update_site_type_key(self, config): return new_config - def get_device_ips_from_hostname(self, hostname_list): + def get_device_ips_from_hostnames(self, hostnames): """ Get the list of unique device IPs for list of specified hostnames of devices in Cisco Catalyst Center. Parameters: self (object): An instance of a class used for interacting with Cisco Catalyst Center. - hostname_list (list): The hostnames of devices for which you want to retrieve the device IPs. + hostnames (list): The hostnames of devices for which you want to retrieve the device IPs. Returns: - list: The list of unique device IPs for the specified devices hostname list. + device_ip_mapping (dict): Provide the dictionary with the mapping of hostname of device to its ip address. Description: Queries Cisco Catalyst Center to retrieve the unique device IP's associated with a device having the specified list of hostnames. If a device is not found in Cisco Catalyst Center, an error log message is printed. """ - device_ips = [] - for hostname in hostname_list: + self.log("Entering 'get_device_ips_from_hostnames' with hostname_list: {0}".format(str(hostnames)), "INFO") + device_ip_mapping = {} + + for hostname in hostnames: try: response = self.dnac._exec( family="devices", @@ -1057,34 +1059,50 @@ def get_device_ips_from_hostname(self, hostname_list): params={"hostname": hostname} ) if response: - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + self.log("Received API response for hostname '{0}': {1}".format(hostname, str(response)), "DEBUG") response = response.get("response") if response: device_ip = response[0]["managementIpAddress"] if device_ip: - device_ips.append(device_ip) + device_ip_mapping[hostname] = device_ip + self.log("Added device IP '{0}' for hostname '{1}'.".format(device_ip, hostname), "INFO") + else: + device_ip_mapping[hostname] = None + self.log("No management IP found for hostname '{0}'.".format(hostname), "WARNING") + else: + device_ip_mapping[hostname] = None + self.log("No response received for hostname '{0}'.".format(hostname), "WARNING") + else: + device_ip_mapping[hostname] = None + self.log("No response received from 'get_device_list' for hostname '{0}'.".format(hostname), "ERROR") + except Exception as e: - error_message = "Exception occurred while fetching device from Cisco Catalyst Center: {0}".format(str(e)) + error_message = "Exception occurred while fetching device IP for hostname '{0}': {1}".format(hostname, str(e)) self.log(error_message, "ERROR") + device_ip_mapping[hostname] = None + self.log("Exiting 'get_device_ips_from_hostnames' with device IP mapping: {0}".format(device_ip_mapping), "INFO") - return device_ips + return device_ip_mapping - def get_device_ips_from_serial_number(self, serial_number_list): + def get_device_ips_from_serial_numbers(self, serial_numbers): """ Get the list of unique device IPs for a specified list of serial numbers in Cisco Catalyst Center. Parameters: self (object): An instance of a class used for interacting with Cisco Catalyst Center. - serial_number_list (list): The list of serial number of devices for which you want to retrieve the device IPs. + serial_numbers (list): The list of serial number of devices for which you want to retrieve the device IPs. Returns: - list: The list of unique device IPs for the specified devices with serial numbers. + device_ip_mapping (dict): Provide the dictionary with the mapping of serial number of device to its ip address. Description: Queries Cisco Catalyst Center to retrieve the unique device IPs associated with a device having the specified serial numbers.If a device is not found in Cisco Catalyst Center, an error log message is printed. """ - device_ips = [] - for serial_number in serial_number_list: + self.log("Entering 'get_device_ips_from_serial_numbers' with serial_numbers: {0}".format(str(serial_numbers)), "INFO") + device_ip_mapping = {} + + for serial_number in serial_numbers: try: + self.log("Fetching device info for serial number: {0}".format(serial_number), "INFO") response = self.dnac._exec( family="devices", function='get_device_list', @@ -1092,34 +1110,50 @@ def get_device_ips_from_serial_number(self, serial_number_list): params={"serialNumber": serial_number} ) if response: - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + self.log("Received API response for serial number '{0}': {1}".format(serial_number, str(response)), "DEBUG") response = response.get("response") if response: device_ip = response[0]["managementIpAddress"] if device_ip: - device_ips.append(device_ip) + device_ip_mapping[serial_number] = device_ip + self.log("Added device IP '{0}' for serial number '{1}'.".format(device_ip, serial_number), "INFO") + else: + device_ip_mapping[serial_number] = None + self.log("No management IP found for serial number '{0}'.".format(serial_number), "WARNING") + else: + device_ip_mapping[serial_number] = None + self.log("No response received for serial number '{0}'.".format(serial_number), "WARNING") + else: + device_ip_mapping[serial_number] = None + self.log("No response received from 'get_device_list' for serial number '{0}'.".format(serial_number), "ERROR") + except Exception as e: - error_message = "Exception occurred while fetching device from Cisco Catalyst Center - {0}".format(str(e)) + error_message = "Exception occurred while fetching device IP for serial number '{0}': {1}".format(serial_number, str(e)) self.log(error_message, "ERROR") + device_ip_mapping[serial_number] = None + self.log("Exiting 'get_device_ips_from_serial_numbers' with device IP mapping: {0}".format(device_ip_mapping), "INFO") - return device_ips + return device_ip_mapping - def get_device_ips_from_mac_address(self, mac_address_list): + def get_device_ips_from_mac_addresses(self, mac_addresses): """ Get the list of unique device IPs for list of specified mac address of devices in Cisco Catalyst Center. Parameters: self (object): An instance of a class used for interacting with Cisco Catalyst Center. - mac_address_list (list): The list of mac address of devices for which you want to retrieve the device IPs. + mac_addresses (list): The list of mac address of devices for which you want to retrieve the device IPs. Returns: - list: The list of unique device IPs for the specified devices. + device_ip_mapping (dict): Provide the dictionary with the mapping of mac address of device to its ip address. Description: Queries Cisco Catalyst Center to retrieve the unique device IPs associated with a device having the specified mac addresses. If a device is not found in Cisco Catalyst Center, an error log message is printed. """ - device_ips = [] - for mac_address in mac_address_list: + self.log("Entering 'get_device_ips_from_mac_addresses' with mac_addresses: {0}".format(str(mac_addresses)), "INFO") + device_ip_mapping = {} + + for mac_address in mac_addresses: try: + self.log("Fetching device info for mac_address: {0}".format(mac_address), "INFO") response = self.dnac._exec( family="devices", function='get_device_list', @@ -1127,20 +1161,30 @@ def get_device_ips_from_mac_address(self, mac_address_list): params={"macAddress": mac_address} ) if response: - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") + self.log("Received API response for mac address '{0}': {1}".format(mac_address, str(response)), "DEBUG") response = response.get("response") if response: device_ip = response[0]["managementIpAddress"] if device_ip: - device_ips.append(device_ip) + device_ip_mapping[mac_address] = device_ip + self.log("Added device IP '{0}' for mac address '{1}'.".format(device_ip, mac_address), "INFO") + else: + device_ip_mapping[mac_address] = None + self.log("No management IP found for mac address '{0}'.".format(mac_address), "WARNING") + else: + device_ip_mapping[mac_address] = None + self.log("No response received for mac address '{0}'.".format(mac_address), "WARNING") + else: + device_ip_mapping[mac_address] = None + self.log("No response received from 'get_device_list' for mac address '{0}'.".format(mac_address), "ERROR") + except Exception as e: - self.status = "failed" - self.msg = "Exception occurred while fetching device from Cisco Catalyst Center - {0}".format(str(e)) - self.result['response'] = self.msg - self.log(self.msg, "ERROR") - self.check_return_status() + error_message = "Exception occurred while fetching device IP for mac address '{0}': {1}".format(mac_address, str(e)) + self.log(error_message, "ERROR") + device_ip_mapping[mac_address] = None + self.log("Exiting 'get_device_ips_from_mac_addresses' with device IP mapping: {0}".format(device_ip_mapping), "INFO") - return device_ips + return device_ip_mapping def get_device_ids_from_device_ips(self, device_ips): """ @@ -1149,39 +1193,49 @@ def get_device_ids_from_device_ips(self, device_ips): self (object): An instance of a class used for interacting with Cisco Catalyst Center. hostname_list (list): The hostnames of devices for which you want to retrieve the device IPs. Returns: - list: The list of unique device IPs for the specified devices hostname list. + device_id_mapping (dict): Provide the dictionary with the mapping of ip address of device to device id. Description: Queries Cisco Catalyst Center to retrieve the unique device IP's associated with a device having the specified list of hostnames. If a device is not found in Cisco Catalyst Center, an error log message is printed. """ - device_ids = [] + self.log("Entering 'get_device_ids_from_device_ips' with device ips: {0}".format(str(device_ips)), "INFO") + device_id_mapping = {} + for device_ip in device_ips: try: + self.log("Fetching device id for device ip: {0}".format(device_ip), "INFO") response = self.dnac._exec( family="devices", function='get_device_list', op_modifies=False, params={"management_ip_address": device_ip} ) - response = response.get("response") - if not response: - self.log("Unable to fetch the device id for the device '{0}' due to absence of device.".format(device_ip), "WARNING") - continue + if response: + self.log("Received API response for device ip '{0}': {1}".format(device_ip, str(response)), "DEBUG") + response = response.get("response") + if response: + device_id = response[0]["id"] + if device_id: + device_id_mapping[device_ip] = device_id + self.log("Added device ID '{0}' for device ip '{1}'.".format(device_id, device_ip), "INFO") + else: + device_id_mapping[device_ip] = None + self.log("No device ID found for device ip '{0}'.".format(device_ip), "WARNING") + else: + device_id_mapping[device_ip] = None + self.log("No response received for device ip '{0}'.".format(device_ip), "WARNING") + else: + device_id_mapping[device_ip] = None + self.log("No response received from 'get_device_list' for device ip '{0}'.".format(device_ip), "ERROR") - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") - device_id = response[0]["id"] - if device_id: - self.log("Received the device id '{0}' for the device {1}".format(device_id, device_ip), "DEBUG") - device_ids.append(device_id) except Exception as e: - error_message = ( - "Exception occurred while fetching device id for the device '{0} 'from " - "Cisco Catalyst Center: {1}" - ).format(device_ip, str(e)) + error_message = "Exception occurred while fetching device ID for device ip '{0}': {1}".format(device_ip, str(e)) self.log(error_message, "ERROR") + device_id_mapping[device_ip] = None + self.log("Exiting 'get_device_ids_from_device_ips' with unique device ID mapping: {0}".format(device_id_mapping), "INFO") - return device_ids + return device_id_mapping def get_device_ips_from_device_ids(self, device_ids): """ @@ -1190,43 +1244,50 @@ def get_device_ips_from_device_ids(self, device_ids): Args: device_ids (list): A list of device IDs for which the management IP addresses need to be fetched. Returns: - device_ips (list): A list of management IP addresses corresponding to the provided device IDs. If a device ID - doesn't have an associated IP or there is an error, the corresponding IP is not included in the list. + device_ip_mapping (dict): Provide the dictionary with the mapping of id of device to its ip address. Description: This function iterates over a list of device IDs, makes an API call to Cisco Catalyst Center to fetch the management IP addresses of the devices, and returns a list of these IPs. If a device is not found or an exception occurs, it logs the error or warning and continues to the next device ID. """ - device_ips = [] + self.log("Entering 'get_device_ips_from_device_ids' with device ips: {0}".format(str(device_ids)), "INFO") + device_ip_mapping = {} for device_id in device_ids: try: + self.log("Fetching device ip for device id: {0}".format(device_id), "INFO") response = self.dnac._exec( family="devices", function='get_device_list', op_modifies=False, params={"id": device_id} ) - response = response.get("response") - if not response: - self.log("Unable to fetch the device ip for the device '{0}' due to absence of device.".format(device_id), "WARNING") - continue - - self.log("Received API response from 'get_device_list': {0}".format(str(response)), "DEBUG") - device_ip = response[0]["managementIpAddress"] - if device_ip: - self.log("Received the device ip '{0}' for the device having id {1}".format(device_ip, device_id), "DEBUG") - device_ips.append(device_ip) + if response: + self.log("Received API response for device id '{0}': {1}".format(device_id, str(response)), "DEBUG") + response = response.get("response") + if response: + device_ip = response[0]["managementIpAddress"] + if device_ip: + device_ip_mapping[device_id] = device_ip + self.log("Added device IP '{0}' for device id '{1}'.".format(device_ip, device_id), "INFO") + else: + device_ip_mapping[device_id] = None + self.log("No device ID found for device id '{0}'.".format(device_id), "WARNING") + else: + device_ip_mapping[device_id] = None + self.log("No response received for device id '{0}'.".format(device_id), "WARNING") + else: + device_ip_mapping[device_id] = None + self.log("No response received from 'get_device_list' for device id '{0}'.".format(device_id), "ERROR") except Exception as e: - error_message = ( - "Exception occurred while fetching device ip with device id'{0} 'from " - "Cisco Catalyst Center: {1}" - ).format(device_id, str(e)) + error_message = "Exception occurred while fetching device ip for device id '{0}': {1}".format(device_id, str(e)) self.log(error_message, "ERROR") + device_ip_mapping[device_id] = None + self.log("Exiting 'get_device_ips_from_device_ids' with device IP mapping: '{0}'".format(device_ip_mapping), "INFO") - return device_ips + return device_ip_mapping def get_network_device_tag_id(self, tag_name): """ @@ -1244,6 +1305,7 @@ def get_network_device_tag_id(self, tag_name): it logs appropriate messages and returns `None`. """ + self.log("Entering 'get_network_device_tag_id' with tag_name: '{0}'".format(tag_name), "INFO") device_tag_id = None try: @@ -1253,14 +1315,21 @@ def get_network_device_tag_id(self, tag_name): op_modifies=False, params={"name": tag_name} ) - response = response.get("response") if not response: + self.log("No response received from 'get_tag' for tag '{0}'.".format(tag_name), "WARNING") + return device_tag_id + + response_data = response.get("response") + if not response_data: self.log("Unable to fetch the tag details for the tag '{0}'.".format(tag_name), "WARNING") return device_tag_id - self.log("Received API response from 'get_tag': {0}".format(str(response)), "DEBUG") - device_tag_id = response[0]["id"] - self.log("Received the tag id '{0}' for the tag: {1}".format(device_tag_id, tag_name), "INFO") + self.log("Received API response from 'get_tag': {0}".format(str(response_data)), "DEBUG") + device_tag_id = response_data[0]["id"] + if device_tag_id: + self.log("Received the tag ID '{0}' for the tag: {1}".format(device_tag_id, tag_name), "INFO") + else: + self.log("Tag ID not found in the response for tag '{0}'.".format(tag_name), "WARNING") except Exception as e: self.msg = ( @@ -1271,6 +1340,41 @@ def get_network_device_tag_id(self, tag_name): return device_tag_id + def get_list_from_dict_values(self, dict_name): + """ + Extracts values from a dictionary and returns a list of non-None values. + + Args: + self (object): An instance of the class used for interacting with Cisco Catalyst Center. + dict_name (dict): The dictionary from which values are extracted. Each key-value pair is + checked, and non-None values are included in the returned list. + Returns: + list: A list containing all non-None values from the dictionary. + Description: + This function iterates over a given dictionary, checking each key-value pair. If the value + is `None`, it logs a debug message and skips that value. Otherwise, it appends the value to + a list. If an exception occurs during this process, it logs the exception message and handles + the operation result. + """ + + values_list = [] + for key, value in dict_name.items(): + try: + if value is None: + self.log("Value for the key {0} is None so not including in the list.".format(key), "DEBUG") + continue + else: + self.log("Fetch the value '{0}' for the key '{1}'".format(value, key), "DEBUG") + values_list.append(value) + except Exception as e: + self.msg = ( + "Exception occurred while fetching value for the key '{0} 'from " + "Cisco Catalyst Center: {1}" + ).format(key, str(e)) + self.set_operation_result("failed", False, self.msg, "INFO").check_return_status() + + return values_list + def is_valid_ipv4(self, ip_address): """ Validates an IPv4 address. @@ -1509,7 +1613,7 @@ def get_task_details_by_id(self, task_id): Call the API 'get_task_details_by_id' to get the details along with the failure reason. Return the details. """ - # Need to handle exception + task_details = None try: response = self.dnac._exec( diff --git a/plugins/modules/accesspoint_workflow_manager.py b/plugins/modules/accesspoint_workflow_manager.py index 96431dfee2..2d48a9b6f1 100644 --- a/plugins/modules/accesspoint_workflow_manager.py +++ b/plugins/modules/accesspoint_workflow_manager.py @@ -2291,7 +2291,8 @@ def get_site_device(self, site_id, ap_mac_address, site_exist=None, current_site not found or if an error occurs during the API call, it returns False. """ try: - device_list = self.get_device_ids_from_site(site_id) + site_name = self.have.get("site_name_hierarchy", self.want.get("site_name")) + device_list = self.get_device_ids_from_site(site_name, site_id) if current_config.get("id") is not None and current_config.get("id") in device_list: self.log("Device with MAC address: {0} found in site: {1} Proceeding with ap_site updation." .format(ap_mac_address, site_id), "INFO") @@ -2301,8 +2302,8 @@ def get_site_device(self, site_id, ap_mac_address, site_exist=None, current_site return False except Exception as e: - self.log("Failed to execute the get_device_ids_from_site function '{}'\ - Error: {}".format(site_id, str(e)), "ERROR") + self.log("Failed to execute the get_device_ids_from_site function '{0}'\ + Error: {1}".format(site_id, str(e)), "ERROR") return False def verify_ap_provision(self, wlc_ip_address): diff --git a/plugins/modules/device_credential_workflow_manager.py b/plugins/modules/device_credential_workflow_manager.py index 79f3a44a76..57cacae4b0 100644 --- a/plugins/modules/device_credential_workflow_manager.py +++ b/plugins/modules/device_credential_workflow_manager.py @@ -2782,7 +2782,7 @@ def apply_credentials_to_site(self): self.status = "success" return self - site_response = self.get_device_ids_from_site(site_id) + site_response = self.get_device_ids_from_site(site_name, site_id) if not site_response: result_apply_credential.update({ "No Apply Credentials": { diff --git a/plugins/modules/inventory_intent.py b/plugins/modules/inventory_intent.py index 25c5c7a012..f66b204d95 100644 --- a/plugins/modules/inventory_intent.py +++ b/plugins/modules/inventory_intent.py @@ -864,7 +864,7 @@ def get_device_ips_from_config_priority(self): If none of the information is available, an empty list is returned. """ # Retrieve device IPs from the configuration - device_ips = self.config[0].get("ip_address_list") + device_ips = self.want.get("device_params").get("ipAddress") if device_ips: return device_ips @@ -872,17 +872,20 @@ def get_device_ips_from_config_priority(self): # If device IPs are not available, check hostnames device_hostnames = self.config[0].get("hostname_list") if device_hostnames: - return self.get_device_ips_from_hostname(device_hostnames) + device_ip_dict = self.get_device_ips_from_hostnames(device_hostnames) + return self.get_list_from_dict_values(device_ip_dict) # If hostnames are not available, check serial numbers device_serial_numbers = self.config[0].get("serial_number_list") if device_serial_numbers: - return self.get_device_ips_from_serial_number(device_serial_numbers) + device_ip_dict = self.get_device_ips_from_serial_numbers(device_serial_numbers) + return self.get_list_from_dict_values(device_ip_dict) # If serial numbers are not available, check MAC addresses device_mac_addresses = self.config[0].get("mac_address_list") if device_mac_addresses: - return self.get_device_ips_from_mac_address(device_mac_addresses) + device_ip_dict = self.get_device_ips_from_mac_addresses(device_mac_addresses) + return self.get_list_from_dict_values(device_ip_dict) # If no information is available, return an empty list return [] @@ -2239,7 +2242,7 @@ def get_device_ids(self, device_ips): return device_ids - def get_device_ips_from_hostname(self, hostname_list): + def get_device_ips_from_hostnames(self, hostname_list): """ Get the list of unique device IPs for list of specified hostnames of devices in Cisco Catalyst Center. Parameters: @@ -2274,7 +2277,7 @@ def get_device_ips_from_hostname(self, hostname_list): return device_ips - def get_device_ips_from_serial_number(self, serial_number_list): + def get_device_ips_from_serial_numbers(self, serial_number_list): """ Get the list of unique device IPs for a specified list of serial numbers in Cisco Catalyst Center. Parameters: @@ -2309,7 +2312,7 @@ def get_device_ips_from_serial_number(self, serial_number_list): return device_ips - def get_device_ips_from_mac_address(self, mac_address_list): + def get_device_ips_from_mac_addresses(self, mac_address_list): """ Get the list of unique device IPs for list of specified mac address of devices in Cisco Catalyst Center. Parameters: diff --git a/plugins/modules/inventory_workflow_manager.py b/plugins/modules/inventory_workflow_manager.py index 9e8fafabe6..a60f913fec 100644 --- a/plugins/modules/inventory_workflow_manager.py +++ b/plugins/modules/inventory_workflow_manager.py @@ -878,17 +878,20 @@ def get_device_ips_from_config_priority(self): # If device IPs are not available, check hostnames device_hostnames = self.config[0].get("hostname_list") if device_hostnames: - return self.get_device_ips_from_hostname(device_hostnames) + device_ip_dict = self.get_device_ips_from_hostnames(device_hostnames) + return self.get_list_from_dict_values(device_ip_dict) # If hostnames are not available, check serial numbers device_serial_numbers = self.config[0].get("serial_number_list") if device_serial_numbers: - return self.get_device_ips_from_serial_number(device_serial_numbers) + device_ip_dict = self.get_device_ips_from_serial_numbers(device_serial_numbers) + return self.get_list_from_dict_values(device_ip_dict) # If serial numbers are not available, check MAC addresses device_mac_addresses = self.config[0].get("mac_address_list") if device_mac_addresses: - return self.get_device_ips_from_mac_address(device_mac_addresses) + device_ip_dict = self.get_device_ips_from_mac_addresses(device_mac_addresses) + return self.get_list_from_dict_values(device_ip_dict) # If no information is available, return an empty list return [] diff --git a/plugins/modules/site_workflow_manager.py b/plugins/modules/site_workflow_manager.py index a7567d29bc..a01c25b930 100644 --- a/plugins/modules/site_workflow_manager.py +++ b/plugins/modules/site_workflow_manager.py @@ -1598,7 +1598,7 @@ def get_diff_deleted(self, config): if self.compare_dnac_versions(self.get_ccc_version(), "2.3.5.3") <= 0: site_id = self.have.get("site_id") - api_response, response = self.get_device_ids_from_site(site_id) + api_response, response = self.get_device_ids_from_site(site_name, site_id) self.log( "Received API response from 'get_membership': {0}".format(str(api_response)), "DEBUG") diff --git a/plugins/modules/template_workflow_manager.py b/plugins/modules/template_workflow_manager.py index 98b8f93bac..2f28205671 100644 --- a/plugins/modules/template_workflow_manager.py +++ b/plugins/modules/template_workflow_manager.py @@ -812,17 +812,17 @@ type: str deploy_template: description: To deploy the template to the devices based on either list of site provisionig details with further filtering - criteria like device family, device role, device tag or by providing the device specific details which includes device_ips_list, - device_hostnames_list, serial_number_list or mac_address_list. + criteria like device family, device role, device tag or by providing the device specific details which includes device_ips, + device_hostnames, serial_numbers or mac_addresses. type: dict suboptions: project_name: - description: Provide the name of project under which template is available. + description: Provide the name of project under which the template is available. type: str template_name: description: Name of the template to be deployed. type: str - force_push_template: + force_push: description: Boolean flag to indicate whether the template should be forcefully pushed to the devices, overriding any existing configuration. type: bool @@ -830,7 +830,7 @@ description: Boolean flag indicating whether the template is composite, which means the template is built using multiple smaller templates. type: bool - device_template_params: + template_parameters: description: A list of parameter name-value pairs used for customizing the template with specific values for each device. type: list elements: dict @@ -841,29 +841,29 @@ param_value: description: Value assigned to the parameter for deployment to devices. type: str - device_specific_details: + device_details: description: Details specific to devices where the template will be deployed, including lists of device IPs, hostnames, serial numbers, or MAC addresses. type: list elements: dict suboptions: - device_ips_list: + device_ips: description: A list of IP addresses of the devices where the template will be deployed. type: list elements: str - device_hostnames_list: + device_hostnames: description: A list of hostnames of the devices where the template will be deployed. type: list elements: str - serial_number_list: + serial_numbers: description: A list of serial numbers of the devices where the template will be deployed. type: list elements: str - mac_address_list: + mac_addresses: description: A list of MAC addresses of the devices where the template will be deployed. type: list elements: str - site_associated_provisioning: + site_provisioning_details: description: Parameters related to site-based provisioning, allowing the deployment of templates to devices associated with specific sites, with optional filtering by device family, role, or tag. type: list @@ -1044,13 +1044,13 @@ deploy_template: project_name: "Sample_Project" template_name: "Sample Template" - force_push_template: true - device_template_params: + force_push: true + template_parameters: - param_name: "vlan_id" param_value: "1431" - param_name: "vlan_name" param_value: "testvlan31" - site_associated_provisioning: + site_provisioning_details: - site_name: "Global/Bangalore/Building14/Floor1" device_family: "Switches and Hubs" @@ -1071,14 +1071,14 @@ deploy_template: project_name: "Sample_Project" template_name: "Sample Template" - force_push_template: true - device_template_params: + force_push: true + template_parameters: - param_name: "vlan_id" param_value: "1431" - param_name: "vlan_name" param_value: "testvlan31" - device_specific_details: - - device_ips_list: ["10.1.2.1", "10.2.3.4"] + device_details: + - device_ips: ["10.1.2.1", "10.2.3.4"] - name: Delete the given project or template from the Cisco Catalyst Center cisco.dnac.template_workflow_manager: @@ -1261,23 +1261,23 @@ def validate_input(self): 'type': 'dict', 'project_name': {'type': 'str'}, 'template_name': {'type': 'str'}, - 'force_push_template': {'type': 'bool'}, + 'force_push': {'type': 'bool'}, 'is_composite': {'type': 'bool'}, - 'device_template_params': { + 'template_parameters': { 'type': 'list', 'elements': 'dict', 'param_name': {'type': 'str'}, 'param_value': {'type': 'str'}, }, - 'device_specific_details': { + 'device_details': { 'type': 'list', 'elements': 'dict', - 'device_ips_list': {'type': 'list', 'elements': 'str'}, - 'device_hostnames_list': {'type': 'list', 'elements': 'str'}, - 'serial_number_list': {'type': 'list', 'elements': 'str'}, - 'mac_address_list': {'type': 'list', 'elements': 'str'}, + 'device_ips': {'type': 'list', 'elements': 'str'}, + 'device_hostnames': {'type': 'list', 'elements': 'str'}, + 'serial_numbers': {'type': 'list', 'elements': 'str'}, + 'mac_addresses': {'type': 'list', 'elements': 'str'}, }, - 'site_associated_provisioning': { + 'site_provisioning_details': { 'type': 'list', 'elements': 'dict', 'site_name': {'type': 'str'}, @@ -1632,25 +1632,32 @@ def get_project_defined_template_details(self, project_name, template_name): project_name (str) - Name of the project under which templates are associated. template_name (str) - Name of the template provided in the playbook. Returns: - result (dict) - Template details for the given template name. + template_details (dict) - Template details for the given template name. """ - result = None - items = self.dnac_apply['exec']( - family="configuration_templates", - function="get_templates_details", - op_modifies=True, - params={ - "project_name": project_name, - "name": template_name - } - ) - if items: - result = items + self.log("Starting to retrieve template details for project '{0}' and template '{1}'.".format(project_name, template_name), "INFO") + template_details = None + try: + items = self.dnac_apply['exec']( + family="configuration_templates", + function="get_templates_details", + op_modifies=True, + params={ + "project_name": project_name, + "name": template_name + } + ) + if items: + template_details = items + self.log("Received template details for '{0}': {1}".format(template_name, template_details), "DEBUG") + else: + self.log("No template details found for project '{0}' and template '{1}'.".format(project_name, template_name), "WARNING") - self.log("Received API response from 'get_templates_details': {0}".format(items), "DEBUG") + self.log("Received API response from 'get_templates_details': {0}".format(template_details), "DEBUG") + except Exception as e: + self.log("Exception occurred while retrieving template details for '{0}': {1}".format(template_name, str(e)), "ERROR") - return result + return template_details def get_containing_templates(self, containing_templates): """ @@ -1883,32 +1890,42 @@ def get_uncommitted_template_id(self, project_name, template_name): or deployed. If the template is unavailable, an appropriate log message is recorded and the function exits early with `None`. """ - + self.log("Retrieving uncommitted template ID for project '{0}' and template " + "'{1}'.".format(project_name, template_name), "INFO" + ) template_id = None - template_list = self.dnac_apply['exec']( - family="configuration_templates", - function="gets_the_templates_available", - op_modifies=False, - params={ - "projectNames": project_name, - "un_committed": True - }, - ) - msg = ( - "Given template '{0}' is not available under the project '{1}' " - "so cannot commit or deploy the template in device(s)." - ).format(template_name, project_name) - - if not template_list: - self.log(msg, "WARNING") - self.msg = msg - return template_id - - for template in template_list: - if template.get("name") == template_name: - template_id = template.get("templateId") + try: + template_list = self.dnac_apply['exec']( + family="configuration_templates", + function="gets_the_templates_available", + op_modifies=False, + params={ + "projectNames": project_name, + "un_committed": True + }, + ) + if not template_list: + msg = ( + "No uncommitted templates available under the project '{0}'. " + "Cannot commit or deploy the template '{1}' in device(s)." + ).format(project_name, template_name) + self.log(msg, "WARNING") return template_id + for template in template_list: + if template.get("name") == template_name: + template_id = template.get("templateId") + self.log("Found uncommitted template '{0}' with ID: '{1}'.".format(template_name, template_id), "INFO") + return template_id + self.log("Template '{0}' not found in the uncommitted templates for project '{1}'.".format(template_name, project_name), "WARNING") + except Exception as e: + error_msg = ( + "Exception occurred while retrieving uncommitted template ID for project '{0}' and " + "template '{1}': {2}." + ).format(project_name, template_name, str(e)) + self.log(error_msg, "ERROR") + self.msg = error_msg + return template_id def versioned_given_template(self, project_name, template_name, template_id): @@ -1929,6 +1946,7 @@ def versioned_given_template(self, project_name, template_name, template_id): The function returns the class instance for further chaining of operations. """ + self.log("Starting the versioning process for template '{0}' in project '{1}'.".format(template_name, project_name), "INFO") try: comments = ( "Given template '{0}' under the project '{1}' versioned successfully." @@ -1938,6 +1956,7 @@ def versioned_given_template(self, project_name, template_name, template_id): "comments": comments, "templateId": template_id } + self.log("Preparing to version template with parameters: {0}".format(version_params), "DEBUG") task_name = "version_template" task_id = self.get_taskid_post_api_call("configuration_templates", task_name, version_params) @@ -2086,15 +2105,18 @@ def get_have(self, config): if deploy_temp_details: template_name = deploy_temp_details.get("template_name") project_name = deploy_temp_details.get("project_name") + self.log("Fetching template details for '{0}' under project '{1}'.".format(template_name, project_name), "INFO") temp_details = self.get_project_defined_template_details(project_name, template_name).get("response") if temp_details: self.log("Given template '{0}' is already committed in the Catalyst Center.".format(template_name), "INFO") have["temp_id"] = temp_details[0].get("id") - self.log("Successfully collect the details for the template '{0}' from the " - "Cisco Catalyst Center.".format(template_name), "INFO" - ) + self.log("Successfully collected the details for the template '{0}' from the " + "Cisco Catalyst Center.".format(template_name), "INFO" + ) + else: + self.log("No details found for template '{0}' under project '{1}'.".format(template_name, project_name), "WARNING") self.have = have @@ -2160,6 +2182,7 @@ def get_want(self, config): self.set_operation_result("failed", False, self.msg, "ERROR") return self + self.log("Project name '{0}' found in the playbook.".format(project_name), "INFO") template_name = deploy_temp_details.get("template_name") if not template_name: self.msg = ( @@ -2169,17 +2192,19 @@ def get_want(self, config): self.set_operation_result("failed", False, self.msg, "ERROR") return self - device_specific_details = deploy_temp_details.get("device_specific_details") - site_associated_provisioning = deploy_temp_details.get("site_associated_provisioning") + self.log("Template name '{0}' found in the playbook.".format(template_name), "INFO") + device_details = deploy_temp_details.get("device_details") + site_provisioning_details = deploy_temp_details.get("site_provisioning_details") - if not (device_specific_details or site_associated_provisioning): + if not (device_details or site_provisioning_details): self.msg = ( - "Either give the parameter 'device_specific_details' or 'site_associated_provisioning' " + "Either give the parameter 'device_details' or 'site_provisioning_details' " "in the playbook to fetch the device ids and proceed for the deployment of template {0}." ).format(template_name) self.set_operation_result("failed", False, self.msg, "ERROR") return self + self.log("Proceeding with deployment details for template '{0}'.".format(template_name), "INFO") want["deploy_tempate"] = deploy_temp_details self.want = want @@ -2775,9 +2800,13 @@ def filter_devices_with_family_role(self, site_assign_device_ids, device_family= """ filtered_device_list = [] + self.log("Filtering devices from the provided site-assigned device IDs: {0}, device_family='{1}', " + "and device_role='{2}'".format(site_assign_device_ids, device_family, device_role), "DEBUG" + ) for device_id in site_assign_device_ids: try: + self.log("Processing device ID: {0}".format(device_id), "DEBUG") response = self.dnac._exec( family="devices", function='get_device_list', @@ -2788,26 +2817,50 @@ def filter_devices_with_family_role(self, site_assign_device_ids, device_family= "role": device_role } ) - response = response.get('response') + if response and "response" in response: + response_data = response.get("response") + else: + self.log("No valid response for device with ID '{0}'.".format(device_id), "INFO") + continue - if not response: + if not response_data: self.log( - "Device with id '{0}' does not belong to the device family '{1}' or not having the " - " device role as {2}".format(device_id, device_family, device_role), "INFO" + "Device with ID '{0}' does not match family '{1}' or role '{2}'.".format(device_id, device_family, device_role), + "INFO" ) continue + self.log("Device with ID '{0}' matches the criteria.".format(device_id), "DEBUG") filtered_device_list.append(device_id) except Exception as e: error_message = "Error while getting the response of device from Cisco Catalyst Center: {0}".format(str(e)) self.log(error_message, "CRITICAL") continue + self.log("Completed filtering. Filtered devices: {0}".format(filtered_device_list), "DEBUG") return filtered_device_list def get_latest_template_version_id(self, template_id, template_name): + """ + Fetches the latest version ID of a specified template from the Cisco Catalyst Center. + + Args: + self (object): An instance of the class interacting with Cisco Catalyst Center. + template_id (str): The unique identifier of the template to retrieve its versions. + template_name (str): The name of the template for logging and reference purposes. + Returns: + str: The ID of the latest version of the template if available; otherwise, returns None. + Description: + This method calls the Cisco Catalyst Center API to fetch all versions of the specified template. + It selects the version with the most recent timestamp and retrieves its version ID. + If no versions are available or an error occurs during the API call, appropriate logs are generated. + """ version_temp_id = None + self.log( + "Fetching the latest version ID for template '{0}' using template_id '{1}'.".format( + template_name, template_id), "DEBUG" + ) try: response = self.dnac._exec( @@ -2819,22 +2872,32 @@ def get_latest_template_version_id(self, template_id, template_name): } ) - if not response: + if not response or not isinstance(response, list) or not response[0].get("versionsInfo"): self.log( - "There is no versioning present for the template {0} in the Cisco " - "Catalyst Center.".format(template_name), "INFO" + "No version information found for template '{0}' in Cisco Catalyst Center.".format(template_name), "INFO" ) - response = response[0].get("versionsInfo") + return version_temp_id + + self.log( + "Successfully retrieved version information for template '{0}'.".format(template_name), "DEBUG" + ) + versions_info = response[0].get("versionsInfo") self.log( - "Received API response from 'get_tempget_template_versionslate_version' for template " - "{0}: {1}".format(str(response), template_name), "DEBUG" + "Processing version details for template '{0}': {1}".format(template_name, str(versions_info)), "DEBUG" ) - latest_version = max(response, key=lambda x: x["versionTime"]) + latest_version = max(versions_info, key=lambda x: x["versionTime"]) version_temp_id = latest_version.get("id") + self.log( + "Identified the latest version for template '{0}'. Version ID: {1}".format( + template_name, version_temp_id), "DEBUG" + ) except Exception as e: - error_message = "Error while getting the latest version id for the template {0}: {1}".format(template_name, str(e)) + error_message = "Error while getting the latest version id for the template '{0}': '{1}'".format(template_name, str(e)) self.log(error_message, "CRITICAL") + self.log( + "Returning latest version ID '{0}' for template '{1}'.".format(version_temp_id, template_name), "DEBUG" + ) return version_temp_id @@ -2859,11 +2922,22 @@ def create_payload_for_template_deploy(self, deploy_temp_details, device_ids): project_name = deploy_temp_details.get("project_name") template_name = deploy_temp_details.get("template_name") + self.log( + "Starting to create deployment payload for template '{0}' in project '{1}'." + .format(template_name, project_name), "DEBUG" + ) # Check if the template is available but not yet committed if self.have.get("temp_id"): - self.log("Given template '{0}' is already committed in the Cisco Catalyst Center".format(template_name), "INFO") + self.log( + "Template '{0}' is already committed in Cisco Catalyst Center. Using the committed template ID." + .format(template_name), "INFO" + ) template_id = self.have.get("temp_id") else: + self.log( + "Fetching uncommitted template ID for template '{0}' in project '{1}'.".format(template_name, project_name), + "DEBUG" + ) template_id = self.get_uncommitted_template_id(project_name, template_name) if not template_id: @@ -2873,24 +2947,34 @@ def create_payload_for_template_deploy(self, deploy_temp_details, device_ids): ).format(template_name) self.set_operation_result("failed", False, self.msg, "ERROR").check_return_status() - self.log("Given template '{0}' is available and is not committed yet.".format(template_name), "INFO") + self.log( + "Template '{0}' is available but not committed yet. Committing template...".format(template_name), + "INFO" + ) # Commit or versioned the given template in the Catalyst Center self.versioned_given_template(project_name, template_name, template_id).check_return_status() deploy_payload = { - "forcePushTemplate": deploy_temp_details.get("force_push_template", False), + "forcePushTemplate": deploy_temp_details.get("force_push", False), "isComposite": deploy_temp_details.get("is_composite", False), "templateId": template_id, } + self.log( + "Handling template parameters for the deployment of template '{0}'.".format(template_name), + "DEBUG" + ) target_info_list = [] template_dict = {} - device_template_params = deploy_temp_details.get("device_template_params") - if not device_template_params: - self.msg = "Template parameters is not given in the playbook so cannot deploy {0} to the devices.".format(template_name) + template_parameters = deploy_temp_details.get("template_parameters") + if not template_parameters: + self.msg = ( + "It appears that no template parameters were provided in the playbook. Unfortunately, this " + "means we cannot proceed with deploying template '{0}' to the devices." + ).format(template_name) self.set_operation_result("failed", False, self.msg, "ERROR").check_return_status() - for param in device_template_params: + for param in template_parameters: name = param["param_name"] value = param["param_value"] self.log("Update the template placeholder for the name '{0}' with value {1}".format(name, value), "DEBUG") @@ -2902,7 +2986,9 @@ def create_payload_for_template_deploy(self, deploy_temp_details, device_ids): self.log("No versioning found for the template: {0}".format(template_name), "INFO") version_template_id = template_id + self.log("Preparing to deploy template '{0}' to the following device IDs: '{1}'".format(template_name, device_ids), "DEBUG") for device_id in device_ids: + self.log("Adding device '{0}' to the deployment payload.".format(device_id), "DEBUG") target_device_dict = { "id": device_id, "type": "MANAGED_DEVICE_UUID", @@ -2913,7 +2999,7 @@ def create_payload_for_template_deploy(self, deploy_temp_details, device_ids): del target_device_dict deploy_payload["targetInfo"] = target_info_list - self.log("Successfully collected the payload for the deploy of template '{0}'.".format(template_name), "INFO") + self.log("Successfully generated deployment payload for template '{0}'.".format(template_name), "INFO") return deploy_payload @@ -2947,14 +3033,32 @@ def deploy_template_to_devices(self, deploy_temp_payload, template_name, device_ task_id = self.get_taskid_post_api_call("configuration_templates", task_name, payload) if not task_id: - self.msg = "Unable to retrive the task_id for the task '{0}'.".format(task_name) + self.msg = "Unable to retrieve the task_id for the task '{0}'.".format(task_name) self.set_operation_result("failed", False, self.msg, "ERROR") return self + loop_start_time = time.time() + sleep_duration = self.params.get('dnac_task_poll_interval') + self.log("Starting task monitoring for '{0}' with task ID '{1}'.".format(task_name, task_id), "DEBUG") + while True: task_details = self.get_task_details_by_id(task_id) + if not task_details: + self.msg = "Error retrieving task status for '{0}' with task ID '{1}'".format(task_name, task_id) + self.set_operation_result("failed", False, self.msg, "ERROR") + return self + + # Check if the elapsed time exceeds the timeout + elapsed_time = time.time() - loop_start_time + if self.check_timeout_and_exit(loop_start_time, task_id, task_name): + self.log( + "Timeout exceeded after {0:.2f} seconds while monitoring task '{1}' with task ID '{2}'.".format( + elapsed_time, task_name, task_id), "DEBUG" + ) + return self + progress = task_details.get("progress") - self.log("Task details for the API {0}: {1}".format(task_name, progress), "DEBUG") + self.log("Task ID '{0}' details for the API '{1}': {2}".format(task_id, task_name, progress), "DEBUG") if "not deploying" in progress: self.log("Deployment of the template {0} gets failed because of: {1}".format(template_name, progress), "WARNING") @@ -2970,7 +3074,8 @@ def deploy_template_to_devices(self, deploy_temp_payload, template_name, device_ self.set_operation_result("success", True, self.msg, "INFO") return self - time.sleep(self.params.get('dnac_task_poll_interval')) + self.log("Waiting for {0} seconds before checking the task status again.".format(sleep_duration), "DEBUG") + time.sleep(sleep_duration) except Exception as e: self.msg = ( @@ -2981,7 +3086,7 @@ def deploy_template_to_devices(self, deploy_temp_payload, template_name, device_ return self - def get_device_ips_from_config_priority(self, device_specific_details): + def get_device_ips_from_config_priority(self, device_details): """ Retrieve device IPs based on the configuration. Parameters: @@ -2995,37 +3100,51 @@ def get_device_ips_from_config_priority(self, device_specific_details): If none of the information is available, an empty list is returned. """ # Retrieve device IPs from the configuration - device_ips = device_specific_details.get("device_ips_list") - - if device_ips: - return device_ips - - # If device IPs are not available, check hostnames - device_hostnames = device_specific_details.get("device_hostnames_list") - if device_hostnames: - return self.get_device_ips_from_hostname(device_hostnames) - - # If hostnames are not available, check serial numbers - device_serial_numbers = device_specific_details.get("serial_number_list") - if device_serial_numbers: - return self.get_device_ips_from_serial_number(device_serial_numbers) - - # If serial numbers are not available, check MAC addresses - device_mac_addresses = device_specific_details.get("mac_address_list") - if device_mac_addresses: - return self.get_device_ips_from_mac_address(device_mac_addresses) + self.log("Retrieving device IPs based on the configuration priority with details: {0}".format(device_details), "INFO") + try: + device_ips = device_details.get("device_ips") + + if device_ips: + self.log("Found device IPs: {0}".format(device_ips), "INFO") + return device_ips + + # If device IPs are not available, check hostnames + device_hostnames = device_details.get("device_hostnames") + if device_hostnames: + self.log("No device IPs found. Checking hostnames: {0}".format(device_hostnames), "INFO") + device_ip_dict = self.get_device_ips_from_hostnames(device_hostnames) + return self.get_list_from_dict_values(device_ip_dict) + + # If hostnames are not available, check serial numbers + device_serial_numbers = device_details.get("serial_numbers") + if device_serial_numbers: + self.log("No device IPs or hostnames found. Checking serial numbers: {0}".format(device_serial_numbers), "INFO") + device_ip_dict = self.get_device_ips_from_serial_numbers(device_serial_numbers) + return self.get_list_from_dict_values(device_ip_dict) + + # If serial numbers are not available, check MAC addresses + device_mac_addresses = device_details.get("mac_addresses") + if device_mac_addresses: + self.log("No device IPs, hostnames, or serial numbers found. Checking MAC addresses: {0}".format(device_mac_addresses), "INFO") + device_ip_dict = self.get_device_ips_from_mac_addresses(device_mac_addresses) + return self.get_list_from_dict_values(device_ip_dict) + + # If no information is available, return an empty list + self.log("No device information available to retrieve IPs.", "WARNING") + return [] - # If no information is available, return an empty list - return [] + except Exception as e: + self.log("No device information available to retrieve IPs.", "WARNING") + return [] - def get_device_ids_from_tag(self, tag_id, tag_name): + def get_device_ids_from_tag(self, tag_name, tag_id): """ Retrieves the device IDs associated with a specific tag from the Cisco Catalyst Center. Args: self (object): An instance of the class used for interacting with Cisco Catalyst Center. - tag_id (str): The unique identifier of the tag from which to retrieve associated device IDs. tag_name (str): The name of the tag, used for logging purposes. + tag_id (str): The unique identifier of the tag from which to retrieve associated device IDs. Returns: list (str): A list of device IDs (strings) associated with the specified tag. If no devices are found or an error occurs, the function returns an empty list. @@ -3038,6 +3157,7 @@ def get_device_ids_from_tag(self, tag_id, tag_name): """ device_ids = [] + self.log("Fetching device IDs associated with the tag '{0}' (ID: {1}).".format(tag_name, tag_id), "INFO") try: response = self.dnac._exec( @@ -3049,14 +3169,20 @@ def get_device_ids_from_tag(self, tag_id, tag_name): "member_type": "networkdevice", } ) - response = response.get("response") - if not response: + if response and "response" in response: + response_data = response.get("response") + else: + self.log("No valid response for device with tag ID '{0}'.".format(tag_id), "INFO") + return device_ids + + if not response_data: self.log("No device(s) are associated with the tag '{0}'.".format(tag_name), "WARNING") return device_ids - self.log("Received API response from 'get_tag_members_by_id' for the tag {0}: {1}".format(tag_name, response), "DEBUG") - for tag in response: + self.log("Received API response from 'get_tag_members_by_id' for the tag {0}: {1}".format(tag_name, response_data), "DEBUG") + for tag in response_data: device_id = tag.get("id") + self.log("Device ID '{0}' found for tag '{1}'.".format(device_id, tag_name), "DEBUG") device_ids.append(device_id) except Exception as e: @@ -3094,25 +3220,33 @@ def get_diff_merged(self, config): export = config.get("export") if export: + self.log("Found export configuration: {0}".format(export), "DEBUG") self.handle_export(export).check_return_status() deploy_temp_details = config.get("deploy_template") if deploy_temp_details: template_name = deploy_temp_details.get("template_name") - device_specific_details = deploy_temp_details.get("device_specific_details") - site_specific_details = deploy_temp_details.get("site_associated_provisioning") - - if device_specific_details: - device_ips = self.get_device_ips_from_config_priority(device_specific_details) + device_details = deploy_temp_details.get("device_details") + site_specific_details = deploy_temp_details.get("site_provisioning_details") + self.log("Deploy template details found for template '{0}'".format(template_name), "DEBUG") + self.log("Device specific details: {0}".format(device_details), "DEBUG") + self.log("Site associated provisioning details: {0}".format(site_specific_details), "DEBUG") + + if device_details: + self.log("Attempting to retrieve device IPs based on priority from device specific details.", "DEBUG") + device_ips = self.get_device_ips_from_config_priority(device_details) if not device_ips: self.msg = ( - "There is no matched device management ip addresss found for the " - "deployment of template '{0}'" + "No matching device management IP addresses found for the " + "deployment of template '{0}'." ).format(template_name) self.set_operation_result("failed", False, self.msg, "ERROR") return self - device_ids = self.get_device_ids_from_device_ips(device_ips) + self.log("Successfully retrieved device IPs for template '{0}': '{1}'".format(template_name, device_ips), "INFO") + device_id_dict = self.get_device_ids_from_device_ips(device_ips) + device_ids = self.get_list_from_dict_values(device_id_dict) + device_missing_msg = ( "There are no device id found for the device(s) '{0}' in the " "Cisco Catalyst Center so cannot deploy the given template '{1}'." @@ -3123,6 +3257,7 @@ def get_diff_merged(self, config): for site in site_specific_details: site_name = site.get("site_name") site_exists, site_id = self.get_site_id(site_name) + self.log("Checking if the site '{0}' exists in Cisco Catalyst Center.".format(site_name), "DEBUG") if not site_exists: self.msg = ( "To Deploy the template in the devices, given site '{0}' must be " @@ -3131,14 +3266,15 @@ def get_diff_merged(self, config): self.set_operation_result("failed", False, self.msg, "ERROR") return self - site_response, site_assign_device_ids = self.get_device_ids_from_site(site_id) + self.log("Retrieving devices associated with site ID '{0}' for site '{1}'.".format(site_id, site_name), "DEBUG") + site_response, site_assign_device_ids = self.get_device_ids_from_site(site_name, site_id) site_name_list.append(site_name) - device_missing_msg = ( - "There is no device currently associated with the site '{0}' in the " - "Cisco Catalyst Center so cannot deploy the given template '{1}'." - ).format(site_name, template_name) if not site_assign_device_ids: + device_missing_msg = ( + "There is no device currently associated with the site '{0}' in the " + "Cisco Catalyst Center so cannot deploy the given template '{1}'." + ).format(site_name, template_name) self.msg = device_missing_msg self.log(device_missing_msg, "WARNING") continue @@ -3148,6 +3284,10 @@ def get_diff_merged(self, config): # Filter devices based on the device family or device role if device_family or device_role: + self.log( + "Filtering devices based on the device family '{0}' or role '{1}' for the site '{2}'.".format( + device_family, device_role, site_name), "DEBUG" + ) self.log("Filtering devices based on the given family/role for the site {0}.".format(site_name), "INFO") site_assign_device_ids = self.filter_devices_with_family_role(site_assign_device_ids, device_family, device_role) @@ -3155,17 +3295,17 @@ def get_diff_merged(self, config): tag_name = site.get("device_tag") tag_device_ids = None if tag_name: - self.log("Filtering out the devices based on the given device tag: {0}".format(tag_name), "INFO") + self.log("Filtering out the devices based on the given device tag: '{0}'".format(tag_name), "INFO") tag_id = self.get_network_device_tag_id(tag_name) - self.log("Successfully collected the tag id {0} for the tag {1}".format(tag_id, tag_name), "INFO") + self.log("Successfully collected the tag id '{0}' for the tag '{1}'".format(tag_id, tag_name), "INFO") # Get the device ids associated with the given tag for given site - tag_device_ids = self.get_device_ids_from_tag(tag_id, tag_name) + tag_device_ids = self.get_device_ids_from_tag(tag_name, tag_id) self.log("Successfully collected the device ids {0} associated with the tag {1}".format(tag_device_ids, tag_name), "INFO") self.log("Getting the device ids based on device assoicated with tag or site or both.", "DEBUG") if tag_device_ids and site_assign_device_ids: - self.log("Getting the common device ids based on devices fetched from site and with tag.", "DEBUG") + self.log("Determining device IDs from site and tag criteria.", "DEBUG") common_device_ids = list(set(tag_device_ids).intersection(set(site_assign_device_ids))) device_ids.extend(common_device_ids) elif site_assign_device_ids and not tag_device_ids: @@ -3189,7 +3329,7 @@ def get_diff_merged(self, config): self.msg = ( "Unable to provision the template '{0}' as device related details are " "not given in the playboook. Please provide it either via the parameter " - "device_specific_details or with site_associated_provisioning." + "device_details or with site_provisioning_details." ).format(self.msg) self.set_operation_result("failed", False, self.msg, "INFO").check_return_status() @@ -3198,10 +3338,13 @@ def get_diff_merged(self, config): self.set_operation_result("failed", False, self.msg, "INFO") return self - device_ips = self.get_device_ips_from_device_ids(device_ids) + device_ip_dict = self.get_device_ips_from_device_ids(device_ids) + device_ips = self.get_list_from_dict_values(device_ip_dict) self.log("Successfully collect the device ips {0} for the device ids {1}.".format(device_ips, device_ids), "INFO") deploy_temp_payload = self.create_payload_for_template_deploy(deploy_temp_details, device_ids) + self.log("Deployment payload created successfully for template '{0}'.".format(template_name), "INFO") self.deploy_template_to_devices(deploy_temp_payload, template_name, device_ips).check_return_status() + self.log("Successfully deployed template '{0}'.".format(template_name), "INFO") self.msg = "Successfully completed merged state execution" self.status = "success" @@ -3237,8 +3380,9 @@ def delete_project_or_template(self, config, is_delete_project=False): params=params_key, ) task_id = response.get("response").get("taskId") + sleep_duration = self.params.get('dnac_task_poll_interval') if not task_id: - self.msg = "Unable to retrive the task_id for the task '{0}'.".format(deletion_value) + self.msg = "Unable to retrieve the task ID for the task '{0}'.".format(deletion_value) self.set_operation_result("failed", False, self.msg, "ERROR") return self @@ -3268,9 +3412,10 @@ def delete_project_or_template(self, config, is_delete_project=False): else: self.msg = "Failed to perform the operation of {0} for {1}.".format(deletion_value, name) self.set_operation_result("failed", False, self.msg, "ERROR") - return self + break - time.sleep(self.params.get('dnac_task_poll_interval')) + self.log("Waiting for {0} seconds before checking the task status again.".format(sleep_duration), "DEBUG") + time.sleep(sleep_duration) return self @@ -3324,7 +3469,7 @@ def get_diff_deleted(self, config): if deploy_temp_details: template_name = deploy_temp_details.get("template_name") self.msg = ( - "Deleting/removing the device configuration using deployment of template is not supported " + "Deleting or removing the device configuration using deployment of template is not supported " "for the template {0} in the Cisco Catalyst Center." ).format(template_name) self.set_operation_result("failed", False, self.msg, "ERROR").check_return_status() From 75cb1482954cac4f9d0ba9b4cdae627307a91d99 Mon Sep 17 00:00:00 2001 From: Abhishek-121 Date: Thu, 24 Oct 2024 12:43:42 +0530 Subject: [PATCH 3/4] rename the variable in inventory_intent module --- plugins/modules/inventory_intent.py | 44 ++++++++++++++--------------- 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/plugins/modules/inventory_intent.py b/plugins/modules/inventory_intent.py index f66b204d95..5bb3345dce 100644 --- a/plugins/modules/inventory_intent.py +++ b/plugins/modules/inventory_intent.py @@ -89,17 +89,17 @@ or resyncing devices, with Meraki devices being the exception. elements: str type: list - hostname_list: + hostnames: description: "A list of hostnames representing devices. Operations such as updating, deleting, resyncing, or rebooting can be performed as alternatives to using IP addresses." type: list elements: str - serial_number_list: + serial_numbers: description: A list of serial numbers representing devices. Operations such as updating, deleting, resyncing, or rebooting can be performed as alternatives to using IP addresses. type: list elements: str - mac_address_list: + mac_addresses: description: "A list of MAC addresses representing devices. Operations such as updating, deleting, resyncing, or rebooting can be performed as alternatives to using IP addresses." type: list @@ -771,9 +771,9 @@ def validate_input(self): 'http_secure': {'type': 'bool'}, 'http_username': {'type': 'str'}, 'ip_address_list': {'type': 'list', 'elements': 'str'}, - 'hostname_list': {'type': 'list', 'elements': 'str'}, - 'serial_number_list': {'type': 'list', 'elements': 'str'}, - 'mac_address_list': {'type': 'list', 'elements': 'str'}, + 'hostnames': {'type': 'list', 'elements': 'str'}, + 'serial_numbers': {'type': 'list', 'elements': 'str'}, + 'mac_addresses': {'type': 'list', 'elements': 'str'}, 'netconf_port': {'type': 'str'}, 'password': {'type': 'str'}, 'snmp_auth_passphrase': {'type': 'str'}, @@ -870,19 +870,19 @@ def get_device_ips_from_config_priority(self): return device_ips # If device IPs are not available, check hostnames - device_hostnames = self.config[0].get("hostname_list") + device_hostnames = self.config[0].get("hostnames") if device_hostnames: device_ip_dict = self.get_device_ips_from_hostnames(device_hostnames) return self.get_list_from_dict_values(device_ip_dict) # If hostnames are not available, check serial numbers - device_serial_numbers = self.config[0].get("serial_number_list") + device_serial_numbers = self.config[0].get("serial_numbers") if device_serial_numbers: device_ip_dict = self.get_device_ips_from_serial_numbers(device_serial_numbers) return self.get_list_from_dict_values(device_ip_dict) # If serial numbers are not available, check MAC addresses - device_mac_addresses = self.config[0].get("mac_address_list") + device_mac_addresses = self.config[0].get("mac_addresses") if device_mac_addresses: device_ip_dict = self.get_device_ips_from_mac_addresses(device_mac_addresses) return self.get_list_from_dict_values(device_ip_dict) @@ -1468,7 +1468,7 @@ def reboot_access_points(self): return self # Get and store the apEthernetMacAddress of given devices - ap_mac_address_list = [] + ap_mac_addresses = [] for device_ip in input_device_ips: response = self.dnac._exec( family="devices", @@ -1484,9 +1484,9 @@ def reboot_access_points(self): ap_mac_address = response.get('apEthernetMacAddress') if ap_mac_address is not None: - ap_mac_address_list.append(ap_mac_address) + ap_mac_addresses.append(ap_mac_address) - if not ap_mac_address_list: + if not ap_mac_addresses: self.status = "success" self.result['changed'] = False self.msg = "Cannot find the AP devices for rebooting" @@ -1496,7 +1496,7 @@ def reboot_access_points(self): # Now call the Reboot Access Point API reboot_params = { - "apMacAddresses": ap_mac_address_list + "apMacAddresses": ap_mac_addresses } response = self.dnac._exec( family="wireless", @@ -2242,12 +2242,12 @@ def get_device_ids(self, device_ips): return device_ids - def get_device_ips_from_hostnames(self, hostname_list): + def get_device_ips_from_hostnames(self, hostnames): """ Get the list of unique device IPs for list of specified hostnames of devices in Cisco Catalyst Center. Parameters: self (object): An instance of a class used for interacting with Cisco Catalyst Center. - hostname_list (list): The hostnames of devices for which you want to retrieve the device IPs. + hostnames (list): The hostnames of devices for which you want to retrieve the device IPs. Returns: list: The list of unique device IPs for the specified devices hostname list. Description: @@ -2256,7 +2256,7 @@ def get_device_ips_from_hostnames(self, hostname_list): """ device_ips = [] - for hostname in hostname_list: + for hostname in hostnames: try: response = self.dnac._exec( family="devices", @@ -2277,12 +2277,12 @@ def get_device_ips_from_hostnames(self, hostname_list): return device_ips - def get_device_ips_from_serial_numbers(self, serial_number_list): + def get_device_ips_from_serial_numbers(self, serial_numbers): """ Get the list of unique device IPs for a specified list of serial numbers in Cisco Catalyst Center. Parameters: self (object): An instance of a class used for interacting with Cisco Catalyst Center. - serial_number_list (list): The list of serial number of devices for which you want to retrieve the device IPs. + serial_numbers (list): The list of serial number of devices for which you want to retrieve the device IPs. Returns: list: The list of unique device IPs for the specified devices with serial numbers. Description: @@ -2291,7 +2291,7 @@ def get_device_ips_from_serial_numbers(self, serial_number_list): """ device_ips = [] - for serial_number in serial_number_list: + for serial_number in serial_numbers: try: response = self.dnac._exec( family="devices", @@ -2312,12 +2312,12 @@ def get_device_ips_from_serial_numbers(self, serial_number_list): return device_ips - def get_device_ips_from_mac_addresses(self, mac_address_list): + def get_device_ips_from_mac_addresses(self, mac_addresses): """ Get the list of unique device IPs for list of specified mac address of devices in Cisco Catalyst Center. Parameters: self (object): An instance of a class used for interacting with Cisco Catalyst Center. - mac_address_list (list): The list of mac address of devices for which you want to retrieve the device IPs. + mac_addresses (list): The list of mac address of devices for which you want to retrieve the device IPs. Returns: list: The list of unique device IPs for the specified devices. Description: @@ -2326,7 +2326,7 @@ def get_device_ips_from_mac_addresses(self, mac_address_list): """ device_ips = [] - for mac_address in mac_address_list: + for mac_address in mac_addresses: try: response = self.dnac._exec( family="devices", From 21101909dcc113c468e95a269b3520e26593df8a Mon Sep 17 00:00:00 2001 From: Abhishek-121 Date: Thu, 24 Oct 2024 13:11:07 +0530 Subject: [PATCH 4/4] remove extra line --- plugins/module_utils/dnac.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/plugins/module_utils/dnac.py b/plugins/module_utils/dnac.py index ab2b550b51..90e2e438ba 100644 --- a/plugins/module_utils/dnac.py +++ b/plugins/module_utils/dnac.py @@ -1080,8 +1080,8 @@ def get_device_ips_from_hostnames(self, hostnames): error_message = "Exception occurred while fetching device IP for hostname '{0}': {1}".format(hostname, str(e)) self.log(error_message, "ERROR") device_ip_mapping[hostname] = None - self.log("Exiting 'get_device_ips_from_hostnames' with device IP mapping: {0}".format(device_ip_mapping), "INFO") + self.log("Exiting 'get_device_ips_from_hostnames' with device IP mapping: {0}".format(device_ip_mapping), "INFO") return device_ip_mapping def get_device_ips_from_serial_numbers(self, serial_numbers): @@ -1131,8 +1131,8 @@ def get_device_ips_from_serial_numbers(self, serial_numbers): error_message = "Exception occurred while fetching device IP for serial number '{0}': {1}".format(serial_number, str(e)) self.log(error_message, "ERROR") device_ip_mapping[serial_number] = None - self.log("Exiting 'get_device_ips_from_serial_numbers' with device IP mapping: {0}".format(device_ip_mapping), "INFO") + self.log("Exiting 'get_device_ips_from_serial_numbers' with device IP mapping: {0}".format(device_ip_mapping), "INFO") return device_ip_mapping def get_device_ips_from_mac_addresses(self, mac_addresses): @@ -1182,8 +1182,8 @@ def get_device_ips_from_mac_addresses(self, mac_addresses): error_message = "Exception occurred while fetching device IP for mac address '{0}': {1}".format(mac_address, str(e)) self.log(error_message, "ERROR") device_ip_mapping[mac_address] = None - self.log("Exiting 'get_device_ips_from_mac_addresses' with device IP mapping: {0}".format(device_ip_mapping), "INFO") + self.log("Exiting 'get_device_ips_from_mac_addresses' with device IP mapping: {0}".format(device_ip_mapping), "INFO") return device_ip_mapping def get_device_ids_from_device_ips(self, device_ips): @@ -1233,8 +1233,8 @@ def get_device_ids_from_device_ips(self, device_ips): error_message = "Exception occurred while fetching device ID for device ip '{0}': {1}".format(device_ip, str(e)) self.log(error_message, "ERROR") device_id_mapping[device_ip] = None - self.log("Exiting 'get_device_ids_from_device_ips' with unique device ID mapping: {0}".format(device_id_mapping), "INFO") + self.log("Exiting 'get_device_ids_from_device_ips' with unique device ID mapping: {0}".format(device_id_mapping), "INFO") return device_id_mapping def get_device_ips_from_device_ids(self, device_ids): @@ -1285,8 +1285,8 @@ def get_device_ips_from_device_ids(self, device_ids): error_message = "Exception occurred while fetching device ip for device id '{0}': {1}".format(device_id, str(e)) self.log(error_message, "ERROR") device_ip_mapping[device_id] = None - self.log("Exiting 'get_device_ips_from_device_ids' with device IP mapping: '{0}'".format(device_ip_mapping), "INFO") + self.log("Exiting 'get_device_ips_from_device_ids' with device IP mapping: '{0}'".format(device_ip_mapping), "INFO") return device_ip_mapping def get_network_device_tag_id(self, tag_name):