path: root/reactive/graylog.py

import hashlib
import os
import re
import sys
import time
import yaml
from urllib.parse import urlparse

from charms.reactive import hook, when, when_not, remove_state, set_state
from charms.reactive.helpers import data_changed
from charmhelpers.core import host, hookenv, unitdata
from charmhelpers.contrib.charmsupport import nrpe

libs_dir = os.path.abspath(os.path.join(os.path.dirname(__file__),
                                        '..', 'lib'))
if libs_dir not in sys.path:
    sys.path.append(libs_dir)

from graylogapi import GraylogAPI  # NOQA: E402


API_PORT = '9001'  # This is the default set by the snap
API_URL = 'http://0.0.0.0:9001/api/'  # This is the default set by the snap
CONF_FILE = '/var/snap/graylog/common/server.conf'
ELASTICSEARCH_DISCOVERY_PORT = '9300'
SERVICE_NAME = 'snap.graylog.graylog'


@hook('upgrade-charm')
def upgrade_charm():
    remove_state('graylog.configured')


@when('config.changed')
def update_config():
    remove_state('graylog.configured')


rotation_strategies = {
    'time': {
        'class': 'org.graylog2.indexer.rotation.strategies.TimeBasedRotationStrategy',
        'special': 'rotation_period',
        'juju-conf': 'index_rotation_period',
    },
    'size': {
        'class': 'org.graylog2.indexer.rotation.strategies.SizeBasedRotationStrategy',
        'special': 'max_size',
        'juju-conf': 'index_rotation_size',
    },
    'msg_count': {
        'class': 'org.graylog2.indexer.rotation.strategies.MessageCountRotationStrategy',
        'special': 'max_docs_per_index',
        'juju-conf': 'index_rotation_msg_count',
    }
}


@when('snap.installed.graylog')
@when_not('graylog.configured')
def configure_graylog():
    db = unitdata.kv()
    if not os.path.exists(CONF_FILE):
        hookenv.log('Configuration file "{}" missing, skipping configuration run.'.format(CONF_FILE))
        return
    conf = hookenv.config()
    admin_password = db.get('admin_password')
    if not admin_password:
        admin_password = host.pwgen(18)
        db.set('admin_password', admin_password)
    pw_hash = hashlib.sha256(admin_password.encode('utf-8')).hexdigest()
    set_conf('root_password_sha2', pw_hash)
    if conf['web_listen_uri']:
        set_conf('web_listen_uri', conf['web_listen_uri'])

    remove_state('graylog_api.configured')
    set_state('graylog.configured')
    hookenv.status_set('active', 'Ready')


@when('graylog.configured')
@when('mongodb.available')
@when('elasticsearch.available')
@when_not('graylog.needs_restart')
@when_not('graylog_api.configured')
def configure_graylog_api(*discard):
    remove_state('graylog_index_sets.configured')
    remove_state('graylog_inputs.configured')
    set_state('graylog_api.configured')


