okd/roles/openshift_health_checker/openshift_checks/logging/logging_index_time.py

"""
Check for ensuring logs from pods can be queried in a reasonable amount of time.
"""

import json
import time

from uuid import uuid4

from openshift_checks import OpenShiftCheckException
from openshift_checks.logging.logging import LoggingCheck


ES_CMD_TIMEOUT_SECONDS = 30


class LoggingIndexTime(LoggingCheck):
    """Check that pod logs are aggregated and indexed in ElasticSearch within a reasonable amount of time."""
    name = "logging_index_time"
    tags = ["health", "logging"]

    logging_namespace = "logging"

    def run(self):
        """Add log entry by making unique request to Kibana. Check for unique entry in the ElasticSearch pod logs."""
        try:
            log_index_timeout = int(
                self.get_var("openshift_check_logging_index_timeout_seconds", default=ES_CMD_TIMEOUT_SECONDS)
            )
        except ValueError:
            return {
                "failed": True,
                "msg": ('Invalid value provided for "openshift_check_logging_index_timeout_seconds". '
                        'Value must be an integer representing an amount in seconds.'),
            }

        running_component_pods = dict()

        # get all component pods
        self.logging_namespace = self.get_var("openshift_logging_namespace", default=self.logging_namespace)
        for component, name in (['kibana', 'Kibana'], ['es', 'Elasticsearch']):
            pods, error = self.get_pods_for_component(self.logging_namespace, component)

            if error:
                msg = 'Unable to retrieve pods for the {} logging component: {}'
                return {"failed": True, "changed": False, "msg": msg.format(name, error)}

            running_pods = self.running_pods(pods)

            if not running_pods:
                msg = ('No {} pods in the "Running" state were found.'
                       'At least one pod is required in order to perform this check.')
                return {"failed": True, "changed": False, "msg": msg.format(name)}

            running_component_pods[component] = running_pods

        uuid = self.curl_kibana_with_uuid(running_component_pods["kibana"][0])
        self.wait_until_cmd_or_err(running_component_pods["es"][0], uuid, log_index_timeout)
        return {}

    def wait_until_cmd_or_err(self, es_pod, uuid, timeout_secs):
        """Retry an Elasticsearch query every second until query success, or a defined
        length of time has passed."""
        deadline = time.time() + timeout_secs
        interval = 1
        while not self.query_es_from_es(es_pod, uuid):
            if time.time() + interval > deadline:
                msg = "expecting match in Elasticsearch for message with uuid {}, but no matches were found after {}s."
                raise OpenShiftCheckException(msg.format(uuid, timeout_secs))
            time.sleep(interval)

    def curl_kibana_with_uuid(self, kibana_pod):
        """curl Kibana with a unique uuid."""
        uuid = self.generate_uuid()
        pod_name = kibana_pod["metadata"]["name"]
        exec_cmd = "exec {pod_name} -c kibana -- curl --max-time 30 -s http://localhost:5601/{uuid}"
        exec_cmd = exec_cmd.format(pod_name=pod_name, uuid=uuid)

        error_str = self.exec_oc(self.logging_namespace, exec_cmd, [])

        try:
            error_code = json.loads(error_str)["statusCode"]
        except KeyError:
            msg = ('invalid response returned from Kibana request (Missing "statusCode" key):\n'
                   'Command: {}\nResponse: {}').format(exec_cmd, error_str)
            raise OpenShiftCheckException(msg)
        except ValueError:
            msg = ('invalid response returned from Kibana request (Non-JSON output):\n'
                   'Command: {}\nResponse: {}').format(exec_cmd, error_str)
            raise OpenShiftCheckException(msg)

        if error_code != 404:
            msg = 'invalid error code returned from Kibana request. Expecting error code "404", but got "{}" instead.'
            raise OpenShiftCheckException(msg.format(error_code))

        return uuid

    def query_es_from_es(self, es_pod, uuid):
        """curl the Elasticsearch pod and look for a unique uuid in its logs."""
        pod_name = es_pod["metadata"]["name"]
        exec_cmd = (
            "exec {pod_name} -- curl --max-time 30 -s -f "
            "--cacert /etc/elasticsearch/secret/admin-ca "
            "--cert /etc/elasticsearch/secret/admin-cert "
            "--key /etc/elasticsearch/secret/admin-key "
            "https://logging-es:9200/project.{namespace}*/_count?q=message:{uuid}"
        )
        exec_cmd = exec_cmd.format(pod_name=pod_name, namespace=self.logging_namespace, uuid=uuid)
        result = self.exec_oc(self.logging_namespace, exec_cmd, [])

        try:
            count = json.loads(result)["count"]
        except KeyError:
            msg = 'invalid response from Elasticsearch query:\n"{}"\nMissing "count" key:\n{}'
            raise OpenShiftCheckException(msg.format(exec_cmd, result))
        except ValueError:
            msg = 'invalid response from Elasticsearch query:\n"{}"\nNon-JSON output:\n{}'
            raise OpenShiftCheckException(msg.format(exec_cmd, result))

        return count

    @staticmethod
    def running_pods(pods):
        """Filter pods that are running."""
        return [pod for pod in pods if pod['status']['phase'] == 'Running']

    @staticmethod
    def generate_uuid():
        """Wrap uuid generator. Allows for testing with expected values."""
        return str(uuid4())