nordugrid-arc-gangliarc 1.0.0-3 / usr / lib / python2.7 / dist-packages / gangliarc.py

#  Copyright 2012 David Cameron, Dmytro Karpenko
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

# gangliarc collects ARC-related information and provides info to ganglia
# through gmetric.

"""gangliarc collects ARC-related information and provides info to ganglia
through gmetric."""

import os
import re
import sys
import time
import signal
import subprocess
import logging
import traceback

# Default gangliarc parameters
#  - path to arc configuration file
#  - frequency (in s) at which to gather info
#  - expiration period of monitored metrics (in s)
#  - path to gmetric executable
#  - path to gm-jobs executable (should eventually be filled by autotools)
#  - metrics to report
gangliarc_conf = {'conffile' : '/etc/arc.conf',
                  'frequency' : '20',
                  'dmax' : '180',
                  'gmetric_exec' : '/usr/bin/gmetric',
                  'metrics' : 'all'}

# job states
arex_job_states = [
    "ACCEPTED",
    "PREPARING",
    "SUBMIT",
    "INLRMS",
    "FINISHING",
    "FINISHED",
    "DELETED",
    "CANCELLING"
]

# data staging states
data_staging_states = [
    "NEW",
    "CHECK_CACHE",
    "RESOLVE",
    "QUERY_REPLICA",
    "PRE_CLEAN",
    "STAGE_PREPARE",
    "TRANSFER_WAIT",
    "TRANSFER",
    "RELEASE_REQUEST",
    "REGISTER_REPLICA",
    "PROCESS_CACHE",
    "CHECKING_CACHE",
    "CACHE_WAIT",
    "CACHE_CHECKED",
    "RESOLVING",
    "RESOLVED",
    "QUERYING_REPLICA",
    "REPLICA_QUERIED",
    "PRE_CLEANING",
    "PRE_CLEANED",
    "STAGING_PREPARING",
    "STAGING_PREPARING_WAIT",
    "STAGED_PREPARED",
    "TRANSFERRING",
    "TRANSFERRING_CANCEL",
    "TRANSFERRED",
    "RELEASING_REQUEST",
    "REQUEST_RELEASED",
    "REGISTERING_REPLICA",
    "REPLICA_REGISTERED",
    "PROCESSING_CACHE",
    "CACHE_PROCESSED"
    ]

# where to get data staging info
data_staging_info = "dtrstate.log"

# dictionary of configuration information, filled by parseConf()
conf = {}

def daemonize(pidfile):
    '''
    Do double fork to create daemon process
    '''
    try:
        pid = os.fork()
        if pid > 0:
            # exit first parent
            sys.exit(0)
    except OSError:
        e = sys.exc_info()[1]
        sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror))
        sys.exit(1)

    # decouple from parent environment
    os.chdir("/")
    os.setsid()
    os.umask(0)

    # do second fork
    try:
        pid = os.fork()
        if pid > 0:
            # exit from second parent
            sys.exit(0)
    except OSError:
        e = sys.exc_info()[1]
        sys.stderr.write("fork #2 failed: %d (%s)\n" % (e.errno, e.strerror))
        sys.exit(1)

    # redirect standard file descriptors
    sys.stdout.flush()
    sys.stderr.flush()
    si = open('/dev/null', 'r')
    so = open('/dev/null', 'a+')
    se = open('/dev/null', 'a+')
    os.dup2(si.fileno(), sys.stdin.fileno())
    os.dup2(so.fileno(), sys.stdout.fileno())
    os.dup2(se.fileno(), sys.stderr.fileno())

    # write pidfile
    pid = str(os.getpid())
    f = open(pidfile,'w+')
    f.write("%s\n" % pid)
    f.close()


