Document Type: Long-form technical reference derived from exhaustive code analysis Generated: 2025-12-27 Source: 8 parallel exploratory agents analyzing 150+ Python files
- Executive Summary
- Platform Capability Inventory
- Core Architecture Foundation
- Kubernetes Resource Abstraction Layer
- Component Architecture
- Networking Architecture
- Virtual Machine Orchestration
- Distributed Storage
- Stack Configuration Patterns
- Metadata and Transformations
- Developer Use Scenarios
- Project Outcomes and Implications
- Complete File Inventory
This document presents an exhaustive analysis of a production-grade Pulumi Python infrastructure platform designed for Kubernetes-first deployments. The platform implements sophisticated patterns for virtual machine orchestration, distributed storage, advanced networking, and developer portal provisioning across bare-metal and cloud environments.
Core Philosophy:
- Zero hardcoding, 100% dynamic discovery
- Fat Models, Thin Controllers pattern throughout
- Fail-fast validation with Pydantic
- Convention over configuration
- Explicit provider management (no ambient credentials)
Key Statistics:
- 150+ Python files
- 100+ Pydantic models
- 32 Kubernetes resource types
- 24+ Kubernetes components
- 25+ stack configurations
- 8 VM profiles
- 6 network attachment definition types
| Module | Status | Capabilities |
|---|---|---|
| aws | Stub (scaffolded) | VPC, EKS, S3, RDS, GuardDuty, Security Hub, CloudTrail |
| k8s | Production | Full Kubernetes orchestration (see sections below) |
AWS Module (src/aws/)
├── Networking
│ ├── VPC with configurable CIDR (default: 10.0.0.0/16)
│ └── Multi-AZ support (1-6 availability zones)
├── Compute
│ ├── EKS cluster with version pinning
│ └── Configurable node instance types
├── Storage
│ ├── S3 with versioning support
│ └── S3 server-side encryption
├── Database
│ ├── RDS with engine version configuration
│ └── Instance class selection
└── Security
├── GuardDuty threat detection
├── Security Hub
└── CloudTrail audit logging
| Component | Category | Key Features |
|---|---|---|
| cilium | CNI/Networking | Gateway API, L2 announcements, Hubble UI, eBPF |
| cert_manager | Security | Self-signed CA, ACME/Let's Encrypt, DNS-01 solvers |
| multus | Networking | Pod multi-homing, thick plugin, Talos patches |
| cluster_network_addons_operator | Networking | macvtap, OVS, linux-bridge, kube-mac-pool |
| envoy_gateway | Edge Gateway | Multi-protocol (HTTP/HTTPS/TCP/UDP), Gateway API |
| external_dns | DNS | Cloudflare, Route53, automated DNS records |
| authelia | Authentication | OIDC, 2FA, session management |
| dex | Identity | OIDC identity broker, LDAP/SAML connectors |
| rook_ceph_operator | Storage | CSI drivers, device discovery |
| rook_ceph_cluster | Storage | MON/MGR/OSD, RBD, CephFS, RGW |
| external_snapshotter | Storage | VolumeSnapshot CRDs, controller HA |
| hostpath_provisioner_operator | Storage | Local path provisioning |
| kubevirt | Virtualization | VM orchestration, live migration, snapshots |
| containerized_data_importer | Virtualization | Disk import (registry, HTTP, S3) |
| kubevirt_manager | Virtualization | Web UI for VM management |
| virtual_machine | Virtualization | VM profiles, cloud-init, performance tuning |
| cloudnative_pg | Database | PostgreSQL operator, HA clusters |
| postgres_database | Database | Individual database provisioning |
| redis_operator | Database | Redis cluster/sentinel |
| prometheus | Observability | Monitoring stack, Grafana |
| zot_registry | Registry | OCI v1.1.1 container registry |
| backstage | Developer Portal | Service catalog, TechDocs |
| forgejo_server | Git | Self-hosted Git forge |
| forgejo_runner | CI/CD | Actions runner for Forgejo |
| namespace | Infrastructure | Namespace with quotas, policies |
| Resource | Location | Purpose |
|---|---|---|
| namespace | resources/namespace/ |
Namespace with finalizers, quotas |
| deployment | resources/deployment/ |
Workload deployment |
| service | resources/service/ |
Service exposure |
| config_map | resources/config_map/ |
Configuration data |
| secret | resources/secret/ |
Sensitive data (TLS, docker-config, opaque) |
| service_account | resources/service_account/ |
Pod identity |
| role | resources/role/ |
Namespaced RBAC |
| role_binding | resources/role_binding/ |
Role attachment |
| cluster_role | resources/cluster_role/ |
Cluster-wide RBAC |
| cluster_role_binding | resources/cluster_role_binding/ |
Cluster role attachment |
| resource_quota | resources/resource_quota/ |
Resource limits |
| limit_range | resources/limit_range/ |
Default limits |
| network_policy | resources/network_policy/ |
Pod network isolation |
| storage_class | resources/storage_class/ |
Dynamic provisioning |
| persistent_volume_claim | resources/persistent_volume_claim/ |
Storage claims |
| volume_snapshot_class | resources/volume_snapshot_class/ |
Snapshot configuration |
| helm_chart | resources/helm_chart/ |
Helm v4 chart deployment |
| config_file | resources/config_file/ |
YAML manifest deployment |
| custom_resource | resources/custom_resource/ |
Generic CRD instances |
| daemon_set | resources/daemon_set/ |
Node-level workloads |
| cilium_httproute | resources/cilium_httproute/ |
Gateway API HTTPRoute |
| cilium_ingress | resources/cilium_ingress/ |
Kubernetes Ingress |
| envoy_httproute | resources/envoy_httproute/ |
Envoy Gateway HTTPRoute |
| envoy_tcproute | resources/envoy_tcproute/ |
Envoy Gateway TCPRoute |
| envoy_reference_grant | resources/envoy_reference_grant/ |
Cross-namespace references |
| network_attachment_definition | resources/network_attachment_definition/ |
Multus NADs |
| golden_image | resources/golden_image/ |
VM golden images |
| kubevirt_image | resources/kubevirt_image/ |
KubeVirt disk images |
| postgresdb | resources/postgresdb/ |
CloudNativePG clusters |
| redis_cluster | resources/redis_cluster/ |
Redis clusters |
| redis_sentinel | resources/redis_sentinel/ |
Redis sentinels |
| validating_webhook_configuration | resources/validating_webhook_configuration/ |
Admission webhooks |
| mutating_webhook_configuration | resources/mutating_webhook_configuration/ |
Mutating webhooks |
| Type | CNI Plugin | Use Case | Key Config |
|---|---|---|---|
| bridge | bridge | Standard L2 | bridge, vlan, mtu |
| macvtap | macvtap | High-performance VMs | master, mode, device plugin |
| linux_bridge | cnv-bridge | CNAO-managed | bridge, vlan |
| ipvlan | ipvlan | Low-overhead L2/L3 | master, mode (l2/l3/l3s) |
| ovs | ovs | Software-defined | bridge, vlan |
| sriov | sriov | Direct hardware | device_id, vlan |
| Type | Purpose |
|---|---|
| dhcp | External DHCP server |
| host-local | Node-local allocation |
| static | Manual IP assignment |
| whereabouts | Cluster-wide unique IPs |
| Profile | OS | Type | Resources |
|---|---|---|---|
minimal_vm_profile |
Generic | Testing | 2c/4Gi/20Gi |
high_performance_vm_profile |
Generic | Production | CPU pinning, hugepages |
debian_server_profile |
Debian 13 | Server | 4c/8Gi/64Gi |
debian_developer_profile |
Debian 13 | Desktop | 8c/16Gi/128Gi, GNOME+XRDP |
kali_server_profile |
Kali | Headless | 4c/8Gi/64Gi |
kali_developer_profile |
Kali | Desktop | 8c/16Gi/256Gi, XFCE+XRDP |
ubuntu_server_profile |
Ubuntu | Server | Configurable |
ubuntu_developer_profile |
Ubuntu | Desktop | Configurable |
File: src/__main__.py
The platform entry point implements a fully dynamic, configuration-driven architecture:
def main() -> None:
"""Main entry point - completely dynamic, configuration-driven."""
# 1. Initialize metadata singleton with validated config
metadata = load_and_validate_config() # GlobalMetadata singleton
# 2. Schema generation in developer mode
if metadata.config.developer_mode:
generate_stack_config_schema(project_root, metadata.discovered_modules)
# 3. Initialize provider registry
registry = ProviderRegistry()
# 4. Load configuration (returns ONLY enabled modules as Pydantic models)
config = load_configuration()
enabled_module_configs = config.get("enabled_modules", {})
# 5. Strict validation - must be dict of Pydantic models
for module_name, module_config in enabled_module_configs.items():
if not isinstance(module_config, BaseModel):
raise RuntimeError(f"Module '{module_name}' config is not a Pydantic model")
if not hasattr(module_config, "enabled"):
raise RuntimeError(f"Module '{module_name}' config missing 'enabled' field")
# 6. Deploy modules in execution order
execution_order = config.get("module_execution_order", [])
prioritized = set(execution_order)
all_modules = [m for m in execution_order if m in enabled_module_configs]
all_modules.extend([m for m in enabled_module_configs if m not in prioritized])
deployed_modules = {}
for module_name in all_modules:
module = registry.load_and_validate_module(module_name, enabled_module_configs[module_name])
if module:
module.initialize()
module.deploy()
deployed_modules[module_name] = module
# 7. Export stack outputs
export_stack_outputs(deployed_modules)File: src/core/base_module.py
Every infrastructure module extends BaseModule:
class BaseModule(ABC):
"""Abstract base class that all provider modules must extend."""
def __init__(
self,
config: BaseModel, # Pre-validated Pydantic model
registry: "ProviderRegistry", # Cross-module provider sharing
module_name: str, # Module identifier
) -> None:
self.config = config
self.registry = registry
self.module_name = module_name
self.metadata = GlobalMetadata() # Singleton access
self.deployed = False
@abstractmethod
def initialize(self) -> None:
"""Initialize provider and register if needed."""
...
@abstractmethod
def deploy(self) -> None:
"""Deploy infrastructure for this provider."""
...
def get_config(self, field_name: str, default: Any = None) -> Any:
"""Get configuration field with deterministic access (SSOT)."""
try:
return getattr(self.config, field_name)
except AttributeError:
return default
def get_outputs(self) -> dict[str, object]:
"""Get standardized module outputs."""
return {
"module_name": self.module_name,
"deployed": self.deployed,
"exports": ModuleExports(),
}File: src/core/registry.py
Thread-safe registry enabling cross-module provider sharing:
class ProviderRegistry:
"""Provider registry for cross-module provider sharing."""
def __init__(self) -> None:
self._providers: dict[str, ProviderInfo] = {}
self._lock = Lock() # Thread-safe operations
def register_provider(
self,
provider_id: str,
provider: object,
platform: str,
metadata: ProviderMetadata | None = None,
) -> None:
"""Register a provider for cross-module use."""
with self._lock:
self._providers[provider_id] = ProviderInfo(
provider=provider, platform=platform, metadata=metadata
)
def load_and_validate_module(
self,
module_name: str,
module_configuration: BaseModel,
) -> Optional["BaseModule"]:
"""Load module with pre-validated configuration."""
# DETERMINISTIC: Module class name MUST be PascalCase(module_name) + "Module"
class_name = f"{self._to_pascal_case(module_name)}Module"
imported_module = importlib.import_module(module_name)
module_class = getattr(imported_module, class_name, None)
if module_class:
return module_class(module_configuration, self, module_name)
return None
def _to_pascal_case(self, snake_str: str) -> str:
"""Convert snake_case to PascalCase with special handling."""
if snake_str == "k8s":
return "K8S"
return "".join(x.capitalize() for x in snake_str.split("_"))File: src/core/metadata.py
Single source of truth for all metadata:
class GlobalMetadata:
"""Singleton source of truth for all metadata."""
_instance: Optional["GlobalMetadata"] = None
_config: InputSchema | None = None
_deployment_id: str
_deployed_at: datetime
_git_metadata: GitMetadata
_discovered_modules: list[str]
_project_root: Path
def __new__(cls, config_data: StackConfig | None = None,
discovered_modules: list[str] | None = None):
"""Ensure only one instance exists."""
if cls._instance is None:
if config_data is None:
raise RuntimeError("GlobalMetadata not initialized.")
if discovered_modules is None:
raise RuntimeError("GlobalMetadata requires discovered_modules.")
instance = super().__new__(cls)
# Validate and freeze configuration
instance._config = InputSchema.model_validate(config_data)
instance._discovered_modules = discovered_modules
instance._git_metadata = GitMetadata.from_repo()
instance._deployment_id = # ... deterministic hash
instance._deployed_at = datetime.now()
cls._instance = instance
return cls._instance
@property
def discovered_modules(self) -> list[str]:
"""Immutable list of discovered modules."""
return self._discovered_modules.copy() # Defensive copyFile: src/core/metadata.py
Strict configuration validation:
class DynamicModuleConfig(BaseModel):
"""Base configuration for any dynamically discovered module."""
enabled: bool = Field(description="Enable this module")
model_config = ConfigDict(
extra="forbid", # STRICT: No undefined fields allowed
populate_by_name=True,
frozen=True, # Immutable after creation
validate_assignment=True,
)
class InputSchema(BaseModel):
"""Root input schema - THE source of truth."""
environment: Environment
debug: bool = Field(default=False)
developer_mode: bool = Field(default=False)
compliance: ComplianceConfig
modules: ModulesConfig
model_config = ConfigDict(
extra="forbid",
alias_generator=to_camel, # snake_case ↔ camelCase
populate_by_name=True,
validate_assignment=True,
frozen=True,
)File: src/core/discovery.py
Centralized module discovery with caching:
_discovered_modules_cache: list[str] | None = None
def discover_modules(force_refresh: bool = False) -> list[str]:
"""Discover all valid modules from filesystem (SSOT)."""
global _discovered_modules_cache
if _discovered_modules_cache is not None and not force_refresh:
return _discovered_modules_cache.copy()
for path in src_dir.iterdir():
if not path.is_dir() or path.name == "core":
continue
init_file = path / "__init__.py"
if not init_file.exists():
continue
# NO SILENT FAILURES
try:
importlib.import_module(path.name)
discovered.append(path.name)
except ImportError as e:
raise RuntimeError(
f"CRITICAL: Module '{path.name}' cannot be imported. "
f"This is a BROKEN module that violates IaC requirements."
) from e
_discovered_modules_cache = discovered
return discovered.copy()File: src/k8s/resources/base.py
Mixin-based architecture eliminating ~82% of boilerplate:
class BaseK8SResource(StandardMetadataMixin, DNS1123ValidationMixin, ABC):
"""Abstract base for all K8S resources."""
config_class: type[BaseModel] # Subclasses declare config type
def __init__(
self,
name: str,
config: BaseModel,
provider: k8s.Provider,
k8s_metadata: K8SMetadata,
):
# Validates config is instance of declared config_class
if not isinstance(config, self.config_class):
raise TypeError(f"Config must be {self.config_class.__name__}")
self.name = name
self.config = config
self.provider = provider # CRITICAL: Explicit provider
self.k8s_metadata = k8s_metadata
self.logger = get_logger(f"k8s.resources.{self.__class__.__name__}")
@abstractmethod
def validate(self) -> bool:
"""Resource-specific validation."""
...
@abstractmethod
def create(self, opts: ResourceOptions | None = None) -> pulumi.Resource:
"""Create the Pulumi resource."""
...
def _merge_resource_options(self, opts: ResourceOptions | None, **kwargs) -> ResourceOptions:
"""Merge user options with base options, ensuring provider is set."""
base_opts = ResourceOptions(
provider=self.provider,
protect=kwargs.get("protect", False),
delete_before_replace=kwargs.get("delete_before_replace"),
ignore_changes=kwargs.get("ignore_changes"),
custom_timeouts=kwargs.get("custom_timeouts"),
)
if opts:
return ResourceOptions.merge(base_opts, opts)
return base_optsFile: src/k8s/resources/mixins.py
class StandardMetadataMixin:
"""Automatic metadata injection."""
def _merge_metadata(
self,
component: str,
user_labels: dict[str, str],
user_annotations: dict[str, str],
version: str = "v1",
description: str | None = None,
) -> tuple[dict[str, str], dict[str, str]]:
"""Single-line metadata retrieval."""
base_labels = self._get_standard_labels(component, version)
base_annotations = self._get_standard_annotations(component, description)
# User overrides take precedence
labels = {**base_labels, **user_labels}
annotations = {**base_annotations, **user_annotations}
return labels, annotationsclass DNS1123ValidationMixin:
"""Kubernetes naming compliance validation."""
_DNS1123_PATTERN = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$")
def _validate_dns1123(self, value: str, max_length: int = 63) -> bool:
"""Validate DNS-1123 compliance."""
if not value:
return False
if len(value) > max_length:
return False
return bool(self._DNS1123_PATTERN.match(value))
def _validate_name(self, max_length: int = 63) -> bool:
"""Validate self.name with actionable error messages."""
if not self._validate_dns1123(self.name, max_length):
self.logger.error(
f"Name '{self.name}' is not DNS-1123 compliant. "
f"Must be lowercase alphanumeric + hyphens, max {max_length} chars."
)
return False
return TrueFile: src/k8s/resources/helm_chart/__init__.py
Feature-rich Helm chart handling:
class HelmChartResource(BaseK8SResource):
"""Helm chart deployment with caching and version resolution."""
config_class = HelmChartConfig
def _resolve_chart_cache(self, config: HelmChartConfig) -> HelmChartConfig:
"""Resolve chart from local cache if available."""
if not config.use_cache:
return config
if config.chart.startswith(("oci://", "file://", "./", "/")):
return config
cache = HelmChartCache()
cached_path = cache.get_cached_chart(
chart=config.chart,
version=config.version,
repository=config.repository,
)
if cached_path:
self.logger.info(f"Using cached chart: {cached_path}")
return config.model_copy(update={
"chart": cached_path,
"repository": None,
})
if config.cache_chart:
try:
cached_path = cache.cache_chart(...)
return config.model_copy(update={...})
except RuntimeError as e:
self.logger.warning(f"Chart caching failed: {e}")
return configFile: src/k8s/resources/config_file/__init__.py
Remote YAML with caching and transformation:
class ConfigFileResource(BaseK8SResource):
"""Deploy YAML manifests from URLs with caching."""
def create(self, opts: ResourceOptions) -> pulumi.Resource:
config: ConfigFileConfig = self.config
# 1. Fetch/load YAML content (with caching)
yaml_content = self._get_yaml_content(config)
# 2. Transform resources
transformed_docs = self._transform_yaml(yaml_content, config)
# 3. Write to temp file
temp_path = self._write_temp_yaml(transformed_docs)
# 4. Deploy via ConfigFile
config_file = k8s.yaml.ConfigFile(
self.name,
file=temp_path,
skip_await=config.skip_await,
opts=ResourceOptions(provider=self.provider),
)
# 5. Schedule cleanup
pulumi.Output.all().apply(lambda _: self._cleanup_temp_files())
return config_file
def _transform_yaml(self, yaml_content: str, config: ConfigFileConfig) -> list[dict]:
"""Transform YAML documents."""
documents = list(yaml.safe_load_all(yaml_content))
for doc in documents:
# Skip namespace resources if requested
if config.skip_namespace_resource and doc.get("kind") == "Namespace":
continue
# Inject namespace (not for cluster-scoped)
if config.namespace and not self._is_cluster_scoped(doc):
doc["metadata"]["namespace"] = config.namespace
# Apply custom transformations
if config.transformations:
for transform_fn in config.transformations:
doc = transform_fn(doc)src/cache/
├── helm_chart/
│ ├── charts.jetstack.io/
│ │ └── cert-manager-v1.19.2.tgz
│ ├── helm.cilium.io/
│ │ └── cilium-1.18.0.tgz
│ └── charts.rook.io/
│ └── rook-ceph-cluster-v1.17.0.tgz
└── config_file/
├── github.com/kubernetes-sigs/gateway-api/releases/download/v1.4.0/
│ └── experimental-install.yaml
├── raw.githubusercontent.com/cloudnative-pg/cloudnative-pg/v1.24.1/
│ └── postgresql.cnpg.io_clusters.yaml
└── github.com/kubevirt/cluster-network-addons-operator/releases/download/v0.99.3/
└── operator.yaml
File: src/k8s/components/base.py
class BaseK8SComponent(Generic[TConfig]):
"""Generic base for K8S components."""
def __init__(
self,
name: str,
config: TConfig | dict,
provider: k8s.Provider,
k8s_metadata: K8SMetadata,
depends_on: list[pulumi.Resource] | None = None,
deployed_resources: dict[str, dict[str, pulumi.Resource]] | None = None,
):
# Accept either Pydantic model or dict
if isinstance(config, dict):
self.config = self.config_class(**config)
else:
self.config = config
self.name = name
self.provider = provider
self.k8s_metadata = k8s_metadata
self.depends_on = depends_on or []
self.deployed_resources = deployed_resources or {}
@abstractmethod
def deploy(self) -> dict[str, pulumi.Resource]:
"""Deploy component resources. Returns resource dict."""
...
def get_outputs(self) -> dict[str, Any]:
"""Default outputs with component metadata."""
return {
"component": self.name,
"namespace": getattr(self.config, "namespace", None),
}File: src/k8s/discovery.py
def discover_k8s_components(force_refresh: bool = False) -> dict[str, type[Any]]:
"""Discover all K8S components from filesystem."""
for path in components_dir.iterdir():
if not path.is_dir() or path.name.startswith("_"):
continue
# STRICT CONVENTION: cert_manager/ → CertManagerComponent
component_name = path.name
class_name = f"{_snake_to_pascal_case(component_name)}Component"
try:
module = importlib.import_module(f"k8s.components.{component_name}")
if not hasattr(module, class_name):
raise RuntimeError(
f"CRITICAL: Component '{component_name}' missing class '{class_name}'"
)
discovered[component_name] = getattr(module, class_name)
except ImportError as e:
raise RuntimeError(f"CRITICAL: Component import failed: {e}") from e
return discovered.copy()
def discover_k8s_resources(force_refresh: bool = False) -> dict[str, type[Any]]:
"""Discover all K8S resource types."""
# Convention: limit_range/ → LimitRangeResource
class_name = _snake_to_pascal_case(resource_type) + "Resource"Every component defines explicit parent chains:
# Cilium Component
k8s-provider
→ gateway-api-crds
→ helm-chart
→ [l2-policy, gateway, httproute]
# KubeVirt Component
provider
→ namespace (with kubevirt.io label)
→ operator (ConfigFile)
→ kubevirt-cr
→ [selinux-workaround, export-ingress]
# Rook Ceph Cluster Component
provider
→ helm-chart (45m timeout)
→ [rbd-storageclass, cephfs-storageclass, dashboard-httproute]
# Backstage Component
[zot_registry-ready]
→ namespace
→ serviceaccount
→ clusterrole
→ clusterrolebinding
→ postgres
→ secret
→ configmap
→ [docker_image (optional)]
→ deployment
→ pdb + service
→ [httproute]# Priority cascade for downstream dependencies
if "hubble-ui-ingress" in resources:
resources["component-ready"] = resources["hubble-ui-ingress"]
elif "hubble-ui-httproute" in resources:
resources["component-ready"] = resources["hubble-ui-httproute"]
elif "shared-gateway" in resources:
resources["component-ready"] = resources["shared-gateway"]
else:
resources["component-ready"] = resources["helm-chart"]# Backstage depends on Zot Registry
def deploy(self) -> dict[str, pulumi.Resource]:
# Extract zot_registry component-ready marker
zot_ready = self.deployed_resources.get("zot_registry", {}).get("component-ready")
if self.config.build_from_source and zot_ready:
# Docker image build depends on registry being ready
image_opts = ResourceOptions(depends_on=[zot_ready])Files: src/k8s/resources/cilium_httproute/, src/k8s/components/cilium/models/gateway.py
class CiliumHTTPRouteSpec(BaseModel):
"""Fat model for HTTPRoute with builder methods."""
# Hostnames
hostname: str | None = None
hostnames: list[str] | None = None
# Backend
service_name: str
service_port: int
service_namespace: str | None = None # Cross-namespace with ReferenceGrant
# Gateway attachment
gateway_name: str = "cilium-gateway"
gateway_namespace: str = "kube-system"
gateway_listener_name: str = "https"
# Path matching
path: str = "/"
path_type: Literal["PathPrefix", "Exact"] = "PathPrefix"
def to_config(self) -> CiliumHTTPRouteConfig:
"""Convert to resource config."""
...class GatewayAPISpec(BaseModel):
"""Gateway API config for Cilium."""
# Network modes
host_network_mode: bool # True=0.0.0.0:80/443, False=LoadBalancer
envoy_keep_cap_net_bind_service: bool = True
# Protocol configuration
enable_proxy_protocol: bool = False
enable_app_protocol: bool = False
enable_alpn: bool = False
# TLS configuration
certificate_ip_addresses: list[str] = ["127.0.0.1"]
# Custom listeners
listeners: list[GatewayListenerSpec] = []
def to_helm_values(self) -> dict[str, Any]:
"""Transform to Cilium Helm values."""
return {
"gatewayAPI": {
"enabled": True,
"enableProxyProtocol": self.enable_proxy_protocol,
"hostNetwork": {"enabled": self.host_network_mode},
}
}File: src/k8s/components/cilium/models/l2.py
class L2AnnouncementsSpec(BaseModel):
"""L2 announcements for bare-metal LoadBalancer support."""
enabled: bool = True
interface: str = "br0"
ip_pool_cidr: str # e.g., "192.168.1.192/28"
# Service selection (OR logic via multiple policies)
service_selectors: list[L2ServiceSelector] | None
def to_cilium_resources(self) -> dict[str, dict[str, Any]]:
"""Generate CiliumLoadBalancerIPPool + CiliumL2AnnouncementPolicy."""
resources = {
"ippool": {
"api_version": "cilium.io/v2",
"kind": "CiliumLoadBalancerIPPool",
"spec": {"blocks": [{"cidr": self.ip_pool_cidr}]},
},
}
# One policy per selector (enables OR logic)
for selector in self.service_selectors:
resources[f"policy-{selector.name}"] = {
"api_version": "cilium.io/v2alpha1",
"kind": "CiliumL2AnnouncementPolicy",
"spec": {
"loadBalancerIPs": True,
"interfaces": [self.interface],
"serviceSelector": {"matchLabels": selector.match_labels},
},
}
return resourcesFile: src/k8s/resources/network_attachment_definition/types/
class BridgeNADSpec(BaseModel):
"""Linux bridge CNI for L2 networking."""
name: str
namespace: str = "default"
bridge: str # Host bridge interface
vlan: int | None = None
mtu: int = 1500
ipam: IPAMConfig | None = None
def to_cni_config(self) -> dict[str, Any]:
return {
"cniVersion": "0.3.1",
"type": "bridge",
"bridge": self.bridge,
"mtu": self.mtu,
"vlan": self.vlan,
"ipam": self.ipam.to_cni_config() if self.ipam else {},
}class MacvtapNADSpec(BaseModel):
"""Macvtap CNI for near-native NIC performance."""
name: str
namespace: str = "default"
master: str # Physical interface (enp3s0)
mode: Literal["bridge", "vepa", "private", "passthru"] = "bridge"
use_device_plugin: bool = True # CNAO device plugin mode
def to_nad_config(self) -> NetworkAttachmentDefinitionConfig:
annotations = {}
if self.use_device_plugin:
annotations["k8s.v1.cni.cncf.io/resourceName"] = (
f"macvtap.network.kubevirt.io/{self.master}"
)
return NetworkAttachmentDefinitionConfig(
name=self.name,
namespace=self.namespace,
config=self.to_cni_config(),
annotations=annotations,
)class SriovNADSpec(BaseModel):
"""SR-IOV CNI for direct hardware access."""
name: str
namespace: str = "default"
vlan: int | None = None
device_id: str | None = None # PCI format: 0000:03:02.0
@field_validator("device_id")
@classmethod
def validate_device_id(cls, v: str | None) -> str | None:
if v and not re.match(r"^[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-9a-fA-F]$", v):
raise ValueError("Device ID must be PCI format")
return vDirectory: src/k8s/components/virtual_machine/models/
class CPUSpec(BaseModel):
"""CPU topology and performance settings."""
cores: int = 2
threads: int = 1
sockets: int = 1
model: str = "host-model" # or "host-passthrough"
# Performance tuning
dedicated_cpu_placement: bool = False # Requires CPU manager
isolate_emulator_thread: bool = False # Separate QEMU housekeeping
# NUMA
numa: NUMASpec | None = Noneclass ResourcesSpec(BaseModel):
"""Memory allocation."""
memory: str = "4Gi"
hugepages: HugepagesSpec | None = None
overcommit_guest_overhead: bool = False
class HugepagesSpec(BaseModel):
"""Hugepages configuration."""
page_size: Literal["2Mi", "1Gi"] = "2Mi"class DiskSpec(BaseModel):
"""Virtual disk with performance tuning."""
name: str
size: str | None = None
# Source types
source_type: Literal["dataVolume", "persistentVolumeClaim",
"containerDisk", "snapshot_clone"]
data_volume_name: str | None = None
pvc_name: str | None = None
container_disk_image: str | None = None
# Performance tuning
cache: Literal["none", "writethrough", "writeback"] = "none"
io: Literal["native", "threads", "default"] = "native"
dedicated_io_thread: bool = False
# Boot order
boot_order: int | None = Noneclass NetworkInterfaceSpec(BaseModel):
"""Network interface with advanced features."""
name: str
network_name: str | None = None # Multus NAD reference
# Interface types (mutually exclusive with binding_plugin)
bridge: bool = False
masquerade: bool = False
sriov: bool = False
# KubeVirt v1.3+ binding plugins
binding_plugin: Literal["macvtap", "passt", "slirp"] | None = None
# Performance
queues: int | None = None # Multi-queue (set to vCPU count)
model: str = "virtio"class CloudInitSpec(BaseModel):
"""Cloud-init configuration."""
use_secret: bool = True # Store in K8s secret
network_data: CloudInitNetworkDataSpec | str | None = None
user_data: CloudInitUserDataSpec | str | None = None
class CloudInitUserDataSpec(BaseModel):
"""Cloud-config user data."""
username: str = "user"
password_hash: str | None = None
ssh_authorized_keys: list[str] = []
packages: list[str] = []
package_upgrade: bool = False
write_files: list[dict] = []
runcmd: list[str | list[str]] = []VirtualMachineSpec(
data_volumes=[DataVolumeSpec(
name="root",
source_type="registry",
url="docker://containercraft/debian:13",
storage_class="ceph-nvme-vm",
size="128Gi",
)],
disks=[DiskSpec(
name="root",
source_type="dataVolume",
data_volume_name="root",
)],
)VirtualMachineComponentConfig(
golden_image=GoldenImageConfig(
source=GoldenImageSourceSpec(
source_type="registry",
url="docker://containercraft/kali:latest",
),
snapshot=GoldenImageSnapshotSpec(
name="kali-base-snapshot",
volume_snapshot_class="ceph-rbd-snapshot",
),
),
)
# Creates: DataVolume → VolumeSnapshot → Clone PVC → VMVirtualMachineComponentConfig(
snapshot_source=SnapshotSourceConfig(
snapshot_name="kali-base-snapshot",
snapshot_namespace="kubevirt", # Central image store
storage_class="ceph-nvme-vm",
size="256Gi",
),
)
# Creates: DataVolume (with spec.source.snapshot) → VMDirectory: src/k8s/components/virtual_machine/profiles/
def debian_developer_profile(
name: str,
namespace: str,
memory: str = "16Gi",
cores: int = 8,
disk_size: str = "128Gi",
storage_class: str = "ceph-nvme-vm",
ssh_keys: list[str] = [],
password_hash: str | None = None,
) -> VirtualMachineSpec:
return VirtualMachineSpec(
name=name,
namespace=namespace,
cpu=CPUSpec(cores=cores),
resources=ResourcesSpec(memory=memory),
firmware=FirmwareSpec(type="uefi"),
disks=[DiskSpec(
name="root",
source_type="dataVolume",
data_volume_name=f"{name}-root",
boot_order=1,
cache="none",
io="native",
)],
data_volumes=[DataVolumeSpec(
name=f"{name}-root",
source_type="registry",
url="docker://docker.io/containercraft/debian:13",
storage_class=storage_class,
size=disk_size,
)],
interfaces=[NetworkInterfaceSpec(
name="default",
masquerade=True,
ports=[PortSpec(name="ssh", port=22)],
)],
cloud_init=CloudInitSpec(
user_data=CloudInitUserDataSpec(
username="debian",
password_hash=password_hash,
ssh_authorized_keys=ssh_keys,
packages=[
# Desktop
"gnome", "gdm3", "xorg",
# XRDP
"xrdp", "xorgxrdp",
# Development
"build-essential", "python3", "python3-pip",
# Containers
"docker.io", "docker-compose",
# 400+ more packages...
],
write_files=[
# GDM3 config: Force X11
{"path": "/etc/gdm3/custom.conf", "content": "WaylandEnable=false"},
# XRDP config: TLS 1.3
{"path": "/etc/xrdp/xrdp.ini", "content": "..."},
# Docker daemon.json
{"path": "/etc/docker/daemon.json", "content": "..."},
],
runcmd=[
# 25+ commands for setup
"systemctl enable xrdp",
"usermod -aG docker debian",
# ...
],
),
),
)CPUSpec(
cores=8,
dedicated_cpu_placement=True, # Pin to physical CPUs
isolate_emulator_thread=True, # Separate QEMU housekeeping
model="host-passthrough", # Full host CPU
numa=NUMASpec(guest_mapping_passthrough=True),
)ResourcesSpec(
memory="32Gi",
hugepages=HugepagesSpec(page_size="2Mi"), # Eliminate TLB misses
)DiskSpec(
cache="none", # O_DIRECT bypass page cache
io="native", # Linux AIO
dedicated_io_thread=True, # Prevent contention
)
VirtualMachineSpec(
io_threads_policy="auto", # 1 IOThread per disk
block_multi_queue=True, # Parallel I/O submission
)Files: src/k8s/components/rook_ceph_operator/, src/k8s/components/rook_ceph_cluster/
Rook Operator Stack
├── Operator Helm chart
├── CSI drivers (RBD + CephFS)
└── Device discovery
Rook Cluster Stack
├── CephCluster CR via Helm
├── MON, MGR, OSD daemons
├── CephBlockPool (RBD)
├── CephFilesystem (CephFS)
├── StorageClasses
└── VolumeSnapshotClasses
File: src/k8s/components/rook_ceph_cluster/models/storage.py
class StorageSpec(BaseModel):
"""OSD device configuration."""
use_all_nodes: bool = False
use_all_devices: bool = False
device_filter: str | None = None
device_path_filter: str | None = None
nodes: list[NodeStorageConfig] | None = None
class NodeStorageConfig(BaseModel):
"""Per-node storage configuration."""
name: str # Node hostname
devices: list[StorageDeviceConfig] | None = None
device_filter: str | None = None
class StorageDeviceConfig(BaseModel):
"""Individual device configuration."""
name: str # nvme0n1 or /dev/disk/by-id/...File: src/k8s/components/rook_ceph_cluster/models/pools.py
class RBDStorageClassSpec(BaseModel):
"""RBD block storage configuration."""
name: str = "rook-ceph-block"
pool_name: str = "replicapool"
is_default: bool = False
# Replication
replicated_size: int = 3
requires_safe_replica_size: int = 2
pg_num: int = 32
# Device class isolation
device_class: Literal["nvme", "ssd", "hdd"] | None = None
# Volume settings
volume_binding_mode: str = "WaitForFirstConsumer"
allow_volume_expansion: bool = True
reclaim_policy: Literal["Delete", "Retain"] = "Delete"
# RBD image configuration
image_format: str = "2"
image_features: str = "layering,exclusive-lock,object-map"class CephFSStorageClassSpec(BaseModel):
"""CephFS shared filesystem configuration."""
name: str = "rook-cephfs"
fs_name: str = "cephfs"
pool_name: str = "cephfs-data0"
# Pool configuration
replicated_size: int = 3
metadata_replicated_size: int = 3
# MDS configuration
mds_active_count: int = 1
mds_standby_count: int = 0
# Device class isolation
device_class: Literal["nvme", "ssd", "hdd"] | None = NoneFile: src/k8s/components/external_snapshotter/
class ExternalSnapshotterSpec(BaseModel):
"""External snapshotter configuration."""
version: str | None = None # Auto-detects latest
namespace: str = "kube-system"
controller_replicas: int = 2 # HA with leader election
# Features
enable_volume_group_snapshots: bool = False # Alpha
leader_election: bool = True
worker_threads: int = 10
protected: bool = True| Storage Class | Type | Device Class | Use Case |
|---|---|---|---|
ceph-nvme-vm-block |
RBD | NVMe | VM root disks |
ceph-nvme-vm-fs |
RBD | NVMe | VM scratch space |
ceph-nvme-db-block |
RBD | NVMe | Database storage |
cephfs-nvme-vm |
CephFS | NVMe | Shared filesystems |
ceph-ssd-archive |
RBD | SSD | Archived data |
| Stack Type | Purpose | Components |
|---|---|---|
| Bootstrap | Foundation services | Cilium, cert-manager, Envoy Gateway, Authelia, Dex |
| Storage | Distributed storage | Rook Operator, Rook Cluster, External Snapshotter |
| Data Services | Database operators | CloudNativePG, Redis Operator |
| Virtualization | VM platform | KubeVirt, CDI |
| Application | Workload services | Forgejo, Backstage |
# Bootstrap Stack (Dependency Chain)
components_deployment_order:
- cilium # CNI first (installs Gateway API CRDs)
- cert_manager # TLS certificates
- external_dns # DNS automation
- envoy_gateway # Edge gateway
- multus # Secondary networks
- cluster_network_addons_operator # L2 plugins
- authelia # Authentication
- dex # Identity broker# Development
debug: true
developer_mode: true
protected: false
compliance:
fisma:
compliance_level: low
enforcement_mode: warn
# Production
debug: false
developer_mode: false
protected: true
compliance:
fisma:
compliance_level: high
enforcement_mode: enforceconfig:
environment: dev
k8s:
enabled: true
components:
cilium:
enabled: true
spec:
deployment_mode: bare-metal
gateway_api:
enabled: true
crd_version: v1.4.0
l2_announcements:
enabled: true
interface: enp3s0
ip_pool_cidr: "192.168.1.192/28"
cert_manager:
enabled: true
spec:
enable_self_signed_ca: true
acme_issuers:
- name: letsencrypt-cloudflare
email: [email protected]
envoy_gateway:
enabled: true
spec:
gateway:
listeners:
- name: https
port: 443
protocol: HTTPSconfig:
k8s:
enabled: true
components:
kubevirt:
enabled: true
spec:
cpu:
default_cpu_model: Skylake-Client-IBRS
memory:
default_hugepages: 2Mi
live_migration:
bandwidth_per_migration: 5Gi
permitted_host_devices:
pci_host_devices:
- pci_vendor_selector: "8086:1912"
resource_name: devices.kubevirt.io/intel-hd-530Tier 1 (Global): commonLabels for ALL resources
Tier 2 (Pod-level): podAnnotations, podLabels
Tier 3 (Component-specific): Hubble, operator-specific
File: src/k8s/core/transformations.py
PROTECTED_RESOURCE_TYPES = {
"kubernetes:ceph.rook.io/v1:CephCluster",
"kubernetes:storage.k8s.io/v1:StorageClass",
"kubernetes:core/v1:PersistentVolumeClaim",
"kubernetes:core/v1:Secret",
}
VOLATILE_METADATA_IGNORE_PATTERNS = [
'metadata.annotations["deployed.at"]',
'metadata.annotations["deployment.id"]',
'metadata.annotations["provenance.attestation"]',
'metadata.labels["app.kubernetes.io/instance"]',
]
def create_metadata_transformation(k8s_metadata: K8SMetadata):
"""Create closure for global resource transformation."""
base_labels = k8s_metadata.get_base_labels()
base_annotations = k8s_metadata.get_base_annotations()
def transform_kubernetes_resource(args: ResourceTransformArgs):
if not args.type_.startswith("kubernetes:"):
return None
# Initialize null metadata
if args.props["metadata"] is None:
args.props["metadata"] = {}
# Merge labels/annotations
props["metadata"]["labels"] = {**base_labels, **existing_labels}
props["metadata"]["annotations"] = {**base_annotations, **existing_annotations}
# Protected resources: add ignore_changes
if args.type_ in PROTECTED_RESOURCE_TYPES:
result_opts = ResourceOptions(
ignore_changes=VOLATILE_METADATA_IGNORE_PATTERNS
)
return ResourceTransformResult(props=props, opts=result_opts)
return transform_kubernetes_resourceFile: src/k8s/core/query_github_release_version.py
class GitHubReleaseVersionQuery:
"""Generic interface for GitHub release version queries."""
GITHUB_API_BASE = "https://api.github.com"
def __init__(self, owner: str, repo: str):
self.owner = owner
self.repo = repo
def resolve_version(self, version: str | None, is_production: bool) -> str:
if version and version.lower() not in {"latest", ""}:
return version.lstrip("v") # No API call
if is_production:
raise ValueError("Cannot use 'latest' in production")
return self.get_latest_stable() # GET /releases/latest
def get_latest_stable(self) -> str | None:
url = f"{self.GITHUB_API_BASE}/repos/{self.owner}/{self.repo}/releases/latest"
release = self._request(url)
return self._normalize_version(release.get("tag_name", ""))# 1. Create component directory
mkdir -p src/k8s/components/my_component/
# 2. Create models.py
cat > src/k8s/components/my_component/models.py << 'EOF'
from pydantic import BaseModel, Field
from k8s.components.base import DynamicComponentConfig
class MyComponentSpec(BaseModel):
namespace: str = "default"
replicas: int = Field(default=1, ge=1, le=10)
model_config = ConfigDict(extra="forbid", frozen=True)
class MyComponentComponentConfig(DynamicComponentConfig):
spec: MyComponentSpec
EOF
# 3. Create __init__.py
cat > src/k8s/components/my_component/__init__.py << 'EOF'
from k8s.components.base import BaseK8SComponent
from .models import MyComponentComponentConfig, MyComponentSpec
class MyComponentComponent(BaseK8SComponent[MyComponentComponentConfig]):
config_class = MyComponentComponentConfig
def deploy(self) -> dict[str, pulumi.Resource]:
spec: MyComponentSpec = self.config.spec
resources = {}
# Create resources here
resources["component-ready"] = resources["deployment"]
return resources
EOF
# Component is auto-discovered!# profiles/fedora/developer.py
from k8s.components.virtual_machine.models import *
def fedora_developer_profile(
name: str,
namespace: str,
memory: str = "16Gi",
cores: int = 8,
ssh_keys: list[str] = [],
) -> VirtualMachineSpec:
return VirtualMachineSpec(
name=name,
namespace=namespace,
cpu=CPUSpec(cores=cores),
resources=ResourcesSpec(memory=memory),
firmware=FirmwareSpec(type="uefi"),
disks=[...],
interfaces=[...],
cloud_init=CloudInitSpec(
user_data=CloudInitUserDataSpec(
username="fedora",
ssh_authorized_keys=ssh_keys,
packages=["gnome-desktop", "xrdp", ...],
),
),
)# types/vxlan.py
class VxlanNADSpec(BaseModel):
"""VXLAN overlay network."""
name: str
namespace: str = "default"
vni: int # VXLAN Network Identifier
remote_ip: str
mtu: int = 1450
def to_cni_config(self) -> dict:
return {
"cniVersion": "0.3.1",
"type": "vxlan",
"vni": self.vni,
"remote_ip": self.remote_ip,
"mtu": self.mtu,
}# Pulumi.my-custom-stack.yaml
config:
environment: dev
k8s:
enabled: true
components:
my_component:
enabled: true
spec:
namespace: my-app
replicas: 3
components_deployment_order:
- cilium
- cert_manager
- my_component- Multi-Cluster Support: Context-aware provider management
- Bare-Metal Ready: L2 announcements, SR-IOV, macvtap networking
- Cloud Portable: AWS, Azure, GCP provider integration (scaffolded)
- Offline Capable: Helm chart and manifest caching
- Production Grade: Compliance tracking, immutable configurations
- Golden Image Pattern: Import once, clone many (Ceph snapshots)
- Cross-Namespace Cloning: Centralized image management
- Desktop VMs: GNOME, KDE, XFCE with XRDP remote access
- Performance Tuning: CPU pinning, hugepages, I/O threading, NUMA
- Multi-Tier Storage: Device class isolation (NVMe/SSD/HDD)
- Dual Protocol: RBD block + CephFS shared filesystem
- Volume Snapshots: RBD and CephFS snapshot classes
- CSI Integration: Full Kubernetes native storage
- Gateway API: Modern HTTP/HTTPS routing (v1.4.0+)
- Multi-Protocol Edge: HTTP, HTTPS, TCP, UDP via Envoy Gateway
- VM Networking: 6 NAD types for diverse workloads
- L2 LoadBalancer: ARP/NDP without BGP
- Module Isolation: Components developed independently
- Schema Validation: IDE support via generated JSON schemas
- Fail-Fast: All validation before resource creation
- Audit Trail: Git metadata in all resources
- Self-Service VMs: Profile-based VM provisioning
- Database-as-a-Service: CloudNativePG integration
- Registry Access: Zot OCI registry with Gateway API routing
- Developer Portals: Backstage with Kubernetes integration
- Observability: Prometheus integration across components
- Security: RBAC, TLS, OIDC authentication (Authelia/Dex)
- Disaster Recovery: Volume snapshots for data protection
- Maintenance: Rolling updates via Helm upgrade patterns
src/core/
├── __init__.py # Module exports
├── base_module.py # BaseModule abstract class
├── config.py # Configuration loading
├── discovery.py # Module discovery with caching
├── exceptions.py # Exception hierarchy
├── git.py # Git metadata collection
├── logging.py # Structured logging
├── metadata.py # GlobalMetadata singleton, Pydantic schemas
├── outputs.py # Stack output management
└── registry.py # ProviderRegistry
src/aws/
├── __init__.py # AwsModule class (stub)
└── core/
├── __init__.py
└── models.py # AWSModuleConfig schema
src/k8s/
├── __init__.py # K8SModule class
├── discovery.py # Component/resource discovery
├── README.md
├── core/
│ ├── __init__.py
│ ├── annotations.py # K8SAnnotations
│ ├── exceptions.py # K8S exceptions
│ ├── labels.py # K8SLabels
│ ├── metadata.py # K8SMetadata
│ ├── models.py # K8SModuleConfig
│ ├── provider.py # K8SProviderManager
│ ├── query_github_release_version.py
│ └── transformations.py # Stack transformations
├── components/
│ ├── base.py # BaseK8SComponent
│ ├── authelia/
│ ├── backstage/
│ ├── cert_manager/
│ ├── cilium/
│ │ ├── __init__.py
│ │ ├── builders/
│ │ ├── gateway_api_crds.py
│ │ ├── httproute.py
│ │ ├── ingress.py
│ │ ├── models/
│ │ ├── profiles.py
│ │ ├── routes.py
│ │ └── values.yaml
│ ├── cloudflare_tunnels/
│ ├── cloudnative_pg/
│ ├── cluster_network_addons_operator/
│ ├── containerized_data_importer/
│ ├── dex/
│ ├── envoy_gateway/
│ ├── external_dns/
│ ├── external_snapshotter/
│ ├── forgejo_runner/
│ ├── forgejo_server/
│ ├── hostpath_provisioner_operator/
│ ├── kubevirt/
│ ├── kubevirt_manager/
│ ├── multus/
│ ├── namespace/
│ ├── postgres_database/
│ ├── prometheus/
│ ├── redis_operator/
│ ├── rook_ceph_cluster/
│ │ ├── __init__.py
│ │ ├── builders/
│ │ ├── httproute.py
│ │ ├── ingress.py
│ │ ├── models/
│ │ └── routes.py
│ ├── rook_ceph_operator/
│ ├── virtual_machine/
│ │ ├── __init__.py
│ │ ├── helpers.py
│ │ ├── models/
│ │ │ ├── cloud_init.py
│ │ │ ├── component_config.py
│ │ │ ├── cpu.py
│ │ │ ├── desktop.py
│ │ │ ├── disk.py
│ │ │ ├── firmware.py
│ │ │ ├── host_device.py
│ │ │ ├── memory.py
│ │ │ ├── network.py
│ │ │ ├── snapshot_source.py
│ │ │ ├── ssh.py
│ │ │ └── vm_spec.py
│ │ └── profiles/
│ │ ├── common.py
│ │ ├── debian/
│ │ ├── kali/
│ │ └── ubuntu/
│ └── zot_registry/
└── resources/
├── base.py # BaseK8SResource
├── mixins.py # StandardMetadataMixin, DNS1123ValidationMixin
├── cilium_httproute/
├── cilium_ingress/
├── cluster_role/
├── cluster_role_binding/
├── config_file/
├── config_map/
├── custom_resource/
├── daemon_set/
├── deployment/
├── envoy_httproute/
├── envoy_reference_grant/
├── envoy_tcproute/
├── golden_image/
├── helm_chart/
│ ├── __init__.py
│ ├── cache.py
│ ├── models.py
│ └── version.py
├── kubevirt_image/
├── limit_range/
├── mutating_webhook_configuration/
├── namespace/
├── network_attachment_definition/
│ ├── __init__.py
│ ├── ipam/
│ ├── models.py
│ └── types/
│ ├── bridge.py
│ ├── ipvlan.py
│ ├── linux_bridge.py
│ ├── macvtap.py
│ ├── ovs.py
│ └── sriov.py
├── network_policy/
├── persistent_volume_claim/
├── postgresdb/
├── redis_cluster/
├── redis_sentinel/
├── resource_quota/
├── role/
├── role_binding/
├── secret/
├── service/
├── service_account/
├── storage_class/
├── validating_webhook_configuration/
└── volume_snapshot_class/
stacks/
├── pulumi-stack-config.schema.json # Generated JSON schema (619KB)
├── Pulumi.dev.yaml
├── Pulumi.dev-backstage.yaml
├── Pulumi.dev-bootstrap.yaml
├── Pulumi.dev-cloudnativepg.yaml
├── Pulumi.dev-kubevirt.yaml
├── Pulumi.optiplex-bootstrap.yaml
├── Pulumi.optiplex-cloudnativepg.yaml
├── Pulumi.optiplex-forgejo.yaml
├── Pulumi.optiplex-kubevirt.yaml
├── Pulumi.optiplex-rook-ceph.yaml
├── Pulumi.ucs-backstage.yaml
├── Pulumi.ucs-bootstrap.yaml
├── Pulumi.ucs-kubevirt.yaml
├── Pulumi.ucs-rook-ceph.yaml
└── ... (25+ stack files)
src/cache/
├── helm_chart/
│ ├── charts.jetstack.io/
│ ├── helm.cilium.io/
│ ├── charts.rook.io/
│ └── ... (cached .tgz files)
└── config_file/
├── github.com/kubernetes-sigs/gateway-api/
├── raw.githubusercontent.com/cloudnative-pg/
└── ... (cached YAML files)
- Fat Models, Thin Controllers: All business logic in Pydantic models
- Convention over Configuration: Filesystem-based discovery
- Explicit Provider: Always set provider (no ambient credentials)
- Immutable Configuration:
frozen=Trueon all Pydantic models - Production Safety: Forbid "latest" versions, enforce explicit configs
- Caching Strategy: Offline-capable with local chart/manifest caches
- Parent Chain Hierarchy: Explicit dependency ordering
- Protected Resources: Prevent accidental deletion of critical resources
- Volatile Metadata Patterns: Prevent recreation from timestamp changes
- Type-Safe Configuration: Strict Pydantic validation with
extra="forbid"
This document was generated from exhaustive code analysis of 150+ Python files across the infrastructure platform.