(
        self,
        cluster: str,
        pool: str,
        scheduler: str,
        app: str,
        config_namespace: str,
        metrics_client: ClustermanMetricsBotoClient,
        signal_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,
        cluster: str,
        pool: str,
        scheduler: str,
        app: str,
        config_namespace: str,
        metrics_client: ClustermanMetricsBotoClient,
        signal_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)
        try:
            signal_name = reader.read_string("autoscale_signal.name")
        except ConfigurationError as e:
            raise NoSignalConfiguredException from e
        super().__init__(signal_name, cluster, pool, scheduler, app, config_namespace)
        self.required_metrics: list = reader.read_list("autoscale_signal.required_metrics", default=[])

        self.metrics_client: ClustermanMetricsBotoClient = metrics_client
        self.signal_namespace = signal_namespace
        self._signal_conn: socket.socket = self._connect_to_signal_process()
def __init__ ( self , cluster : str , pool : str , scheduler : str , app : str , config_namespace : str , metrics_client : ClustermanMetricsBotoClient , signal_namespace : str , ) -> None : reader = staticconf . NamespaceReaders ( config_namespace ) try : signal_name = reader . read_string ( "autoscale_signal.name" ) except ConfigurationError as e : raise NoSignalConfiguredException from e super ( ) . __init__ ( signal_name , cluster , pool , scheduler , app , config_namespace ) self . required_metrics : list = reader . read_list ( "autoscale_signal.required_metrics" , default = [ ] ) self . metrics_client : ClustermanMetricsBotoClient = metrics_client self . signal_namespace = signal_namespace self . _signal_conn : socket . socket = self . _connect_to_signal_process ( )