def parseConf(conffile):
    '''
    Parse arc.conf and fill the conf dictionary of parameters in the format
    {section : {parameter : [value(s)]}} eg
    {"gangliarc" : {"logfile" : ["/tmp/gangliarc.log"], "pidfile" : ["/tmp/gangliarc.pid"]}}
    Value is a list since some parameters can be used multiple times
    '''
    try:
        f = open(conffile)
    except IOError:
        e = sys.exc_info()[1]
        logging.error("Cannot open file %s: %s", conffile, str(e))
        return

    current_section = ''
    for line in f:
        line = line.strip() # strip new line and leading/trailing spaces
        if len(line) == 0 or line[0] == '#':
            continue
        if line[0] == '[':
            # new section
            if line[-1] != ']':
                logging.error("Bad format of section name in line "+line)
                continue
            current_section = line[1:-1]
            conf[current_section] = {}
        else:
            if len(current_section) == 0:
                logging.error("Found configuration parameter %s outside a defined section", line)
                continue

            # here we have param="value" (sometimes without quotes)
            equals_pos = line.find('=')
            if equals_pos == -1 or equals_pos == len(line)-1:
                logging.error("Bad format for parameter %s", line)
                continue
            param = line[:equals_pos]
            value = line[equals_pos+1:]
            # remove surrounding quotes
            if value[0] == '"':
                if value[-1] != '"':
                    logging.error("Bad format for parameter %s", line)
                    continue
                value = value[1:-1]
            elif value[0] == "'":
                if value[-1] != "'":
                    logging.error("Bad format for parameter %s", line)
                    continue
                value = value[1:-1]

            # check if already stored
            if param in conf[current_section]:
                conf[current_section][param].append(value)
            else:
                conf[current_section][param] = [value]

    f.close()


def parseURL(url):
    """
    Split url into (protocol, host, port, path) and return this tuple.
    """
    match = re.match(r'(\w*)://([^/?#:]*):?(\d*)/?(.*)', url)
    if match is None:
        return (None, None, None, None)

    port_s = match.group(3)
    if (len(port_s) > 0):
        port = int(port_s)
    else:
        port = None

    urltuple = (match.group(1), match.group(2), port, '/'+match.group(4))
    return urltuple


def runGmetric(metric, value, unit, unit_type):
    '''
    Run the gmetric executable for the given metric.
    '''
    try:
        res = subprocess.call([gangliarc_conf['gmetric_exec'], "-n", metric, "-v", str(value), "-t", unit_type,
                              "-u", unit, "-d", gangliarc_conf['dmax']])
    except OSError:
        e = sys.exc_info()[1]
        logging.error("Error calling gmetric process: %s", str(e))
        return
    if res != 0:
        logging.error("Error returned from gmetric")


def getDataStagingInfo():
    '''
    Pull out number of files in each data staging state from
    the info dumped by A-REX
    '''

    # most important states to monitor
    data_staging_important_states = [
        "TRANSFER_WAIT",
        "CACHE_WAIT",
        "STAGE_PREPARE",
        "STAGING_PREPARING_WAIT",
    ]

    # get possible staging hosts for TRANSFERRING state
    if 'data-staging' in conf:
        if 'deliveryservice' in conf['data-staging']:
            for url in conf['data-staging']['deliveryservice']:
                (protocol, host, port, path) = parseURL(url)
                if host is not None:
                    data_staging_important_states.append('TRANSFERRING_' + host)

        # there is always a chance local transfer is used so always display it
        data_staging_important_states.append('TRANSFERRING_local')

    # get DTR log location
    if 'data-staging' in conf and 'dtrlog' in conf['data-staging']:
        dtrlog = conf['data-staging']['dtrlog'][0]
    elif 'controldir' in conf['grid-manager']:
        control = conf['grid-manager']['controldir'][0]
        dtrlog = control+'/'+data_staging_info
    else:
        logging.error("No control dir or DTR log found in configuration")
        return

    try:
        f = open(dtrlog)
    except IOError:
        e = sys.exc_info()[1]
        logging.error("Cannot open file %s: %s", dtrlog, str(e))
        return

    states = {}
    num_lines = 0
    for line in f:
        l = line.split()
        state = l[1]
        # if TRANSFERRING, get the host doing the transfer
        if state == 'TRANSFERRING':
            if len(l) == 6:
                state = state + '_' + l[5]
            else:
                state = state + '_local'
        if state in states:
            states[state] = states[state] + 1
        else:
            states[state] = 1
        num_lines = num_lines + 1

    f.close()
    for state in data_staging_important_states:
        if state == 'STAGE_PREPARE' and 'PRE_CLEANED' in states:
            # In arc > 2.0.0 files waiting to be staged are in STAGE_PREPARE
            # instead of PRE_CLEANED
            value = states['PRE_CLEANED']
            if state in states:
                value += states[state]
        elif state == 'TRANSFER_WAIT' and 'TRANSFER' in states:
            # In arc > 1.1.0 TRANSFER_WAIT was merged into TRANSFER but calling
            # the metric TRANSFER_WAIT is more intuitive
            value = states['TRANSFER']
        elif state in states:
            value = states[state]
        else:
            value = 0

        state = 'ARC_STAGING_' + state
        logging.info("%s %s", state, value)
        runGmetric(state, value, 'files', 'int32')

    runGmetric('ARC_STAGING_TOTAL', num_lines, 'files', 'int32')


