0
# Azure Container Services
1

2
Complete Azure Container Services integration providing comprehensive container orchestration capabilities including Azure Container Instances management, Container Registry operations, and Container Volume handling for containerized workloads.
3

4
## Capabilities
5

6
### Azure Container Instances Hook
7

8
Primary interface for Azure Container Instances management, providing authenticated connections and container lifecycle operations.
9

10
```python { .api }
11
class AzureContainerInstanceHook(BaseHook):
12
    """
13
    Hook for Azure Container Instances management.
14
    
15
    Provides methods for creating, managing, and monitoring container instances
16
    with support for various container configurations and networking options.
17
    """
18
    
19
    def get_conn(self) -> ContainerInstanceManagementClient:
20
        """
21
        Get authenticated Azure Container Instance management client.
22
        
23
        Returns:
24
            ContainerInstanceManagementClient: Container management client instance
25
        """
26
    
27
    def create_or_update(
28
        self,
29
        resource_group_name: str,
30
        container_group_name: str,
31
        container_group: ContainerGroup
32
    ) -> ContainerGroup:
33
        """
34
        Create or update a container group.
35
        
36
        Args:
37
            resource_group_name (str): Name of the resource group
38
            container_group_name (str): Name of the container group
39
            container_group (ContainerGroup): Container group configuration
40
            
41
        Returns:
42
            ContainerGroup: Created or updated container group
43
        """
44
    
45
    def get_state_exitcode_details(
46
        self,
47
        resource_group_name: str,
48
        container_group_name: str,
49
        container_name: str
50
    ) -> tuple[str, int, str]:
51
        """
52
        Get container state, exit code, and detailed information.
53
        
54
        Args:
55
            resource_group_name (str): Name of the resource group
56
            container_group_name (str): Name of the container group
57
            container_name (str): Name of the container
58
            
59
        Returns:
60
            tuple[str, int, str]: State, exit code, and details
61
        """
62
    
63
    def get_logs(
64
        self,
65
        resource_group_name: str,
66
        container_group_name: str,
67
        container_name: str,
68
        tail: int | None = None
69
    ) -> str:
70
        """
71
        Get logs from a running or terminated container.
72
        
73
        Args:
74
            resource_group_name (str): Name of the resource group
75
            container_group_name (str): Name of the container group
76
            container_name (str): Name of the container
77
            tail (int | None): Number of log lines to retrieve from the end
78
            
79
        Returns:
80
            str: Container logs
81
        """
82
    
83
    def delete(
84
        self,
85
        resource_group_name: str,
86
        container_group_name: str
87
    ) -> None:
88
        """
89
        Delete a container group.
90
        
91
        Args:
92
            resource_group_name (str): Name of the resource group
93
            container_group_name (str): Name of the container group to delete
94
        """
95
    
96
    def exists(
97
        self,
98
        resource_group_name: str,
99
        container_group_name: str
100
    ) -> bool:
101
        """
102
        Check if a container group exists.
103
        
104
        Args:
105
            resource_group_name (str): Name of the resource group
106
            container_group_name (str): Name of the container group
107
            
108
        Returns:
109
            bool: True if container group exists, False otherwise
110
        """
111
    
112
    def get_state(
113
        self,
114
        resource_group_name: str,
115
        container_group_name: str,
116
        container_name: str
117
    ) -> str:
118
        """
119
        Get the current state of a container.
120
        
121
        Args:
122
            resource_group_name (str): Name of the resource group
123
            container_group_name (str): Name of the container group
124
            container_name (str): Name of the container
125
            
126
        Returns:
127
            str: Current container state (Running, Terminated, Waiting, etc.)
128
        """
129
    
130
    def get_messages(
131
        self,
132
        resource_group_name: str,
133
        container_group_name: str,
134
        container_name: str
135
    ) -> list[str]:
136
        """
137
        Get status messages from a container.
138
        
139
        Args:
140
            resource_group_name (str): Name of the resource group
141
            container_group_name (str): Name of the container group
142
            container_name (str): Name of the container
143
            
144
        Returns:
145
            list[str]: List of container status messages
146
        """
147
    
148
    def test_connection(self) -> tuple[bool, str]:
149
        """
150
        Test the Azure Container Instance connection.
151
        
152
        Returns:
153
            tuple[bool, str]: Success status and message
154
        """
155
```
156

