0
# Snapshots
1
2
Node pool snapshot management for backup, restore, template creation scenarios, and consistent deployments enabling disaster recovery, infrastructure-as-code patterns, and standardized cluster configurations. Snapshots capture the exact state of node pools for reliable reproduction and operational consistency.
3
4
## Capabilities
5
6
### Snapshot Lifecycle Management
7
8
Create, update, and delete node pool snapshots for backup and template scenarios.
9
10
```python { .api }
11
def create_or_update(
12
resource_group_name: str,
13
resource_name: str,
14
parameters: Snapshot
15
) -> Snapshot:
16
"""
17
Creates or updates a node pool snapshot.
18
19
Args:
20
resource_group_name: Azure resource group name
21
resource_name: Snapshot name
22
parameters: Complete snapshot configuration
23
24
Returns:
25
Snapshot: Created or updated snapshot
26
"""
27
28
def delete(resource_group_name: str, resource_name: str) -> None:
29
"""
30
Deletes a snapshot.
31
32
Args:
33
resource_group_name: Azure resource group name
34
resource_name: Snapshot name
35
"""
36
37
def update_tags(
38
resource_group_name: str,
39
resource_name: str,
40
parameters: TagsObject
41
) -> Snapshot:
42
"""
43
Updates snapshot tags for organization and management.
44
45
Args:
46
resource_group_name: Azure resource group name
47
resource_name: Snapshot name
48
parameters: Tags to update
49
50
Returns:
51
Snapshot: Updated snapshot with new tags
52
"""
53
```
54
55
### Snapshot Discovery
56
57
List and retrieve snapshots for backup management and template discovery.
58
59
```python { .api }
60
def list() -> ItemPaged[Snapshot]:
61
"""
62
Lists all snapshots in the subscription.
63
64
Returns:
65
ItemPaged[Snapshot]: Paginated snapshot list across all resource groups
66
"""
67
68
def list_by_resource_group(resource_group_name: str) -> ItemPaged[Snapshot]:
69
"""
70
Lists snapshots in a specific resource group.
71
72
Args:
73
resource_group_name: Azure resource group name
74
75
Returns:
76
ItemPaged[Snapshot]: Paginated snapshot list for the resource group
77
"""
78
79
def get(resource_group_name: str, resource_name: str) -> Snapshot:
80
"""
81
Gets a specific snapshot with complete configuration details.
82
83
Args:
84
resource_group_name: Azure resource group name
85
resource_name: Snapshot name
86
87
Returns:
88
Snapshot: Complete snapshot configuration and metadata
89
"""
90
```
91
92
## Core Types
93
94
```python { .api }
95
class Snapshot:
96
"""
97
Node pool snapshot definition with configuration and metadata.
98
99
Attributes:
100
location: Azure region where snapshot is stored
101
tags: Resource tags for organization
102
creation_data: Source information for snapshot creation
103
snapshot_type: Type of snapshot ('NodePool')
104
kubernetes_version: Kubernetes version captured in snapshot (read-only)
105
node_image_version: Node image version captured (read-only)
106
vm_size: VM size of nodes in snapshot (read-only)
107
os_type: Operating system type (read-only)
108
os_sku: Operating system SKU (read-only)
109
enable_fips: FIPS compliance setting (read-only)
110
system_data: System metadata (read-only)
111
id: Resource ID (read-only)
112
name: Snapshot name (read-only)
113
type: Resource type (read-only)
114
"""
115
location: str
116
tags: Optional[Dict[str, str]]
117
creation_data: Optional[CreationData]
118
snapshot_type: Optional[str]
119
kubernetes_version: Optional[str]
120
node_image_version: Optional[str]
121
vm_size: Optional[str]
122
os_type: Optional[str]
123
os_sku: Optional[str]
124
enable_fips: Optional[bool]
125
system_data: Optional[SystemData]
126
id: Optional[str]
127
name: Optional[str]
128
type: Optional[str]
129
130
class CreationData:
131
"""
132
Snapshot creation metadata identifying the source.
133
134
Attributes:
135
source_resource_id: Resource ID of source agent pool
136
"""
137
source_resource_id: Optional[str]
138
139
class TagsObject:
140
"""
141
Resource tags for organization and management.
142
143
Attributes:
144
tags: Dictionary of key-value tag pairs
145
"""
146
tags: Optional[Dict[str, str]]
147
```
148
149
## Usage Examples
150
151
### Basic Snapshot Creation
152
153
```python
154
from azure.identity import DefaultAzureCredential
155
from azure.mgmt.containerservice import ContainerServiceClient
156
from azure.mgmt.containerservice.models import Snapshot, CreationData
157
158
credential = DefaultAzureCredential()
159
client = ContainerServiceClient(credential, subscription_id)
160
161
# Create snapshot of existing agent pool
162
source_agent_pool_id = "/subscriptions/sub-id/resourceGroups/myRG/providers/Microsoft.ContainerService/managedClusters/myCluster/agentPools/nodepool1"
163
164
snapshot = Snapshot(
165
location="eastus",
166
creation_data=CreationData(
167
source_resource_id=source_agent_pool_id
168
),
169
tags={
170
"environment": "production",
171
"backup-date": "2024-01-15",
172
"cluster": "myCluster",
173
"purpose": "disaster-recovery"
174
}
175
)
176
177
result = client.snapshots.create_or_update(
178
"myResourceGroup",
179
"nodepool1-backup-20240115",
180
snapshot
181
)
182
print(f"Snapshot created: {result.name}")
183
print(f"Kubernetes version: {result.kubernetes_version}")
184
print(f"Node image version: {result.node_image_version}")
185
```
186
187
### Template Snapshots for Standardization
188
189
```python
190
# Create template snapshots for different environments
191
environments = ["dev", "staging", "prod"]
192
configurations = {
193
"dev": {
194
"cluster": "dev-cluster",
195
"agent_pool": "nodepool1",
196
"tags": {"environment": "dev", "cost-center": "engineering"}
197
},
198
"staging": {
199
"cluster": "staging-cluster",
200
"agent_pool": "nodepool1",
201
"tags": {"environment": "staging", "cost-center": "qa"}
202
},
203
"prod": {
204
"cluster": "prod-cluster",
205
"agent_pool": "nodepool1",
206
"tags": {"environment": "prod", "cost-center": "operations"}
207
}
208
}
209
210
for env in environments:
211
config = configurations[env]
212
source_id = f"/subscriptions/sub-id/resourceGroups/{env}-rg/providers/Microsoft.ContainerService/managedClusters/{config['cluster']}/agentPools/{config['agent_pool']}"
213
214
template_snapshot = Snapshot(
215
location="eastus",
216
creation_data=CreationData(source_resource_id=source_id),
217
tags={
218
**config["tags"],
219
"template": "true",
220
"created-date": "2024-01-15"
221
}
222
)
223
224
result = client.snapshots.create_or_update(
225
f"{env}-rg",
226
f"{env}-nodepool-template",
227
template_snapshot
228
)
229
print(f"Template snapshot created for {env}: {result.name}")
230
```
231
232
### Snapshot Management Operations
233
234
```python
235
# List all snapshots in subscription
236
print("All snapshots:")
237
for snapshot in client.snapshots.list():
238
print(f" {snapshot.name} - {snapshot.location}")
239
if snapshot.tags:
240
env = snapshot.tags.get("environment", "unknown")
241
purpose = snapshot.tags.get("purpose", "unknown")
242
print(f" Environment: {env}, Purpose: {purpose}")
243
244
# List snapshots in specific resource group
245
prod_snapshots = client.snapshots.list_by_resource_group("prod-rg")
246
print("\nProduction snapshots:")
247
for snapshot in prod_snapshots:
248
print(f" {snapshot.name}")
249
print(f" K8s version: {snapshot.kubernetes_version}")
250
print(f" Node image: {snapshot.node_image_version}")
251
print(f" VM size: {snapshot.vm_size}")
252
253
# Get detailed snapshot information
254
detailed_snapshot = client.snapshots.get("prod-rg", "prod-nodepool-template")
255
print(f"\nSnapshot details for {detailed_snapshot.name}:")
256
print(f" OS Type: {detailed_snapshot.os_type}")
257
print(f" OS SKU: {detailed_snapshot.os_sku}")
258
print(f" FIPS enabled: {detailed_snapshot.enable_fips}")
259
if detailed_snapshot.creation_data:
260
print(f" Source: {detailed_snapshot.creation_data.source_resource_id}")
261
```
262
263
### Automated Backup Strategy
264
265
```python
266
import datetime
267
from azure.mgmt.containerservice.models import TagsObject
268
269
def create_automated_backup(client, cluster_rg, cluster_name, agent_pool_name):
270
"""Create automated daily backup of agent pool."""
271
272
today = datetime.date.today()
273
backup_name = f"{cluster_name}-{agent_pool_name}-backup-{today.strftime('%Y%m%d')}"
274
275
source_id = f"/subscriptions/{subscription_id}/resourceGroups/{cluster_rg}/providers/Microsoft.ContainerService/managedClusters/{cluster_name}/agentPools/{agent_pool_name}"
276
277
backup_snapshot = Snapshot(
278
location="eastus",
279
creation_data=CreationData(source_resource_id=source_id),
280
tags={
281
"backup-type": "automated",
282
"backup-date": today.isoformat(),
283
"cluster": cluster_name,
284
"agent-pool": agent_pool_name,
285
"retention": "30-days"
286
}
287
)
288
289
try:
290
result = client.snapshots.create_or_update(
291
f"{cluster_rg}-backups",
292
backup_name,
293
backup_snapshot
294
)
295
print(f"Automated backup created: {result.name}")
296
return result
297
except Exception as e:
298
print(f"Backup failed: {e}")
299
return None
300
301
# Create backups for multiple clusters
302
clusters = [
303
{"rg": "prod-rg", "name": "prod-cluster", "pools": ["system", "apps"]},
304
{"rg": "staging-rg", "name": "staging-cluster", "pools": ["nodepool1"]}
305
]
306
307
for cluster in clusters:
308
for pool in cluster["pools"]:
309
create_automated_backup(
310
client,
311
cluster["rg"],
312
cluster["name"],
313
pool
314
)
315
```
316
317
### Snapshot Restoration Workflow
318
319
```python
320
def restore_from_snapshot(client, snapshot_rg, snapshot_name, target_cluster_rg, target_cluster, new_pool_name):
321
"""Create new agent pool from snapshot configuration."""
322
323
# Get snapshot details
324
snapshot = client.snapshots.get(snapshot_rg, snapshot_name)
325
326
from azure.mgmt.containerservice.models import AgentPool
327
328
# Create agent pool configuration based on snapshot
329
restored_pool = AgentPool(
330
count=3, # Default count, adjust as needed
331
vm_size=snapshot.vm_size,
332
os_type=snapshot.os_type,
333
os_sku=snapshot.os_sku,
334
mode="User", # Default to User mode
335
orchestrator_version=snapshot.kubernetes_version,
336
# Reference the snapshot for node image
337
creation_data=CreationData(
338
source_resource_id=f"/subscriptions/{subscription_id}/resourceGroups/{snapshot_rg}/providers/Microsoft.ContainerService/snapshots/{snapshot_name}"
339
)
340
)
341
342
# Create the agent pool
343
operation = client.agent_pools.begin_create_or_update(
344
target_cluster_rg,
345
target_cluster,
346
new_pool_name,
347
restored_pool
348
)
349
350
result = operation.result()
351
print(f"Agent pool restored from snapshot: {result.name}")
352
return result
353
354
# Restore production configuration to staging
355
restore_from_snapshot(
356
client,
357
"prod-rg",
358
"prod-nodepool-template",
359
"staging-rg",
360
"staging-cluster",
361
"restored-nodepool"
362
)
363
```
364
365
### Snapshot Tag Management
366
367
```python
368
from azure.mgmt.containerservice.models import TagsObject
369
370
# Update snapshot tags for better organization
371
update_tags = TagsObject(
372
tags={
373
"environment": "production",
374
"backup-type": "manual",
375
"retention": "90-days",
376
"compliance": "required",
377
"owner": "platform-team"
378
}
379
)
380
381
updated_snapshot = client.snapshots.update_tags(
382
"prod-rg",
383
"important-snapshot",
384
update_tags
385
)
386
print(f"Tags updated for snapshot: {updated_snapshot.name}")
387
388
# Bulk tag updates for compliance
389
compliance_tags = TagsObject(
390
tags={
391
"compliance-reviewed": "2024-01-15",
392
"gdpr-compliant": "true",
393
"retention-policy": "applied"
394
}
395
)
396
397
snapshots = client.snapshots.list_by_resource_group("compliance-rg")
398
for snapshot in snapshots:
399
if "prod" in snapshot.name:
400
client.snapshots.update_tags(
401
"compliance-rg",
402
snapshot.name,
403
compliance_tags
404
)
405
print(f"Compliance tags applied to {snapshot.name}")
406
```
407
408
### Cross-Region Snapshot Strategy
409
410
```python
411
# Create snapshots across multiple regions for disaster recovery
412
regions = ["eastus", "westus", "northeurope"]
413
critical_clusters = ["prod-cluster-1", "prod-cluster-2"]
414
415
for region in regions:
416
for cluster in critical_clusters:
417
# Create snapshot in each region
418
regional_snapshot = Snapshot(
419
location=region,
420
creation_data=CreationData(
421
source_resource_id=f"/subscriptions/{subscription_id}/resourceGroups/{region}-rg/providers/Microsoft.ContainerService/managedClusters/{cluster}/agentPools/nodepool1"
422
),
423
tags={
424
"region": region,
425
"cluster": cluster,
426
"dr-purpose": "cross-region-backup",
427
"backup-date": datetime.date.today().isoformat()
428
}
429
)
430
431
result = client.snapshots.create_or_update(
432
f"{region}-dr-rg",
433
f"{cluster}-{region}-dr-snapshot",
434
regional_snapshot
435
)
436
print(f"DR snapshot created: {result.name} in {region}")
437
```
438
439
### Cleanup Old Snapshots
440
441
```python
442
import datetime
443
444
def cleanup_old_snapshots(client, resource_group, retention_days=30):
445
"""Remove snapshots older than retention period."""
446
447
cutoff_date = datetime.date.today() - datetime.timedelta(days=retention_days)
448
deleted_count = 0
449
450
snapshots = client.snapshots.list_by_resource_group(resource_group)
451
452
for snapshot in snapshots:
453
if snapshot.tags and "backup-date" in snapshot.tags:
454
backup_date = datetime.datetime.strptime(
455
snapshot.tags["backup-date"], "%Y-%m-%d"
456
).date()
457
458
if backup_date < cutoff_date:
459
# Check if it's not a template or important snapshot
460
if snapshot.tags.get("template") != "true" and snapshot.tags.get("retention") != "permanent":
461
try:
462
client.snapshots.delete(resource_group, snapshot.name)
463
print(f"Deleted old snapshot: {snapshot.name}")
464
deleted_count += 1
465
except Exception as e:
466
print(f"Failed to delete {snapshot.name}: {e}")
467
468
print(f"Cleanup completed. Deleted {deleted_count} old snapshots.")
469
470
# Run cleanup for backup resource groups
471
backup_rgs = ["prod-backups-rg", "staging-backups-rg"]
472
for rg in backup_rgs:
473
cleanup_old_snapshots(client, rg, retention_days=30)
474
```
475
476
## Error Handling
477
478
```python
479
from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
480
481
try:
482
snapshot = client.snapshots.get("myRG", "nonexistent-snapshot")
483
except ResourceNotFoundError:
484
print("Snapshot not found")
485
except HttpResponseError as e:
486
print(f"HTTP error: {e.status_code}")
487
488
# Handle snapshot creation failures
489
try:
490
result = client.snapshots.create_or_update("myRG", "new-snapshot", snapshot)
491
except HttpResponseError as e:
492
if e.status_code == 400:
493
print("Invalid snapshot configuration")
494
elif e.status_code == 409:
495
print("Snapshot with this name already exists")
496
elif "SourceAgentPoolNotFound" in str(e):
497
print("Source agent pool no longer exists")
498
else:
499
print(f"Snapshot creation failed: {e}")
500
501
# Handle deletion errors
502
try:
503
client.snapshots.delete("myRG", "snapshot-to-delete")
504
except HttpResponseError as e:
505
if e.status_code == 409:
506
print("Snapshot is currently being used and cannot be deleted")
507
else:
508
print(f"Deletion failed: {e}")
509
```
510
511
## Best Practices
512
513
### Snapshot Naming Convention
514
515
```python
516
# Consistent naming for better management
517
def generate_snapshot_name(cluster_name, agent_pool, purpose, date=None):
518
"""Generate consistent snapshot names."""
519
if date is None:
520
date = datetime.date.today()
521
522
date_str = date.strftime("%Y%m%d")
523
return f"{cluster_name}-{agent_pool}-{purpose}-{date_str}"
524
525
# Examples
526
backup_name = generate_snapshot_name("prod-cluster", "nodepool1", "backup")
527
template_name = generate_snapshot_name("gold-cluster", "system", "template")
528
dr_name = generate_snapshot_name("prod-cluster", "nodepool1", "dr")
529
```
530
531
### Tagging Strategy
532
533
```python
534
# Comprehensive tagging for snapshot management
535
standard_tags = {
536
"created-by": "platform-automation",
537
"backup-schedule": "daily",
538
"retention-policy": "30-days",
539
"environment": "production",
540
"cost-center": "platform",
541
"compliance": "sox-required",
542
"notification": "platform-team@company.com"
543
}
544
545
# Apply tags based on snapshot type
546
def get_snapshot_tags(purpose, environment, cluster_name):
547
"""Get appropriate tags for snapshot."""
548
base_tags = {
549
"purpose": purpose,
550
"environment": environment,
551
"cluster": cluster_name,
552
"created-date": datetime.date.today().isoformat()
553
}
554
555
if purpose == "backup":
556
base_tags.update({
557
"retention": "30-days",
558
"automated": "true"
559
})
560
elif purpose == "template":
561
base_tags.update({
562
"retention": "permanent",
563
"template": "true"
564
})
565
elif purpose == "dr":
566
base_tags.update({
567
"retention": "90-days",
568
"disaster-recovery": "true"
569
})
570
571
return base_tags
572
```