(
        self, timestamp: arrow.Arrow, retry_on_broken_pipe: bool = True,
    )
Communicate over a Unix socket with the signal to evaluate its result

        :param timestamp: a Unix timestamp to pass to the signal as the "current time"
        :param retry_on_broken_pipe: if the signal socket pipe is broken, restart the signal process and try again
        :returns: a dict of resource_name -> requested resources from the signal
        :raises SignalConnectionError: if the signal connection fails for some reason
Communicate over a Unix socket with the signal to evaluate its result
Communicate over a Unix socket with the signal to evaluate its result
def evaluate(
        self, timestamp: arrow.Arrow, retry_on_broken_pipe: bool = True,
    ) -> Union[SignalResourceRequest, List[SignalResourceRequest]]:
        """ Communicate over a Unix socket with the signal to evaluate its result

        :param timestamp: a Unix timestamp to pass to the signal as the "current time"
        :param retry_on_broken_pipe: if the signal socket pipe is broken, restart the signal process and try again
        :returns: a dict of resource_name -> requested resources from the signal
        :raises SignalConnectionError: if the signal connection fails for some reason
        """
        # Get the required metrics for the signal
        metrics = get_metrics_for_signal(
            self.cluster, self.pool, self.scheduler, self.app, self.metrics_client, self.required_metrics, timestamp,
        )

        try:
            # First send the length of the metrics data
            metric_bytes = json.dumps({"metrics": metrics, "timestamp": timestamp.timestamp}).encode()
            len_metrics = struct.pack(">I", len(metric_bytes))  # bytes representation of the length, packed big-endian
            self._signal_conn.send(len_metrics)
            response = self._signal_conn.recv(SOCKET_MESG_SIZE)
            if response != ACK:
                raise SignalConnectionError(f"Error occurred sending metric length to signal (response={response})")

            # Then send the actual metrics data, broken up into chunks
            for i in range(0, len(metric_bytes), SOCKET_MESG_SIZE):
                self._signal_conn.send(metric_bytes[i : i + SOCKET_MESG_SIZE])
            response = self._signal_conn.recv(SOCKET_MESG_SIZE)
            ack_bit = response[:1]
            if ack_bit != ACK:
                raise SignalConnectionError(f"Error occurred sending metric data to signal (response={response})")

            # Sometimes the signal sends the ack and the reponse "too quickly" so when we call
            # recv above it gets both values.  This should handle that case, or call recv again
            # if there's no more data in the previous message
            response = response[1:] or self._signal_conn.recv(SOCKET_MESG_SIZE)
            logger.info(response)

            return SignalResourceRequest(**json.loads(response)["Resources"])

        except JSONDecodeError as e:
            raise ClustermanSignalError("Signal evaluation failed") from e
        except BrokenPipeError as e:
            if retry_on_broken_pipe:
                logger.error("Signal connection failed; reloading the signal and trying again")
                time.sleep(5)  # give supervisord some time to restart the signal
                self._signal_conn = self._connect_to_signal_process()
                return self.evaluate(timestamp, retry_on_broken_pipe=False)
            else:
                raise ClustermanSignalError("Signal evaluation failed") from e
def evaluate ( self , timestamp : arrow . Arrow , retry_on_broken_pipe : bool = True , ) -> Union [ SignalResourceRequest , List [ SignalResourceRequest ] ] : # Get the required metrics for the signal metrics = get_metrics_for_signal ( self . cluster , self . pool , self . scheduler , self . app , self . metrics_client , self . required_metrics , timestamp , ) try : # First send the length of the metrics data metric_bytes = json . dumps ( { "metrics" : metrics , "timestamp" : timestamp . timestamp } ) . encode ( ) len_metrics = struct . pack ( ">I" , len ( metric_bytes ) ) # bytes representation of the length, packed big-endian self . _signal_conn . send ( len_metrics ) response = self . _signal_conn . recv ( SOCKET_MESG_SIZE ) if response != ACK : raise SignalConnectionError ( f"Error occurred sending metric length to signal (response={response})" ) # Then send the actual metrics data, broken up into chunks for i in range ( 0 , len ( metric_bytes ) , SOCKET_MESG_SIZE ) : self . _signal_conn . send ( metric_bytes [ i : i + SOCKET_MESG_SIZE ] ) response = self . _signal_conn . recv ( SOCKET_MESG_SIZE ) ack_bit = response [ : 1 ] if ack_bit != ACK : raise SignalConnectionError ( f"Error occurred sending metric data to signal (response={response})" ) # Sometimes the signal sends the ack and the reponse "too quickly" so when we call # recv above it gets both values.  This should handle that case, or call recv again # if there's no more data in the previous message response = response [ 1 : ] or self . _signal_conn . recv ( SOCKET_MESG_SIZE ) logger . info ( response ) return SignalResourceRequest ( * * json . loads ( response ) [ "Resources" ] ) except JSONDecodeError as e : raise ClustermanSignalError ( "Signal evaluation failed" ) from e except BrokenPipeError as e : if retry_on_broken_pipe : logger . error ( "Signal connection failed; reloading the signal and trying again" ) time . sleep ( 5 ) # give supervisord some time to restart the signal self . _signal_conn = self . _connect_to_signal_process ( ) return self . evaluate ( timestamp , retry_on_broken_pipe = False ) else : raise ClustermanSignalError ( "Signal evaluation failed" ) from e