hello
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/ |
# 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)