vmware-nsx

Instructions

Implementation Reference

• vmware_nsx/ops/troubleshoot.py:105-195 (handler)

  The core implementation of get_segment_port_for_vm, which queries the NSX fabric for VM information and cross-references it with segment ports.

  def get_segment_port_for_vm(
      client: NsxClient,
      vm_display_name: str,
  ) -> dict:
      """Find the segment port(s) associated with a VM by display name.
  
      Queries the NSX fabric for VM info, then cross-references with
      segment ports to find connectivity.
  
      Args:
          client: Authenticated NSX API client.
          vm_display_name: VM display name to search for.
  
      Returns:
          Dict with VM info and associated segment ports.
      """
      sanitized_name = _sanitize(vm_display_name, max_len=200)
  
      # Step 1: Find the VM in NSX fabric
      vm_data = client.get(
          "/api/v1/fabric/virtual-machines",
          params={"display_name": sanitized_name},
      )
      vms = vm_data.get("results", [])
  
      if not vms:
          return {
              "vm_display_name": sanitized_name,
              "found": False,
              "hint": (
                  f"No VM found with display name '{sanitized_name}'. "
                  "Verify the VM exists and NSX has discovered it."
              ),
          }
  
      vm = vms[0]
      vm_external_id = vm.get("external_id", "")
  
      # Step 2: Get VIF attachments for this VM
      vifs = vm.get("virtual_interfaces", [])
      vif_external_ids = [
          vif.get("external_id", "") for vif in vifs if vif.get("external_id")
      ]
  
      # Step 3: Search for segment ports with matching VIF attachments
      # Query all segments and their ports to find the match
      segments = client.get_all("/policy/api/v1/infra/segments")
  
      matched_ports: list[dict] = []
      for seg in segments:
          seg_id = seg.get("id", "")
          try:
              ports = client.get_all(
                  f"/policy/api/v1/infra/segments/{seg_id}/ports"
              )
          except Exception:
              continue
  
          for p in ports:
              attachment = p.get("attachment", {})
              attachment_id = attachment.get("id", "")
  
              # Match by VIF external ID or VM external ID
              if attachment_id in vif_external_ids or (
                  attachment.get("type") == "VIF"
                  and attachment_id == vm_external_id
              ):
                  matched_ports.append(
                      {
                          "segment_id": _sanitize(seg_id),
                          "segment_name": _sanitize(
                              seg.get("display_name", "")
                          ),
                          "port_id": _sanitize(p.get("id", "")),
                          "port_name": _sanitize(
                              p.get("display_name", "")
                          ),
                          "attachment_id": _sanitize(attachment_id),
                      }
                  )
  
      return {
          "vm_display_name": sanitized_name,
          "found": True,
          "vm_external_id": _sanitize(vm_external_id),
          "host_id": _sanitize(vm.get("host_id", "")),
          "power_state": vm.get("power_state", ""),
          "matched_ports": matched_ports,
          "port_count": len(matched_ports),
      }