zerto/Zerto-Python-SDK-Hands-On-Labs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

simplified version

Browse Source
This commit is contained in:
Kosta Mushkin
2025-06-22 17:43:43 -07:00
parent 3eee7d623b
commit 3cfde61792
23 changed files with 425 additions and 778 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+26
View File
@@ -0,0 +1,26 @@
# Python cache files
__pycache__/
*.py[cod]
*$py.class
*.so
# Virtual environment
venv/
env/
ENV/
# IDE files
.vscode/
.idea/
*.swp
*.swo
# OS files
.DS_Store
Thumbs.db
# Logs
*.log
# Local configuration (contains secrets)
prerequisites/config.py
diagrams/01-keykloack-login.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 66 KiB

diagrams/02-keykloack-login.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 58 KiB

diagrams/03-keykloack-realm.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 110 KiB

diagrams/04-keykloack-create-new-client.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 145 KiB

diagrams/05-keykloack-create-new-client.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 102 KiB

diagrams/06-keykloack-client-attributes.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 119 KiB

diagrams/07-keykloack-client-credentials.jpg
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 169 KiB

exercises/01_introduction/README.md
+59 -1
View File
@@ -28,5 +28,63 @@ This exercise introduces you to Zerto's REST API and the Python SDK. You'll lear
- API endpoints
- SDK architecture
## 2. Understand the Authentication Flow
Zerto uses **Keycloak** as its authentication provider. Understanding this flow is crucial for working with the Zerto APIs. Here's the step-by-step process:
### Step 1: Log in to Keycloak
From any browser, navigate to https://<zvmipaddress>/auth.
![Keycloak Login Screen](../../diagrams/01-keykloack-login.jpg)
If this is your first time logging in, and you didn’t already change the password, use the default login information:
• Username: admin
• Password: admin
![Keycloak Admin Login](../../diagrams/02-keykloack-login.jpg)
Once logged in, make sure you are in the "Zerto realm," as per the screenshot below:
### Step 2: Create a New Client
Next, click on the "Clients" link in the left menu. Then, click "Create Client.".
![Create New Realm](../../diagrams/03-keykloack-realm.jpg)
Now, provide a client_ID. The rest of the info on this page is optional. Click "Next".
![Create New Client - Step 1](../../diagrams/04-keykloack-create-new-client.jpg)
On the two next screen, we need to check a few boxes.
The arrows point out the things you want to have enabled.
These options allow scripts to get a JSON Web Token (JWT) without interactively logging in.
![Create New Client - Step 2](../../diagrams/05-keykloack-create-new-client.jpg)
![Client Attributes Configuration](../../diagrams/06-keykloack-client-attributes.jpg)
### Step 3: Gather Your Credentials
Once Keycloak has created the client, it will bring you to a screen with many tabs for this client.
Click on the "Credentials" tab, and then copy the client secret.
![Client Credentials](../../diagrams/07-keykloack-client-credentials.jpg)
Now you have your username (client_id) and password (client_secret) for your script.
### Authentication Flow Summary
1. **Client Credentials Flow**: Your application uses the Client ID and Client Secret to obtain an access token
2. **Token Exchange**: The access token is exchanged for API access
3. **API Requests**: Include the access token in the Authorization header for all API calls
### What You'll Need for the Exercises
For the hands-on exercises, you'll need to:
1. Create a Keycloak client following the steps above
2. Note down your Client ID and Client Secret
3. Update the `prerequisites/config.py` file with these credentials
4. Ensure your client has the necessary permissions to access Zerto APIs
## Next Steps
Proceed to Exercise 2: Authentication to start working with the SDK.
Proceed to Exercise 2: Authentication to start working with the SDK and implement the authentication flow in code.
exercises/01_introduction/solution/.gitkeep
-1
View File
@@ -1 +0,0 @@
exercises/01_introduction/working/.gitkeep
-1
View File
@@ -1 +0,0 @@
exercises/02_authentication/solution/auth.py
+5 -1
View File
@@ -20,6 +20,7 @@ import os
import logging
import json
from pathlib import Path
import urllib3
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
@@ -28,6 +29,9 @@ sys.path.append(str(prerequisites_path))
# Import the SDK modules
from zvml import ZVMLClient
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Import configuration
try:
from config import (
@@ -68,9 +72,9 @@ def main():
# Step 2: Test the connection by getting local site info
logging.info("Testing connection by retrieving local site information...")
local_site = client.localsite.get_local_site()
# Extract and log version information
local_site = client.localsite.get_local_site()
version = local_site.get('Version')
logging.info(f"Successfully connected to ZVM version: {version}")
exercises/02_authentication/working/auth.py
+23 -32
View File
@@ -3,9 +3,13 @@
Exercise 2: Authentication - Beginner-Friendly Instructions
This script demonstrates how to authenticate with Zerto API using Keycloak.
PREREQUISITES (Complete these first):
1. Make sure you have the zvml package installed (see main README)
2. Update prerequisites/config.py with your ZVM details:
Prerequisites:
1. Install the zvml package in development mode:
cd /path/to/zvml-python-sdk
pip install -e .
2. Update prerequisites/config.py with your ZVM details
3. Update prerequisites/config.py with your ZVM details:
- ZVM_HOST: Your Zerto Virtual Manager IP address or hostname
- CLIENT_ID: Your Keycloak client ID
- CLIENT_SECRET: Your Keycloak client secret
@@ -33,15 +37,19 @@ import os
import logging
import json
from pathlib import Path
import urllib3
# Add prerequisites to Python path (this helps Python find your config file)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
sys.path.append(str(prerequisites_path))
# Import the Zerto SDK - this gives us the ZVMLClient class
# Import the SDK modules
from zvml import ZVMLClient
# Import your configuration settings
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Import configuration
try:
from config import (
ZVM_HOST, # Your ZVM IP address (e.g., "192.168.1.100")
@@ -51,25 +59,24 @@ try:
CLIENT_SECRET # Your Keycloak client secret
)
except ImportError:
print("❌ ERROR: Configuration file not found!")
print("Please copy config.example.py to config.py and update with your values")
print("Error: Please copy config.example.py to config.py and update with your values")
print("Expected path:", prerequisites_path / "config.py")
sys.exit(1)
def main():
"""
Main function - this is where your code goes!
Follow the step-by-step instructions below.
Main function to demonstrate Zerto authentication.
Shows how to:
1. Initialize ZVMLClient with Keycloak credentials
2. Test connection by retrieving local site info
3. Handle authentication and connection errors
"""
# Set up logging so you can see what's happening
# Set up logging with timestamp
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
print("🚀 Starting Zerto Authentication Exercise")
print("=" * 50)
try:
# ========================================
# STEP 1: Create a ZVMLClient instance
@@ -86,6 +93,7 @@ def main():
# client_secret=CLIENT_SECRET,
# verify_certificate=ZVM_SSL_VERIFY
# )
# logging.info("Testing connection by retrieving local site information...")
#
# EXPLANATION:
# - zvm_address: Where your ZVM is located (from config.py)
@@ -93,7 +101,7 @@ def main():
# - client_secret: Your Keycloak client secret (from config.py)
# - verify_certificate: Whether to check SSL certificates (from config.py)
client = None # ← REPLACE THIS LINE WITH YOUR CODE
# ← ADD YOUR CODE HERE
# ========================================
# STEP 2: Test the connection
@@ -115,25 +123,8 @@ def main():
# ← ADD YOUR CODE HERE
# ========================================
# STEP 3: Display success message
# ========================================
print("\n📝 STEP 3: Displaying success message...")
print("You need to add a final success message.")
# TODO: Add a success message
# HINT: Use this syntax:
# logging.info("🎉 Connection successful!")
#
# EXPLANATION:
# This confirms that everything worked correctly
# ← ADD YOUR CODE HERE
except Exception as e:
# This catches any errors that might occur
print(f"\n❌ ERROR: Something went wrong!")
print(f"Error details: {str(e)}")
logging.error(f"Authentication failed: {str(e)}")
sys.exit(1)
exercises/03_site_discovery/solution/sites.py
+5 -8
View File
@@ -20,6 +20,10 @@ import os
import logging
import json
from pathlib import Path
import urllib3
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
@@ -66,17 +70,10 @@ def main():
)
# Step 2: List all available sites
logging.info("Retrieving list of available sites...")
sites = client.virtualization_sites.get_virtualization_sites()
if not sites:
logging.warning("No sites found!")
else:
logging.info(f"Found {len(sites)} site(s):")
logging.info(f'Sites Info: {json.dumps(sites, indent=4)}')
logging.info(f"Sites Info: {json.dumps(sites, indent=4)}")
# Step 3: Get and display local site information
logging.info("\nRetrieving local site information...")
local_site = client.localsite.get_local_site()
logging.info(f"Local site details: {json.dumps(local_site, indent=4)}")
exercises/03_site_discovery/working/sites.py
+19 -35
View File
@@ -3,10 +3,11 @@
Exercise 3: Site Discovery - Beginner-Friendly Instructions
This script demonstrates how to discover and work with Zerto virtualization sites.
PREREQUISITES (Complete these first):
1. ✅ Completed Exercise 2 (Authentication)
2. ✅ Make sure you have the zvml package installed
3. ✅ Updated prerequisites/config.py with your ZVM details
Prerequisites:
1. Install the zvml package in development mode:
cd /path/to/zvml-python-sdk
pip install -e .
2. Update prerequisites/config.py with your ZVM details
WHAT YOU NEED TO DO:
In this exercise, you will:
@@ -32,15 +33,19 @@ import os
import logging
import json
from pathlib import Path
import urllib3
# Add prerequisites to Python path (this helps Python find your config file)
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
sys.path.append(str(prerequisites_path))
# Import the Zerto SDK - this gives us the ZVMLClient class
# Import the SDK modules
from zvml import ZVMLClient
# Import your configuration settings
# Import configuration
try:
from config import (
ZVM_HOST, # Your ZVM IP address (e.g., "192.168.1.100")
@@ -57,18 +62,17 @@ except ImportError:
def main():
"""
Main function - this is where your code goes!
Follow the step-by-step instructions below.
Main function to demonstrate site discovery.
Shows how to:
1. List all available virtualization sites
2. Get and display local site information
"""
# Set up logging so you can see what's happening
# Set up logging with timestamp
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
print("🚀 Starting Zerto Site Discovery Exercise")
print("=" * 50)
try:
# ========================================
# STEP 1: Create a ZVMLClient instance
@@ -88,7 +92,7 @@ def main():
# EXPLANATION:
# This creates a connection to your ZVM (same as Exercise 2)
client = None # ← REPLACE THIS LINE WITH YOUR CODE
# ← ADD YOUR CODE HERE
# ========================================
# STEP 2: List all available sites
@@ -99,6 +103,7 @@ def main():
# TODO: Add code to get the list of sites
# HINT: Use this syntax:
# sites = client.virtualization_sites.get_virtualization_sites()
# logging.info(f"Sites Info: {json.dumps(sites, indent=4)}")
#
# EXPLANATION:
# - client.virtualization_sites.get_virtualization_sites() gets all sites
@@ -106,27 +111,9 @@ def main():
# ← ADD YOUR CODE HERE
# TODO: Add code to display the sites
# HINT: Use this syntax:
# if not sites:
# logging.warning("No sites found!")
# else:
# logging.info(f"Found {len(sites)} site(s):")
# logging.info(f'Sites Info: {json.dumps(sites, indent=4)}')
#
# EXPLANATION:
# - len(sites) counts how many sites were found
# - json.dumps(sites, indent=4) formats the site data nicely
# - logging.info() displays the information
# ← ADD YOUR CODE HERE
# ========================================
# STEP 3: Get local site information
# ========================================
print("\n📝 STEP 3: Getting local site details...")
print("You need to get detailed information about your local site.")
# TODO: Add code to get local site information
# HINT: Use this syntax:
# local_site = client.localsite.get_local_site()
@@ -140,9 +127,6 @@ def main():
# ← ADD YOUR CODE HERE
except Exception as e:
# This catches any errors that might occur
print(f"\n❌ ERROR: Something went wrong!")
print(f"Error details: {str(e)}")
logging.error(f"Site discovery failed: {str(e)}")
sys.exit(1)
exercises/04_resource_discovery/solution/resources.py
+6 -17
View File
@@ -20,6 +20,10 @@ import os
import logging
import json
from pathlib import Path
import urllib3
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
@@ -69,12 +73,7 @@ def main():
# Step 2.1: List all available sites
logging.info("Retrieving list of available sites...")
sites = client.virtualization_sites.get_virtualization_sites()
if not sites:
logging.warning("No sites found!")
else:
logging.info(f"Found {len(sites)} site(s):")
logging.info(f'Sites Info: {json.dumps(sites, indent=4)}')
logging.info(f"Sites: {json.dumps(sites, indent=4)}")
# Step 2.2: Get local and peer site Identifiers
local_site_identifier = client.localsite.get_local_site().get('SiteIdentifier')
@@ -82,38 +81,28 @@ def main():
# Get peer site identifier (first non-local site)
peer_site = next((site for site in sites if site.get('SiteIdentifier') != local_site_identifier), None)
if not peer_site:
logging.warning("No peer site found!")
sys.exit(1)
peer_site_identifier = peer_site.get('SiteIdentifier')
logging.info(f"Peer site identifier: {peer_site_identifier}")
# Step 3: Get local site resources
# Step 3.1: Get local site vms
logging.info("\nRetrieving local site vms...")
# Step 3: Get local site vms
local_vms = client.virtualization_sites.get_virtualization_site_vms(site_identifier=local_site_identifier)
logging.info(f"Local Vms Info: {json.dumps(local_vms, indent=4)}")
# in order to create a DR solution, we need to get information about the local site vms and peer site datastores, hosts, folders and networks
# Step 3.2: Get peer site datastores
logging.info("\nRetrieving peer site datastores...")
peer_datastores = client.virtualization_sites.get_virtualization_site_datastores(site_identifier=peer_site_identifier)
logging.info(f"Peer Datastores Info: {json.dumps(peer_datastores, indent=4)}")
# Step 3.3: Get peer site hosts
logging.info("\nRetrieving peer site hosts...")
peer_hosts = client.virtualization_sites.get_virtualization_site_hosts(site_identifier=peer_site_identifier)
logging.info(f"Peer Hosts Info: {json.dumps(peer_hosts, indent=4)}")
# Step 3.4: Get peer site folders
logging.info("\nRetrieving peer site folders...")
peer_folders = client.virtualization_sites.get_virtualization_site_folders(site_identifier=peer_site_identifier)
logging.info(f"Peer Folders Info: {json.dumps(peer_folders, indent=4)}")
# Step 3.5: Get peer site networks
logging.info("\nRetrieving peer site networks...")
peer_networks = client.virtualization_sites.get_virtualization_site_networks(site_identifier=peer_site_identifier)
logging.info(f"Peer Networks Info: {json.dumps(peer_networks, indent=4)}")
exercises/04_resource_discovery/working/resources.py
+51 -48
View File
@@ -3,10 +3,11 @@
Exercise 4: Resource Discovery - Beginner-Friendly Instructions
This script demonstrates how to discover and work with resources in your Zerto environment.
PREREQUISITES (Complete these first):
1. ✅ Completed Exercise 3 (Site Discovery)
2. ✅ Make sure you have the zvml package installed
3. ✅ Updated prerequisites/config.py with your ZVM details
Prerequisites:
1. Install the zvml package in development mode:
cd /path/to/zvml-python-sdk
pip install -e .
2. Update prerequisites/config.py with your ZVM details
WHAT YOU NEED TO DO:
In this exercise, you will:
@@ -38,15 +39,19 @@ import os
import logging
import json
from pathlib import Path
import urllib3
# Add prerequisites to Python path (this helps Python find your config file)
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
sys.path.append(str(prerequisites_path))
# Import the Zerto SDK - this gives us the ZVMLClient class
# Import the SDK modules
from zvml import ZVMLClient
# Import your configuration settings
# Import configuration
try:
from config import (
ZVM_HOST, # Your ZVM IP address (e.g., "192.168.1.100")
@@ -56,32 +61,27 @@ try:
CLIENT_SECRET # Your Keycloak client secret
)
except ImportError:
print("❌ ERROR: Configuration file not found!")
print("Please copy config.example.py to config.py and update with your values")
print("Error: Please copy config.example.py to config.py and update with your values")
print("Expected path:", prerequisites_path / "config.py")
sys.exit(1)
def main():
"""
Main function - this is where your code goes!
Follow the step-by-step instructions below.
Main function to demonstrate resource discovery.
Shows how to:
1. Discover local site resources
2. Work with peer site resources
"""
# Set up logging so you can see what's happening
# Set up logging with timestamp
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
print("🚀 Starting Zerto Resource Discovery Exercise")
print("=" * 50)
try:
# ========================================
# STEP 1: Create a ZVMLClient instance
# ========================================
print("\n📝 STEP 1: Creating ZVMLClient...")
print("This is the same as previous exercises - you need to create a client to connect to ZVM.")
# TODO: Replace this line with actual ZVMLClient creation
# HINT: Use this syntax (same as previous exercises):
# client = ZVMLClient(
@@ -91,60 +91,64 @@ def main():
# verify_certificate=ZVM_SSL_VERIFY
# )
client = None # ← REPLACE THIS LINE WITH YOUR CODE
# EXPLANATION:
# connection to ZVM is established using the ZVMLClient class
# ← ADD YOUR CODE HERE
# ========================================
# STEP 2: Identify local and peer sites
# ========================================
print("\n📝 STEP 2.1: Getting list of sites...")
print("You need to get a list of all available sites (same as Exercise 3).")
# ========================================
# STEP 2.1: Get virtualization sites
# ========================================
# TODO: Add code to get the list of sites
# HINT: Use this syntax:
# sites = client.virtualization_sites.get_virtualization_sites()
#
# logging.info(f"Sites: {json.dumps(sites, indent=4)}")
# EXPLANATION:
# This gets all sites available to your ZVM
# ← ADD YOUR CODE HERE
print("\n📝 STEP 2.2: Getting local site identifier...")
print("You need to get the identifier for your local site.")
# ========================================
# STEP 2.2: Get local site identifier
# TODO: Add code to get local site identifier
# HINT: Use this syntax:
# local_site_identifier = client.localsite.get_local_site().get('SiteIdentifier')
# logging.info(f"Local site identifier: {local_site_identifier}")
#
# EXPLANATION:
# - client.localsite.get_local_site() gets your local site info
# - .get('SiteIdentifier') extracts the unique identifier
local_site_identifier = None # ← REPLACE THIS LINE WITH YOUR CODE
print("\n📝 STEP 2.3: Getting peer site identifier...")
print("You need to find a peer site (any site that's not your local site).")
# ← ADD YOUR CODE HERE
# ========================================
# STEP 2.3: Get peer site identifier
# TODO: Add code to get peer site identifier
# HINT: Use this syntax:
# peer_site = next((site for site in sites if site.get('SiteIdentifier') != local_site_identifier), None)
# peer_site_identifier = peer_site.get('SiteIdentifier')
#
# logging.info(f"Peer site identifier: {peer_site_identifier}")
# EXPLANATION:
# - This finds the first site that's not your local site
# - next() gets the first matching site from the list
# - .get('SiteIdentifier') extracts the unique identifier
peer_site_identifier = None # ← REPLACE THIS LINE WITH YOUR CODE
# ← ADD YOUR CODE HERE
# ========================================
# STEP 3: Get local site resources
# ========================================
print("\n📝 STEP 3.1: Getting local site VMs...")
print("You need to get all virtual machines from your local site.")
# TODO: Add code to get local site VMs
# HINT: Use this syntax:
# local_vms = client.virtualization_sites.get_virtualization_site_vms(site_identifier=local_site_identifier)
# logging.info(f"Local VMs: {json.dumps(local_vms, indent=4)}")
#
# EXPLANATION:
# This gets all VMs that can be protected/replicated from your local site
@@ -154,48 +158,50 @@ def main():
# ========================================
# STEP 4: Get peer site resources
# ========================================
print("\n📝 STEP 4.1: Getting peer site datastores...")
print("You need to get datastores from the peer site.")
# ========================================
# STEP 4.1: Get peer site datastores
# ========================================
# TODO: Add code to get peer site datastores
# HINT: Use this syntax:
# peer_datastores = client.virtualization_sites.get_virtualization_site_datastores(site_identifier=peer_site_identifier)
# logging.info(f"Peer datastores: {json.dumps(peer_datastores, indent=4)}")
#
# EXPLANATION:
# Datastores are storage locations where VMs can be stored on the peer site
# ← ADD YOUR CODE HERE
print("\n📝 STEP 4.2: Getting peer site hosts...")
print("You need to get hosts from the peer site.")
# ========================================
# STEP 4.2: Get peer site hosts
# TODO: Add code to get peer site hosts
# HINT: Use this syntax:
# peer_hosts = client.virtualization_sites.get_virtualization_site_hosts(site_identifier=peer_site_identifier)
# logging.info(f"Peer hosts: {json.dumps(peer_hosts, indent=4)}")
#
# EXPLANATION:
# Hosts are physical servers that can run VMs on the peer site
# ← ADD YOUR CODE HERE
print("\n📝 STEP 4.3: Getting peer site folders...")
print("You need to get folders from the peer site.")
# ========================================
# STEP 4.3: Get peer site folders
# TODO: Add code to get peer site folders
# HINT: Use this syntax:
# peer_folders = client.virtualization_sites.get_virtualization_site_folders(site_identifier=peer_site_identifier)
# logging.info(f"Peer folders: {json.dumps(peer_folders, indent=4)}")
#
# EXPLANATION:
# Folders are organizational containers for VMs on the peer site
# ← ADD YOUR CODE HERE
print("\n📝 STEP 4.4: Getting peer site networks...")
print("You need to get networks from the peer site.")
# ========================================
# STEP 4.4: Get peer site networks
# TODO: Add code to get peer site networks
# HINT: Use this syntax:
# peer_networks = client.virtualization_sites.get_virtualization_site_networks(site_identifier=peer_site_identifier)
# logging.info(f"Peer networks: {json.dumps(peer_networks, indent=4)}")
#
# EXPLANATION:
# Networks are network connections that VMs can use on the peer site
@@ -203,9 +209,6 @@ def main():
# ← ADD YOUR CODE HERE
except Exception as e:
# This catches any errors that might occur
print(f"\n❌ ERROR: Something went wrong!")
print(f"Error details: {str(e)}")
logging.error(f"Resource discovery failed: {str(e)}")
sys.exit(1)
exercises/05_vpg_operations/solution/create_vpg.py
+41 -144
View File
@@ -56,46 +56,6 @@ except ImportError:
print("Expected path:", prerequisites_path / "config.py")
sys.exit(1)
def parse_arguments():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description='Create VPG and add specified VMs')
parser.add_argument('--vm-names', nargs='+', required=True,
help='List of VM names to add to the VPG (can be space-separated or comma-separated)')
parser.add_argument('--vpg-name', default="Test-VPG-Python",
help='Name of the VPG to create (default: Test-VPG-Python)')
return parser.parse_args()
def find_vms_by_names(client, site_identifier, vm_names):
"""Find VMs by their names in the specified site."""
vms = client.virtualization_sites.get_virtualization_site_vms(site_identifier=site_identifier)
logging.info(f"find_vms_by_names: found vms {json.dumps(vms, indent=4)} VMs in site {site_identifier}")
if not vms:
return [], []
# Create a dictionary of VM name to VM object for easy lookup
vm_dict = {vm.get('VmName'): vm for vm in vms}
# Find requested VMs
found_vms = []
not_found = []
for vm_name in vm_names:
if vm_name in vm_dict:
found_vms.append(vm_dict[vm_name])
logging.info(f"Found VM: {vm_name} (ID: {vm_dict[vm_name].get('VmIdentifier')})")
else:
not_found.append(vm_name)
logging.warning(f"VM not found: {vm_name}")
return found_vms, not_found
def remove_vm_from_vpg(client, vpg_name, vm):
"""Remove a VM from the VPG."""
vm_name = vm.get('VmName')
# vm_id = vm.get('VmIdentifier')
logging.info(f"\nRemoving VM {vm_name} from VPG...")
client.vpgs.remove_vm_from_vpg(vpg_name=vpg_name, vm_name=vm_name)
def main():
"""
Main function to demonstrate VPG creation.
@@ -105,20 +65,18 @@ def main():
3. Add specified VMs to the VPG
4. Remove the last added VM from the VPG
"""
# Parse command line arguments
args = parse_arguments()
# Handle both space-separated and comma-separated VM names
vm_names = []
for name in args.vm_names:
vm_names.extend([n.strip() for n in name.split(',') if n.strip()])
# Update args.vm_names with the processed list
args.vm_names = vm_names
try:
# Step 1: Create a ZVMLClient instance
logging.info(f"Initializing ZVMLClient for ZVM at {ZVM_HOST}")
parser = argparse.ArgumentParser(description='Create VPG and add specified VMs')
parser.add_argument('--vm-name', required=True,
help='VM name to add to the VPG')
parser.add_argument('--vpg-name', default="Test-VPG-Python",
help='Name of the VPG to create (default: Test-VPG-Python)')
args = parser.parse_args()
# Step 2: Create a ZVMLClient instance
client = ZVMLClient(
zvm_address=ZVM_HOST,
client_id=CLIENT_ID,
@@ -126,68 +84,42 @@ def main():
verify_certificate=ZVM_SSL_VERIFY
)
# Step 2: Identify local and peer sites
# Step 3: Identify local and peer sites
logging.info("Retrieving list of available sites...")
sites = client.virtualization_sites.get_virtualization_sites()
if not sites:
logging.warning("No sites found!")
sys.exit(1)
logging.info(f"Found {len(sites)} site(s):")
logging.info(f'Sites Info: {json.dumps(sites, indent=4)}')
# Get local and peer site identifiers
local_site = client.localsite.get_local_site()
local_site_identifier = local_site.get('SiteIdentifier')
logging.info(f"Local site identifier: {local_site_identifier}")
peer_site = next((site for site in sites if site.get('SiteIdentifier') != local_site_identifier), None)
if not peer_site:
logging.warning("No peer site found!")
sys.exit(1)
peer_site_identifier = peer_site.get('SiteIdentifier')
logging.info(f"Peer site identifier: {peer_site_identifier}")
# Step 3: Get peer site resources for VPG configuration
logging.info("\nRetrieving peer site resources for VPG configuration...")
# Get peer datastores
# Step 4: Get peer resources
peer_datastores = client.virtualization_sites.get_virtualization_site_datastores(site_identifier=peer_site_identifier)
if not peer_datastores:
logging.warning("No datastores found in peer site!")
sys.exit(1)
target_datastore = peer_datastores[0] # Use first available datastore
logging.info(f"Peer datastores: {json.dumps(peer_datastores, indent=4)}")
target_datastore = peer_datastores[2] # Use first available
# Get peer folders
peer_folders = client.virtualization_sites.get_virtualization_site_folders(site_identifier=peer_site_identifier)
if not peer_folders:
logging.warning("No folders found in peer site!")
sys.exit(1)
target_folder = peer_folders[0] # Use first available folder
logging.info(f"Peer folders: {json.dumps(peer_folders, indent=4)}")
target_folder = peer_folders[0] # Use first available
# Get peer networks
peer_networks = client.virtualization_sites.get_virtualization_site_networks(site_identifier=peer_site_identifier)
if not peer_networks:
logging.warning("No networks found in peer site!")
sys.exit(1)
target_network = peer_networks[0] # Use first available network
logging.info(f"Peer networks: {json.dumps(peer_networks, indent=4)}")
target_network = peer_networks[0] # Use first available
# Get peer hosts
peer_hosts = client.virtualization_sites.get_virtualization_site_hosts(site_identifier=peer_site_identifier)
if not peer_hosts:
logging.warning("No hosts found in peer site!")
sys.exit(1)
target_host = peer_hosts[0] # Use first available host
logging.info(f"Peer hosts: {json.dumps(peer_hosts, indent=4)}")
target_host = peer_hosts[0] # Use first available
# Step 4: Create VPG configuration
# Step 5: Create VPG configuration
logging.info("\nCreating VPG configuration...")
vpg_name = args.vpg_name
# Basic VPG settings
basic = {
"Name": vpg_name,
"Name": args.vpg_name,
"VpgType": "Remote",
"RpoInSeconds": 300,
"JournalHistoryInHours": 24,
@@ -196,17 +128,24 @@ def main():
"ProtectedSiteIdentifier": local_site_identifier,
"RecoverySiteIdentifier": peer_site_identifier
}
# Journal settings, keep the default settings
journal = {
}
# Recovery settings
journal = {} # Keep default settings
recovery = {
"DefaultHostIdentifier": target_host.get('HostIdentifier'),
"DefaultDatastoreIdentifier": target_datastore.get('DatastoreIdentifier'),
"DefaultFolderIdentifier": target_folder.get('FolderIdentifier')
}
networks = {
"Failover": {
"Hypervisor": {
"DefaultNetworkIdentifier": target_network.get('NetworkIdentifier')
}
},
"FailoverTest": {
"Hypervisor": {
"DefaultNetworkIdentifier": target_network.get('NetworkIdentifier')
}
}
}
# Network settings
networks = {
@@ -223,62 +162,20 @@ def main():
}
# Step 5: Create VPG
logging.info(f"\nCreating VPG '{vpg_name}'...")
logging.info("VPG Settings:")
logging.info(f"Basic: {json.dumps(basic, indent=4)}")
logging.info(f"Journal: {json.dumps(journal, indent=4)}")
logging.info(f"Recovery: {json.dumps(recovery, indent=4)}")
logging.info(f"Networks: {json.dumps(networks, indent=4)}")
# Create VPG
vpg_id = client.vpgs.create_vpg(basic=basic, journal=journal, recovery=recovery, networks=networks, sync=True)
logging.info(f"VPG created successfully with ID: {vpg_id}")
logging.info(f'vpg {args.vpg_name} successfully created, vpg_id is {vpg_id}')
# Step 6: Get specified VMs for protection
logging.info(f"\nRetrieving specified VMs for protection: {args.vm_names}")
found_vms, not_found = find_vms_by_names(client, local_site_identifier, args.vm_names)
if not_found:
logging.warning(f"The following VMs were not found: {not_found}")
if not found_vms:
logging.error("No VMs found for protection!")
sys.exit(1)
logging.info(f"Found {len(found_vms)} VM(s) for protection")
# Step 7: Add VMs to VPG
for vm in found_vms:
vm_name = vm.get('VmName')
vm_id = vm.get('VmIdentifier')
logging.info(f"\nAdding VM {vm_name} (ID: {vm_id}) to VPG...")
vm_payload = {
"VmIdentifier": vm_id,
"Recovery": {
"HostIdentifier": target_host.get('HostIdentifier'),
"DatastoreIdentifier": target_datastore.get('DatastoreIdentifier'),
"FolderIdentifier": target_folder.get('FolderIdentifier')
}
}
task_id = client.vpgs.add_vm_to_vpg(args.vpg_name, vm_list_payload=vm_payload)
logging.info(f"Task ID: {task_id} to add VM {vm_name} to VPG")
# Step 7: Add Vm to the VPG
task_id = client.vpgs.add_vm_to_vpg_by_name(args.vpg_name, args.vm_name)
logging.info(f'vm {args.vm_name} successfully added to vpg {args.vpg_name}')
# Step 8: Interactive VM removal
if found_vms:
last_vm = found_vms[-1]
vm_name = last_vm.get('VmName')
while True:
response = input(f"\nWould you like to remove the last added VM ({vm_name}) from the VPG? (yes/no): ").lower()
if response in ['yes', 'y']:
remove_vm_from_vpg(client, args.vpg_name, last_vm)
logging.info(f"Successfully removed VM {vm_name} from VPG {args.vpg_name}")
break
elif response in ['no', 'n']:
logging.info("Skipping VM removal.")
break
else:
print("Please answer 'yes' or 'no'")
response = input(f"Remove VPG{args.vpg_name}? (yes/no): ").lower()
if response in ['yes', 'y']:
client.vpgs.delete_vpg(args.vpg_name)
except Exception as e:
logging.error(f"VPG operation failed: {str(e)}")
exercises/05_vpg_operations/working/create_vpg.py
+110 -159
View File
@@ -18,7 +18,7 @@ In this exercise, you will:
5. Create a VPG configuration with all necessary settings
6. Create the VPG
7. Add VMs to the VPG
8. Optionally remove a VM from the VPG
8. Optionally delete the VPG
STEP-BY-STEP INSTRUCTIONS:
1. Look at the TODO comments below - they tell you exactly what to do
@@ -33,12 +33,17 @@ WHAT IS A VPG?
- You can replicate VMs from one site to another using VPGs
USAGE EXAMPLE:
python create_vpg.py --vm-names "vm1" "vm2" "vm3" --vpg-name "My-VPG"
python create_vpg.py --vm-name "vm1" --vpg-name "My-VPG"
"""
import sys
import os
import logging
# Set up logging with timestamp
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
import json
import argparse
from pathlib import Path
@@ -47,14 +52,14 @@ import urllib3
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Add prerequisites to Python path (this helps Python find your config file)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
sys.path.append(str(prerequisites_path))
# Import the Zerto SDK - this gives us the ZVMLClient class
# Import the SDK modules
from zvml import ZVMLClient
# Import your configuration settings
# Import configuration
try:
from config import (
ZVM_HOST, # Your ZVM IP address (e.g., "192.168.1.100")
@@ -64,100 +69,37 @@ try:
CLIENT_SECRET # Your Keycloak client secret
)
except ImportError:
print("❌ ERROR: Configuration file not found!")
print("Please copy config.example.py to config.py and update with your values")
print("Error: Please copy config.example.py to config.py and update with your values")
print("Expected path:", prerequisites_path / "config.py")
sys.exit(1)
def parse_arguments():
"""
Parse command line arguments.
TODO: Implement argument parsing
HINT: Use this syntax:
parser = argparse.ArgumentParser(description='Create VPG and add specified VMs')
parser.add_argument('--vm-names', nargs='+', required=True,
help='List of VM names to add to the VPG')
parser.add_argument('--vpg-name', default="Test-VPG-Python",
help='Name of the VPG to create (default: Test-VPG-Python)')
return parser.parse_args()
EXPLANATION:
- argparse helps you get command line arguments
- --vm-names: List of VM names (required)
- --vpg-name: Name for the VPG (optional, has default)
"""
pass # ← REPLACE WITH YOUR CODE
def find_vms_by_names(client, site_identifier, vm_names):
"""
Find VMs by their names in the specified site.
TODO: Implement the function to:
1. Get all VMs from the site using client.virtualization_sites.get_virtualization_site_vms()
2. Create a dictionary for easy lookup: {vm.get('VmName'): vm for vm in vms}
3. Find requested VMs and return found/not found lists
HINT: Use this syntax:
vms = client.virtualization_sites.get_virtualization_site_vms(site_identifier=site_identifier)
vm_dict = {vm.get('VmName'): vm for vm in vms}
found_vms = []
not_found = []
for vm_name in vm_names:
if vm_name in vm_dict:
found_vms.append(vm_dict[vm_name])
else:
not_found.append(vm_name)
return found_vms, not_found
"""
pass # ← REPLACE WITH YOUR CODE
def remove_vm_from_vpg(client, vpg_name, vm):
"""
Remove a VM from the VPG.
TODO: Implement the function to:
1. Get VM name from vm object
2. Call client.vpgs.remove_vm_from_vpg() to remove the VM
HINT: Use this syntax:
vm_name = vm.get('VmName')
client.vpgs.remove_vm_from_vpg(vpg_name=vpg_name, vm_name=vm_name)
"""
pass # ← REPLACE WITH YOUR CODE
def main():
"""
Main function - this is where your code goes!
Follow the step-by-step instructions below.
Main function to demonstrate VPG creation.
Shows how to:
1. Create a new VPG with basic settings
2. Configure journal, recovery, and network settings
3. Add specified VMs to the VPG
4. Remove the last added VM from the VPG
"""
# Set up logging so you can see what's happening
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
print("🚀 Starting Zerto VPG Operations Exercise")
print("=" * 50)
try:
# ========================================
# STEP 1: Parse command line arguments
# ========================================
print("\n📝 STEP 1: Parsing command line arguments...")
print("You need to call the parse_arguments() function you created above.")
# TODO: Add code to parse arguments
# HINT: Use this syntax:
# args = parse_arguments()
parser = argparse.ArgumentParser(description='Create VPG and add specified VMs')
parser.add_argument('--vm-name', required=True,
help='VM name to add to the VPG')
parser.add_argument('--vpg-name', default="Test-VPG-Python",
help='Name of the VPG to create (default: Test-VPG-Python)')
args = parser.parse_args()
#
# EXPLANATION:
# This gets the VM names and VPG name from command line
# ← ADD YOUR CODE HERE
# This gets the VM name and VPG name from command line
# ========================================
# STEP 2: Create ZVMLClient instance
@@ -167,13 +109,12 @@ def main():
# TODO: Add code to create ZVMLClient
# HINT: Use this syntax:
# client = ZVMLClient(
# zvm_address=ZVM_HOST,
# client_id=CLIENT_ID,
# client_secret=CLIENT_SECRET,
# verify_certificate=ZVM_SSL_VERIFY
# )
client = ZVMLClient(
zvm_address=ZVM_HOST,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
verify_certificate=ZVM_SSL_VERIFY
)
# ← ADD YOUR CODE HERE
# ========================================
@@ -184,11 +125,17 @@ def main():
# TODO: Add code to get sites and site identifiers
# HINT: Use this syntax:
# sites = client.virtualization_sites.get_virtualization_sites()
# local_site = client.localsite.get_local_site()
# local_site_identifier = local_site.get('SiteIdentifier')
# peer_site = next((site for site in sites if site.get('SiteIdentifier') != local_site_identifier), None)
# peer_site_identifier = peer_site.get('SiteIdentifier')
sites = client.virtualization_sites.get_virtualization_sites()
local_site = client.localsite.get_local_site()
local_site_identifier = local_site.get('SiteIdentifier')
peer_site = next((site for site in sites if site.get('SiteIdentifier') != local_site_identifier), None)
peer_site_identifier = peer_site.get('SiteIdentifier')
#
# EXPLANATION:
# - client.virtualization_sites.get_virtualization_sites() gets all sites
# - client.localsite.get_local_site() gets local site info
# - next((site for site in sites if site.get('SiteIdentifier') != local_site_identifier), None) gets peer site
# - peer_site_identifier = peer_site.get('SiteIdentifier') gets peer site identifier
# ← ADD YOUR CODE HERE
@@ -200,17 +147,30 @@ def main():
# TODO: Add code to get peer site resources
# HINT: Use this syntax:
# peer_datastores = client.virtualization_sites.get_virtualization_site_datastores(site_identifier=peer_site_identifier)
# target_datastore = peer_datastores[0] # Use first available
#
# peer_folders = client.virtualization_sites.get_virtualization_site_folders(site_identifier=peer_site_identifier)
# target_folder = peer_folders[0] # Use first available
#
# peer_networks = client.virtualization_sites.get_virtualization_site_networks(site_identifier=peer_site_identifier)
# target_network = peer_networks[0] # Use first available
#
# peer_hosts = client.virtualization_sites.get_virtualization_site_hosts(site_identifier=peer_site_identifier)
# target_host = peer_hosts[0] # Use first available
peer_datastores = client.virtualization_sites.get_virtualization_site_datastores(site_identifier=peer_site_identifier)
logging.info(f"Peer datastores: {json.dumps(peer_datastores, indent=4)}")
target_datastore = peer_datastores[2] # Use first available
peer_folders = client.virtualization_sites.get_virtualization_site_folders(site_identifier=peer_site_identifier)
logging.info(f"Peer folders: {json.dumps(peer_folders, indent=4)}")
target_folder = peer_folders[0] # Use first available
peer_networks = client.virtualization_sites.get_virtualization_site_networks(site_identifier=peer_site_identifier)
logging.info(f"Peer networks: {json.dumps(peer_networks, indent=4)}")
target_network = peer_networks[0] # Use first available
peer_hosts = client.virtualization_sites.get_virtualization_site_hosts(site_identifier=peer_site_identifier)
logging.info(f"Peer hosts: {json.dumps(peer_hosts, indent=4)}")
target_host = peer_hosts[0] # Use first available
# EXPLANATION:
# - client.virtualization_sites.get_virtualization_site_datastores(site_identifier=peer_site_identifier) gets datastores
# - client.virtualization_sites.get_virtualization_site_folders(site_identifier=peer_site_identifier) gets folders
# - client.virtualization_sites.get_virtualization_site_networks(site_identifier=peer_site_identifier) gets networks
# - client.virtualization_sites.get_virtualization_site_hosts(site_identifier=peer_site_identifier) gets hosts
# - target_datastore = peer_datastores[2] # Use first available
# - target_folder = peer_folders[0] # Use first available
# - target_network = peer_networks[0] # Use first available
# ← ADD YOUR CODE HERE
@@ -222,7 +182,37 @@ def main():
# TODO: Add code to create VPG configuration
# HINT: Use this syntax:
# basic = {
basic = {
"Name": args.vpg_name,
"VpgType": "Remote",
"RpoInSeconds": 300,
"JournalHistoryInHours": 24,
"Priority": "Medium",
"UseWanCompression": True,
"ProtectedSiteIdentifier": local_site_identifier,
"RecoverySiteIdentifier": peer_site_identifier
}
journal = {} # Keep default settings
recovery = {
"DefaultHostIdentifier": target_host.get('HostIdentifier'),
"DefaultDatastoreIdentifier": target_datastore.get('DatastoreIdentifier'),
"DefaultFolderIdentifier": target_folder.get('FolderIdentifier')
}
networks = {
"Failover": {
"Hypervisor": {
"DefaultNetworkIdentifier": target_network.get('NetworkIdentifier')
}
},
"FailoverTest": {
"Hypervisor": {
"DefaultNetworkIdentifier": target_network.get('NetworkIdentifier')
}
}
}
#
# EXPLANATION:
# - basic = {
# "Name": args.vpg_name,
# "VpgType": "Remote",
# "RpoInSeconds": 300,
@@ -232,27 +222,10 @@ def main():
# "ProtectedSiteIdentifier": local_site_identifier,
# "RecoverySiteIdentifier": peer_site_identifier
# }
#
# journal = {} # Keep default settings
#
# recovery = {
# - journal = {} # Keep default settings
# - recovery = {
# "DefaultHostIdentifier": target_host.get('HostIdentifier'),
# "DefaultDatastoreIdentifier": target_datastore.get('DatastoreIdentifier'),
# "DefaultFolderIdentifier": target_folder.get('FolderIdentifier')
# }
#
# networks = {
# "Failover": {
# "Hypervisor": {
# "DefaultNetworkIdentifier": target_network.get('NetworkIdentifier')
# }
# },
# "FailoverTest": {
# "Hypervisor": {
# "DefaultNetworkIdentifier": target_network.get('NetworkIdentifier')
# }
# }
# }
# ← ADD YOUR CODE HERE
@@ -264,7 +237,8 @@ def main():
# TODO: Add code to create VPG
# HINT: Use this syntax:
# vpg_id = client.vpgs.create_vpg(basic=basic, journal=journal, recovery=recovery, networks=networks, sync=True)
vpg_id = client.vpgs.create_vpg(basic=basic, journal=journal, recovery=recovery, networks=networks, sync=True)
logging.info(f'vpg {args.vpg_name} successfully created, vpg_id is {vpg_id}')
#
# EXPLANATION:
# This creates the VPG with all your settings
@@ -272,51 +246,28 @@ def main():
# ← ADD YOUR CODE HERE
# ========================================
# STEP 7: Find and add VMs to VPG
# STEP 7: Add Vm to the VPG
# ========================================
print("\n📝 STEP 7: Finding and adding VMs...")
print("You need to find the VMs and add them to the VPG.")
# TODO: Add code to find VMs
# HINT: Use this syntax:
# found_vms, not_found = find_vms_by_names(client, local_site_identifier, args.vm_names)
#
# EXPLANATION:
# This finds the VMs you specified in the command line
# ← ADD YOUR CODE HERE
# TODO: Add code to add VMs to VPG
# HINT: Use this syntax:
# for vm in found_vms:
# vm_name = vm.get('VmName')
# vm_id = vm.get('VmIdentifier')
# vm_payload = {
# "VmIdentifier": vm_id,
# "Recovery": {
# "HostIdentifier": target_host.get('HostIdentifier'),
# "DatastoreIdentifier": target_datastore.get('DatastoreIdentifier'),
# "FolderIdentifier": target_folder.get('FolderIdentifier')
# }
# }
# task_id = client.vpgs.add_vm_to_vpg(args.vpg_name, vm_list_payload=vm_payload)
task_id = client.vpgs.add_vm_to_vpg_by_name(args.vpg_name, args.vm_name)
logging.info(f'vm {args.vm_name} successfully added to vpg {args.vpg_name}')
# EXPLANATION:
# This adds vm to an existing VPG by Vm name
# ← ADD YOUR CODE HERE
# ========================================
# STEP 8: Interactive VM removal (optional)
# ========================================
print("\n📝 STEP 8: Interactive VM removal...")
print("You can optionally remove the last VM from the VPG.")
print("\n📝 STEP 8: Interactive VPG deletion...")
print("You can optionally delete the VPG.")
# TODO: Add code for interactive VM removal
# HINT: Use this syntax:
# if found_vms:
# last_vm = found_vms[-1]
# vm_name = last_vm.get('VmName')
# response = input(f"Remove VM {vm_name} from VPG? (yes/no): ").lower()
# if response in ['yes', 'y']:
# remove_vm_from_vpg(client, args.vpg_name, last_vm)
response = input(f"Remove VPG{args.vpg_name}? (yes/no): ").lower()
if response in ['yes', 'y']:
client.vpgs.delete_vpg(args.vpg_name)
# ← ADD YOUR CODE HERE
exercises/06_failover_test/README.md
+5 -16
View File
@@ -17,13 +17,6 @@ In this exercise, you'll learn how to perform and manage failover tests for your
- Working VPG
- Access to test resources
## Exercise Steps
1. Select a VPG for testing
2. Initiate a failover test
3. Monitor test progress
4. Stop the test
5. Review test results
## Working Directory
The `working` directory contains:
- `failover.py` - Template to complete
@@ -33,16 +26,12 @@ The `solution` directory contains:
- `failover.py` - Complete working example
## Key Concepts
- Authentication
- VPG attributes vs VM vs Volume vs NIC attributes
- Resource discovery
- Failover testing
- Test monitoring
- Test management
- Status tracking
## Common Issues
- Test initiation failures
- Resource conflicts
- Test timeout
- Cleanup issues
## Next Steps
Proceed to Exercise 7: Bulk Operations to learn about managing multiple VMs.
## Architecture
![Client Credentials](/Zerto-Python-SDK-Hands-On-Labs/diagrams/vpg-structure.png)
exercises/06_failover_test/solution/failover.py
+11 -94
View File
@@ -52,89 +52,6 @@ except ImportError:
print("Expected path:", prerequisites_path / "config.py")
sys.exit(1)
def parse_arguments():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(description='Perform failover test on a VPG')
parser.add_argument('--vpg-name', required=True,
help='Name of the VPG to test')
return parser.parse_args()
def find_vpg_by_name(client, vpg_name):
"""
Find a VPG by its name.
Args:
client: ZVMLClient instance
vpg_name: Name of the VPG to find
Returns:
dict: VPG object if found, None otherwise
"""
vpg = client.vpgs.list_vpgs(vpg_name=vpg_name)
# logging.info(f"Found vpg {json.dumps(vpg, indent=4)}")
return vpg if vpg else None
def start_failover_test(client, vpg_name):
"""
Start a failover test for the specified VPG using default settings.
Args:
client: ZVMLClient instance
vpg_name: Name of the VPG to test
Returns:
str: Test identifier
"""
logging.info(f"Starting failover test for VPG '{vpg_name}'")
# Start the test with default settings
response = client.vpgs.failover_test(
vpg_name=vpg_name,
sync=True # Wait for the test to start
)
logging.info(f"Faiolver test response: {response}")
return response
def monitor_test_progress(client, vpg_name, test_id):
"""
Monitor the progress of a failover test.
Args:
client: ZVMLClient instance
vpg_name: Name of the VPG
test_id: Test identifier
Returns:
bool: True if test completed successfully, False otherwise
"""
test_status = client.vpgs.get_vpg_test_status(vpg_name, test_id)
status = test_status.get('Status')
progress = test_status.get('Progress', 0)
logging.info(f"Test status: {status} (Progress: {progress}%)")
if status == 'Succeeded':
return True
elif status in ['Failed', 'Stopped']:
logging.error(f"Test {status.lower()}: {test_status.get('Message', 'No message')}")
return False
return False # Test is still running
def stop_failover_test(client, vpg_name):
"""
Stop a running failover test.
Args:
client: ZVMLClient instance
vpg_name: Name of the VPG
"""
logging.info(f"Stopping faiolver test for VPG '{vpg_name}'...")
response = client.vpgs.stop_failover_test(vpg_name=vpg_name)
logging.info(f"Stop failover test response: {response}")
def main():
"""
Main function to demonstrate failover testing.
@@ -147,7 +64,10 @@ def main():
try:
# Step 1: Parse command line arguments
args = parse_arguments()
parser = argparse.ArgumentParser(description='Perform failover test on a VPG')
parser.add_argument('--vpg-name', required=True,
help='Name of the VPG to test')
args = parser.parse_args()
# Step 2: Create ZVMLClient instance
logging.info(f"Initializing ZVMLClient for ZVM at {ZVM_HOST}")
@@ -158,19 +78,16 @@ def main():
verify_certificate=ZVM_SSL_VERIFY
)
# Step 3: Find the VPG
vpg = find_vpg_by_name(client, args.vpg_name)
if not vpg:
logging.error(f"VPG '{args.vpg_name}' not found!")
sys.exit(1)
# Step 3: Start the test with default settings
response = client.vpgs.failover_test(vpg_name=args.vpg_name, sync=True)
logging.info(f"Faiolver test response: {response}")
# Step 4: Start failover test
response = start_failover_test(client, args.vpg_name)
# Step 5: Handle test stop request
# Step 4: Handle test stop request
response = input("\nWould you like to stop the test? (yes/no): ").lower()
if response in ['yes', 'y']:
stop_failover_test(client, args.vpg_name)
logging.info(f"Stopping faiolver test for VPG '{args.vpg_name}'...")
response = client.vpgs.stop_failover_test(vpg_name=args.vpg_name)
logging.info(f"Stop failover test response: {response}")
except Exception as e:
logging.error(f"Failover test failed: {str(e)}")
exercises/06_failover_test/working/failover.py
+43 -200
View File
@@ -1,37 +1,22 @@
#!/usr/bin/env python3
"""
Exercise 6: Failover Testing - Beginner-Friendly Instructions
Exercise 6: Failover Testing - Solution
This script demonstrates how to perform a failover test on a VPG.
PREREQUISITES (Complete these first):
1. ✅ Completed Exercise 5 (VPG Operations)
2. ✅ Make sure you have the zvml package installed
3. ✅ Updated prerequisites/config.py with your ZVM details
4. ✅ Have a VPG created and running (from Exercise 5)
Prerequisites:
1. Install the zvml package in development mode:
cd /path/to/zvml-python-sdk
pip install -e .
2. Update prerequisites/config.py with your ZVM details
WHAT YOU NEED TO DO:
In this exercise, you will:
1. Create a ZVMLClient to connect to your ZVM
2. Parse command line arguments (VPG name)
3. Find a VPG by its name
4. Start a failover test on the VPG
5. Monitor the test progress
6. Optionally stop the test
Usage:
python failover.py --vpg-name "My-VPG"
STEP-BY-STEP INSTRUCTIONS:
1. Look at the TODO comments below - they tell you exactly what to do
2. Replace the placeholder code with the actual code
3. Each step has hints and examples to help you
4. If you get stuck, check the solution file in the solution/ directory
WHAT IS A FAILOVER TEST?
- **Failover Test**: Tests if your VMs can be successfully started at the recovery site
- It creates test VMs at the peer site to verify everything works
- The test VMs are isolated and don't affect production
- You can stop the test at any time to clean up the test VMs
USAGE EXAMPLE:
python failover.py --vpg-name "My-VPG"
This solution demonstrates:
- Finding a VPG by name
- Starting a failover test with default settings
- Monitoring test progress
- Stopping the test when requested
"""
import sys
@@ -46,219 +31,77 @@ import urllib3
# Suppress SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Add prerequisites to Python path (this helps Python find your config file)
# Add prerequisites to Python path
prerequisites_path = Path(__file__).parent.parent.parent.parent / "prerequisites"
sys.path.append(str(prerequisites_path))
# Import the Zerto SDK - this gives us the ZVMLClient class
# Import the SDK modules
from zvml import ZVMLClient
# Import your configuration settings
# Import configuration
try:
from config import (
ZVM_HOST, # Your ZVM IP address (e.g., "192.168.1.100")
ZVM_PORT, # Usually 443 for HTTPS
ZVM_SSL_VERIFY, # True/False for SSL certificate verification
CLIENT_ID, # Your Keycloak client ID (e.g., "my-api-client")
CLIENT_SECRET # Your Keycloak client secret
ZVM_HOST,
ZVM_PORT,
ZVM_SSL_VERIFY,
CLIENT_ID,
CLIENT_SECRET
)
except ImportError:
print("❌ ERROR: Configuration file not found!")
print("Please copy config.example.py to config.py and update with your values")
print("Error: Please copy config.example.py to config.py and update with your values")
print("Expected path:", prerequisites_path / "config.py")
sys.exit(1)
def parse_arguments():
"""
Parse command line arguments.
TODO: Implement argument parsing
HINT: Use this syntax:
parser = argparse.ArgumentParser(description='Perform failover test on a VPG')
parser.add_argument('--vpg-name', required=True,
help='Name of the VPG to test')
return parser.parse_args()
EXPLANATION:
- argparse helps you get command line arguments
- --vpg-name: Name of the VPG to test (required)
"""
pass # ← REPLACE WITH YOUR CODE
def find_vpg_by_name(client, vpg_name):
"""
Find a VPG by its name.
TODO: Implement the function to find a VPG by name
HINT: Use this syntax:
vpg = client.vpgs.list_vpgs(vpg_name=vpg_name)
return vpg if vpg else None
EXPLANATION:
- client.vpgs.list_vpgs() gets VPGs with a specific name
- Returns the VPG if found, None if not found
"""
pass # ← REPLACE WITH YOUR CODE
def start_failover_test(client, vpg_name):
"""
Start a failover test for the specified VPG.
TODO: Implement the function to start a failover test
HINT: Use this syntax:
response = client.vpgs.failover_test(
vpg_name=vpg_name,
sync=True # Wait for the test to start
)
return response
EXPLANATION:
- client.vpgs.failover_test() starts a failover test
- sync=True means wait for the test to start before returning
- Returns the response from the API
"""
pass # ← REPLACE WITH YOUR CODE
def monitor_test_progress(client, vpg_name, test_id):
"""
Monitor the progress of a failover test.
TODO: Implement the function to monitor test progress
HINT: Use this syntax:
test_status = client.vpgs.get_vpg_test_status(vpg_name, test_id)
status = test_status.get('Status')
progress = test_status.get('Progress', 0)
logging.info(f"Test status: {status} (Progress: {progress}%)")
if status == 'Succeeded':
return True
elif status in ['Failed', 'Stopped']:
return False
return False # Test is still running
EXPLANATION:
- client.vpgs.get_vpg_test_status() gets the current test status
- Returns True if test succeeded, False if failed/stopped/still running
"""
pass # ← REPLACE WITH YOUR CODE
def stop_failover_test(client, vpg_name):
"""
Stop a running failover test.
TODO: Implement the function to stop a failover test
HINT: Use this syntax:
response = client.vpgs.stop_failover_test(vpg_name=vpg_name)
EXPLANATION:
- client.vpgs.stop_failover_test() stops the running test
- This cleans up the test VMs at the recovery site
"""
pass # ← REPLACE WITH YOUR CODE
def main():
"""
Main function - this is where your code goes!
Follow the step-by-step instructions below.
Main function to demonstrate failover testing.
"""
# Set up logging so you can see what's happening
# Set up logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
print("🚀 Starting Zerto Failover Testing Exercise")
print("=" * 50)
try:
# ========================================
# STEP 1: Parse command line arguments
# ========================================
print("\n📝 STEP 1: Parsing command line arguments...")
print("You need to call the parse_arguments() function you created above.")
# TODO: Add code to parse arguments
# Step 1: Parse command line arguments
# TODO: Add code to get sites and site identifiers
# HINT: Use this syntax:
# args = parse_arguments()
#
# EXPLANATION:
# This gets the VPG name from command line
# parser = argparse.ArgumentParser(description='Perform failover test on a VPG')
# parser.add_argument('--vpg-name', required=True,
# help='Name of the VPG to test')
# args = parser.parse_args()
# ← ADD YOUR CODE HERE
# ========================================
# STEP 2: Create ZVMLClient instance
# ========================================
print("\n📝 STEP 2: Creating ZVMLClient...")
print("This is the same as previous exercises.")
# TODO: Add code to create ZVMLClient
# Step 2: Create ZVMLClient instance
# TODO: Add code to get sites and site identifiers
# HINT: Use this syntax:
logging.info(f"Initializing ZVMLClient for ZVM at {ZVM_HOST}")
# client = ZVMLClient(
# zvm_address=ZVM_HOST,
# client_id=CLIENT_ID,
# client_secret=CLIENT_SECRET,
# verify_certificate=ZVM_SSL_VERIFY
# )
# ← ADD YOUR CODE HERE
# ========================================
# STEP 3: Find the VPG
# ========================================
print("\n📝 STEP 3: Finding the VPG...")
print("You need to find the VPG by its name.")
# TODO: Add code to find VPG
# Step 3: Start the test with default settings
# TODO: Add code to get sites and site identifiers
# HINT: Use this syntax:
# vpg = find_vpg_by_name(client, args.vpg_name)
# if not vpg:
# logging.error(f"VPG '{args.vpg_name}' not found!")
# sys.exit(1)
#
# EXPLANATION:
# This checks if the VPG exists before trying to test it
# response = client.vpgs.failover_test(vpg_name=args.vpg_name, sync=True)
# logging.info(f"Faiolver test response: {response}")
# ← ADD YOUR CODE HERE
# ← ADD YOUR CODE HERE
# ========================================
# STEP 4: Start failover test
# ========================================
print("\n📝 STEP 4: Starting failover test...")
print("You need to start a failover test on the VPG.")
# TODO: Add code to start failover test
# HINT: Use this syntax:
# response = start_failover_test(client, args.vpg_name)
#
# EXPLANATION:
# This starts the failover test and waits for it to begin
# ← ADD YOUR CODE HERE
# ========================================
# STEP 5: Handle test stop request
# ========================================
print("\n📝 STEP 5: Handling test stop request...")
print("You can optionally stop the test.")
# TODO: Add code for interactive test stopping
# Step 4: Handle test stop request
# TODO: Add code to get sites and site identifiers
# HINT: Use this syntax:
# response = input("\nWould you like to stop the test? (yes/no): ").lower()
# if response in ['yes', 'y']:
# stop_failover_test(client, args.vpg_name)
#
# EXPLANATION:
# This asks the user if they want to stop the test and cleans up if yes
# logging.info(f"Stopping faiolver test for VPG '{args.vpg_name}'...")
# response = client.vpgs.stop_failover_test(vpg_name=args.vpg_name)
# logging.info(f"Stop failover test response: {response}")
# ← ADD YOUR CODE HERE
except Exception as e:
# This catches any errors that might occur
print(f"\n❌ ERROR: Something went wrong!")
print(f"Error details: {str(e)}")
logging.error(f"Failover test failed: {str(e)}")
sys.exit(1)
prerequisites/__pycache__/config.cpython-313.pyc
BIN
View File
Binary file not shown.