Source code for Happyface4.helpers

import json
import requests
import pycurl
from io import BytesIO




def get_data_from_elasticsearch(
    base_url, header, query_head, query_body, logger, timeout=600
):
    logger.debug("Starting data extraction from elastic API")
    # next build the http data string from the query_header and the query_body
    # important: There has to be a newline between header and body
    # see here for more information: https://www.elastic.co/guide/en/elasticsearch/reference/master/search-multi-search.html
    request = "\n" + json.dumps(query_head) + "\n" + json.dumps(query_body) + "\n"
    logger.debug(f"using query:\n{request}")
    # get the data with pycurl (http://pycurl.io/docs/latest/quickstart.html)
    # pycurl needs a function to write the http response to, we use BytesIO
    # the correspondig curl command is: curl -X POST -H 'Content-Type: application/json' -H "Authorization: Bearer $CERN_BEARER_TOKEN" "https://monit-grafana.cern.ch/api/datasources/proxy/9582/_msearch" --data $request
    response_body = BytesIO()
    response_header = BytesIO()
    c = pycurl.Curl()
    c.setopt(c.URL, base_url)
    c.setopt(c.POSTFIELDS, request)
    c.setopt(c.HTTPHEADER, header)
    c.setopt(c.WRITEDATA, response_body)
    c.setopt(c.WRITEHEADER, response_header)
    c.setopt(c.TIMEOUT, timeout)
    c.perform()
    c.close()
    logger.debug(f"recieved haeder:\n{response_header.getvalue()}\n")
    body_bytes = response_body.getvalue()
    logger.debug(f"recieved data:\n{body_bytes}\n")
    # check if the http code is 200 else raise an error
    if "200" not in response_header.getvalue().decode("utf8").split("\n")[0]:
        logger.error(f"recieved haeder:\n{response_header.getvalue().decode('utf8')}\n")
        logger.error(f"recieved data:\n{body_bytes.decode('utf8')}\n")
        raise Exception(
            f"Server retourned an error code:\n{response_header.getvalue().decode('utf8')}"
        )

    # try to interpret the resieved data as json
    try:
        result = json.loads(body_bytes.decode("utf8"))
    except Exception as e:
        logger.error(e)
        logger.error(f"Response body size: {len(body_bytes)} bytes")
        logger.error("The HTTP response didn't contain the expected JSON.")
        raise
    return result





def get_data_from_grafana(
    url,
    datasource,
    queries,
    logger,
    token=None,
    time_range={"from": "now-24h", "to": "now"},
) -> dict:
    """Simple helper function to get the resource data from a grafana service in JSON format.

    Args:
        url (str): Url to the grafana instance. Should look like `https://grafana_instance.com/api/ds/query`.
        datasource (dict): Datasource to fetch data from. Usually something like `{'uid': 'foobar'}`
        queries (dict): The queries to run against the datasource. The keys are used as refId, values are the queries.
        logger (logging.logger): A logger. Usually `self.logger`.
        token (str, optional): The identification token for the grafana instance if necessary. Defaults to None.
        time_range (dict): Time range. Defaults to `{"from": "now-24h", "to": "now"}`.

    Returns:
        dict: The ``results`` member of the Grafana response JSON.
    """
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"

    if "from" not in time_range or "to" not in time_range:
        logger.error(
            'get_data_from_grafana argument "time_range" has to define "from" and "to".'
        )

    data = {
        "queries": [
            {
                "refId": ref_id,
                "datasource": datasource,
                "query": query,
                "rawQuery": True,
                "resultFormat": "time_series",
            }
            for ref_id, query in queries.items()
        ],
        **time_range,
    }

    logger.debug(f"Get data from Grafana ({url}) with request")
    logger.debug(data)

    req = requests.post(url, headers=headers, json=data)

    logger.debug(f"Response: {req.text[:150]}{'...' if len(req.text) > 150 else ''}")

    return req.json()["results"]





def field_index(name, fields, key="name") -> int | None:
    return next((i for i, f in enumerate(fields) if f[key] == name), None)