Getting CDP (Cisco Discovery Protocol) information via Python on specified list of ESXi hosts

get_cdp.py

#!/usr/bin/env python
import sys
import atexit
from pyVim.connect import SmartConnectNoSSL, Disconnect
import pyVmomi
from pyVmomi import vim
from typing import List, Dict, Union
from tabulate import tabulate
from getpass import getpass


def collect_properties(content, view_ref, obj_type, path_set=None, incl_ref=False, key=None):
    """
    Collect properties for managed objects from a view ref
    Check the vSphere API documentation for example on retrieving
    object properties: http://goo.gl/erbFDz
    Args:
        content     (ServiceInstance): ServiceInstance connection
        view_ref (pyVmomi.vim.view.*): Starting point of inventory navigation
        obj_type      (pyVmomi.vim.*): Type of managed object
        path_set               (list): List of properties to retrieve
        incl_ref               (bool): If True include the managed objects refs in the result
        key                     (str): Optional prop to use as dict key. If not specified a list will be returned.
    Returns:
        A list (or dict if key is present) of properties for the managed objects.
    """
    if key is not None and key not in path_set:
        raise ValueError('Key is not one of the paths in path_set.')

    collector = content.propertyCollector

    # Create object specification to define the starting point of
    # inventory navigation
    obj_spec = pyVmomi.vmodl.query.PropertyCollector.ObjectSpec()
    obj_spec.obj = view_ref
    obj_spec.skip = True

    # Create a traversal specification to identify the path for collection
    traversal_spec = pyVmomi.vmodl.query.PropertyCollector.TraversalSpec()
    traversal_spec.name = 'traverseEntities'
    traversal_spec.path = 'view'
    traversal_spec.skip = False
    traversal_spec.type = view_ref.__class__
    obj_spec.selectSet = [traversal_spec]

    # Identify the properties to the retrieved
    property_spec = pyVmomi.vmodl.query.PropertyCollector.PropertySpec()
    property_spec.type = obj_type

    if not path_set:
        property_spec.all = True

    property_spec.pathSet = path_set

    # Add the object and property specification to the
    # property filter specification
    filter_spec = pyVmomi.vmodl.query.PropertyCollector.FilterSpec()
    filter_spec.objectSet = [obj_spec]
    filter_spec.propSet = [property_spec]

    # Retrieve properties
    props = collector.RetrieveContents([filter_spec])

    data = {} if key else []
    for obj in props:
        properties = {}
        for prop in obj.propSet:
            properties[prop.name] = prop.val

        if incl_ref:
            properties['obj'] = obj.obj

        if key is None:
            data.append(properties)
        else:
            data[properties[key]] = properties

    return data


def exception_handler(ex_type, ex_val, ex_traceback):
    if serviceInst: Disconnect(serviceInst)
    sys.__excepthook__(ex_type, ex_val, ex_traceback)


def connect_viserver(hostname: str, username: str, password: str):
    global serviceInst, viServer
    serviceInst = SmartConnectNoSSL(host=hostname, user=username, pwd=password)
    atexit.register(Disconnect, serviceInst)
    # sys.excepthook = exception_handler

    viServer = serviceInst.RetrieveContent()
    print('Connected to {}. Variables and methods defined:\n'
          'serviceInst: ServiceInstance\n'
          'viServer: ServiceInstance content\n'
          'list_hosts(list_only=True)\n'
          'get_cdp(host: Union[str, vim.HostSystem], list_only=True)'.format(hostname))


def list_hosts(list_only=True):
    """
    Print a formatted list of all the hosts or return a dict of all the hosts.
    Action performed depends on the value of list_only.
    """
    if viServer is None:
        print('Use ConnectVIServer first.')
        return
    view = viServer.viewManager.CreateContainerView(viServer.rootFolder, [vim.HostSystem], True)
    props = ['name',
             'summary.overallStatus',
             'summary.runtime.powerState',
             'summary.quickStats.uptime',
             'summary.quickStats.overallCpuUsage',
             'summary.quickStats.overallMemoryUsage',
             'summary.hardware.cpuMhz',
             'summary.hardware.numCpuCores',
             'summary.hardware.memorySize']

    hosts: Dict[str, Dict[str, object]] = collect_properties(viServer, view, vim.HostSystem, props, True, 'name')
    if list_only:
        hostsFormated = sorted([(
            host['name'],
            host.get('summary.overallStatus', 'unknown'),
            host.get('summary.runtime.powerState', 'unknown'),
            f"{host.get('summary.quickStats.uptime', 0) / 86400:.1f} days",
            'unknown' if None in (host.get('summary.quickStats.overallCpuUsage'),
                                  host.get('summary.hardware.cpuMhz'),
                                  host.get('summary.hardware.numCpuCores')) else
            f"{host['summary.quickStats.overallCpuUsage'] / host['summary.hardware.cpuMhz'] / host['summary.hardware.numCpuCores']:.0%}",
            'unknown' if None in (host.get('summary.quickStats.overallMemoryUsage'),
                                  host.get('summary.hardware.memorySize')) else
            f"{host['summary.quickStats.overallMemoryUsage'] / host['summary.hardware.memorySize'] * 1048576:.0%}")
            for host in hosts.values()], key=lambda h: h[0])
        print(
            tabulate(hostsFormated, headers=('Name', 'Status', 'Power State', 'Up Time', 'CPU Usage', 'Memory Usage'), stralign='right'))
    else:
        return hosts


def get_cdp(host: Union[str, vim.HostSystem], list_only=True):
    if viServer is None:
        print('Use ConnectVIServer first.')
        return
    if type(host) is str:
        view = viServer.viewManager.CreateContainerView(viServer.rootFolder, [vim.HostSystem], True)
        host = collect_properties(viServer, view, vim.HostSystem, ['name'], True, 'name').get(host)['obj']

    table: List[dict] = []

    # print('HostName', 'PhysicalNic', 'LinkSpeed (Mb)', 'SwitchID', 'Platform', 'IpAddress', 'PortId', 'Vlan', sep='\t')
    for pnic in host.configManager.networkSystem.networkInfo.pnic:
        entry = {
            'host': host.name,
            'device': pnic.device,
            'pci': pnic.pci,
            'mac': pnic.mac,
            'linkSpeedMb': '-' if pnic.linkSpeed is None else pnic.linkSpeed.speedMb
        }
        for hint in host.configManager.networkSystem.QueryNetworkHint(pnic.device):
            csp = hint.connectedSwitchPort
            if csp is not None:
                entry['switchId'] = csp.devId
                entry['platform'] = csp.hardwarePlatform
                entry['ipAddress'] = csp.address
                entry['portId'] = csp.portId
                entry['vlan'] = csp.vlan
        table.append(entry)
        # print(entry['host'], entry['device'], entry['linkSpeedMb'], entry['switchId'], entry['platform'],
        #       entry['ipAddress'], entry['portId'], entry['vlan'], sep='\t')
    if list_only:
        print(tabulate(table, headers='keys'))
    else:
        return table


if __name__ == "__main__":
    serviceInst = None
    viServer = None
    connect_viserver(input('Host address: '), input('Username: '), getpass())