@when('graylog_api.configured')
@when_not('graylog_index_sets.configured')
def configure_index_sets(*discard):
    conf = hookenv.config()
    db = unitdata.kv()
    g = GraylogAPI(base_url=API_URL,
                   username='admin',
                   password=db.get('admin_password'),
                   token_name='graylog-charm')
    index_sets = g.index_set_get()
    if index_sets is None:
        return
    index_shards = conf['index_shards']
    if index_shards == 0:
        # Automatically work out the index shards based on how
        # many Elasticsearch units.
        for relid in hookenv.relation_ids('elasticsearch'):
            index_shards += len(hookenv.related_units(relid))
        if not index_shards:
            hookenv.log("Can't work out number of elasticsearch units")
            return

    for iset in index_sets:
        log = ''
        if iset['shards'] != index_shards:
            log += ' shards from {} to {}'.format(iset['shards'], index_shards)
        iset['shards'] = index_shards
        if iset['replicas'] != conf['index_replicas']:
            log += ' replicas from {} to {}'.format(iset['replicas'], conf['index_replicas'])
        iset['replicas'] = conf['index_replicas']
        if conf['index_rotation_strategy'] in rotation_strategies:
            rs = rotation_strategies[conf['index_rotation_strategy']]
            if iset['rotation_strategy_class'] != rs['class']:
                log += ' rotation strategy from {} to {} ({}={})'.format(
                    iset['rotation_strategy_class'], rs['class'], rs['special'], conf[rs['juju-conf']]
                )
            iset['rotation_strategy_class'] = rs['class']
            iset['rotation_strategy'] = {
                'type': '{}Config'.format(rs['class']),
                rs['special']: conf[rs['juju-conf']],
            }
        if iset['retention_strategy']['max_number_of_indices'] != conf['index_retention_count']:
            log += ' retention max_number_of_indices={}'.format(conf['index_retention_count'])
        iset['retention_strategy_class'] = 'org.graylog2.indexer.retention.strategies.DeletionRetentionStrategy'
        iset['retention_strategy'] = {
            'type': 'org.graylog2.indexer.retention.strategies.DeletionRetentionStrategyConfig',
            'max_number_of_indices': conf['index_retention_count'],
        }
        if not g.index_set_update(iset['id'], iset):
            return
        if log:
            hookenv.log('Updated{}'.format(log))

    set_state('graylog_index_sets.configured')


def _check_input_exists(current_inputs, new):
    """
    Given the current list of configured Graylog inputs, check if new input
    exists and return an ID as well as "changed" flag to indicate that the
    new input provided has changed (input type, bind address, and port).
    """
    changed = True
    cur = None
    for input in current_inputs:
        if input['title'] == new['title']:
            cur = input
            break
    if not cur:
        return (True, None)
    current = [cur['type'], cur['attributes']['bind_address'], str(cur['attributes']['port'])]
    new = [new['type'], new['configuration']['bind_address'], str(new['configuration']['port'])]
    if sorted(new) == sorted(current):
        changed = False
    return (changed, cur['id'])


def _close_uneeded_ports(current_inputs, new):
    for cur in current_inputs:
        title = cur['title']
        if title.lower().startswith('custom'):
            continue
        type = cur['type']
        if 'udp' in type.lower():
            proto = 'UDP'
        else:
            proto = 'TCP'
        port = cur['attributes']['port']
        if '{}/{}'.format(port, proto) not in new:
            hookenv.close_port(port, proto)


def _remove_old_inputs(inputs, new_inputs):
    """
    Removes any old/previously defined inputs, or any unknown. We'll keep those
    prefixed with 'Custom' to allow configuring additional inputs via the web
    UI.
    """
    db = unitdata.kv()
    g = GraylogAPI(base_url=API_URL,
                   username='admin',
                   password=db.get('admin_password'),
                   token_name='graylog-charm')
    success = True
    for cur in inputs:
        title = cur['title']
        if title.lower().startswith('custom'):
            continue
        if title in new_inputs:
            continue
        if g.log_input_remove(cur['id']):
            hookenv.log('Removed old input: {} ({})'.format(cur['title'], cur['id']))
        else:
            success = False
    return success


