def _evaluate_launch_params()

in src/slurm_plugin/fleet_manager.py [0:0]

• 54 lines of code
• 19 McCabe index (conditional complexity)

    def _evaluate_launch_params(self, count):
        """Evaluate parameters to be passed to create_fleet call."""
        try:
            common_launch_options = {
                "SingleInstanceType": self._uses_single_instance_type(),
                "SingleAvailabilityZone": self._uses_single_az(),  # If using Multi-AZ (by specifying multiple subnets),
                # set SingleAvailabilityZone to False
            }
            allocation_strategy = self._compute_resource_config.get("AllocationStrategy")
            if allocation_strategy:
                # AllocationStrategy can assume different values for SpotOptions and OnDemandOptions
                # and is not set for Capacity Block
                common_launch_options.update({"AllocationStrategy": allocation_strategy})

            if self._uses_single_az() or self._uses_single_instance_type():
                # If the minimum target capacity is not reached, the fleet launches no instances
                common_launch_options.update({"MinTargetCapacity": count if self._all_or_nothing else 1})

            if not self._uses_single_az() and not self._uses_single_instance_type() and self._all_or_nothing:
                logger.warning(
                    "All-or-Nothing is only available with single instance type compute resources or "
                    "single subnet queues"
                )

            if self._compute_resource_config["CapacityType"] == "spot":
                launch_options = {"SpotOptions": common_launch_options}
            else:
                # This is valid for both on-demand and capacity-block
                launch_options = {
                    "OnDemandOptions": {
                        **common_launch_options,
                        "CapacityReservationOptions": {"UsageStrategy": "use-capacity-reservations-first"},
                    },
                }

            template_overrides = self._evaluate_template_overrides()

            launch_params = {
                "LaunchTemplateConfigs": [
                    {
                        "LaunchTemplateSpecification": {
                            # LaunchTemplate is different for every compute resources in every queue
                            "LaunchTemplateName": f"{self._cluster_name}-{self._queue}-{self._compute_resource}",
                            "Version": "$Latest",
                        },
                        "Overrides": template_overrides,
                    }
                ],
                "TargetCapacitySpecification": {
                    "TotalTargetCapacity": count,
                    "DefaultTargetCapacityType": self._compute_resource_config["CapacityType"],
                },
                "Type": "instant",
                **launch_options,
                # TODO verify if we need to add user's tag in "TagSpecifications": []
            }
        except KeyError as e:
            message = (
                f"Unable to find key {e} in the configuration of queue: {self._queue}, "
                f"compute resource {self._compute_resource}"
            )
            logger.error(message)
            raise FleetManagerException(message)

        launch_params.update(self._launch_overrides)
        if self._launch_overrides:
            logger.info("Found CreateFleet parameters override. Launching instances with: %s", launch_params)
        return launch_params