fff_monitoring.py

Go to the documentation of this file.

#!/usr/bin/env python

import os
import sys
import logging
import re
import datetime
import subprocess
import socket
import time
import select
import json

log = logging.getLogger(__name__)

def prepare_imports():
    # minihack
    sys.path.append('/opt/hltd/python')
    sys.path.append('/opt/hltd/lib')

    global inotify, watcher, es_client

    import _inotify as inotify
    import watcher
    import pyelasticsearch.client as es_client


class DQMMonitor(object):
    def __init__(self, top_path, rescan_timeout=30):
        self.path = top_path
        self.rescan_timeout = rescan_timeout
        self.es = es_client.ElasticSearch("http://127.0.0.1:9200")
        self.index_name = "dqm_online_monitoring"

        try:
            os.makedirs(self.path)
        except OSError:
            pass

        self.mask = inotify.IN_CLOSE_WRITE | inotify.IN_MOVED_TO
        self.w = watcher.Watcher()
        self.w.add(self.path, self.mask)

    def recreate_index(self):
        self.delete_index()
        self.create_index()

    def delete_index(self):
        log.info("Deleting index: %s", self.index_name)
        self.es.delete_index(self.index_name)

    def create_index(self):
        log.info("Creating index: %s", self.index_name)

        self.settings = {
            "analysis": {
                "analyzer": {
                    "prefix-test-analyzer": {
                        "type": "custom",
                        "tokenizer": "prefix-test-tokenizer"
                    }
                },
                "tokenizer": {
                    "prefix-test-tokenizer": {
                        "type": "path_hierarchy",
                        "delimiter": "_"
                    }
                }
            },
            "index":{
                'number_of_shards' : 16,
                'number_of_replicas' : 1
            }
        }

        self.mappings = {
            'dqm-source-state' : {
                'properties' : {
                    'type' : {'type' : 'string' },
                    'pid' : { 'type' : 'integer' },
                    'hostname' : { 'type' : 'string' },
                    'sequence' : { 'type' : 'integer', "index" : "not_analyzed" },
                    'run' : { 'type' : 'integer' },
                    'lumi' : { 'type' : 'integer' },
                },
                '_timestamp' : { 'enabled' : True, 'store' : True, },
                '_ttl' : { 'enabled' : True, 'default' : '24h' }
            },
            'dqm-diskspace' : {
                'properties' : {
                    'type' : {'type' : 'string' },
                    'pid' : { 'type' : 'integer' },
                    'hostname' : { 'type' : 'string' },
                    'sequence' : { 'type' : 'integer', "index" : "not_analyzed" },
                },
                '_timestamp' : { 'enabled' : True, 'store' : True, },
                '_ttl' : { 'enabled' : True, 'default' : '24h' }
            },
        }

        try:
            self.es.create_index(self.index_name, settings={ 'settings': self.settings, 'mappings': self.mappings })
        except es_client.IndexAlreadyExistsError:
            logger.info("Index already exists.", exc_info=True)
            pass
        except:
            logger.warning("Cannot create index", exc_info=True)

        log.info("Created index: %s", self.index_name)

    def upload_file(self, fp):
        log.info("Uploading: %s", fp)

        try:
            f = open(fp, "r")
            document = json.load(f)
            f.close()

            ret = self.es.index(self.index_name, document["type"], document, id=document["_id"])
            print ret
            
        except:
            log.warning("Failure to upload the document: %s", fp, exc_info=True)    

    def process_file(self, fp):
        fname = os.path.basename(fp)

        if fname.startswith("."):
            return

        if not fname.endswith(".jsn"):
            return

        self.upload_file(fp)
        os.unlink(fp)

    def process_dir(self):
        for f in os.listdir(self.path):
            fp = os.path.join(self.path, f)
            self.process_file(fp) 

    def run(self):
        poll = select.poll()
        poll.register(self.w, select.POLLIN)
        poll.poll(self.rescan_timeout*1000)
        
        # clear the events
        for event in self.w.read(bufsize=0):
            pass
            #print event

        self.process_dir()

    def run_daemon(self):
        self.process_dir()

        while True:
            service.run()
    
# use a named socket check if we are running
# this is very clean and atomic and leave no files
# from: http://stackoverflow.com/a/7758075
def lock(pname):
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    try:
        sock.bind('\0' + pname)
        return sock
    except socket.error:
        return None

# same as in fff_deleter.py
def daemonize(logfile, pidfile):
    # do the double fork
    pid = os.fork()
    if pid != 0:
        sys.exit(0)

    os.setsid()
    sys.stdin.close()
    sys.stdout.close()
    sys.stderr.close()

    fl = open(logfile, "a")
    sys.stdout = fl
    sys.stderr = fl

    pid = os.fork()
    if pid != 0:
        sys.exit(0)

    if pidfile:
        f = open(pidfile, "w")
        f.write("%d\n" % os.getpid())
        f.close()
   
if __name__ == "__main__":
    do_reindex = False
    if len(sys.argv) > 1 and sys.argv[1] == "reindex":
        do_reindex = True
        
    # try to take the lock or quit
    sock = lock("fff_dqmmon")
    if sock is None:
        sys.stderr.write("Already running, exitting.\n")
        sys.stderr.flush()
        sys.exit(1)

    # setup logging
    if not do_reindex:
        daemonize("/var/log/fff_monitoring.log", "/var/run/fff_monitoring.pid")

    # log to stderr (it might be redirected)
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    flog_ch = logging.StreamHandler()
    flog_ch.setLevel(logging.INFO)
    flog_ch.setFormatter(formatter)
    log.setLevel(logging.INFO)
    log.addHandler(flog_ch)
   
    # write the pid file
    log.info("Pid is %d", os.getpid())

    prepare_imports()

    service = DQMMonitor(
        top_path = "/tmp/dqm_monitoring/",
    )

    if do_reindex:
        service.recreate_index()
        sys.exit(0)
     
    service.run_daemon()