157
### Azure Container Registry Hook
158

159
Hook for Azure Container Registry operations providing image management and credential handling capabilities.
160

161
```python { .api }
162
class AzureContainerRegistryHook(BaseHook):
163
    """
164
    Hook for Azure Container Registry operations.
165
    
166
    Provides methods for managing container registries, retrieving credentials,
167
    and handling image operations.
168
    """
169
    
170
    def get_conn(self) -> ImageRegistryCredential:
171
        """
172
        Get registry credentials for authentication.
173
        
174
        Returns:
175
            ImageRegistryCredential: Registry credentials for authentication
176
        """
177
    
178
    def get_registry_credentials(
179
        self,
180
        registry_name: str,
181
        resource_group_name: str
182
    ) -> dict[str, str]:
183
        """
184
        Get container registry admin credentials.
185
        
186
        Args:
187
            registry_name (str): Name of the container registry
188
            resource_group_name (str): Name of the resource group
189
            
190
        Returns:
191
            dict[str, str]: Registry credentials including username and passwords
192
        """
193
    
194
    def get_login_server(
195
        self,
196
        registry_name: str,
197
        resource_group_name: str
198
    ) -> str:
199
        """
200
        Get the login server URL for the container registry.
201
        
202
        Args:
203
            registry_name (str): Name of the container registry
204
            resource_group_name (str): Name of the resource group
205
            
206
        Returns:
207
            str: Login server URL
208
        """
209
    
210
    def list_repositories(
211
        self,
212
        registry_name: str,
213
        resource_group_name: str
214
    ) -> list[str]:
215
        """
216
        List all repositories in the container registry.
217
        
218
        Args:
219
            registry_name (str): Name of the container registry
220
            resource_group_name (str): Name of the resource group
221
            
222
        Returns:
223
            list[str]: List of repository names
224
        """
225
    
226
    def list_tags(
227
        self,
228
        registry_name: str,
229
        resource_group_name: str,
230
        repository_name: str
231
    ) -> list[str]:
232
        """
233
        List all tags for a specific repository.
234
        
235
        Args:
236
            registry_name (str): Name of the container registry
237
            resource_group_name (str): Name of the resource group
238
            repository_name (str): Name of the repository
239
            
240
        Returns:
241
            list[str]: List of image tags
242
        """
243
    
244
    def delete_repository(
245
        self,
246
        registry_name: str,
247
        resource_group_name: str,
248
        repository_name: str
249
    ) -> None:
250
        """
251
        Delete a repository from the container registry.
252
        
253
        Args:
254
            registry_name (str): Name of the container registry
255
            resource_group_name (str): Name of the resource group
256
            repository_name (str): Name of the repository to delete
257
        """
258
    
259
    def delete_tag(
260
        self,
261
        registry_name: str,
262
        resource_group_name: str,
263
        repository_name: str,
264
        tag: str
265
    ) -> None:
266
        """
267
        Delete a specific tag from a repository.
268
        
269
        Args:
270
            registry_name (str): Name of the container registry
271
            resource_group_name (str): Name of the resource group
272
            repository_name (str): Name of the repository
273
            tag (str): Tag to delete
274
        """
275
    
276
    def test_connection(self) -> tuple[bool, str]:
277
        """
278
        Test the Azure Container Registry connection.
279
        
280
        Returns:
281
            tuple[bool, str]: Success status and message
282
        """
283
```
284

285
### Azure Container Volume Hook
286

287
Hook for Azure Container Volume management providing persistent storage capabilities for containers.
288