def getJobsStatesInfo():
    '''
    Pull out number of jobs in each A-REX internal state by scanning
    job.id.status files in control subdirectories.
    '''

    if 'controldir' in conf['grid-manager']:
        control = conf['grid-manager']['controldir'][0]
    else:
        logging.error("No control dir found in configuration")
        return

    control_subdirs = ['accepting', 'finished', 'processing', 'restarting']

    states = {}

    for control_subdir in control_subdirs:
        subdir = os.path.join(control, control_subdir)

        # subdirs only exist in versions >= 1.0.0
        if not os.path.isdir(subdir):
            logging.warning('Control sub-directory %s not found. ' \
                            'Job state information is only supported in ARC >= 1.0.0', subdir)
            return

        # scan files in the sub dir
        try:
            for status_file in os.listdir(subdir):
                try:
                    f = open(os.path.join(subdir, status_file))
                except IOError:
                    e = sys.exc_info()[1]
                    logging.warning('Could not open status file %s: %s', status_file, str(e))
                    continue

                status = f.readline().strip()
                if status in states:
                    states[status] += 1
                else:
                    states[status] = 1
                f.close()
        except OSError:
            e = sys.exc_info()[1]
            logging.warning('Could not list status files in %s: %s', subdir, str(e))

    # first get total jobs including pending
    totaljobs = 0
    for state in states:
        totaljobs += states[state]
    logging.info('ARC_JOBS_TOTAL %i', totaljobs)
    runGmetric('ARC_JOBS_TOTAL', totaljobs, 'jobs', 'int32')

    # now number for each specific state we are interested in
    for state in arex_job_states:
        if state in states:
            value = states[state]
        else:
            value = 0
        state = 'ARC_JOBS_' + state
        logging.info("%s %s", state, value)
        runGmetric(state, value, 'jobs', 'int32')


def getHeartBeatInfo():
    '''
    use the modification timestamp of the gm-heartbeat file to
    send to gmetric the time since the last heartbeat
    '''

    if 'controldir' in conf['grid-manager']:
        control = conf['grid-manager']['controldir'][0]
    else:
        logging.error("No control dir found in configuration")
        return

    heartbeat = control + "/gm-heartbeat"
    try:
        statinfo = os.stat(heartbeat)
    except OSError:
        e = sys.exc_info()[1]
        logging.error("Error with heartbeat file: %s", str(e))
        return

    mtime = statinfo.st_mtime
    now = time.time()
    heartbeat_time = now - mtime

    logging.info("heartbeat time %s", heartbeat_time)
    runGmetric('ARC_AREX_HEARTBEAT_LAST_SEEN', heartbeat_time, 'sec', 'int32')


def getCacheFree():
    '''
    go through each cache dir and find the free space on the filesystem
    '''

    if 'cachedir' not in conf['grid-manager']:
        logging.warn('No caches defined in configuration')
        return
    caches = conf['grid-manager']['cachedir']

    totalfree = 0
    for cache in caches:
        if cache.find(' ') != -1:
            cache = cache[:cache.find(' ')]
        # If substitutions are present, truncate the path
        if cache.find('%') != -1:
            cache = cache[:cache.find('%')]
            logging.info('cache dir with substitution truncated to %s', cache)
        # Note: Not portable!
        try:
            s = os.statvfs(cache)
        except OSError:
            e = sys.exc_info()[1]
            logging.error("Error finding space in cache %s: %s", cache, str(e))
            continue

        free = float(s.f_bsize) * float(s.f_bavail) / float(1024) / float(1024) / float(1024) # space in GiB
        totalfree += free

        logging.info('cache %s free space %s GB', cache, free)

        # ganglia doesn't allow slashes in metric names
        cache = cache.replace('/', '_')
        runGmetric('ARC_CACHE_FREE'+cache, free, 'GB', 'float')

    if len(caches) > 1:
        runGmetric('ARC_CACHE_FREE_TOTAL', totalfree, 'GB', 'float')


