Icinga plugin to test that certain known facet values exist when doing an ESGF search.

check_index_node.py

#!/usr/bin/env python3

"""Icinga plugin to test that certain known facet values exist when doing an
ESGF search.

The most important check is on the values for index_node.  Because of the way
that the value for index_node is set by the publisher (it matches the index
hostname that it is publishing to), the set of facet values for index_node 
can be used to test whether all of the Solr shards are responding properly.

Some values of a few other facets are also included.
"""

import sys
import requests
from urllib.parse import urlencode


esg_search_url = 'https://esgf.ceda.ac.uk/esg-search/search/'

expected_facet_vals = {
    'index_node': {
        'must_exist': {'esgf-node.llnl.gov',
                       'esgf-data.dkrz.de',
                       'esgf-node.ipsl.upmc.fr',
                       'esgf.ceda.ac.uk',
                       'esg-dn1.nsc.liu.se',
                       'esgdata.gfdl.noaa.gov',
                       'esgf.nci.org.au'},
        'allow_others': False
    },
    'data_node': {
        'must_exist': {'esgf.ceda.ac.uk'},
        'allow_others': True
    },
    'project': {
        'must_exist': {'CMIP5', 'CMIP6'},
        'allow_others': True
    },
    'institute': {
        'must_exist': {'MOHC', 'ECMWF'},
        'allow_others': True
    }
}


class GetFacetsError(Exception): pass

def get_facet_vals(url, facets):

    params = {'limit': '0',
              'type': 'Dataset',
              'facets': ','.join(facets),
              'format': 'application/solr+json'}

    try:
        resp = requests.get(esg_search_url, params)
    except requests.exceptions.RequestException as e:
        raise GetFacetsError(f'requests error: {e} URL {url} params {params}')

    try:
        resp_fields = resp.json()
    except ValueError:
        raise GetFacetsError('cannot parse JSON response')
        
    counts_dict = resp_fields['facet_counts']['facet_fields']

    return {facet: set(counts_dict[facet][::2])
            for facet in facets}


def check_empty(vals, label, errors):
    if vals:
        errors.append(f'{label}: {", ".join(sorted(vals))}.')


def check_facet_vals(url, expected_facet_vals):

    facets = list(expected_facet_vals.keys())

    try:
        facet_vals = get_facet_vals(url, facets)
    except GetFacetsError as e:
        return (False, f'error obtaining facet values for check:\n{e}')

    errors = []
    for facet in sorted(facets):
        d = expected_facet_vals[facet]
        must_exist = d['must_exist']
        allow_others = d['allow_others']

        check_empty(must_exist - facet_vals[facet],
                    f'Missing vals for {facet}',
                    errors)
        
        if not allow_others:
            check_empty(facet_vals[facet] - must_exist,
                        f'Unrecognised vals for {facet}',
                        errors)

    if errors:
        return (False, 'facet value check failed\n' + '\n'.join(errors))
    else:
        return (True, 'facet value check succeeded')


def main():
    success, message = check_facet_vals(esg_search_url, expected_facet_vals)
    print(message)    
    sys.exit(0 if success else 2)
    
    
if __name__ == '__main__':
    main()