289
```python { .api }
290
class AzureContainerVolumeHook(BaseHook):
291
    """
292
    Hook for Azure Container Volume management.
293
    
294
    Provides methods for creating and managing persistent volumes
295
    for Azure Container Instances with various storage backends.
296
    """
297
    
298
    def get_conn(self) -> Any:
299
        """
300
        Get authenticated connection for volume operations.
301
        
302
        Returns:
303
            Any: Volume management client
304
        """
305
    
306
    def create_file_share_volume(
307
        self,
308
        volume_name: str,
309
        share_name: str,
310
        storage_account_name: str,
311
        storage_account_key: str,
312
        read_only: bool = False
313
    ) -> Volume:
314
        """
315
        Create a volume backed by Azure File Share.
316
        
317
        Args:
318
            volume_name (str): Name of the volume
319
            share_name (str): Name of the Azure File Share
320
            storage_account_name (str): Storage account name
321
            storage_account_key (str): Storage account key
322
            read_only (bool): Whether the volume is read-only (default: False)
323
            
324
        Returns:
325
            Volume: Created volume configuration
326
        """
327
    
328
    def create_secret_volume(
329
        self,
330
        volume_name: str,
331
        secrets: dict[str, str]
332
    ) -> Volume:
333
        """
334
        Create a volume for storing secrets.
335
        
336
        Args:
337
            volume_name (str): Name of the volume
338
            secrets (dict[str, str]): Dictionary of secret files and content
339
            
340
        Returns:
341
            Volume: Created secret volume configuration
342
        """
343
    
344
    def create_empty_dir_volume(
345
        self,
346
        volume_name: str
347
    ) -> Volume:
348
        """
349
        Create an empty directory volume.
350
        
351
        Args:
352
            volume_name (str): Name of the volume
353
            
354
        Returns:
355
            Volume: Created empty directory volume configuration
356
        """
357
    
358
    def create_git_repo_volume(
359
        self,
360
        volume_name: str,
361
        repository_url: str,
362
        directory_name: str | None = None,
363
        revision: str | None = None
364
    ) -> Volume:
365
        """
366
        Create a volume from a Git repository.
367
        
368
        Args:
369
            volume_name (str): Name of the volume
370
            repository_url (str): URL of the Git repository
371
            directory_name (str | None): Directory to clone into
372
            revision (str | None): Specific revision to clone
373
            
374
        Returns:
375
            Volume: Created Git repository volume configuration
376
        """
377
    
378
    def test_connection(self) -> tuple[bool, str]:
379
        """
380
        Test the Azure Container Volume connection.
381
        
382
        Returns:
383
            tuple[bool, str]: Success status and message
384
        """
385
```
386

387
## Container Services Operators
388

389
Execute Azure Container Services operations as Airflow tasks with comprehensive container management capabilities.
390

391
### Azure Container Instances Operator
392

393
```python { .api }
394
class AzureContainerInstancesOperator(BaseOperator):
395
    """
396
    Runs containers on Azure Container Instances.
397
    
398
    Supports creating and managing containerized workloads with custom
399
    configurations, networking, and volume mounting options.
400
    """
401
    
402
    def __init__(
403
        self,
404
        *,
405
        ci_conn_id: str = "azure_container_instance_default",
406
        registry_conn_id: str | None = None,
407
        resource_group: str,
408
        name: str,
409
        image: str,
410
        region: str = "West US",
411
        environment_variables: dict[str, str] | None = None,
412
        secured_variables: list[str] | None = None,
413
        volumes: list[Volume] | None = None,
414
        volume_mounts: list[VolumeMount] | None = None,
415
        memory_in_gb: float = 1.5,
416
        cpu: float = 1.0,
417
        gpu: GpuResource | None = None,
418
        command: list[str] | None = None,
419
        restart_policy: str = "Never",
420
        os_type: str = "Linux",
421
        ports: list[ContainerPort] | None = None,
422
        dns_name_label: str | None = None,
423
        ip_address_type: str = "Public",
424
        network_profile: ContainerGroupNetworkProfile | None = None,
425
        tags: dict[str, str] | None = None,
426
        **kwargs
427
    ):
428
        """
429
        Initialize Azure Container Instances operator.
430
        
431
        Args:
432
            ci_conn_id (str): Airflow connection ID for Container Instances
433
            registry_conn_id (str | None): Connection ID for container registry
434
            resource_group (str): Azure resource group name
435
            name (str): Container group name
436
            image (str): Container image to run
437
            region (str): Azure region (default: "West US")
438
            environment_variables (dict[str, str] | None): Environment variables
439
            secured_variables (list[str] | None): Secured environment variables
440
            volumes (list[Volume] | None): Volumes to mount
441
            volume_mounts (list[VolumeMount] | None): Volume mount configurations
442
            memory_in_gb (float): Memory allocation in GB (default: 1.5)
443
            cpu (float): CPU allocation (default: 1.0)
444
            gpu (GpuResource | None): GPU resource configuration
445
            command (list[str] | None): Command to run in container
446
            restart_policy (str): Restart policy (default: "Never")
447
            os_type (str): Operating system type (default: "Linux")
448
            ports (list[ContainerPort] | None): Port configurations
449
            dns_name_label (str | None): DNS name label for public IP
450
            ip_address_type (str): IP address type (default: "Public")
451
            network_profile (ContainerGroupNetworkProfile | None): Network profile
452
            tags (dict[str, str] | None): Resource tags
453
        """
454
    
455
    def execute(self, context: Context) -> str:
456
        """
457
        Execute container instance creation and management.
458
        
459
        Args:
460
            context (Context): Airflow task context
461
            
462
        Returns:
463
            str: Container execution results or status
464
        """
465
    
466
    def on_kill(self) -> None:
467
        """Clean up container resources on task termination."""
468
```
469

