import base64
import pulumi
import pulumi_azure_native as azure_native
import pulumi_kubernetes as k8s
import pulumi_random
class Kubernetes(pulumi.ComponentResource):
def __init__(
self,
name: str,
subnet_id: pulumi.Input[str],
system_pool: dict,
kubernetes_version: str,
availability_zones: list[str] = None,
opts: pulumi.ResourceOptions = None
):
super().__init__("workshop:azure:Kubernetes", name, {}, opts)
self._subnet_id = subnet_id
self._availability_zones = availability_zones
self.child_opts = pulumi.ResourceOptions(parent=self)
self.rg = azure_native.resources.ResourceGroup(
name,
resource_group_name=f"rg-{name}",
opts=self.child_opts
)
self.egress_ip = azure_native.network.PublicIPAddress(
f"egress-{name}",
resource_group_name=self.rg.name,
public_ip_address_name=f"pip-egress-{name}",
sku=azure_native.network.PublicIPAddressSkuArgs(name="Standard"),
public_ip_allocation_method="Static",
opts=self.child_opts
)
self.cluster = azure_native.containerservice.ManagedCluster(
name,
resource_name_=name,
resource_group_name=self.rg.name,
dns_prefix=name,
node_resource_group=f"rg-nodes-{name}",
kubernetes_version=kubernetes_version,
identity=azure_native.containerservice.ManagedClusterIdentityArgs(
type=azure_native.containerservice.ResourceIdentityType.SYSTEM_ASSIGNED
),
aad_profile=azure_native.containerservice.ManagedClusterAADProfileArgs(
managed=True,
enable_azure_rbac=True
),
network_profile=azure_native.containerservice.ContainerServiceNetworkProfileArgs(
network_plugin="azure",
load_balancer_sku="standard",
load_balancer_profile=azure_native.containerservice.ManagedClusterLoadBalancerProfileArgs(
outbound_ips=azure_native.containerservice.ManagedClusterLoadBalancerProfileOutboundIPsArgs(
public_ips=[azure_native.containerservice.ResourceReferenceArgs(
id=self.egress_ip.id
)]
)
)
),
agent_pool_profiles=[
azure_native.containerservice.ManagedClusterAgentPoolProfileArgs(
**system_pool,
vnet_subnet_id=subnet_id,
availability_zones=availability_zones
)
],
opts=self.child_opts
)
client = azure_native.authorization.get_client_config_output()
azure_native.authorization.RoleAssignment(
f"{name}-egress-network-contributor",
scope=self.egress_ip.id,
principal_id=self.cluster.identity.principal_id,
principal_type="ServicePrincipal",
role_definition_id=client.subscription_id.apply(
lambda sub: (
f"/subscriptions/{sub}/providers/Microsoft.Authorization"
f"/roleDefinitions/4d97b98b-1d4f-4787-a291-c67834d212e7" # Network Contributor
)
),
opts=self.child_opts
)
creds = azure_native.containerservice.list_managed_cluster_admin_credentials_output(
resource_group_name=self.rg.name,
resource_name=self.cluster.name,
opts=pulumi.InvokeOptions(parent=self)
)
self.kubeconfig = creds.kubeconfigs[0].value.apply(
lambda v: base64.b64decode(v).decode()
)
self._k8s_provider = k8s.Provider(
f"{name}-k8s",
kubeconfig=self.kubeconfig,
opts=self.child_opts
)
def install_helm_chart(
self,
name: str,
chart: str,
repo: str,
namespace: str,
values: dict = None,
create_namespace: bool = True,
version: str = None,
timeout: int = None
) -> k8s.helm.v3.Release:
args = k8s.helm.v3.ReleaseArgs(
name=name,
chart=chart,
repository_opts=k8s.helm.v3.RepositoryOptsArgs(repo=repo),
version=version,
namespace=namespace,
create_namespace=create_namespace,
values=values or {},
timeout=timeout
)
return k8s.helm.v3.Release(
name,
args,
opts=pulumi.ResourceOptions.merge(
self.child_opts,
pulumi.ResourceOptions(provider=self._k8s_provider)
)
)
def grant_role(
self,
role_definition_id: str,
principal_id: str | pulumi.Output[str]
) -> azure_native.authorization.RoleAssignment:
client = azure_native.authorization.get_client_config_output()
return azure_native.authorization.RoleAssignment(
f"role-{role_definition_id}",
scope=self.cluster.id,
principal_id=principal_id,
role_definition_id=client.subscription_id.apply(
lambda sub: (
f"/subscriptions/{sub}/providers/Microsoft.Authorization"
f"/roleDefinitions/{role_definition_id}"
)
),
opts=self.child_opts
)
def add_registry(self, name: str) -> azure_native.containerregistry.Registry:
rand = pulumi_random.RandomInteger(
f"{name}-suffix",
min=10000000,
max=99999999,
opts=self.child_opts
)
registry = azure_native.containerregistry.Registry(
name,
registry_name=rand.result.apply(lambda n: f"cr{n}"),
resource_group_name=self.rg.name,
sku=azure_native.containerregistry.SkuArgs(name="Basic"),
admin_user_enabled=True,
opts=self.child_opts
)
client = azure_native.authorization.get_client_config_output()
kubelet_object_id = self.cluster.identity_profile.apply(
lambda profile: profile["kubeletidentity"].object_id
)
azure_native.authorization.RoleAssignment(
f"{name}-acrpull",
scope=registry.id,
principal_id=kubelet_object_id,
principal_type="ServicePrincipal",
role_definition_id=client.subscription_id.apply(
lambda sub: (
f"/subscriptions/{sub}/providers/Microsoft.Authorization"
f"/roleDefinitions/7f951dda-4ed3-4680-a7ca-43fe172d538d" # AcrPull
)
),
opts=self.child_opts
)
return registry
def add_public_ip(self, name: str) -> azure_native.network.PublicIPAddress:
return azure_native.network.PublicIPAddress(
name,
resource_group_name=self.cluster.node_resource_group,
public_ip_address_name=f"pip-{name}",
sku=azure_native.network.PublicIPAddressSkuArgs(name="Standard"),
public_ip_allocation_method="Static",
opts=self.child_opts
)
def add_node_pool(
self,
name: str,
**kwargs
) -> azure_native.containerservice.AgentPool:
return azure_native.containerservice.AgentPool(
name,
agent_pool_name=name,
resource_group_name=self.rg.name,
resource_name_=self.cluster.name,
vnet_subnet_id=self._subnet_id,
availability_zones=self._availability_zones,
**kwargs,
opts=self.child_opts
)