def getSessionFree():
    '''
    go through each session dir and find the free space on the filesystem
    '''

    if 'sessiondir' not in conf['grid-manager']:
        logging.warn('No sessiondir defined in configuration')
        return
    sessiondirs = conf['grid-manager']['sessiondir']

    totalfree = 0
    for session in sessiondirs:
        if session.find(' ') != -1:
            session = session[:session.find(' ')]
        # If substitutions are present, truncate the path
        if session.find('%') != -1:
            session = session[:session.find('%')]
            logging.info('session dir with substitution truncated to %s', session)
        # Special case of session dirs in users' home directories
        # Guessing that /home is ok
        if session == '*':
            session = '/home'
            logging.info('using /home for session dir info')
        # Note: Not portable!
        try:
            s = os.statvfs(session)
        except OSError:
            e = sys.exc_info()[1]
            logging.error("Error finding space in session dir %s: %s", session, str(e))
            continue

        free = float(s.f_bsize) * float(s.f_bavail) / float(1024) / float(1024) / float(1024) # space in GiB
        totalfree += free

        logging.info('session dir %s free space %s GB', session, free)

        # ganglia doesn't allow slashes in metric names
        session = session.replace('/', '_')
        runGmetric('ARC_SESSION_FREE'+session, free, 'GB', 'float')

    if len(sessiondirs) > 1:
        runGmetric('ARC_SESSION_FREE_TOTAL', totalfree, 'GB', 'float')


def getProcessingJobs():
    '''
    Count the number of jobs in the 'processing' subdir of the
    control dir. This will only work on ARC1 clusters.
    '''

    if 'controldir' in conf['grid-manager']:
        controldir = conf['grid-manager']['controldir'][0]
    else:
        logging.error("No control dir found in configuration")
        return

    processing = 0

    processing_dir = controldir + "/processing"

    try:
        entries = os.listdir(processing_dir)
    except OSError:
        e = sys.exc_info()[1]
        logging.error("Error listing dir %s: %s", processing_dir, str(e))
        return
    processing += len(entries)

    logging.info("%i jobs in processing", processing)
    runGmetric('ARC_JOBS_PROCESSING', processing, 'jobs', 'int32')


def getFailedJobs():
    '''
    Look into the grid manager jobs log file to extract
    the quantity of failed jobs among last 100 finished
    and send to gmetric
    '''

    if 'joblog' not in conf['grid-manager']:
        logging.warn('No grid manager jobs log defined in configuration')
        return

    gmjlog = conf['grid-manager']['joblog'][0]

    try:
        # Linux specific commands, but ganglia is only for *nix systems
        # Get last 400 lines from the log file, select only concerning finished jobs
        # and select failed in the finished
        p1 = subprocess.Popen(['tail', '-400', gmjlog], stdout = subprocess.PIPE)
        p2 = subprocess.Popen(['grep', '-i','finish'], stdin = p1.stdout, stdout = subprocess.PIPE)
        p1.stdout.close()
        p3 = subprocess.Popen(['tail', '-100'], stdin = p2.stdout, stdout = subprocess.PIPE)
        p2.stdout.close()
        p4 = subprocess.Popen(['grep', '-i', 'fail'], stdin = p3.stdout, stdout = subprocess.PIPE)
        p3.stdout.close()
        p5 = subprocess.Popen(['wc', '-l'], stdin = p4.stdout, stdout = subprocess.PIPE)
        p4.stdout.close()
        num_of_jobs = p5.stdout.readline()
        num_of_jobs = num_of_jobs.strip()
        logging.info('%s jobs failed among last 100', num_of_jobs)
        p5.stdout.close()
    except OSError:
        e = sys.exc_info()[1]
        logging.error("Error with running OS commands: %s", str(e))
        return

    runGmetric('ARC_JOBS_FAILED_PER_100', num_of_jobs, 'jobs', 'int32')


