Added logic to export Tags and Tag Categories, Issue #12.

README.md

@@ -381,15 +381,21 @@ python3 sddc_import_export.py -o import-nsx
 
 ### 1.4.6. Export vCenter
 
-To export your vCenter server folder structure, set the export_vcenter_folders flag in config.ini to True. Then run the export command:
+To export your vCenter server folder structure, set the export_vcenter_folders flag in config.ini to True.
+To export your vCenter server Tag Categories, set the export_vcenter_catagories flag in config.ini to True. 
+To export your vCenter server Tags, set the export_vcenter_tags flag in config.ini to True. 
+Then run the export command:
 
 ```
 python3 sddc_import_export.py -o export-vcenter
 ```
 
 ### 1.4.7. Import vCenter
 
-To Import your vCenter server folder structure, set the import_vcenter_folders flag in config.ini to True. Then run the import command:
+To Import your vCenter server folder structure, set the import_vcenter_folders flag in config.ini to True.
+To Import your vCenter server catagories, set the import_vcenter_catagories flag in config.ini to True.
+To Import your vCenter server tags, set the import_vcenter_tags flag in config.ini to True.
+Then run the import command:
 
 ```
 python3 sddc_import_export.py -o import-vcenter
@@ -501,4 +507,4 @@ python sddc_import_export.py -o testbed --test-name delete-cgw-groups --num-obje
 This will delete ALL CGW GROUPS. Use with extreme caution.
 ```bash
 python sddc_import_export.py -o testbed --test-name delete-all-cgw-groups
-```
+```

VMCImportExport.py