470
## Usage Examples
471

472
### Basic Container Execution
473

474
```python
475
from airflow import DAG
476
from airflow.providers.microsoft.azure.operators.container_instances import AzureContainerInstancesOperator
477
from airflow.operators.python import PythonOperator
478
from datetime import datetime, timedelta
479

480
def check_container_results(**context):
481
    """Check and process container execution results."""
482
    result = context['task_instance'].xcom_pull(task_ids='run_data_processing')
483
    print(f"Container execution result: {result}")
484

485
dag = DAG(
486
    'azure_container_workflow',
487
    default_args={
488
        'owner': 'container-team',
489
        'retries': 2,
490
        'retry_delay': timedelta(minutes=5)
491
    },
492
    description='Azure Container Instances workflow',
493
    schedule_interval=timedelta(hours=6),
494
    start_date=datetime(2024, 1, 1),
495
    catchup=False
496
)
497

498
# Run data processing container
499
run_container = AzureContainerInstancesOperator(
500
    task_id='run_data_processing',
501
    ci_conn_id='azure_container_conn',
502
    resource_group='data-processing-rg',
503
    name='data-processor-{{ ds_nodash }}',
504
    image='myregistry.azurecr.io/data-processor:latest',
505
    region='East US',
506
    memory_in_gb=4.0,
507
    cpu=2.0,
508
    environment_variables={
509
        'INPUT_PATH': '/data/input',
510
        'OUTPUT_PATH': '/data/output',
511
        'PROCESSING_MODE': 'batch'
512
    },
513
    command=[
514
        'python',
515
        'process_data.py',
516
        '--input', '/data/input',
517
        '--output', '/data/output'
518
    ],
519
    restart_policy='Never',
520
    dag=dag
521
)
522

523
# Process results
524
check_results = PythonOperator(
525
    task_id='check_results',
526
    python_callable=check_container_results,
527
    dag=dag
528
)
529

530
run_container >> check_results
531
```
532

533
### Advanced Container Configuration
534

