(self, app: str)
Load the signal object to use for autoscaling for a particular app

        :param app: the name of the app to load a Signal for
        :returns: the configured app signal, or the default signal in case of an error
Load the signal object to use for autoscaling for a particular app
Load the signal object to use for autoscaling for a particular app
def _get_signal_for_app(self, app: str) -> Signal:
        """Load the signal object to use for autoscaling for a particular app

        :param app: the name of the app to load a Signal for
        :returns: the configured app signal, or the default signal in case of an error
        """
        logger.info(f"Loading autoscaling signal for {app} on {self.pool} in {self.cluster}")

        # TODO (CLUSTERMAN-126, CLUSTERMAN-195) apps will eventually have separate namespaces from pools
        pool_namespace = POOL_NAMESPACE.format(pool=app, scheduler=self.scheduler)

        try:
            # see if the pool has set up a custom signal correctly; if not, fall back to the default signal
            if staticconf.read_bool("autoscale_signal.internal", default=False, namespace=pool_namespace):
                # we should never get here unless we're on Kubernetes; this assert makes mypy happy
                assert isinstance(self.pool_manager.cluster_connector, KubernetesClusterConnector)
                return PendingPodsSignal(
                    self.cluster,
                    self.pool,
                    self.scheduler,
                    app,
                    pool_namespace,
                    self.metrics_client,
                    self.pool_manager.cluster_connector,
                )
            return ExternalSignal(
                self.cluster,
                self.pool,
                self.scheduler,
                app,
                pool_namespace,
                self.metrics_client,
                signal_namespace=staticconf.read_string(
                    "autoscale_signal.namespace", default=app, namespace=pool_namespace,
                ),
            )
        except NoSignalConfiguredException:
            logger.info(f"No signal configured for {app}, falling back to default")
            return self.default_signal
        except Exception:
            msg = f"WARNING: loading signal for {app} failed, falling back to default"
            logger.exception(msg)
            sensu_checkin(
                check_name=SIGNAL_LOAD_CHECK_NAME,
                status=Status.WARNING,
                output=msg,
                source=self.cluster,
                scheduler=self.scheduler,
                page=False,
                ttl=None,
                app=app,
                noop=not self.monitoring_enabled,
                pool=self.pool,
            )
            return self.default_signal
def _get_signal_for_app ( self , app : str ) -> Signal : logger . info ( f"Loading autoscaling signal for {app} on {self.pool} in {self.cluster}" ) # TODO (CLUSTERMAN-126, CLUSTERMAN-195) apps will eventually have separate namespaces from pools pool_namespace = POOL_NAMESPACE . format ( pool = app , scheduler = self . scheduler ) try : # see if the pool has set up a custom signal correctly; if not, fall back to the default signal if staticconf . read_bool ( "autoscale_signal.internal" , default = False , namespace = pool_namespace ) : # we should never get here unless we're on Kubernetes; this assert makes mypy happy assert isinstance ( self . pool_manager . cluster_connector , KubernetesClusterConnector ) return PendingPodsSignal ( self . cluster , self . pool , self . scheduler , app , pool_namespace , self . metrics_client , self . pool_manager . cluster_connector , ) return ExternalSignal ( self . cluster , self . pool , self . scheduler , app , pool_namespace , self . metrics_client , signal_namespace = staticconf . read_string ( "autoscale_signal.namespace" , default = app , namespace = pool_namespace , ) , ) except NoSignalConfiguredException : logger . info ( f"No signal configured for {app}, falling back to default" ) return self . default_signal except Exception : msg = f"WARNING: loading signal for {app} failed, falling back to default" logger . exception ( msg ) sensu_checkin ( check_name = SIGNAL_LOAD_CHECK_NAME , status = Status . WARNING , output = msg , source = self . cluster , scheduler = self . scheduler , page = False , ttl = None , app = app , noop = not self . monitoring_enabled , pool = self . pool , ) return self . default_signal