def __init__()

in perfkitbenchmarker/providers/gcp/gce_network.py [0:0]

• 132 lines of code
• 25 McCabe index (conditional complexity)

  def __init__(self, network_spec: GceNetworkSpec):
    super().__init__(network_spec)
    self.project: str | None = network_spec.project
    self.vpn_gateway: Dict[str, GceVpnGateway] = {}

    #  Figuring out the type of network here.
    #  Precedence: User Managed > MULTI > SINGLE > DEFAULT
    self.net_type = network.NetType.DEFAULT.value
    self.cidr = NETWORK_RANGE
    if FLAGS.gce_subnet_region:
      self.net_type = network.NetType.SINGLE.value
      self.cidr = FLAGS.gce_subnet_addr
    if network_spec.cidr:
      self.net_type = network.NetType.MULTI.value
      self.cidr = network_spec.cidr
    self.mtu = network_spec.mtu

    # TODO(user): Create separate Network objects for each network name.
    self.is_existing_network = True
    self.subnet_names = []
    self.primary_subnet_name = None
    if network_spec.subnet_names:
      self.subnet_names = network_spec.subnet_names
    elif gcp_flags.GCE_NETWORK_NAMES.value:
      self.subnet_names = gcp_flags.GCE_NETWORK_NAMES.value
    else:
      self.subnet_names = self._MakeGceNetworkName()
      self.is_existing_network = False
    if not isinstance(self.subnet_names, list):
      self.subnet_names = [self.subnet_names]
    self.primary_subnet_name = self.subnet_names[0]

    self.network_resources = []
    self.subnet_resources = []
    mode = gcp_flags.GCE_NETWORK_TYPE.value
    self.subnet_resource = None
    if not self.is_existing_network or mode == 'legacy':
      for name in self.subnet_names:
        self.network_resources.append(
            GceNetworkResource(name, mode, self.project, self.mtu)
        )
    if (self.is_existing_network and mode != 'legacy') or (mode == 'custom'):
      subnet_region = util.GetRegionFromZone(network_spec.zone)
      for name in self.subnet_names:
        self.subnet_resources.append(
            GceSubnetResource(
                name, name, subnet_region, self.cidr, self.project
            )
        )
      self.subnet_resource = GceSubnetResource(
          self.primary_subnet_name,
          self.primary_subnet_name,
          subnet_region,
          self.cidr,
          self.project,
      )
    self.network_resource = GceNetworkResource(
        self.primary_subnet_name, mode, self.project, self.mtu
    )
    # Stage FW rules.
    self.all_nets = self._GetNetworksFromSpec(
        network_spec
    )  # Holds the different networks in this run.
    # Holds FW rules for any external subnets.
    self.external_nets_rules: Dict[str, GceFirewallRule] = {}

    #  Set the default rule to allow all traffic within this network's subnet.
    firewall_name = self._MakeGceFWRuleName()
    self.default_firewall_rule = GceFirewallRule(
        firewall_name,
        self.project,
        ALLOW_ALL,
        self.primary_subnet_name,
        self.cidr,
    )

    # Set external rules to allow traffic from other subnets in this benchmark.
    for ext_net in self.all_nets:
      if ext_net == self.cidr:
        continue  # We've already added our own network to the default rule.
      rule_name = self._MakeGceFWRuleName(dst_cidr=ext_net)
      self.external_nets_rules[rule_name] = GceFirewallRule(
          rule_name, self.project, ALLOW_ALL, self.primary_subnet_name, ext_net
      )

    # Add VpnGateways to the network.
    if FLAGS.use_vpn:
      for gatewaynum in range(0, FLAGS.vpn_service_gateway_count):
        vpn_gateway_name = 'vpngw-%s-%s-%s' % (
            util.GetRegionFromZone(network_spec.zone),
            gatewaynum,
            FLAGS.run_uri,
        )
        self.vpn_gateway[vpn_gateway_name] = GceVpnGateway(
            vpn_gateway_name,
            self.primary_subnet_name,
            util.GetRegionFromZone(network_spec.zone),
            network_spec.cidr,
            self.project,
        )

    # Placement Group
    no_placement_group = (
        not FLAGS.placement_group_style
        or FLAGS.placement_group_style == placement_group.PLACEMENT_GROUP_NONE
    )
    has_optional_pg = (
        FLAGS.placement_group_style
        == placement_group.PLACEMENT_GROUP_CLOSEST_SUPPORTED
    )
    if no_placement_group:
      self.placement_group = None
    elif has_optional_pg and not IsPlacementGroupCompatible(
        network_spec.machine_type
    ):
      logging.warning(
          'machine type %s does not support placement groups. '
          'Placement group style set to none.',
          network_spec.machine_type,
      )
      self.placement_group = None
    elif has_optional_pg and len(set(FLAGS.zone)) > 1:
      logging.warning(
          'inter-zone/inter-region tests do not support placement groups. '
          'Placement group style set to none.'
      )
      self.placement_group = None
    elif not IsPlacementGroupCompatible(network_spec.machine_type):
      raise errors.Benchmarks.UnsupportedConfigError(
          f'machine type {network_spec.machine_type} does not support '
          'placement groups. Use placement group style none.'
      )
    elif len(set(FLAGS.zone)) > 1:
      raise errors.Benchmarks.UnsupportedConfigError(
          'inter-zone/inter-region tests do not support placement groups. '
          'Use placement group style closest_supported.'
      )
    else:
      placement_group_spec = gce_placement_group.GcePlacementGroupSpec(
          'GcePlacementGroupSpec',
          flag_values=FLAGS,
          zone=network_spec.zone,
          project=self.project,
          num_vms=self._GetNumberVms(),
      )
      self.placement_group = gce_placement_group.GcePlacementGroup(
          placement_group_spec
      )