10.0.0

library/decort_vm_snapshot.py

@@ -0,0 +1,188 @@
+#!/usr/bin/python
+
+DOCUMENTATION = r'''
+---
+module: decort_vm_snapshot
+
+description: See L(Module Documentation,https://repository.basistech.ru/BASIS/decort-ansible/wiki/Home). # noqa: E501
+'''
+
+import time
+from ansible.module_utils.basic import AnsibleModule
+from ansible.module_utils.decort_utils import DecortController
+
+
+class DecortVMSnapshot(DecortController):
+    def __init__(self):
+        super().__init__(AnsibleModule(**self.amodule_init_args))
+        self.check_amodule_args()
+
+        self.vm_id: int
+        self.vm_facts: dict
+        self.aparams_label = self.aparams['label']
+        self.aparams_vm_id = self.aparams['vm_id']
+
+        self.vm_id, self.vm_facts, _ = self._compute_get_by_id(
+            comp_id=self.aparams_vm_id,
+        )
+        if not self.vm_id:
+            self.message(f'VM {self.aparams_vm_id} not found')
+            self.exit(fail=True)
+
+        self.vm_name = self.vm_facts['name']
+        self.vm_snapshots = self.vm_facts['snapSets']
+        self.vm_snapshot_labels = [
+            snapshot['label'] for snapshot in self.vm_snapshots
+        ]
+
+        self.new_snapshot_label = None
+        if self.aparams['state'] == 'present':
+            if self.aparams_label is None:
+                self.new_snapshot_label = (
+                    f'{self.vm_name}_{self.sec_to_dt_str(time.time())}'
+                )
+            elif self.aparams_label not in self.vm_snapshot_labels:
+                self.new_snapshot_label = self.aparams_label
+
+        if (
+            self.new_snapshot_label is None
+            and self.aparams_label is not None
+            and self.aparams_label not in self.vm_snapshot_labels
+        ):
+            self.message(
+                f'Snapshot {self.aparams_label} '
+                f'not found for VM {self.aparams_vm_id}'
+            )
+            self.exit(fail=True)
+
+    @property
+    def amodule_init_args(self) -> dict:
+        return self.pack_amodule_init_args(
+            argument_spec=dict(
+                label=dict(
+                    type='str',
+                ),
+                state=dict(
+                    type='str',
+                    choices=(
+                        'absent',
+                        'present',
+                        'merge_aborted',
+                    ),
+                ),
+                usage=dict(
+                    type='bool',
+                    default=False,
+                ),
+                vm_id=dict(
+                    type='int',
+                    required=True,
+                ),
+            ),
+            supports_check_mode=True,
+            required_if=[
+                ('state', 'absent', ('label',)),
+            ],
+            required_one_of=[
+                ('label', 'state'),
+            ],
+        )
+
+    def check_amodule_args(self):
+        check_error = False
+        if (
+            self.aparams['state'] == 'absent'
+            and self.aparams['usage']
+        ):
+            self.message(
+                'Parameter "usage" is not supported when deleting snapshot'
+            )
+            check_error = True
+
+        if check_error:
+            self.exit(fail=True)
+
+    def run(self):
+        self.get_info(first_run=True)
+        self.check_amodule_args_for_change()
+        self.change()
+        self.exit()
+
+    def get_info(self, first_run: bool = False):
+        if not first_run:
+            self.vm_snapshots = self.snapshot_list(
+                compute_id=self.aparams_vm_id,
+            )
+        label = self.new_snapshot_label or self.aparams_label
+        for snapshot in self.vm_snapshots:
+            if snapshot['label'] == label:
+                self.facts = snapshot
+                if self.aparams['usage']:
+                    self.facts['stored'] = self.get_snapshot_usage()
+                self.facts['vm_id'] = self.aparams_vm_id
+                break
+
+    def change(self):
+        match self.aparams['state']:
+            case 'present':
+                if self.new_snapshot_label:
+                    self.create()
+            case 'absent':
+                if self.aparams_label in self.vm_snapshot_labels:
+                    self.delete()
+            case 'merge_aborted':
+                self.abort_merge()
+
+    def create(self):
+        self.snapshot_create(
+            compute_id=self.aparams_vm_id,
+            label=self.new_snapshot_label,
+        )
+        self.get_info()
+
+    def delete(self):
+        self.snapshot_delete(
+            compute_id=self.aparams_vm_id,
+            label=self.aparams_label,
+        )
+        self.facts = {}
+
+    def abort_merge(self):
+        self.snapshot_abort_merge(
+            vm_id=self.aparams_vm_id, 
+            label=self.aparams_label,
+        )
+        self.get_info()
+
+    def get_snapshot_usage(self) -> int:
+        label = self.new_snapshot_label or self.aparams_label
+        common_snapshots_usage_info, _ = self.snapshot_usage(
+            compute_id=self.aparams_vm_id,
+            label=label,
+        )
+        return common_snapshots_usage_info['stored']
+    
+    def check_amodule_args_for_change(self):
+        check_errors = False
+
+        if (
+            self.aparams['state'] == 'merge_aborted'
+            and self.vm_facts['techStatus'] != 'MERGE'
+        ):
+            check_errors = True
+            self.message(
+                f'Check for parameter "state" failed: '
+                'Merge can be aborted only for VM in "MERGE" tech status.'
+            )
+
+        if check_errors:
+            self.exit(fail=True)
+
+
+
+def main():
+    DecortVMSnapshot().run()
+
+
+if __name__ == '__main__':
+    main()