(
        self,
        cluster: str,
        pool: str,
        scheduler: str,
        apps: List[str],
        pool_manager: Optional[PoolManager] = None,
        metrics_client: Optional[ClustermanMetricsBotoClient] = None,
        monitoring_enabled: bool = True,
    )
Class containing the core logic for autoscaling a cluster

        :param cluster: the name of the cluster to autoscale
        :param pool: the name of the pool to autoscale
        :param apps: a list of apps running on the pool
        :param pool_manager: a PoolManager object (used for simulations)
        :param metrics_client: a ClustermanMetricsBotoClient object (used for simulations)
        :param monitoring_enabled: set to False to disable sensu alerts during scaling
Class containing the core logic for autoscaling a cluster
Class containing the core logic for autoscaling a cluster
def __init__(
        self,
        cluster: str,
        pool: str,
        scheduler: str,
        apps: List[str],
        pool_manager: Optional[PoolManager] = None,
        metrics_client: Optional[ClustermanMetricsBotoClient] = None,
        monitoring_enabled: bool = True,
    ) -> None:
        """ Class containing the core logic for autoscaling a cluster

        :param cluster: the name of the cluster to autoscale
        :param pool: the name of the pool to autoscale
        :param apps: a list of apps running on the pool
        :param pool_manager: a PoolManager object (used for simulations)
        :param metrics_client: a ClustermanMetricsBotoClient object (used for simulations)
        :param monitoring_enabled: set to False to disable sensu alerts during scaling
        """
        self.cluster = cluster
        self.pool = pool
        self.scheduler = scheduler
        self.apps = apps
        self.monitoring_enabled = monitoring_enabled

        # TODO: handle multiple apps in the autoscaler (CLUSTERMAN-126)
        if len(self.apps) > 1:
            raise NotImplementedError("Scaling multiple apps in a cluster is not yet supported")

        logger.info(f"Initializing autoscaler engine for {self.pool} in {self.cluster}...")

        gauge_dimensions = {"cluster": cluster, "pool": pool}
        monitoring_client = get_monitoring_client()
        self.target_capacity_gauge = monitoring_client.create_gauge(TARGET_CAPACITY_GAUGE_NAME, gauge_dimensions)
        self.resource_request_gauges: Dict[str, Any] = {}
        for resource in SignalResourceRequest._fields:
            self.resource_request_gauges[resource] = monitoring_client.create_gauge(
                RESOURCE_GAUGE_BASE_NAME.format(resource=resource), gauge_dimensions,
            )

        self.autoscaling_config = get_autoscaling_config(
            POOL_NAMESPACE.format(pool=self.pool, scheduler=self.scheduler),
        )
        self.pool_manager = pool_manager or PoolManager(self.cluster, self.pool, self.scheduler)

        self.mesos_region = staticconf.read_string("aws.region")
        self.metrics_client = metrics_client or ClustermanMetricsBotoClient(self.mesos_region)
        self.default_signal: Signal
        if staticconf.read_bool("autoscale_signal.internal", default=False):
            # we should never get here unless we're on Kubernetes; this assert makes mypy happy
            assert isinstance(self.pool_manager.cluster_connector, KubernetesClusterConnector)
            self.default_signal = PendingPodsSignal(
                self.cluster,
                self.pool,
                self.scheduler,
                "__default__",
                DEFAULT_NAMESPACE,
                self.metrics_client,
                self.pool_manager.cluster_connector,
            )
        else:
            self.default_signal = ExternalSignal(
                self.cluster,
                self.pool,
                self.scheduler,
                "__default__",
                DEFAULT_NAMESPACE,
                self.metrics_client,
                signal_namespace=staticconf.read_string("autoscaling.default_signal_role"),
            )
        self.signal = self._get_signal_for_app(self.apps[0])
        logger.info("Initialization complete")
def __init__ ( self , cluster : str , pool : str , scheduler : str , apps : List [ str ] , pool_manager : Optional [ PoolManager ] = None , metrics_client : Optional [ ClustermanMetricsBotoClient ] = None , monitoring_enabled : bool = True , ) -> None : self . cluster = cluster self . pool = pool self . scheduler = scheduler self . apps = apps self . monitoring_enabled = monitoring_enabled # TODO: handle multiple apps in the autoscaler (CLUSTERMAN-126) if len ( self . apps ) > 1 : raise NotImplementedError ( "Scaling multiple apps in a cluster is not yet supported" ) logger . info ( f"Initializing autoscaler engine for {self.pool} in {self.cluster}..." ) gauge_dimensions = { "cluster" : cluster , "pool" : pool } monitoring_client = get_monitoring_client ( ) self . target_capacity_gauge = monitoring_client . create_gauge ( TARGET_CAPACITY_GAUGE_NAME , gauge_dimensions ) self . resource_request_gauges : Dict [ str , Any ] = { } for resource in SignalResourceRequest . _fields : self . resource_request_gauges [ resource ] = monitoring_client . create_gauge ( RESOURCE_GAUGE_BASE_NAME . format ( resource = resource ) , gauge_dimensions , ) self . autoscaling_config = get_autoscaling_config ( POOL_NAMESPACE . format ( pool = self . pool , scheduler = self . scheduler ) , ) self . pool_manager = pool_manager or PoolManager ( self . cluster , self . pool , self . scheduler ) self . mesos_region = staticconf . read_string ( "aws.region" ) self . metrics_client = metrics_client or ClustermanMetricsBotoClient ( self . mesos_region ) self . default_signal : Signal if staticconf . read_bool ( "autoscale_signal.internal" , default = False ) : # we should never get here unless we're on Kubernetes; this assert makes mypy happy assert isinstance ( self . pool_manager . cluster_connector , KubernetesClusterConnector ) self . default_signal = PendingPodsSignal ( self . cluster , self . pool , self . scheduler , "__default__" , DEFAULT_NAMESPACE , self . metrics_client , self . pool_manager . cluster_connector , ) else : self . default_signal = ExternalSignal ( self . cluster , self . pool , self . scheduler , "__default__" , DEFAULT_NAMESPACE , self . metrics_client , signal_namespace = staticconf . read_string ( "autoscaling.default_signal_role" ) , ) self . signal = self . _get_signal_for_app ( self . apps [ 0 ] ) logger . info ( "Initialization complete" )