@when('graylog_api.configured')
@when_not('graylog_inputs.configured')
def configure_inputs(*discard):
    """
    Configure log inputs in Graylog via the API.
    """
    conf = hookenv.config()
    db = unitdata.kv()
    g = GraylogAPI(base_url=API_URL,
                   username='admin',
                   password=db.get('admin_password'),
                   token_name='graylog-charm')
    inputs = g.log_input_get()
    if inputs is None:
        return
    new_opened_ports = []
    new_inputs = []
    for new in yaml.safe_load(conf['log_inputs']) or {}:
        type = g.map_log_input_type(new['type'])
        if not type:
            hookenv.log('Input type "{}" not supported'.format(new['type']))
            continue
        d = {
            'title': new['name'],
            'type': type,
            'global': "true",
            'configuration': {
                'bind_address': new['bind_address'],
                'port': new['bind_port'],
            }
        }
        if 'udp' in type.lower():
            proto = 'UDP'
        else:
            proto = 'TCP'
            d['configuration']['tcp_keepalive'] = 'true'

        hookenv.open_port(new['bind_port'], proto)
        new_opened_ports.append('{}/{}'.format(new['bind_port'], proto))

        (changed, input_id) = _check_input_exists(inputs, d)
        if not changed:
            new_inputs.append(new['name'])
            continue
        if input_id:
            change_text = 'Updated existing input'
        else:
            change_text = 'Adding new input'
        ret = g.log_input_update(input_id, d)
        if not ret:
            # Can't add input, let's try again later
            return
        hookenv.log('{}: {} ({})'.format(change_text, new['name'], ret['id']))
        new_inputs.append(new['name'])

    # Now remove inputs that should no longer be there as well as close ports
    _close_uneeded_ports(inputs, new_opened_ports)
    refreshed_inputs = g.log_input_get()
    if not _remove_old_inputs(refreshed_inputs, new_inputs):
        # Can't delete input, let's try again later
        return
    set_state('graylog_inputs.configured')


@when('graylog.configured')
@when('elasticsearch.available')
def configure_elasticsearch_connection(elasticsearch):
    cluster = ""
    # Pre-2.3.x discovery unicast hosts (binary ElasticSearch protocol)
    discovery_hosts = []
    # 2.3.x and above HTTP REST API
    http_hosts = []
    for unit in elasticsearch.list_unit_data():
        cluster_name = unit['cluster_name']
        if cluster_name is not None:
            cluster = cluster_name
        http_hosts.append('http://{}:{}'.format(unit['host'], unit['port']))
        discovery_hosts.append('{}:{}'.format(unit['host'], ELASTICSEARCH_DISCOVERY_PORT))

    if not cluster:
        conf = hookenv.config()
        cluster = conf['elasticsearch_cluster_name']

    if not data_changed('elasticsearch.relation', {'cluster_name': cluster, 'hosts': http_hosts}):
        return

    set_conf('elasticsearch_cluster_name', cluster)
    set_conf('elasticsearch_discovery_zen_ping_unicast_hosts ', ', '.join(discovery_hosts))
    set_conf('elasticsearch_hosts', ', '.join(http_hosts))
    # Elastic search does not reliably pick the right ip to listen on
    set_conf('elasticsearch_network_host', hookenv.unit_private_ip())

    hookenv.log('Updated elasticsearch configuration')
    hookenv.status_set('active', 'Ready')


@when('graylog.configured')
@when('mongodb.available')
def configure_mongodb_connection(mongodb):
    mongodb_hosts = []
    for mongo_host in mongodb.connection_strings():
        mongodb_hosts.append(mongo_host)

    mongodb_uri = 'mongodb://{}/graylog'.format(','.join(mongodb_hosts))
    if not data_changed('mongodb.uri', mongodb_uri):
        return

    set_conf('mongodb_uri', mongodb_uri)
    hookenv.log('Updated mongodb configuration')
    hookenv.status_set('active', 'Ready')


@when('graylog.configured')
@when('website.available')
def update_reverseproxy_config(website):
    services = ""
    conf = hookenv.config()
    if conf['web_listen_uri']:
        url = urlparse(conf['web_listen_uri'])
        website.configure(port=url.port)
        services += "- {service_name: web, service_port: " + str(url.port) + "}\n"

    services += "- {service_name: api, service_port: " + API_PORT + "}\n"
    website.set_remote(all_services=services)


@when('graylog.configured')
@when('graylog.needs_restart')
def restart_service(service=SERVICE_NAME):
    """ Restart the service and on failure wait 15 seconds and try again.
        This handles situations when relations are adding quick enough to trigger
        systemd's standard protection against frequent restarts.
    """
    host.service_restart(service)
    remove_state('graylog.needs_restart')
    if host.service_running(service):
        return

    time.sleep(15)
    host.service_restart(service)