@@ -73,6 +73,10 @@ def __init__(self,configPath="./config_ini/config.ini", vmcConfigPath="./config_
         self.network_import_exclude_list = []
         self.export_vcenter_folders = False
         self.import_vcenter_folders = False
+        self.export_vcenter_catagories = False
+        self.import_vcenter_catagories = False
+        self.export_vcenter_tags = False
+        self.import_vcenter_tags = False
         self.user_search_results_json = ""
         self.convertedServiceRolePayload = ""
         self.RoleSyncSourceUserEmail = ""
@@ -143,6 +147,16 @@ def ConfigLoader(self):
 
         self.vcenter_folders_filename = self.loadConfigFilename(config,"exportConfig","vcenter_folders_filename")
 
+        self.export_vcenter_categories = self.loadConfigFlag(config,"exportConfig","export_vcenter_categories")
+        self.import_vcenter_categories = self.loadConfigFlag(config,"importConfig","import_vcenter_categories")
+
+        self.vcenter_categories_filename = self.loadConfigFilename(config,"exportConfig","vcenter_categories_filename")
+
+        self.export_vcenter_tags = self.loadConfigFlag(config,"exportConfig","export_vcenter_tags")
+        self.import_vcenter_tags = self.loadConfigFlag(config,"importConfig","import_vcenter_tags")
+
+        self.vcenter_tags_filename = self.loadConfigFilename(config,"exportConfig","vcenter_tags_filename")
+
         #NSX manager
         self.srcNSXmgrURL               =  vCenterConfig.get("nsxConfig","srcNSXmgrURL")
         self.srcNSXmgrUsername          =  vCenterConfig.get("nsxConfig","srcNSXmgrUsername")
@@ -3679,6 +3693,10 @@ def loadConfigFilename(self,config,section,key):
                 return 'vpn-local-bgp.json'
             elif (key == 'vcenter_folders_filename'):
                 return 'vcenterfolderpaths.json'
+            elif (key == 'vcenter_categories_filename'):
+                return 'vcentercategories.json'
+            elif (key == 'vcenter_tags_filename'):
+                return 'vcentertags.json'
             elif (key == 'network_dhcp_static_binding_filename'):
                 return 'dhcp-static-binding.json'
 

config_ini/config.ini

@@ -115,6 +115,8 @@ export_purge_before_run = True
 # vCenter Export Options
 # Must configure vcenter.ini
 export_vcenter_folders = False
+export_vcenter_categories = False
+export_vcenter_tags = False
 
 # CSP Role sync options
 # The source user - the template user account with roles that need to be synced to other user accounts
@@ -270,6 +272,8 @@ nsx_l7_context_profile_import_filename = nsx_l7_context_profile.json
 # vCenter Import Options
 # Must configure vcenter.ini
 import_vcenter_folders = False
+import_vcenter_categories = False
+import_vcenter_tags = False
 
 # CSP Role sync options
 # A pipe delimited list of email addresses - these accounts will have the roles synchronized with roles attached to the source user

sddc_import_export.py

@@ -101,6 +101,12 @@ def main(args):
     ap.add_argument("-s3b","--aws-s3-export-bucket", required=False,help="AWS bucket name for export to S3")
     ap.add_argument("-rss","--role-sync-source-user-email", required=False, help="The source email address used as a template for syncing roles")
     ap.add_argument("-rsd","--role-sync-dest-user-emails", required=False, help="The dest email addresses used as a target for syncing roles, formatted as a set")
+    ap.add_argument("-sv", "--source-vcenter-url", required=False, help="Source vCenter URL")
+    ap.add_argument("-dv", "--dest-vcenter-url", required=False, help="Destination vCenter URL")
+    ap.add_argument("-svu","--source-vcenter-username", required=False,help="Source vCenter Username")
+    ap.add_argument("-dvu","--dest-vcenter-username", required=False,help="Destination vCenter Username")
+    ap.add_argument("-svp","--source-vcenter-password", required=False,help="Source vCenter Password")
+    ap.add_argument("-dvp","--dest-vcenter-password", required=False,help="Destination vCenter Password")
 
     args = ap.parse_args(args)
 
@@ -175,6 +181,31 @@ def main(args):
         ioObj.RoleSyncDestUserEmails = args.role_sync_dest_user_emails.split(',')
         print('Loaded role sync dest user emails from command line')
 
+    # Check the optional command-line arguments to override the values in vcenter.ini
+    if args.source_vcenter_url:
+        ioObj.srcvCenterURL = args.source_vcenter_url
+        print('Loaded source vcenter URL from command line')
+
+    if args.dest_vcenter_url:
+        ioObj.destvCenterURL = args.dest_vcenter_url
+        print('Loaded source vcenter URL from command line')
+
+    if args.source_vcenter_username:
+        ioObj.srcvCenterUsername = args.source_vcenter_username
+        print('Loaded source vcenter username from command line')
+
+    if args.dest_vcenter_username:
+        ioObj.destvCenterUsername = args.dest_vcenter_username
+        print('Loaded source vcenter username from command line')
+
+    if args.source_vcenter_password:
+        ioObj.srcvCenterPassword = args.source_vcenter_password
+        print('Loaded source vcenter password from command line')
+
+    if args.dest_vcenter_password:
+        ioObj.destvCenterPassword = args.dest_vcenter_password
+        print('Loaded source vcenter password from command line')
+
     # Variable added so we can have an intent run multiple operations
     no_intent_found = True
 
@@ -226,6 +257,49 @@ def main(args):
             print('Exporting folder paths from source vCenter...')
             srcdc.export_folder_paths(ioObj.export_path / ioObj.vcenter_folders_filename)
             print('Export complete.')
+        if ioObj.export_vcenter_categories:
+            srcvc = vcenter.vCenter(ioObj.srcvCenterURL,ioObj.srcvCenterUsername,ioObj.srcvCenterPassword,ioObj.srcvCenterSSLVerify)
+            srcdc = srcvc.get_datacenter(ioObj.srcvCenterDatacenter)
+            print('Exporting tag categories from source vCenter...')
+            srcdc.export_tag_categories(ioObj.export_path / ioObj.vcenter_categories_filename)
+            print('Export complete.')
+        if ioObj.export_vcenter_tags:
+            srcvc = vcenter.vCenter(ioObj.srcvCenterURL,ioObj.srcvCenterUsername,ioObj.srcvCenterPassword,ioObj.srcvCenterSSLVerify)
+            srcdc = srcvc.get_datacenter(ioObj.srcvCenterDatacenter)
+            print('Exporting tags from source vCenter...')
+            srcdc.export_tags(ioObj.export_path / ioObj.vcenter_tags_filename)
+            print('Export complete.')
+        if ioObj.export_history is True:
+            retval = ioObj.zipJSONfiles()
+            if retval is False:
+                print('JSON files were not successfully zipped.')
+            else:
+                print('JSON files successfully zipped into', ioObj.export_zip_name)
+                if ioObj.export_type == 's3':
+                    print('Uploading to s3 bucket',ioObj.aws_s3_export_bucket)
+                    if len(ioObj.aws_s3_export_access_id) == 0:
+                        #Blank access ID - running in Lambda mode, do not pass the key and secret, the Lambda role will grant access to the bucket
+                        s3 = boto3.client('s3')
+                    else:
+                        s3 = boto3.client('s3',aws_access_key_id=ioObj.aws_s3_export_access_id,aws_secret_access_key=ioObj.aws_s3_export_access_secret)
+                    try:
+                        fname = ioObj.export_folder + '/' + ioObj.export_zip_name
+                        with open(fname, "rb") as f:
+                            response = s3.upload_fileobj(f,ioObj.aws_s3_export_bucket,ioObj.export_zip_name)
+                        print('S3 upload successful')
+                    except Exception as e:
+                        print('Failed to upload file.')
+                        print(e)
+
+                if ioObj.export_purge_after_zip == True:
+                    print('export_purge_after_zip flag is true, deleting JSON files')
+                    retval = ioObj.purgeJSONfiles()
+                    if retval is False:
+                        print('Unable to purge JSON files.')
+
+            retval = ioObj.purgeJSONzipfiles()
+            if retval is True:
+                print('Zipfile maintenance completed with no errors.')
 
     if intent_name == "testbed":
         no_intent_found = False
@@ -304,6 +378,35 @@ def main(args):
 
     if intent_name == "import-vcenter":
         no_intent_found = False
+
+        if import_first_file != "":
+            files = glob.glob(import_first_file + '/*.zip')
+            if len(files) > 0:
+                import_file_path = files[0]
+                print('Found',import_file_path,'in folder.')
+            else:
+                print('Found no zipfiles in',import_first_file)
+
+        # User passed a zipfile path to use as the import source
+        if import_file_path != "":
+            ioObj.import_folder = os.path.dirname(import_file_path)
+            ioObj.import_path = Path(ioObj.import_folder)
+            ioObj.export_folder = os.path.dirname(import_file_path)
+            ioObj.export_path = Path(ioObj.export_folder)
+            retval = ioObj.purgeJSONfiles()
+            if retval is False:
+                stop_script = yes_or_no("Errors purging old files. Stop running script?")
+                if stop_script is True:
+                    sys.exit()
+            retval = ioObj.unzipJSONfiles(import_file_path)
+            if retval is False:
+                stop_script = yes_or_no("Could not unzip archive. Stop running script?")
+                if stop_script is True:
+                    sys.exit()
+            else:
+                print('Extracted JSON from zip archive',import_file_path,"- continuing with import.")
+                print('Loaded import and export folder from command line:', ioObj.import_path )
+
         if ioObj.import_vcenter_folders:
             destvc = vcenter.vCenter(ioObj.destvCenterURL,ioObj.destvCenterUsername,ioObj.destvCenterPassword,ioObj.destvCenterSSLVerify)
             destdc = destvc.get_datacenter(ioObj.destvCenterDatacenter)
@@ -314,6 +417,26 @@ def main(args):
                 test_mode = True
             destdc.import_folder_paths(ioObj.import_path / ioObj.vcenter_folders_filename,test_mode=test_mode)
             print('Import complete.')
+        if ioObj.import_vcenter_categories:
+            destvc = vcenter.vCenter(ioObj.destvCenterURL,ioObj.destvCenterUsername,ioObj.destvCenterPassword,ioObj.destvCenterSSLVerify)
+            destdc = destvc.get_datacenter(ioObj.destvCenterDatacenter)
+            print('Importing tag categories into destination vCenter...')
+            if ioObj.import_mode == 'live':
+                test_mode = False
+            else:
+                test_mode = True
+            destdc.import_tag_categories(ioObj.import_path / ioObj.vcenter_categories_filename,test_mode=test_mode)
+            print('Import complete.')
+        if ioObj.import_vcenter_tags:
+            destvc = vcenter.vCenter(ioObj.destvCenterURL,ioObj.destvCenterUsername,ioObj.destvCenterPassword,ioObj.destvCenterSSLVerify)
+            destdc = destvc.get_datacenter(ioObj.destvCenterDatacenter)
+            print('Importing tags into destination vCenter...')
+            if ioObj.import_mode == 'live':
+                test_mode = False
+            else:
+                test_mode = True
+            destdc.import_tags(ioObj.import_path / ioObj.vcenter_tags_filename,test_mode=test_mode)
+            print('Import complete.')
 
     if intent_name == "export-nsx":
         no_intent_found = False

vcenter.py

@@ -136,6 +136,96 @@ def _get_vm_by_path(self, vm_path: str) -> vim.VirtualMachine:
         vm = self._get_vm_by_name(vm_name, parent_folder=folder)
         return vm
 
+    def _create_tag_category(self, name, description, cardinality, associable_types):
+        create_spec = self._vsphere_client.tagging.Category.CreateSpec()
+        create_spec.name = name
+        create_spec.description = description
+        create_spec.cardinality = cardinality
+        create_spec.associable_types = set(associable_types)
+        return self._vsphere_client.tagging.Category.create(create_spec)
+
+    def _create_tag(self, name, description, category_id):
+        create_spec = self._vsphere_client.tagging.Tag.CreateSpec()
+        create_spec.name = name
+        create_spec.description = description
+        create_spec.category_id = category_id
+        return self._vsphere_client.tagging.Tag.create(create_spec)
+
+    def _get_tag_categories(self):
+        categories = set()
+        for id in self._vsphere_client.tagging.Category.list():
+            category = self._vsphere_client.tagging.Category.get(id)
+            categories.add (category.name)
+        return categories
+
+    def _get_tags(self):
+        tags = set()
+        for id in self._vsphere_client.tagging.Tag.list():
+            tag = self._vsphere_client.tagging.Tag.get(id)
+            tags.add (tag.name)
+        return tags
+
+    def export_tag_categories(self, export_file_path: str) -> None:
+        categories = []
+        for id in self._vsphere_client.tagging.Category.list():
+            category = self._vsphere_client.tagging.Category.get(id)
+            new_category = {}
+            new_category['name'] = category.name
+            new_category['description'] = category.description
+            new_category['cardinality'] = category.cardinality
+            new_category['associable_types'] = list(category.associable_types)
+            categories.append(new_category)
+
+        with open(export_file_path, 'w') as paths_file:
+            json.dump(categories, paths_file)
+
+    def export_tags(self, export_file_path: str) -> None:
+        categories = {}
+        tags = []
+        for id in self._vsphere_client.tagging.Category.list():
+            category = self._vsphere_client.tagging.Category.get(id)
+            categories[category.id] = category.name
+        for id in self._vsphere_client.tagging.Tag.list():
+            tag = self._vsphere_client.tagging.Tag.get(id)
+            new_tag = {}
+            new_tag['name'] = tag.name
+            new_tag['category'] = categories[tag.category_id]
+            new_tag['description'] = tag.description
+            tags.append(new_tag)
+
+        with open(export_file_path, 'w') as paths_file:
+            json.dump(tags, paths_file)
+
+    def import_tag_categories(self, import_file_path: str, test_mode: bool = False) -> None:
+        with open(import_file_path) as paths_file:
+            categories = json.load(paths_file)
+
+        existing_categories = self._get_tag_categories()
+
+        for category in categories:
+            if test_mode:
+                print(f'TEST MODE: would have created category {category}')
+            else:
+                if category['name'] not in existing_categories:
+                    self._create_tag_category(category['name'], category['description'], category['cardinality'], category['associable_types'] )
+
+    def import_tags(self, import_file_path: str, test_mode: bool = False) -> None:
+        with open(import_file_path) as paths_file:
+            tags = json.load(paths_file)
+
+        categories = {}
+        for id in self._vsphere_client.tagging.Category.list():
+            category = self._vsphere_client.tagging.Category.get(id)
+            categories[category.name] = category.id
+
+        existing_tags = self._get_tags()
+
+        for tag in tags:
+            if test_mode:
+                print(f'TEST MODE: would have created tag {tag}')
+            else:
+                if tag['name'] not in existing_tags:
+                    self._create_tag(tag['name'], tag['description'],categories[tag['category']])
 
 class vCenter:
     def __init__(self, address, username, password, ssl_verification=False):