Source code for axonius_api_client.api.system.instances

# -*- coding: utf-8 -*-
"""API for working with instances."""
import datetime
from typing import List, Optional, Union

from ...exceptions import NotFoundError
from .. import json_api
from ..api_endpoints import ApiEndpoints
from ..mixins import ModelMixins


[docs]class Instances(ModelMixins): """API for working with instances. Examples: * Get all instances: :meth:`get` * Get the core instance: :meth:`get_core` * Get all collector instances: :meth:`get_collectors` * Get an instance by name: :meth:`get_by_name` * Set the name of an instance: :meth:`set_name` * Get the hostname of an instance: :meth:`get_hostname` * Set the hostname of an instance: :meth:`get_hostname` * Get the setting for is environment name of an instance: :meth:`get_is_env_name` * Set the setting for is environment name of an instance: :meth:`set_is_env_name` * Get the central core mode: :meth:`get_central_core_mode` * Set the central core mode: :meth:`get_central_core_mode` * Get the core delete backups mode: :meth:`get_core_delete_mode` * Set the core delete backups mode: :meth:`set_core_delete_mode` * Get the central core config: :meth:`get_central_core_config` * Perform a restore on a core from a file in an AWS S3 Bucket: :meth:`restore_from_aws_s3` Notes: Certain functionality is not yet exposed, please submit a feature request via support@axonius.com if you would like to: * Deactivate an instance * Connect an instance * Tag an instance """
[docs] def get(self) -> dict: """Get metadata about all instances. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> data = client.instances.get() >>> len(data) 1 """ return [x.to_dict() for x in self._get()]
[docs] def get_core(self, key: Optional[str] = None) -> Union[dict, Union[str, bool, int, float]]: """Get the core instance. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> instance = client.instances.get_core() >>> instance['node_name'] 'Master' >>> instance['cpu_usage'] 39 >>> instance['data_disk_free_space'] 74431500 >>> instance['hostname'] 'builds-vm-jim-pre-3-10-1601596999-000' >>> instance['ips'] ['10.20.0.61'] >>> instance['data_disk_free_space_percent'] 36.62 >>> instance['memory_free_space_percent'] 1.46 >>> instance['swap_free_space_percent'] 0.0 Args: key: key to return or just return whole object """ instances = self.get() for instance in instances: if instance["is_master"]: return instance[key] if key else instance
[docs] def get_collectors(self) -> List[dict]: """Get the collector instances. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> collectors = client.instances.get_collectors() >>> len(collectors) 0 """ instances = self.get() return [x for x in instances if not x.get("is_master")]
[docs] def get_by_name( self, name: str, key: Optional[str] = None ) -> Union[dict, Union[str, bool, int, float]]: """Get an instance by name. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> instance = client.instances.get_by_name("Master") >>> instance['cpu_usage'] 39 >>> instance['data_disk_free_space'] 74431500 >>> instance['hostname'] 'builds-vm-jim-pre-3-10-1601596999-000' >>> instance['ips'] ['10.20.0.61'] >>> instance['data_disk_free_space_percent'] 36.62 >>> instance['memory_free_space_percent'] 1.46 >>> instance['swap_free_space_percent'] 0.0 Args: name: name of instance key: key to return or just return whole object """ instances = self.get() valid = [] for instance in instances: instance_name = instance["name"] valid.append(instance_name) if instance_name == name: return instance[key] if key else instance valid = "\n - " + "\n - ".join(valid) raise NotFoundError(f"No instance (node) named {name!r} found, valid: {valid}")
[docs] def get_by_name_id_core(self, value: Optional[str] = None, serial: bool = True) -> dict: """Pass.""" data = None instances = self._get() for instance in instances: if not value and instance.is_master: data = instance break if instance.name == value or instance.id == value: data = instance break if not data: valid = [f"Name: {x.name}, ID: {x.id}" for x in instances] valid = "\n - " + "\n - ".join(valid) raise NotFoundError(f"No instance with ID or name of {value!r} found, valid:{valid}") if serial: data = data.to_dict() data = {k: str(v) if isinstance(v, datetime.datetime) else v for k, v in data.items()} return data
[docs] def set_name(self, name: str, new_name: str) -> str: """Set the name of an instance. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` Get the current name >>> orig_value = client.instances.get_core(key='name') >>> orig_value 'Master' Set a new name >>> new_value = client.instances.set_name(name=orig_value, new_name='Temp') >>> new_value 'Temp' Revert back to the original name >>> reset_value = client.instances.set_name(name=new_value, new_name=orig_value) >>> reset_value 'Master' Args: name: name of instance new_name: new name to set on instance """ instance = self.get_by_name(name=name) self._update_attrs( node_id=instance["id"], node_name=new_name, hostname=instance["hostname"], use_as_environment_name=instance["use_as_environment_name"], ) return self.get_by_name(name=new_name, key="name")
[docs] def get_hostname(self, name: str) -> str: """Get the host name of an instance. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` Get the hostname >>> hostname = client.instances.get_hostname(name='Master') >>> hostname 'builds-vm-jim-pre-3-10-1601596999-000' Args: name: name of instance value: new hostname to set on instance """ return self.get_by_name(name=name, key="hostname")
[docs] def set_hostname(self, name: str, value: str) -> str: """Set the hostname of an instance. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` Get the current hostname >>> orig_value = client.instances.get_hostname(name='Master') >>> orig_value 'builds-vm-jim-pre-3-10-1601596999-000' Set a new hostname >>> new_value = client.instances.set_hostname(name='Master', value="hostname") >>> new_value "hostname" Revert to the old hostname >>> reset_value = client.instances.set_hostname(name='Master', value=orig_value) >>> reset_value 'builds-vm-jim-pre-3-10-1601596999-000' Args: name: name of instance value: new hostname to set on instance """ instance = self.get_by_name(name=name) self._update_attrs( node_id=instance["id"], node_name=instance["node_name"], hostname=value, use_as_environment_name=instance["use_as_environment_name"], ) return self.get_hostname(name=name)
[docs] def get_is_env_name(self, name: str) -> bool: """See if an instance name is being used for the environment name. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> client.instances.get_is_env_name(name='Master') False Notes: Environment name is the name shown in the banner at the bottom of the GUI. Args: name: name of instance """ return self.get_by_name(name=name, key="use_as_environment_name")
[docs] def set_is_env_name(self, name: str, enabled: bool) -> bool: """Set if an instance name is being used for the environment name. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` Enable an instance name as the environment name >>> client.instances.set_is_env_name(name='Master', enabled=True) True Disable an instance name as the environment name >>> client.instances.set_is_env_name(name='Master', enabled=False) False Notes: Environment name is the name shown in the banner at the bottom of the GUI. Args: name: name of instance enabled: enable/disable instance name as the environment name """ instance = self.get_by_name(name=name) self._update_attrs( node_id=instance["id"], node_name=instance["node_name"], hostname=instance["hostname"], use_as_environment_name=enabled, ) return self.get_is_env_name(name=name)
[docs] def get_central_core_mode(self) -> bool: """Get a bool that shows if a core is in central core mode. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> client.instances.get_central_core_mode() False """ data = self.get_central_core_config() return data["enabled"]
[docs] def set_central_core_mode(self, enabled: bool) -> bool: """Convert a normal core into a central core. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` Enable central core mode on a core >>> client.instances.set_central_core_mode(enabled=True) True Disable central core mode on a core >>> client.instances.set_central_core_mode(enabled=False) False Args: enabled: enable/disable central core mode """ data = self.get_central_core_config() self._update_central_core_config(enabled=enabled, delete_backups=data["delete_backups"]) return self.get_central_core_mode()
[docs] def get_core_delete_mode(self) -> bool: """Get a bool that shows if a core is in central core mode. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> client.instances.get_core_delete_mode() True """ data = self.get_central_core_config() return data["delete_backups"]
[docs] def set_core_delete_mode(self, enabled: bool) -> bool: """Configure a normal core to delete backups after they have been restored. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` Set backups to delete after restore >>> client.instances.set_core_delete_mode(enabled=True) True Set backups to NOT delete after restore >>> client.instances.set_core_delete_mode(enabled=False) False Args: enabled: enable/disable deletion of backups after they have been restored by a core """ data = self.get_central_core_config() self._update_central_core_config(enabled=data["enabled"], delete_backups=enabled) return self.get_core_delete_mode()
[docs] def factory_reset( self, reset: bool = False ) -> json_api.instances.FactoryReset: # pragma: no cover """Perform a factory reset on an instance. Notes: Can not run in test suite! Args: reset: actually do it... be careful! """ return self._factory_reset(approve_not_recoverable_action=reset)
[docs] def get_central_core_config(self) -> dict: """Get the current central core configuration. Examples: Create a ``client`` using :obj:`axonius_api_client.connect.Connect` >>> client.instances.get_central_core_config() {'core_delete_backups': False, 'central_core_enabled': False} """ data = self._get_central_core_config() return data.to_dict()["config"]
[docs] def restore_from_aws_s3( self, key_name: str, bucket_name: Optional[str] = None, access_key_id: Optional[str] = None, secret_access_key: Optional[str] = None, preshared_key: Optional[str] = None, allow_re_restore: Optional[bool] = None, delete_backups: Optional[bool] = None, ) -> dict: # pragma: no cover """Perform a restore on a core from a file in an AWS S3 Bucket. Notes: Can not run in test suite! Args: key_name: Name of backup file from central core in [bucket_name] to restore to this core bucket_name: Name of bucket in S3 to get [key_name] from (Overrides ``Global Settings > Amazon S3 Settings > Amazon S3 bucket name``) access_key_id: AWS Access Key Id to use to access [bucket_name] (Overrides ``Global Settings > Amazon S3 Settings > AWS Access Key Id``) secret_access_key: AWS Secret Access Key to use to access [bucket_name] (Overrides ``Global Settings > Amazon S3 Settings > AWS Secret Access Key``) preshared_key: Password to use to decrypt [key_name] (Overrides: ``Global Settings > Amazon S3 Settings > Backup encryption passphrase``) allow_re_restore: Restore [key_name] even if it has already been restored delete_backups: Delete [key_name] from [bucket_name] after restore has finished (Overrides the current value of :meth:`get_core_delete_mode`) """ restore_opts = {} restore_opts["key_name"] = key_name restore_opts["allow_re_restore"] = allow_re_restore restore_opts["delete_backups"] = delete_backups restore_opts["bucket_name"] = bucket_name restore_opts["access_key_id"] = access_key_id restore_opts["secret_access_key"] = secret_access_key restore_opts["preshared_key"] = preshared_key response = self._restore_aws(**restore_opts) return response.to_dict()
@property def feature_flags(self) -> json_api.system_settings.FeatureFlags: """Get the feature flags for the core.""" return self._feature_flags() @property def has_cloud_compliance(self) -> bool: """Get the status of cloud compliance module being enabled.""" return self.feature_flags.has_cloud_compliance @property def trial_expiry(self) -> Optional[datetime.datetime]: """Get the trial expiration date.""" return self.feature_flags.trial_expiry_dt @property def trial_days_left(self) -> Optional[int]: """Get the number of days left for the trial.""" return self.feature_flags.trial_expiry_in_days @property def license_expiry(self) -> Optional[datetime.datetime]: """Get the license expiration date.""" return self.feature_flags.license_expiry_dt @property def license_days_left(self) -> Optional[int]: """Get the number of days left for the license.""" return self.feature_flags.license_expiry_in_days
[docs] def _get(self) -> List[json_api.instances.Instance]: """Direct API method to get instances.""" api_endpoint = ApiEndpoints.instances.get return api_endpoint.perform_request(http=self.auth.http)
[docs] def _factory_reset( self, approve_not_recoverable_action: bool = False ) -> json_api.instances.FactoryReset: # pragma: no cover """Direct API method to do a factory reset on an instance. Notes: Can not run in test suite! Args: approve_not_recoverable_action: actually do it... be careful! """ api_endpoint = ApiEndpoints.instances.factory_reset request_obj = api_endpoint.load_request( approve_not_recoverable_action=approve_not_recoverable_action ) return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _delete(self, node_id: str) -> str: # pragma: no cover """Direct API method to delete an instance. Notes: Can not run in test suite! Args: node_id: node id of instance """ api_endpoint = ApiEndpoints.instances.delete request_obj = api_endpoint.load_request(nodeIds=[node_id]) return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _update_attrs( self, node_id: str, node_name: str, hostname: str, use_as_environment_name: bool ) -> str: """Direct API method to update an instance. Args: node_id: node id of instance node_name: node name of instance hostname: hostname of instance use_as_environment_name: instance name is being used for the environment name """ api_endpoint = ApiEndpoints.instances.update_attrs request_obj = api_endpoint.load_request( nodeIds=node_id, node_name=node_name, hostname=hostname, use_as_environment_name=use_as_environment_name, ) return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _update_active(self, node_id: str, status: bool) -> str: """Direct API method to update an instance. Notes: Can not run in test suite! Args: node_id: node id of instance status: enabled or disabled an instance """ api_endpoint = ApiEndpoints.instances.update_active request_obj = api_endpoint.load_request(nodeIds=node_id, status=status) return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _get_central_core_config(self) -> json_api.system_settings.SystemSettings: """Direct API method to get the current central core configuration.""" api_endpoint = ApiEndpoints.central_core.settings_get return api_endpoint.perform_request(http=self.auth.http)
[docs] def _update_central_core_config( self, enabled: bool, delete_backups: bool ) -> json_api.system_settings.SystemSettings: """Direct API method to set the current central core configuration. Args: enabled: enable/disable central core mode (ignored if not True/False) delete_backups: enable/disable deletion of backups on a core after resture (ignored if not True/False) """ api_endpoint = ApiEndpoints.central_core.settings_update request_obj = api_endpoint.load_request(enabled=enabled, delete_backups=delete_backups) return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _restore_aws( self, key_name: str, bucket_name: Optional[str] = None, preshared_key: Optional[str] = None, access_key_id: Optional[str] = None, secret_access_key: Optional[str] = None, delete_backups: Optional[bool] = None, allow_re_restore: Optional[bool] = None, ) -> json_api.central_core.CentralCoreRestore: # pragma: no cover """Direct API method to perform a central core restore operation. Notes: Can not run in test suite! Args: restore_type: currently only AWS supported? additional_data: options specific to restore_type """ api_endpoint = ApiEndpoints.central_core.restore_aws request_obj = api_endpoint.load_request( additional_data={ "key_name": key_name, "bucket_name": bucket_name, "preshared_key": preshared_key, "access_key_id": access_key_id, "secret_access_key": secret_access_key, "delete_backups": delete_backups, "allow_re_restore": allow_re_restore, } ) return api_endpoint.perform_request(http=self.auth.http, request_obj=request_obj)
[docs] def _feature_flags(self) -> json_api.system_settings.FeatureFlags: """Direct API method to get the feature flags for the core.""" api_endpoint = ApiEndpoints.system_settings.feature_flags_get return api_endpoint.perform_request(http=self.auth.http)