535
```python
536
from airflow import DAG
537
from airflow.providers.microsoft.azure.operators.container_instances import AzureContainerInstancesOperator
538
from airflow.providers.microsoft.azure.hooks.container_volume import AzureContainerVolumeHook
539
from azure.mgmt.containerinstance.models import (
540
    Volume, VolumeMount, ContainerPort, GpuResource,
541
    ContainerGroupNetworkProfile, ResourceRequests, ResourceRequirements
542
)
543
from datetime import datetime, timedelta
544

545
def setup_volumes():
546
    """Set up volumes for container workload."""
547
    volume_hook = AzureContainerVolumeHook(azure_container_volume_conn_id='volume_conn')
548
    
549
    # Create file share volume for persistent storage
550
    data_volume = volume_hook.create_file_share_volume(
551
        volume_name='data-volume',
552
        share_name='container-data',
553
        storage_account_name='mystorageaccount',
554
        storage_account_key='storage-key-here',
555
        read_only=False
556
    )
557
    
558
    # Create secret volume for configuration
559
    config_volume = volume_hook.create_secret_volume(
560
        volume_name='config-volume',
561
        secrets={
562
            'config.json': '{"database": "prod", "debug": false}',
563
            'secrets.env': 'DB_PASSWORD=secret123\nAPI_KEY=key456'
564
        }
565
    )
566
    
567
    return [data_volume, config_volume]
568

569
dag = DAG(
570
    'advanced_container_workflow',
571
    default_args={
572
        'owner': 'ml-team',
573
        'retries': 1,
574
        'retry_delay': timedelta(minutes=3)
575
    },
576
    description='Advanced container workflow with GPU and volumes',
577
    schedule_interval=timedelta(days=1),
578
    start_date=datetime(2024, 1, 1),
579
    catchup=False
580
)
581

582
# Configure volumes
583
volumes = [
584
    Volume(
585
        name='data-volume',
586
        azure_file={
587
            'share_name': 'ml-data',
588
            'storage_account_name': 'mlstorageaccount',
589
            'storage_account_key': 'storage-key'
590
        }
591
    ),
592
    Volume(
593
        name='model-volume',
594
        azure_file={
595
            'share_name': 'ml-models', 
596
            'storage_account_name': 'mlstorageaccount',
597
            'storage_account_key': 'storage-key',
598
            'read_only': True
599
        }
600
    )
601
]
602

603
# Configure volume mounts
604
volume_mounts = [
605
    VolumeMount(
606
        name='data-volume',
607
        mount_path='/data',
608
        read_only=False
609
    ),
610
    VolumeMount(
611
        name='model-volume',
612
        mount_path='/models',
613
        read_only=True
614
    )
615
]
616

617
# Configure GPU resources for ML workload
618
gpu_resource = GpuResource(
619
    count=1,
620
    sku='K80'
621
)
622

623
# Configure network ports
624
ports = [
625
    ContainerPort(port=8080, protocol='TCP'),
626
    ContainerPort(port=8443, protocol='TCP')
627
]
628

629
# Run ML training container with GPU
630
ml_training = AzureContainerInstancesOperator(
631
    task_id='run_ml_training',
632
    ci_conn_id='azure_container_conn',
633
    registry_conn_id='azure_container_registry_conn',
634
    resource_group='ml-training-rg',
635
    name='ml-trainer-{{ ds_nodash }}',
636
    image='mlregistry.azurecr.io/trainer:v2.1',
637
    region='East US',
638
    memory_in_gb=16.0,
639
    cpu=4.0,
640
    gpu=gpu_resource,
641
    volumes=volumes,
642
    volume_mounts=volume_mounts,
643
    ports=ports,
644
    environment_variables={
645
        'DATASET_PATH': '/data/training_set.csv',
646
        'MODEL_OUTPUT_PATH': '/data/models',
647
        'EPOCHS': '100',
648
        'BATCH_SIZE': '32',
649
        'LEARNING_RATE': '0.001'
650
    },
651
    secured_variables=['DB_PASSWORD', 'API_SECRET_KEY'],
652
    command=[
653
        'python',
654
        'train_model.py',
655
        '--config', '/config/training_config.json',
656
        '--data-path', '/data',
657
        '--model-path', '/models',
658
        '--gpu'
659
    ],
660
    restart_policy='Never',
661
    dns_name_label='ml-trainer-{{ ds_nodash }}',
662
    tags={
663
        'project': 'machine-learning',
664
        'environment': 'production',
665
        'cost-center': 'research'
666
    },
667
    dag=dag
668
)
669

670
# Run inference container
671
ml_inference = AzureContainerInstancesOperator(
672
    task_id='run_ml_inference',
673
    ci_conn_id='azure_container_conn',
674
    registry_conn_id='azure_container_registry_conn',
675
    resource_group='ml-inference-rg',
676
    name='ml-inference-{{ ds_nodash }}',
677
    image='mlregistry.azurecr.io/inference:v2.1',
678
    region='East US',
679
    memory_in_gb=8.0,
680
    cpu=2.0,
681
    volumes=[volumes[0]],  # Only data volume needed
682
    volume_mounts=[volume_mounts[0]],
683
    environment_variables={
684
        'MODEL_PATH': '/data/models/latest_model.pkl',
685
        'INPUT_PATH': '/data/inference_input.csv',
686
        'OUTPUT_PATH': '/data/inference_output.csv'
687
    },
688
    command=[
689
        'python',
690
        'run_inference.py',
691
        '--model', '/data/models/latest_model.pkl',
692
        '--input', '/data/inference_input.csv',
693
        '--output', '/data/inference_output.csv'
694
    ],
695
    restart_policy='Never',
696
    dag=dag
697
)
698

699
ml_training >> ml_inference
700
```
701

