(self, name: str, cluster: str, pool: str, scheduler: str, app: str, config_namespace: str,)
Create an encapsulation of the Unix sockets via which we communicate with signals

        :param cluster: the name of the cluster this signal is for
        :param pool: the name of the pool this signal is for
        :param app: the name of the application this signal is for
        :param config_namespace: the staticconf namespace we can find the signal config in
        :param metrics_client: the metrics client to use to populate signal metrics
        :param signal_namespace: the namespace in the signals repo to find the signal class
            (if this is None, we default to the app name)
Create an encapsulation of the Unix sockets via which we communicate with signals
Create an encapsulation of the Unix sockets via which we communicate with signals
def __init__(self, name: str, cluster: str, pool: str, scheduler: str, app: str, config_namespace: str,) -> None:
        """ Create an encapsulation of the Unix sockets via which we communicate with signals

        :param cluster: the name of the cluster this signal is for
        :param pool: the name of the pool this signal is for
        :param app: the name of the application this signal is for
        :param config_namespace: the staticconf namespace we can find the signal config in
        :param metrics_client: the metrics client to use to populate signal metrics
        :param signal_namespace: the namespace in the signals repo to find the signal class
            (if this is None, we default to the app name)
        """
        reader = staticconf.NamespaceReaders(config_namespace)

        self.name = name
        self.cluster: str = cluster
        self.pool: str = pool
        self.scheduler: str = scheduler
        self.app: str = app

        self.period_minutes: int = reader.read_int("autoscale_signal.period_minutes")
        if self.period_minutes <= 0:
            raise SignalValidationError(f"Length of signal period must be positive, got {self.period_minutes}")

        self.parameters: Dict = {
            key: value
            for param_dict in reader.read_list("autoscale_signal.parameters", default=[])
            for (key, value) in param_dict.items()
        }
        # Even if cluster and pool were set in parameters, we override them here
        # as we want to preserve a single source of truth
        self.parameters.update(dict(cluster=self.cluster, pool=self.pool,))
def __init__ ( self , name : str , cluster : str , pool : str , scheduler : str , app : str , config_namespace : str , ) -> None : reader = staticconf . NamespaceReaders ( config_namespace ) self . name = name self . cluster : str = cluster self . pool : str = pool self . scheduler : str = scheduler self . app : str = app self . period_minutes : int = reader . read_int ( "autoscale_signal.period_minutes" ) if self . period_minutes <= 0 : raise SignalValidationError ( f"Length of signal period must be positive, got {self.period_minutes}" ) self . parameters : Dict = { key : value for param_dict in reader . read_list ( "autoscale_signal.parameters" , default = [ ] ) for ( key , value ) in param_dict . items ( ) } # Even if cluster and pool were set in parameters, we override them here # as we want to preserve a single source of truth self . parameters . update ( dict ( cluster = self . cluster , pool = self . pool , ) )