Skip to content

[AKS] az aks: Fix outbound type validation error msg #30886

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhoxing-ms merged 1 commit into from
Feb 28, 2025
Merged

[AKS] az aks: Fix outbound type validation error msg #30886

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
172 changes: 66 additions & 106 deletions src/azure-cli/azure/cli/command_modules/acs/managed_cluster_decorator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2053,19 +2053,6 @@ def get_nat_gateway_managed_outbound_ip_count(self) -> Union[int, None]:
"""
# read the original value passed by the command
nat_gateway_managed_outbound_ip_count = self.raw_param.get("nat_gateway_managed_outbound_ip_count")
# In create mode, try to read the property value corresponding to the parameter from the `mc` object.
if nat_gateway_managed_outbound_ip_count is None and self.decorator_mode == DecoratorMode.UPDATE:
if (
self.mc and
self.mc.network_profile and
self.mc.network_profile.nat_gateway_profile and
self.mc.network_profile.nat_gateway_profile.managed_outbound_ip_profile and
self.mc.network_profile.nat_gateway_profile.managed_outbound_ip_profile.count is not None
):
nat_gateway_managed_outbound_ip_count = (
self.mc.network_profile.nat_gateway_profile.managed_outbound_ip_profile.count
)

# this parameter does not need dynamic completion
# this parameter does not need validation
return nat_gateway_managed_outbound_ip_count
Expand All @@ -2079,17 +2066,6 @@ def get_nat_gateway_idle_timeout(self) -> Union[int, None]:
"""
# read the original value passed by the command
nat_gateway_idle_timeout = self.raw_param.get("nat_gateway_idle_timeout")
# In create mode, try to read the property value corresponding to the parameter from the `mc` object.
if nat_gateway_idle_timeout is None and self.decorator_mode == DecoratorMode.UPDATE:
if (
self.mc and
self.mc.network_profile and
self.mc.network_profile.nat_gateway_profile and
self.mc.network_profile.nat_gateway_profile.idle_timeout_in_minutes is not None
):
nat_gateway_idle_timeout = (
self.mc.network_profile.nat_gateway_profile.idle_timeout_in_minutes
)