702
### Container Registry Integration
703

704
```python
705
from airflow import DAG
706
from airflow.providers.microsoft.azure.hooks.container_registry import AzureContainerRegistryHook
707
from airflow.operators.python import PythonOperator
708
from datetime import datetime, timedelta
709

710
def manage_container_images():
711
    """Manage container registry images."""
712
    registry_hook = AzureContainerRegistryHook(
713
        azure_container_registry_conn_id='registry_conn'
714
    )
715
    
716
    registry_name = 'myregistry'
717
    resource_group = 'container-rg'
718
    
719
    # Get registry credentials
720
    credentials = registry_hook.get_registry_credentials(
721
        registry_name=registry_name,
722
        resource_group_name=resource_group
723
    )
724
    print(f"Registry username: {credentials['username']}")
725
    
726
    # Get login server
727
    login_server = registry_hook.get_login_server(
728
        registry_name=registry_name,
729
        resource_group_name=resource_group
730
    )
731
    print(f"Login server: {login_server}")
732
    
733
    # List repositories
734
    repositories = registry_hook.list_repositories(
735
        registry_name=registry_name,
736
        resource_group_name=resource_group
737
    )
738
    print(f"Found repositories: {repositories}")
739
    
740
    # List tags for each repository
741
    for repo in repositories:
742
        tags = registry_hook.list_tags(
743
            registry_name=registry_name,
744
            resource_group_name=resource_group,
745
            repository_name=repo
746
        )
747
        print(f"Repository {repo} tags: {tags}")
748
        
749
        # Clean up old tags (keep only latest 5)
750
        if len(tags) > 5:
751
            old_tags = sorted(tags)[:-5]  # Keep latest 5
752
            for old_tag in old_tags:
753
                try:
754
                    registry_hook.delete_tag(
755
                        registry_name=registry_name,
756
                        resource_group_name=resource_group,
757
                        repository_name=repo,
758
                        tag=old_tag
759
                    )
760
                    print(f"Deleted old tag: {repo}:{old_tag}")
761
                except Exception as e:
762
                    print(f"Failed to delete tag {old_tag}: {e}")
763

764
def monitor_container_execution():
765
    """Monitor running container instances."""
766
    from airflow.providers.microsoft.azure.hooks.container_instance import AzureContainerInstanceHook
767
    
768
    aci_hook = AzureContainerInstanceHook(azure_container_instance_conn_id='aci_conn')
769
    
770
    resource_group = 'container-rg'
771
    container_group = 'data-processor'
772
    container_name = 'processor'
773
    
774
    # Check if container exists
775
    if aci_hook.exists(resource_group, container_group):
776
        # Get container state
777
        state = aci_hook.get_state(resource_group, container_group, container_name)
778
        print(f"Container state: {state}")
779
        
780
        # Get logs if container is running or terminated
781
        if state in ['Running', 'Terminated']:
782
            logs = aci_hook.get_logs(
783
                resource_group,
784
                container_group,
785
                container_name,
786
                tail=100
787
            )
788
            print(f"Container logs:\n{logs}")
789
        
790
        # Get detailed state information
791
        state, exit_code, details = aci_hook.get_state_exitcode_details(
792
            resource_group,
793
            container_group,
794
            container_name
795
        )
796
        print(f"State: {state}, Exit Code: {exit_code}, Details: {details}")
797
        
798
        # Get status messages
799
        messages = aci_hook.get_messages(resource_group, container_group, container_name)
800
        for message in messages:
801
            print(f"Status message: {message}")
802

803
dag = DAG(
804
    'container_registry_management',
805
    default_args={
806
        'owner': 'devops-team',
807
        'retries': 1,
808
        'retry_delay': timedelta(minutes=2)
809
    },
810
    description='Container registry and instance management',
811
    schedule_interval=timedelta(hours=12),
812
    start_date=datetime(2024, 1, 1),
813
    catchup=False
814
)
815

816
manage_images = PythonOperator(
817
    task_id='manage_registry_images',
818
    python_callable=manage_container_images,
819
    dag=dag
820
)
821

822
monitor_containers = PythonOperator(
823
    task_id='monitor_containers',
824
    python_callable=monitor_container_execution,
825
    dag=dag
826
)
827

828
manage_images >> monitor_containers
829
```
830

