(api_token, program, start_time, end_time, dimensions=None, resolution=60)
return datapoints
Execute an arbitrary SignalFlow program

    :param api_token: a valid SFX API query token (you can get this from the SignalFX dashboard)
    :param program: a valid signalflow program to execute
    :param start_time: beginning of program execution range, as an Arrow object
    :param end_time: end of program execution range, as an Arrow object
    :param dimensions: list of strings to group the returned timeseries by
    :param resolution: smallest time interval (in seconds) to evaluate the program on
        note: SignalFX has a maximum resolution of 1 minute, and only for the most recent data;
              setting a resolution higher than this (or even 1 minute for older data) will be ignored
    :returns: a list of (timestamp, data_points) tuples, where data_points is a dict of timeseries_name -> value
Execute an arbitrary SignalFlow program
Execute an arbitrary SignalFlow program
def execute_sfx_program(api_token, program, start_time, end_time, dimensions=None, resolution=60):
    """ Execute an arbitrary SignalFlow program

    :param api_token: a valid SFX API query token (you can get this from the SignalFX dashboard)
    :param program: a valid signalflow program to execute
    :param start_time: beginning of program execution range, as an Arrow object
    :param end_time: end of program execution range, as an Arrow object
    :param dimensions: list of strings to group the returned timeseries by
    :param resolution: smallest time interval (in seconds) to evaluate the program on
        note: SignalFX has a maximum resolution of 1 minute, and only for the most recent data;
              setting a resolution higher than this (or even 1 minute for older data) will be ignored
    :returns: a list of (timestamp, data_points) tuples, where data_points is a dict of timeseries_name -> value
    """
    with signalfx.SignalFx().signalflow(api_token) as sfx:
        curr_time = start_time
        datapoints = []
        while curr_time < end_time:
            # To prevent overloading SignalFX we grab a maximum of 5 days worth of data at a time
            next_time = min(curr_time.shift(days=5), end_time)
            logger.info(f"Querying SignalFX from {curr_time} to {next_time}")
            raw_data = sfx.execute(
                program,
                # SignalFX operates on millisecond timescales
                start=curr_time.timestamp * 1000,
                stop=next_time.timestamp * 1000,
                resolution=resolution * 1000,
            )

            # We can only call _make_ts_label after all of the entries in the raw_data.stream() have been processed
            data_messages = [msg for msg in raw_data.stream() if isinstance(msg, DataMessage)]
            new_datapoints = sorted(
                [
                    (
                        Arrow.utcfromtimestamp(msg.logical_timestamp_ms / 1000),
                        {_make_ts_label(raw_data, key, dimensions): value for key, value in msg.data.items()},
                    )
                    for msg in data_messages
                ]
            )

            # SignalFX sometimes gives us duplicate datapoints at the beginning of one chunk/the start of
            # the next chunk.  This doesn't play nicely with the metrics client so detect and remove those here
            if datapoints and new_datapoints[0][0] == datapoints[-1][0]:
                new_datapoints = new_datapoints[1:]
            datapoints.extend(new_datapoints)

            curr_time = next_time
    return datapoints
def execute_sfx_program ( api_token , program , start_time , end_time , dimensions = None , resolution = 60 ) : with signalfx . SignalFx ( ) . signalflow ( api_token ) as sfx : curr_time = start_time datapoints = [ ] while curr_time < end_time : # To prevent overloading SignalFX we grab a maximum of 5 days worth of data at a time next_time = min ( curr_time . shift ( days = 5 ) , end_time ) logger . info ( f"Querying SignalFX from {curr_time} to {next_time}" ) raw_data = sfx . execute ( program , # SignalFX operates on millisecond timescales start = curr_time . timestamp * 1000 , stop = next_time . timestamp * 1000 , resolution = resolution * 1000 , ) # We can only call _make_ts_label after all of the entries in the raw_data.stream() have been processed data_messages = [ msg for msg in raw_data . stream ( ) if isinstance ( msg , DataMessage ) ] new_datapoints = sorted ( [ ( Arrow . utcfromtimestamp ( msg . logical_timestamp_ms / 1000 ) , { _make_ts_label ( raw_data , key , dimensions ) : value for key , value in msg . data . items ( ) } , ) for msg in data_messages ] ) # SignalFX sometimes gives us duplicate datapoints at the beginning of one chunk/the start of # the next chunk.  This doesn't play nicely with the metrics client so detect and remove those here if datapoints and new_datapoints [ 0 ] [ 0 ] == datapoints [ - 1 ] [ 0 ] : new_datapoints = new_datapoints [ 1 : ] datapoints . extend ( new_datapoints ) curr_time = next_time return datapoints