web_stats.py

#!/usr/bin/env python
'''Collect web stats from elasticsearch for later parsing'''

import os
from datetime import datetime
from datetime import timedelta
import tempfile
import logging
import pwd
# pylint: disable=import-error
from elasticsearch import Elasticsearch
from funcy import compose, partial

LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
LOGGER.addHandler(logging.StreamHandler())

ELASTICSEARCH_URL = os.environ['ELASTICSEARCH_URL']
LOG_OUTPUT_DIR = os.environ.get('LOG_OUTPUT_DIR')


def collect_and_log():
    '''Collect web stats and write to log'''
    es_conn = __create_es_connection__()
    access_log = os.path.join(LOG_OUTPUT_DIR, 'access_log')
    assert es_conn.ping()
    compose(partial(write_access_log_file, access_log),
            extract_messages_from_results,
            search_es)(es_conn)


def __create_es_connection__():
    return Elasticsearch([ELASTICSEARCH_URL],
                         use_ssl=True,
                         ssl_assert_hostname=False)


def search_es(es_conn):
    '''perform es search'''
    assert es_conn
    now = datetime.now()
    yesterday = now - timedelta(days=1)
    day_before_yesterday = yesterday - timedelta(days=1)
    query_body = {
        "query": {
            "bool": {
                "filter": [
                    {"match": {"image_tag": "odyssey_web"}},
                    {"range": {"timestamp": {
                        # pylint: disable=no-member
                        "gte": day_before_yesterday.isoformat(),
                        "lte": yesterday.isoformat()
                    }}},
                ]
            }
        }
    }
    return es_conn.search(body=query_body)


def extract_messages_from_results(results):
    '''pull messages from es search results

       Returns a generator for results.
    '''
    return (hit['_source']['message'] for hit in results['hits']['hits'])


def map_messages_into_bins(messages):
    '''map log items (messages) into "bins"'''
    bins = {}
    for message in messages:
        # Probably an improper split...
        key = message.split('-')[1].strip() or "default"
        if key in bins:
            bins[key].extend([message])
        else:
            bins[key] = [message]
    return bins


def process_access_log_bins(bins):
    '''process bins and write access logs'''
    if not os.path.exists(LOG_OUTPUT_DIR):
        os.mkdir(LOG_OUTPUT_DIR)
    for key in bins.keys():
        filename = os.path.join(LOG_OUTPUT_DIR, '%s_access.log' % key)
        write_access_log_file(filename, bins[key])


def write_access_log_file(filename, lines):
    '''Write out access log'''
    with open(filename, 'w') as access_log:
        access_log.write('\n'.join(lines))
    return filename


if __name__ == '__main__':
    UID = pwd.getpwnam('www-data').pw_uid
    os.seteuid(UID)
    if not LOG_OUTPUT_DIR:
        LOG_OUTPUT_DIR = tempfile.mkdtemp()
    LOGGER.info("Starting Elasticsearch web-stats collection ...")
    collect_and_log()
    LOGGER.info("Done.")