# this parameter does not need dynamic completion
# this parameter does not need validation
Expand Down Expand Up @@ -2171,23 +2147,19 @@ def _get_outbound_type(
CONST_OUTBOUND_TYPE_LOAD_BALANCER.

This function supports the option of enable_validation. When enabled, if the value of outbound_type is
CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY, CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY or
CONST_OUTBOUND_TYPE_LOAD_BALANCER,CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY, CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY or
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING, the following checks will be performed. If load_balancer_sku is set
to basic, an InvalidArgumentValueError will be raised. If vnet_subnet_id is not assigned,
a RequiredArgumentMissingError will be raised. If any of load_balancer_managed_outbound_ip_count,
load_balancer_outbound_ips or load_balancer_outbound_ip_prefixes is assigned, a MutuallyExclusiveArgumentError
will be raised.
This function supports the option of read_only. When enabled, it will skip dynamic completion and validation.
This function supports the option of load_balancer_profile, if provided, when verifying loadbalancer-related
parameters, the value in load_balancer_profile will be used for validation.

:return: string or None
"""
# read the original value passed by the command
outbound_type = self.raw_param.get("outbound_type")
# try to read the property value corresponding to the parameter from the `mc` object
read_from_mc = False
if outbound_type is None and self.decorator_mode != DecoratorMode.CREATE:
if outbound_type is None:
if (
self.mc and
self.mc.network_profile and
Expand All @@ -2200,58 +2172,79 @@ def _get_outbound_type(
if read_only:
return outbound_type

isBasicSKULb = safe_lower(self._get_load_balancer_sku(enable_validation=False)) == CONST_LOAD_BALANCER_SKU_BASIC
# dynamic completion
if (
self.decorator_mode == DecoratorMode.CREATE and
not read_from_mc and
outbound_type != CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY and
outbound_type != CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY and
outbound_type != CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING
):
if not read_from_mc and not isBasicSKULb and outbound_type is None:
outbound_type = CONST_OUTBOUND_TYPE_LOAD_BALANCER

# validation
# Note: The parameters involved in the validation are not verified in their own getters.
if enable_validation:
if outbound_type in [
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
if not read_from_mc and outbound_type is not None and outbound_type not in [
CONST_OUTBOUND_TYPE_LOAD_BALANCER,
CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
"none"
]:
if safe_lower(self._get_load_balancer_sku(enable_validation=False)) == CONST_LOAD_BALANCER_SKU_BASIC:
raise InvalidArgumentValueError(
"Invalid outbound type, supported values are loadBalancer, managedNATGateway, userAssignedNATGateway and "
"userDefinedRouting. Please refer to "
"https://learn.microsoft.com/en-us/azure/aks/egress-outboundtype#updating-outboundtype-after-cluster-creation " # pylint:disable=line-too-long
"for more details."
)
if isBasicSKULb:
if outbound_type is not None:
raise InvalidArgumentValueError(
"userDefinedRouting doesn't support basic load balancer sku"
"{outbound_type} doesn't support basic load balancer sku".format(outbound_type=outbound_type)
)
return outbound_type # basic sku lb doesn't support outbound type

if outbound_type in [
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY,
]:
if self.get_vnet_subnet_id() in ["", None]:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userDefinedRouting and it must "
"be pre-configured with a route table with egress rules"
)

if outbound_type != CONST_OUTBOUND_TYPE_LOAD_BALANCER:
if (
self.get_load_balancer_managed_outbound_ip_count() or
self.get_load_balancer_managed_outbound_ipv6_count() or
self.get_load_balancer_outbound_ips() or
self.get_load_balancer_outbound_ip_prefixes()
):
raise MutuallyExclusiveArgumentError(
outbound_type + " doesn't support customizing "
"a standard load balancer with IP addresses"
)
if outbound_type != CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY:
if (
self.get_nat_gateway_managed_outbound_ip_count()
):
raise MutuallyExclusiveArgumentError(
outbound_type + " doesn't support customizing "
"a standard nat gateway with IP addresses"
)
if outbound_type == CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING:
if self.get_vnet_subnet_id() in ["", None]:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userDefinedRouting and it must "
"be pre-configured with a route table with egress rules"
)
if outbound_type == CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY:
if self.get_vnet_subnet_id() in ["", None]:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userAssignedNATGateway and it must "
"be pre-configured with a NAT gateway with outbound ips"
)
if outbound_type == CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY:
if self.get_vnet_subnet_id() not in ["", None]:
raise InvalidArgumentValueError(
"--vnet-subnet-id cannot be specified for managedNATGateway"
)
if outbound_type != CONST_OUTBOUND_TYPE_LOAD_BALANCER:
if (
self.get_load_balancer_managed_outbound_ip_count() or
self.get_load_balancer_managed_outbound_ipv6_count() or
self.get_load_balancer_outbound_ips() or
self.get_load_balancer_outbound_ip_prefixes()
):
raise MutuallyExclusiveArgumentError(
outbound_type + " type doesn't support customizing "
"the standard load balancer ips"
)
if (
self.get_load_balancer_idle_timeout() or
self.get_load_balancer_outbound_ports()
):
raise MutuallyExclusiveArgumentError(
outbound_type + " type doesn't support customizing "
"the standard load balancer config"
)
if outbound_type != CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY:
if (
self.get_nat_gateway_managed_outbound_ip_count() or
self.get_nat_gateway_idle_timeout()
):
raise MutuallyExclusiveArgumentError(
outbound_type + " type doesn't support customizing "
"the standard nat gateway ips"
)
return outbound_type

def get_outbound_type(
Expand All @@ -2264,16 +2257,6 @@ def get_outbound_type(
When outbound_type is not assigned, dynamic completion will be triggerd. By default, the value is set to
CONST_OUTBOUND_TYPE_LOAD_BALANCER.

This function will verify the parameter by default. If the value of outbound_type is
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING, the following checks will be performed. If load_balancer_sku is set
to basic, an InvalidArgumentValueError will be raised. If vnet_subnet_id is not assigned,
a RequiredArgumentMissingError will be raised. If any of load_balancer_managed_outbound_ip_count,
load_balancer_outbound_ips or load_balancer_outbound_ip_prefixes is assigned, a MutuallyExclusiveArgumentError
will be raised.

This function supports the option of load_balancer_profile, if provided, when verifying loadbalancer-related
parameters, the value in load_balancer_profile will be used for validation.

:return: string or None
"""
return self._get_outbound_type(
Expand Down Expand Up @@ -7189,29 +7172,6 @@ def update_outbound_type_in_network_profile(self, mc: ManagedCluster) -> Managed

outboundType = self.context.get_outbound_type()
if outboundType:
vnet_subnet_id = self.context.get_vnet_subnet_id()
if vnet_subnet_id is None and outboundType not in [
CONST_OUTBOUND_TYPE_LOAD_BALANCER,
CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING
]:
raise InvalidArgumentValueError(
"Invalid outbound type, supported values are loadBalancer, managedNATGateway and "
"userDefinedRouting. Please refer to "
"https://learn.microsoft.com/en-us/azure/aks/egress-outboundtype#updating-outboundtype-after-cluster-creation " # pylint:disable=line-too-long
"for more details."
)
if vnet_subnet_id is not None and outboundType not in [
CONST_OUTBOUND_TYPE_LOAD_BALANCER,
CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING
]:
raise InvalidArgumentValueError(
"Invalid outbound type, supported values are loadBalancer, userAssignedNATGateway and "
"userDefinedRouting. Please refer to "
"https://learn.microsoft.com/en-us/azure/aks/egress-outboundtype#updating-outboundtype-after-cluster-creation " # pylint:disable=line-too-long
"for more details."
)
mc.network_profile.outbound_type = outboundType
return mc

Expand Down Expand Up @@ -7262,9 +7222,9 @@ def update_nat_gateway_profile(self, mc: ManagedCluster) -> ManagedCluster:
mc.network_profile.nat_gateway_profile = None
else:
mc.network_profile.nat_gateway_profile = _update_nat_gateway_profile(
self.context.get_nat_gateway_managed_outbound_ip_count(),
self.context.get_nat_gateway_idle_timeout(),
mc.network_profile.nat_gateway_profile,
managed_outbound_ip_count=self.context.get_nat_gateway_managed_outbound_ip_count(),
idle_timeout=self.context.get_nat_gateway_idle_timeout(),
profile=mc.network_profile.nat_gateway_profile,
models=self.models.nat_gateway_models,
)
return mc
Expand Down
33 changes: 30 additions & 3 deletions src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_managed_cluster_decorator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@
CONST_MONITORING_LOG_ANALYTICS_WORKSPACE_RESOURCE_ID,
CONST_OPEN_SERVICE_MESH_ADDON_NAME,
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_LOAD_BALANCER,
CONST_PRIVATE_DNS_ZONE_NONE,
CONST_PRIVATE_DNS_ZONE_SYSTEM,
CONST_ROTATION_POLL_INTERVAL,
Expand Down Expand Up @@ -1747,7 +1749,7 @@ def test_get_nat_gateway_managed_outbound_ip_count(self):
network_profile=network_profile,
)
ctx_2.attach_mc(mc)
self.assertEqual(ctx_2.get_nat_gateway_managed_outbound_ip_count(), 10)
self.assertEqual(ctx_2.get_nat_gateway_managed_outbound_ip_count(), None)

ctx_2_notnull = AKSManagedClusterContext(
self.cmd,
Expand Down Expand Up @@ -1790,7 +1792,7 @@ def test_get_nat_gateway_idle_timeout(self):
network_profile=network_profile,
)
ctx_2.attach_mc(mc)
self.assertEqual(ctx_2.get_nat_gateway_idle_timeout(), 20)
self.assertEqual(ctx_2.get_nat_gateway_idle_timeout(), None)

def test_get_outbound_type(self):
# default
Expand All @@ -1805,7 +1807,7 @@ def test_get_outbound_type(self):
DecoratorMode.UPDATE,
)
self.assertEqual(ctx_1._get_outbound_type(read_only=True), None)
self.assertEqual(ctx_1.get_outbound_type(), None)
self.assertEqual(ctx_1.get_outbound_type(), CONST_OUTBOUND_TYPE_LOAD_BALANCER) # auto-fill
network_profile_1 = self.models.ContainerServiceNetworkProfile(outbound_type="test_outbound_type")
mc = self.models.ManagedCluster(location="test_location", network_profile=network_profile_1)
ctx_1.attach_mc(mc)
Expand Down Expand Up @@ -1968,6 +1970,31 @@ def test_get_outbound_type(self):
existingOutboundType = ctx_8.get_outbound_type()
self.assertEqual(existingOutboundType, "test_outbound_type")

network_profile_1 = self.models.ContainerServiceNetworkProfile(outbound_type=CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY)
mc = self.models.ManagedCluster(location="test_location", network_profile=network_profile_1)
ctx_9 = AKSManagedClusterContext(
self.cmd,
AKSManagedClusterParamDict(
{
"outbound_type": CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY,
"vnet_subnet_id": "test_vnet_subnet_id"
}
),
self.models,
DecoratorMode.UPDATE,
)
agentpool_ctx_9 = AKSAgentPoolContext(
self.cmd,
AKSAgentPoolParamDict({"vnet_subnet_id": "test_vnet_subnet_id"}),
self.models,
DecoratorMode.UPDATE,
AgentPoolDecoratorMode.MANAGED_CLUSTER,
)
ctx_9.attach_agentpool_context(agentpool_ctx_9)
ctx_9.attach_mc(mc)
with self.assertRaises(InvalidArgumentValueError):
ctx_9.get_outbound_type()

def test_get_network_plugin_mode(self):
# default
ctx_1 = AKSManagedClusterContext(
Expand Down