Provider Abstraction Patterns
Creating truly portable infrastructure requires abstraction layers that hide provider-specific differences while exposing common functionality. This part covers patterns for building cloud-agnostic modules that can deploy the same logical infrastructure across AWS, Azure, and GCP.
Universal Compute Module
Create a compute module that works across all providers:
# modules/universal-compute/variables.tf
variable "provider_type" {
description = "Cloud provider (aws, azure, gcp)"
type = string
validation {
condition = contains(["aws", "azure", "gcp"], var.provider_type)
error_message = "Provider must be aws, azure, or gcp."
}
}
variable "instance_config" {
description = "Instance configuration"
type = object({
name = string
size = string
image = string
subnet_id = string
key_name = optional(string)
user_data = optional(string)
tags = optional(map(string), {})
})
}
variable "gcp_region" {
description = "GCP region (required when provider_type is gcp)"
type = string
default = ""
}
variable "gcp_project_id" {
description = "GCP project ID (required when provider_type is gcp)"
type = string
default = ""
}
# modules/universal-compute/main.tf
locals {
# Size mapping across providers
size_mapping = {
aws = {
small = "t3.micro"
medium = "t3.small"
large = "t3.medium"
}
azure = {
small = "Standard_B1s"
medium = "Standard_B2s"
large = "Standard_B4ms"
}
gcp = {
small = "e2-micro"
medium = "e2-small"
large = "e2-medium"
}
}
actual_size = local.size_mapping[var.provider_type][var.instance_config.size]
}
# AWS EC2 Instance
resource "aws_instance" "this" {
count = var.provider_type == "aws" ? 1 : 0
ami = var.instance_config.image
instance_type = local.actual_size
subnet_id = var.instance_config.subnet_id
key_name = var.instance_config.key_name
user_data = var.instance_config.user_data
tags = merge(var.instance_config.tags, {
Name = var.instance_config.name
})
}
# Azure Virtual Machine
resource "azurerm_network_interface" "this" {
count = var.provider_type == "azure" ? 1 : 0
name = "${var.instance_config.name}-nic"
location = data.azurerm_subnet.this[0].location
resource_group_name = data.azurerm_subnet.this[0].resource_group_name
ip_configuration {
name = "internal"
subnet_id = var.instance_config.subnet_id
private_ip_address_allocation = "Dynamic"
}
}
resource "azurerm_linux_virtual_machine" "this" {
count = var.provider_type == "azure" ? 1 : 0
name = var.instance_config.name
resource_group_name = data.azurerm_subnet.this[0].resource_group_name
location = data.azurerm_subnet.this[0].location
size = local.actual_size
disable_password_authentication = true
admin_username = "adminuser"
network_interface_ids = [
azurerm_network_interface.this[0].id,
]
os_disk {
caching = "ReadWrite"
storage_account_type = "Standard_LRS"
}
source_image_reference {
publisher = "Canonical"
offer = "0001-com-ubuntu-server-focal"
sku = "20_04-lts-gen2"
version = "latest"
}
admin_ssh_key {
username = "adminuser"
public_key = file("~/.ssh/id_rsa.pub")
}
custom_data = base64encode(var.instance_config.user_data)
tags = var.instance_config.tags
}
# GCP Compute Instance
resource "google_compute_instance" "this" {
count = var.provider_type == "gcp" ? 1 : 0
name = var.instance_config.name
machine_type = local.actual_size
zone = "${data.google_compute_subnetwork.this[0].region}-a" # Use first zone in region
boot_disk {
initialize_params {
image = var.instance_config.image
}
}
network_interface {
subnetwork = var.instance_config.subnet_id
}
metadata = {
ssh-keys = "adminuser:${file("~/.ssh/id_rsa.pub")}"
}
metadata_startup_script = var.instance_config.user_data
labels = var.instance_config.tags
}
# Data sources for provider-specific information
data "azurerm_subnet" "this" {
count = var.provider_type == "azure" ? 1 : 0
name = split("/", var.instance_config.subnet_id)[10] # Extract subnet name from resource ID
virtual_network_name = split("/", var.instance_config.subnet_id)[8] # Extract VNet name
resource_group_name = split("/", var.instance_config.subnet_id)[4] # Extract RG name
}
data "google_compute_subnetwork" "this" {
count = var.provider_type == "gcp" ? 1 : 0
name = var.instance_config.subnet_id
region = var.gcp_region
project = var.gcp_project_id
}
# modules/universal-compute/outputs.tf
output "instance_id" {
description = "Instance ID"
value = var.provider_type == "aws" ? aws_instance.this[0].id : (
var.provider_type == "azure" ? azurerm_linux_virtual_machine.this[0].id :
google_compute_instance.this[0].id
)
}
output "private_ip" {
description = "Private IP address"
value = var.provider_type == "aws" ? aws_instance.this[0].private_ip : (
var.provider_type == "azure" ? azurerm_linux_virtual_machine.this[0].private_ip_address :
google_compute_instance.this[0].network_interface[0].network_ip
)
}
Universal Storage Module
Create storage that works across providers:
# modules/universal-storage/variables.tf
variable "provider_type" {
description = "Cloud provider"
type = string
}
variable "bucket_config" {
description = "Storage bucket configuration"
type = object({
name = string
versioning_enabled = optional(bool, false)
encryption_enabled = optional(bool, true)
public_access = optional(bool, false)
lifecycle_rules = optional(list(object({
days = number
action = string
})), [])
tags = optional(map(string), {})
})
}
variable "resource_group_name" {
description = "Azure resource group name (required when provider_type is azure)"
type = string
default = ""
}
variable "location" {
description = "Azure location (required when provider_type is azure)"
type = string
default = ""
}
variable "gcp_region" {
description = "GCP region (required when provider_type is gcp)"
type = string
default = ""
}
# modules/universal-storage/main.tf
# AWS S3 Bucket
resource "aws_s3_bucket" "this" {
count = var.provider_type == "aws" ? 1 : 0
bucket = var.bucket_config.name
tags = var.bucket_config.tags
}
resource "aws_s3_bucket_versioning" "this" {
count = var.provider_type == "aws" && var.bucket_config.versioning_enabled ? 1 : 0
bucket = aws_s3_bucket.this[0].id
versioning_configuration {
status = "Enabled"
}
}
resource "aws_s3_bucket_server_side_encryption_configuration" "this" {
count = var.provider_type == "aws" && var.bucket_config.encryption_enabled ? 1 : 0
bucket = aws_s3_bucket.this[0].id
rule {
apply_server_side_encryption_by_default {
sse_algorithm = "AES256"
}
}
}
resource "aws_s3_bucket_lifecycle_configuration" "this" {
count = var.provider_type == "aws" && length(var.bucket_config.lifecycle_rules) > 0 ? 1 : 0
bucket = aws_s3_bucket.this[0].id
dynamic "rule" {
for_each = var.bucket_config.lifecycle_rules
content {
id = "rule-${rule.key}"
status = "Enabled"
expiration {
days = rule.value.days
}
}
}
}
# Azure Storage Account
resource "azurerm_storage_account" "this" {
count = var.provider_type == "azure" ? 1 : 0
name = replace(var.bucket_config.name, "-", "")
resource_group_name = var.resource_group_name
location = var.location
account_tier = "Standard"
account_replication_type = "LRS"
blob_properties {
versioning_enabled = var.bucket_config.versioning_enabled
}
tags = var.bucket_config.tags
}
resource "azurerm_storage_container" "this" {
count = var.provider_type == "azure" ? 1 : 0
name = "data"
storage_account_name = azurerm_storage_account.this[0].name
container_access_type = var.bucket_config.public_access ? "blob" : "private"
}
# GCP Storage Bucket
resource "google_storage_bucket" "this" {
count = var.provider_type == "gcp" ? 1 : 0
name = var.bucket_config.name
location = var.gcp_region
versioning {
enabled = var.bucket_config.versioning_enabled
}
dynamic "lifecycle_rule" {
for_each = var.bucket_config.lifecycle_rules
content {
condition {
age = lifecycle_rule.value.days
}
action {
type = lifecycle_rule.value.action == "delete" ? "Delete" : "SetStorageClass"
}
}
}
labels = var.bucket_config.tags
}
Configuration Factory Pattern
Generate provider-specific configurations from common definitions:
#!/usr/bin/env python3
# scripts/config_factory.py
import json
import yaml
from typing import Dict, Any, List
from pathlib import Path
class MultiCloudConfigFactory:
def __init__(self):
self.provider_mappings = {
'compute': {
'aws': self._generate_aws_compute,
'azure': self._generate_azure_compute,
'gcp': self._generate_gcp_compute
},
'storage': {
'aws': self._generate_aws_storage,
'azure': self._generate_azure_storage,
'gcp': self._generate_gcp_storage
},
'network': {
'aws': self._generate_aws_network,
'azure': self._generate_azure_network,
'gcp': self._generate_gcp_network
}
}
def generate_configs(self, spec_file: str, output_dir: str) -> Dict[str, str]:
"""Generate provider-specific configs from universal spec"""
with open(spec_file, 'r') as f:
spec = yaml.safe_load(f)
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)
generated_files = {}
for provider in spec.get('providers', []):
provider_name = provider['name']
provider_config = {
'terraform': {
'required_providers': {
provider_name: provider.get('version_constraint', {})
}
},
'provider': {
provider_name: provider.get('config', {})
}
}
# Generate resources for each service
for service_name, service_config in spec.get('services', {}).items():
if service_name in self.provider_mappings:
generator = self.provider_mappings[service_name].get(provider_name)
if generator:
resources = generator(service_config)
provider_config.update(resources)
# Write provider-specific configuration
config_file = output_path / f"{provider_name}.tf.json"
with open(config_file, 'w') as f:
json.dump(provider_config, f, indent=2)
generated_files[provider_name] = str(config_file)
return generated_files
def _generate_aws_compute(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate AWS compute resources"""
resources = {'resource': {'aws_instance': {}}}
for instance_name, instance_config in config.get('instances', {}).items():
resources['resource']['aws_instance'][instance_name] = {
'ami': instance_config['image'],
'instance_type': self._map_instance_size('aws', instance_config['size']),
'subnet_id': instance_config['subnet_id'],
'tags': instance_config.get('tags', {})
}
if 'user_data' in instance_config:
resources['resource']['aws_instance'][instance_name]['user_data'] = instance_config['user_data']
return resources
def _generate_azure_compute(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate Azure compute resources"""
resources = {
'resource': {
'azurerm_linux_virtual_machine': {},
'azurerm_network_interface': {}
}
}
for instance_name, instance_config in config.get('instances', {}).items():
# Network interface
resources['resource']['azurerm_network_interface'][f"{instance_name}_nic"] = {
'name': f"{instance_name}-nic",
'location': '${var.location}',
'resource_group_name': '${var.resource_group_name}',
'ip_configuration': [{
'name': 'internal',
'subnet_id': instance_config['subnet_id'],
'private_ip_address_allocation': 'Dynamic'
}]
}
# Virtual machine
resources['resource']['azurerm_linux_virtual_machine'][instance_name] = {
'name': instance_name,
'resource_group_name': '${var.resource_group_name}',
'location': '${var.location}',
'size': self._map_instance_size('azure', instance_config['size']),
'disable_password_authentication': True,
'network_interface_ids': [f"${{azurerm_network_interface.{instance_name}_nic.id}}"],
'os_disk': [{
'caching': 'ReadWrite',
'storage_account_type': 'Standard_LRS'
}],
'source_image_reference': [{
'publisher': 'Canonical',
'offer': '0001-com-ubuntu-server-focal',
'sku': '20_04-lts-gen2',
'version': 'latest'
}],
'tags': instance_config.get('tags', {})
}
return resources
def _generate_gcp_compute(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate GCP compute resources"""
resources = {'resource': {'google_compute_instance': {}}}
for instance_name, instance_config in config.get('instances', {}).items():
resources['resource']['google_compute_instance'][instance_name] = {
'name': instance_name,
'machine_type': self._map_instance_size('gcp', instance_config['size']),
'zone': '${var.zone}',
'boot_disk': [{
'initialize_params': [{
'image': instance_config['image']
}]
}],
'network_interface': [{
'subnetwork': instance_config['subnet_id']
}],
'labels': instance_config.get('tags', {})
}
return resources
def _generate_aws_storage(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate AWS storage resources"""
resources = {'resource': {'aws_s3_bucket': {}}}
for bucket_name, bucket_config in config.get('buckets', {}).items():
resources['resource']['aws_s3_bucket'][bucket_name] = {
'bucket': bucket_config['name'],
'tags': bucket_config.get('tags', {})
}
return resources
def _generate_azure_storage(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate Azure storage resources"""
resources = {'resource': {'azurerm_storage_account': {}}}
for bucket_name, bucket_config in config.get('buckets', {}).items():
resources['resource']['azurerm_storage_account'][bucket_name] = {
'name': bucket_config['name'].replace('-', ''),
'resource_group_name': '${var.resource_group_name}',
'location': '${var.location}',
'account_tier': 'Standard',
'account_replication_type': 'LRS',
'tags': bucket_config.get('tags', {})
}
return resources
def _generate_gcp_storage(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate GCP storage resources"""
resources = {'resource': {'google_storage_bucket': {}}}
for bucket_name, bucket_config in config.get('buckets', {}).items():
resources['resource']['google_storage_bucket'][bucket_name] = {
'name': bucket_config['name'],
'location': '${var.region}',
'labels': bucket_config.get('tags', {})
}
return resources
def _generate_aws_network(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate AWS network resources"""
resources = {
'resource': {
'aws_vpc': {},
'aws_subnet': {}
}
}
for vpc_name, vpc_config in config.get('vpcs', {}).items():
resources['resource']['aws_vpc'][vpc_name] = {
'cidr_block': vpc_config['cidr'],
'enable_dns_hostnames': True,
'enable_dns_support': True,
'tags': vpc_config.get('tags', {})
}
for subnet_name, subnet_config in vpc_config.get('subnets', {}).items():
resources['resource']['aws_subnet'][subnet_name] = {
'vpc_id': f"${{aws_vpc.{vpc_name}.id}}",
'cidr_block': subnet_config['cidr'],
'availability_zone': subnet_config.get('az', '${data.aws_availability_zones.available.names[0]}'),
'tags': subnet_config.get('tags', {})
}
return resources
def _generate_azure_network(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate Azure network resources"""
resources = {
'resource': {
'azurerm_virtual_network': {},
'azurerm_subnet': {}
}
}
for vpc_name, vpc_config in config.get('vpcs', {}).items():
resources['resource']['azurerm_virtual_network'][vpc_name] = {
'name': vpc_name,
'address_space': [vpc_config['cidr']],
'location': '${var.location}',
'resource_group_name': '${var.resource_group_name}',
'tags': vpc_config.get('tags', {})
}
for subnet_name, subnet_config in vpc_config.get('subnets', {}).items():
resources['resource']['azurerm_subnet'][subnet_name] = {
'name': subnet_name,
'resource_group_name': '${var.resource_group_name}',
'virtual_network_name': f"${{azurerm_virtual_network.{vpc_name}.name}}",
'address_prefixes': [subnet_config['cidr']]
}
return resources
def _generate_gcp_network(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Generate GCP network resources"""
resources = {
'resource': {
'google_compute_network': {},
'google_compute_subnetwork': {}
}
}
for vpc_name, vpc_config in config.get('vpcs', {}).items():
resources['resource']['google_compute_network'][vpc_name] = {
'name': vpc_name,
'auto_create_subnetworks': False
}
for subnet_name, subnet_config in vpc_config.get('subnets', {}).items():
resources['resource']['google_compute_subnetwork'][subnet_name] = {
'name': subnet_name,
'ip_cidr_range': subnet_config['cidr'],
'region': '${var.region}',
'network': f"${{google_compute_network.{vpc_name}.id}}"
}
return resources
def _map_instance_size(self, provider: str, size: str) -> str:
"""Map universal size to provider-specific instance type"""
size_mappings = {
'aws': {
'small': 't3.micro',
'medium': 't3.small',
'large': 't3.medium',
'xlarge': 't3.large'
},
'azure': {
'small': 'Standard_B1s',
'medium': 'Standard_B2s',
'large': 'Standard_B4ms',
'xlarge': 'Standard_B8ms'
},
'gcp': {
'small': 'e2-micro',
'medium': 'e2-small',
'large': 'e2-medium',
'xlarge': 'e2-standard-2'
}
}
return size_mappings.get(provider, {}).get(size, size)
def main():
import argparse
parser = argparse.ArgumentParser(description='Multi-Cloud Configuration Factory')
parser.add_argument('--spec-file', required=True, help='Universal specification file')
parser.add_argument('--output-dir', default='./generated', help='Output directory')
args = parser.parse_args()
factory = MultiCloudConfigFactory()
generated_files = factory.generate_configs(args.spec_file, args.output_dir)
print("Generated configurations:")
for provider, file_path in generated_files.items():
print(f" {provider}: {file_path}")
if __name__ == "__main__":
main()
What’s Next
Provider abstraction patterns enable you to write infrastructure code once and deploy it across multiple clouds. However, data often needs to move between these environments. In the next part, we’ll explore data and storage strategies for multi-cloud architectures, including replication, backup, and disaster recovery patterns.