def getTransferInfo():
    '''
    Get the number of downloader and uploader processes.
    Function is used if the new data staging is disabled
    '''

    try:
    # Linux specific commands, but ganglia is only for *nix systems
        ulp1 = subprocess.Popen(['ps', 'auxw'], stdout = subprocess.PIPE)
        ulp2 = subprocess.Popen(['grep', 'uploader'], stdin = ulp1.stdout, stdout = subprocess.PIPE)
        ulp1.stdout.close()
        ulp3 = subprocess.Popen(['grep', '-v', 'grep'], stdin = ulp2.stdout, stdout = subprocess.PIPE)
        ulp2.stdout.close()
        ulp4 = subprocess.Popen(['wc', '-l'], stdin = ulp3.stdout, stdout = subprocess.PIPE)
        ulp3.stdout.close()
        ul = ulp4.stdout.readline()
        ul = ul.strip()
        ulp4.stdout.close()

        dlp1 = subprocess.Popen(['ps', 'auxw'], stdout = subprocess.PIPE)
        dlp2 = subprocess.Popen(['grep', 'downloader'], stdin = dlp1.stdout, stdout = subprocess.PIPE)
        dlp1.stdout.close()
        dlp3 = subprocess.Popen(['grep', '-v','grep'], stdin = dlp2.stdout, stdout = subprocess.PIPE)
        dlp2.stdout.close()
        dlp4 = subprocess.Popen(['wc', '-l'], stdin = dlp3.stdout, stdout = subprocess.PIPE)
        dlp3.stdout.close()
        dl = dlp4.stdout.readline()
        dl = dl.strip()
        dlp4.stdout.close()
        logging.info('%s downloaders and %s uploaders running', dl, ul)
    except OSError:
        e = sys.exc_info()[1]
        logging.error("Error with running OS commands: %s", str(e))
        return

    runGmetric('ARC_UPLOADING', ul, 'jobs', 'int32')
    runGmetric('ARC_DOWNLOADING', dl, 'jobs', 'int32')


def handler(signum, frame):
    '''
    shut down nicely
    '''
    logging.info('*** gangliarc stopped ***')
    sys.exit(0)



def run(conffile, pidfile, logfile):
    '''
    Main program. It creates a daemon which loops indefinitely. In each loop
    several metrics are gathered and sent to gmetric.

    run() takes three arguments, the path to the ARC configuration file and
    the paths to the pid and log files for this process.
    '''

    # path to arc.conf
    gangliarc_conf['conffile'] = conffile

    # set up logging
    logging.basicConfig(filename=logfile, level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')

    # create daemon
    daemonize(pidfile)

    # add signal handler
    signal.signal(signal.SIGTERM, handler)

    logging.info('*** gangliarc started ***')

    # main loop - loop forever until interrupted
    while (True):
        try:
            # read arc conf file
            # TODO parse once, but enable reloading via /etc/init.d/gangliarc reload
            parseConf(conffile)

            # sanity check
            if 'grid-manager' not in conf:
                logging.critical("No [grid-manager] section found in configuration, exiting")
                break

            # check if some gangliarc defaults were overridden in configuration
            if 'gangliarc' in conf:
                for key in gangliarc_conf:
                    if key in conf['gangliarc']:
                        gangliarc_conf[key] = conf['gangliarc'][key][0]

            # metrics to measure
            metrics = gangliarc_conf['metrics'].split(',')

            if 'staging' in metrics or 'all' in metrics:
                # data staging info
                if 'newdatastaging' in conf['grid-manager'] and conf['grid-manager']['newdatastaging'][0] == 'no' \
                        or 'enable_dtr' in conf['grid-manager'] and conf['grid-manager']['enable_dtr'][0] == 'no' :
                    # monitor old downloader/uploader
                    getTransferInfo()
                else:
                    # monitor DTR
                    getDataStagingInfo()

            if 'jobstates' in metrics or 'all' in metrics:
                # jobs in different states
                getJobsStatesInfo()

            if 'heartbeat' in metrics or 'all' in metrics:
                # gm-heartbeat info
                getHeartBeatInfo()

            if 'cache' in metrics or 'all' in metrics:
                # cache free space
                getCacheFree()

            if 'session' in metrics or 'all' in metrics:
                # session dir free space
                getSessionFree()

            if 'processingjobs' in metrics or 'all' in metrics:
                # processing jobs
                getProcessingJobs()

            if 'failedjobs' in metrics or 'all' in metrics:
                # number of failed jobs among last 100
                getFailedJobs()

            time.sleep(float(gangliarc_conf['frequency']))
        except SystemExit: # raised by signal handler
            break
        except:
            logging.critical("Unexpected exception, exiting gangliarc")
            logging.critical(traceback.format_exc())
            logging.info('*** gangliarc stopped ***')
            break


if __name__ == '__main__':
    run('/etc/arc.conf', '/tmp/gangliarc.pid', '/tmp/gangliarc.log')