def set_conf(key, value, conf_path=CONF_FILE):
    """ Updates the graylog configuration setting lines matching 'key = value'.
        The configuration contains randomly generated values specific to each install
        so setting the configuration this way avoids having to regenerate those values.
    """
    conf = []
    key_re = re.compile('^#*\s*{}'.format(key))

    changed = False
    replaced = False
    with open(conf_path, 'rb') as conf_file:
        found = False
        for line in conf_file:
            line = line.decode().rstrip()
            if not key_re.match(line):
                conf.append(line)
                continue
            found = True
            if not replaced:
                new = '{} = {}'.format(key, value)
                if line != new:
                    hookenv.log('Updating configuration file option "{}".'.format(key))
                    changed = True
                conf.append(new)
                replaced = True
            elif not line.startswith('#'):
                conf.append('#' + line)
            else:
                conf.append(line)
        if not found:
            hookenv.log('Updating configuration file new option "{}".'.format(key))
            changed = True
            conf.append('{} = {}'.format(key, value))

    if changed:
        set_state('graylog.needs_restart')
        with open(conf_path, 'wb') as conf_file:
            conf_file.write('\n'.join(conf).encode())
        return True
    return False


@when('graylog.configured')
@when('nrpe-external-master.available')
def configure_nagios(nagios):
    if hookenv.hook_name() == 'update-status':
        return

    db = unitdata.kv()
    nagios_password = db.get('nagios_password')
    if not nagios_password:
        nagios_password = host.pwgen(18)
        db.set('nagios_password', nagios_password)

    # Ask charmhelpers.contrib.charmsupport's nrpe to work out our hostname
    hostname = nrpe.get_nagios_hostname()
    nrpe_setup = nrpe.NRPE(hostname=hostname, primary=True)

    conf = hookenv.config()
    if conf['web_listen_uri']:
        url = urlparse(conf['web_listen_uri'])
        # LP:#1706265: Work around this bug by escaping some characters.
        check_string = '\<title\>Graylog\ Web\ Interface\</title\>'
        nrpe_setup.add_check(
            'graylog_http',
            'Graylog Web UI check',
            '/usr/lib/nagios/plugins/check_http -I {} -p {} -s "{}"'.format(url.hostname, str(url.port), check_string)
        )
    check_string = 'cluster_id'
    nrpe_setup.add_check(
        'graylog_api',
        'Greylog API check',
        '/usr/lib/nagios/plugins/check_http -I 127.0.0.1 -p {} -u /api -s "{}"'.format(API_PORT, check_string)
    )

    # For nagios checks via API, we need to create a user and token
    g = GraylogAPI(base_url=API_URL,
                   username='admin',
                   password=db.get('admin_password'),
                   token_name='graylog-charm')
    g.user_create('nagios', nagios_password)
    g.user_permissions_set('nagios', ['users:tokenlist', 'users:tokencreate', 'indexercluster:read', 'indices:failures', 'notifications:read', 'journal:read'])
    g = GraylogAPI(base_url=API_URL,
                   username='nagios',
                   password=nagios_password,
                   token_name='nagios')
    nagios_token = g.token_get()
    if not nagios_token:
        # API not ready, let's write out what we have anyways.
        nrpe_setup.write()
        return
    host.write_file('/var/lib/nagios/graylog-api-token', nagios_token.encode(),
                    group='nagios', perms=0o640)
    nagios_plugins = '/usr/local/lib/nagios/plugins/'
    for f in ('files/check_graylog_health.py', 'lib/graylogapi.py'):
        with open(f, 'rb') as content:
            fname = os.path.join(nagios_plugins, os.path.basename(f))
            host.write_file(fname, content.read(), perms=0o755)
    nrpe_setup.add_check(
        'graylog_health',
        'Graylog Health check',
        '{} {}'.format(os.path.join(nagios_plugins, 'check_graylog_health.py'), API_URL)
    )

    nrpe_setup.write()