831
## Connection Configuration
832

833
### Container Instances Connection (`azure_container_instance`)
834

835
Configure Azure Container Instances connections in Airflow:
836

837
```python
838
# Connection configuration for Container Instances
839
{
840
    "conn_id": "azure_container_instance_default",
841
    "conn_type": "azure_container_instance",
842
    "extra": {
843
        "subscriptionId": "your-subscription-id",
844
        "tenantId": "your-tenant-id",
845
        "clientId": "your-client-id",
846
        "clientSecret": "your-client-secret"
847
    }
848
}
849
```
850

851
### Container Registry Connection (`azure_container_registry`)
852

853
Configure Azure Container Registry connections in Airflow:
854

855
```python
856
# Connection configuration for Container Registry
857
{
858
    "conn_id": "azure_container_registry_default",
859
    "conn_type": "azure_container_registry",
860
    "login": "myregistry",  # Registry name
861
    "host": "myregistry.azurecr.io",  # Registry URL
862
    "extra": {
863
        "subscriptionId": "your-subscription-id",
864
        "resourceGroupName": "your-resource-group",
865
        "tenantId": "your-tenant-id",
866
        "clientId": "your-client-id",
867
        "clientSecret": "your-client-secret"
868
    }
869
}
870
```
871

872
### Container Volume Connection (`azure_container_volume`)
873

874
Configure Azure Container Volume connections in Airflow:
875

876
```python
877
# Connection configuration for Container Volume
878
{
879
    "conn_id": "azure_container_volume_default",
880
    "conn_type": "azure_container_volume",
881
    "extra": {
882
        "tenantId": "your-tenant-id",
883
        "clientId": "your-client-id",
884
        "clientSecret": "your-client-secret"
885
    }
886
}
887
```
888

889
### Authentication Methods
890

891
All container services support multiple authentication methods:
892

893
1. **Service Principal Authentication**:
894
   ```python
895
   extra = {
896
       "tenantId": "your-tenant-id",
897
       "clientId": "your-client-id",
898
       "clientSecret": "your-client-secret"
899
   }
900
   ```
901

902
2. **Managed Identity Authentication**:
903
   ```python
904
   extra = {
905
       "managedIdentityClientId": "your-managed-identity-client-id"
906
   }
907
   ```
908

909
3. **Azure CLI Authentication**:
910
   ```python
911
   extra = {
912
       "use_azure_cli": True
913
   }
914
   ```
915

916
## Error Handling
917

918
### Common Exception Patterns
919

920
```python
921
from azure.core.exceptions import ResourceNotFoundError, ResourceExistsError
922
from azure.mgmt.containerinstance.models import ContainerGroup
923
from airflow.providers.microsoft.azure.hooks.container_instance import AzureContainerInstanceHook
924

925
def robust_container_operations():
926
    """Demonstrate error handling patterns for container operations."""
927
    hook = AzureContainerInstanceHook(azure_container_instance_conn_id='aci_conn')
928
    
929
    resource_group = 'my-rg'
930
    container_group_name = 'my-container-group'
931
    
932
    try:
933
        # Check if container group exists before creating
934
        if hook.exists(resource_group, container_group_name):
935
            print("Container group already exists, deleting first...")
936
            hook.delete(resource_group, container_group_name)
937
            # Wait for deletion to complete
938
            import time
939
            time.sleep(30)
940
        
941
        # Create container group
942
        container_group = ContainerGroup(
943
            location='East US',
944
            containers=[],
945
            os_type='Linux'
946
        )
947
        
948
        result = hook.create_or_update(
949
            resource_group_name=resource_group,
950
            container_group_name=container_group_name,
951
            container_group=container_group
952
        )
953
        print(f"Container group created: {result.name}")
954
        
955
    except ResourceNotFoundError as e:
956
        print(f"Resource not found: {e}")
957
        # Handle missing resource group or other dependencies
958
        
959
    except ResourceExistsError as e:
960
        print(f"Resource already exists: {e}")
961
        # Handle existing resources
962
        
963
    except Exception as e:
964
        print(f"Unexpected error: {e}")
965
        # Cleanup on failure
966
        try:
967
            hook.delete(resource_group, container_group_name)
968
        except:
969
            pass  # Ignore cleanup errors
970
        raise
971
```
972

973
### Connection Testing
974

975
```python
976
def test_container_connections():
977
    """Test all container service connections."""
978
    
979
    # Test Container Instances connection
980
    try:
981
        aci_hook = AzureContainerInstanceHook(azure_container_instance_conn_id='aci_conn')
982
        success, message = aci_hook.test_connection()
983
        print(f"Container Instances: {message}")
984
    except Exception as e:
985
        print(f"Container Instances connection failed: {e}")
986
    
987
    # Test Container Registry connection
988
    try:
989
        acr_hook = AzureContainerRegistryHook(azure_container_registry_conn_id='acr_conn')
990
        success, message = acr_hook.test_connection()
991
        print(f"Container Registry: {message}")
992
    except Exception as e:
993
        print(f"Container Registry connection failed: {e}")
994
    
995
    # Test Container Volume connection
996
    try:
997
        volume_hook = AzureContainerVolumeHook(azure_container_volume_conn_id='volume_conn')
998
        success, message = volume_hook.test_connection()
999
        print(f"Container Volume: {message}")
1000
    except Exception as e:
1001
        print(f"Container Volume connection failed: {e}")
1002
```
1003

1004
## Performance Considerations
1005

1006
### Optimizing Container Workloads
1007

1008
```python
1009
def optimize_container_performance():
1010
    """Demonstrate performance optimization techniques."""
1011
    
1012
    # Use appropriate resource sizing
1013
    memory_optimized_config = {
1014
        'memory_in_gb': 8.0,  # Adequate memory for workload
1015
        'cpu': 2.0,           # Balanced CPU allocation
1016
    }
1017
    
1018
    # Configure restart policies appropriately
1019
    batch_job_config = {
1020
        'restart_policy': 'Never',  # For batch jobs
1021
    }
1022
    
1023
    long_running_config = {
1024
        'restart_policy': 'Always',  # For services
1025
    }
1026
    
1027
    # Use local SSD storage for temporary files
1028
    volume_config = [
1029
        Volume(
1030
            name='temp-volume',
1031
            empty_dir={}  # Uses local SSD storage
1032
        )
1033
    ]
1034
    
1035
    # Pre-pull images for faster startup
1036
    # This would be done in the registry/image management process
1037
    
1038
    return {
1039
        'optimized_resources': memory_optimized_config,
1040
        'batch_config': batch_job_config,
1041
        'service_config': long_running_config,
1042
        'volume_config': volume_config
1043
    }
1044

1045
def implement_container_monitoring():
1046
    """Implement comprehensive container monitoring."""
1047
    hook = AzureContainerInstanceHook(azure_container_instance_conn_id='aci_conn')
1048
    
1049
    def monitor_container_health(resource_group: str, container_group: str, container_name: str):
1050
        """Monitor container health and performance."""
1051
        
1052
        # Get current state
1053
        state = hook.get_state(resource_group, container_group, container_name)
1054
        
1055
        # Get recent logs for error detection
1056
        logs = hook.get_logs(resource_group, container_group, container_name, tail=50)
1057
        
1058
        # Check for error patterns in logs
1059
        error_keywords = ['ERROR', 'FATAL', 'Exception', 'Failed']
1060
        errors_found = []
1061
        
1062
        for line in logs.split('\n'):
1063
            for keyword in error_keywords:
1064
                if keyword in line:
1065
                    errors_found.append(line.strip())
1066
        
1067
        # Get detailed status information
1068
        state, exit_code, details = hook.get_state_exitcode_details(
1069
            resource_group, container_group, container_name
1070
        )
1071
        
1072
        health_status = {
1073
            'state': state,
1074
            'exit_code': exit_code,
1075
            'details': details,
1076
            'errors_found': errors_found,
1077
            'healthy': state == 'Running' and len(errors_found) == 0
1078
        }
1079
        
1080
        return health_status
1081
    
1082
    return monitor_container_health
1083
```
1084

1085
This comprehensive documentation covers all Azure Container Services capabilities in the Apache Airflow Microsoft Azure Provider, including Container Instances, Container Registry, and Container Volume management with practical usage patterns and optimization techniques.