<commit_before>#
# Copyright (c) 2004-2006 rPath, Inc.
#
# All Rights Reserved
#

import os
import sys
import signal
import urllib
import weakref
import StringIO

import logging
import threading

from conary import conarycfg
from conary import conaryclient
from conary.lib import util, log
import subprocess
from conary import versions

from conary.deps import deps

from jobslave.generators import constants

from mcp import jobstatus

MSG_INTERVAL = 5

class NoConfigFile(Exception):
    def __init__(self, path = ""):
        self._path = path

    def __str__(self):
        return "Unable to access configuration file: %s" % self._path

import time
class LogHandler(logging.Handler):
    def __init__(self, jobId, response):
        self.jobId = jobId
        self.response = weakref.ref(response)
        self._msgs = ''
        self.lastSent = 0
        logging.Handler.__init__(self)

    def flush(self):
        self.lastSent = time.time()
        try:
            msgs = self._msgs
            self._msgs = ''
            self.response().jobLog(self.jobId, msgs)
        except Exception, e:
            print >> sys.stderr, "Warning: log message was lost:", self._msgs
            sys.stderr.flush()

    def emit(self, record):
        self._msgs += record.getMessage() + '\n'
        self.flush()


def logCall(cmd, ignoreErrors = False):
    log.info("+ " + cmd)
    p = subprocess.Popen(cmd, shell = True,
        stdout = subprocess.PIPE, stderr = subprocess.PIPE)
    while p.poll() is None:
        err = p.stdout.read()
        if err:
            [log.info("++ " + errLine) for errLine in err.split("\n")]
        err = p.stderr.readline()
        if err:
            [log.debug("++ " + errLine) for errLine in err.split("\n")]

    if p.returncode and not ignoreErrors:
        raise RuntimeError("Error executing command: %s (return code %d)" % (cmd, p.returncode))
    else:
        return p.returncode


class Generator(threading.Thread):
    configObject = None

    def __init__(self, jobData, parent):
        self.pid = None
        self.jobData = jobData
        self.response = weakref.ref(parent.response)
        self.parent = weakref.ref(parent)
        self.workDir = None

        # FIXME: once conary handles unicode better, remove these items
        # coerce to str if possible due to conary not handling unicode well
        for key, val in jobData.iteritems():
            if type(val) is unicode:
                jobData[key] = str(val)
        self.jobId = str(jobData['UUID'])
        #self.jobData['project']['name'] = str(self.jobData['project']['name'])
        #self.jobData['project']['label'] = str(self.jobData['project']['label'])
        #self.jobData['troveName'] = str(self.jobData['troveName'])
        # end str coercions

        self.UUID = \
            ''.join([hex(ord(os.urandom(1)))[2:] for x in range(16)]).upper()

        self.status('initializing')

        self.conarycfg = conarycfg.ConaryConfiguration(False)
        cfgData = StringIO.StringIO(jobData['project']['conaryCfg'])
        self.conarycfg.readObject(cfgData, cfgData)

        self.conarycfg.configLine('tmpDir %s' % constants.tmpDir)
        self.conarycfg.configLine('entitlementDirectory %s' % constants.entDir)

        util.mkdirChain(constants.entDir)
        for serverName, (entClass, entKey) in jobData['entitlements'].items():
            f = open(os.path.join(constants.entDir, serverName), 'w')
            f.write(conarycfg.emitEntitlement(serverName, entClass, entKey))
            f.close()

        if parent and parent.cfg.proxy:
            self.conarycfg.configLine('proxy %s' % parent.cfg.proxy)

        self.conarycfg.display() # dump conary config for debugging 
        self.cc = conaryclient.ConaryClient(self.conarycfg)
        self.nc = self.cc.getRepos()

        rootLogger = logging.getLogger('')
        for handler in rootLogger.handlers[:]:
            rootLogger.removeHandler(handler)
        self.logger = LogHandler(self.jobId, parent.response)
        rootLogger.addHandler(self.logger)
        log.setVerbosity(logging.DEBUG)

        self.doneStatus = jobstatus.FINISHED
        self.doneStatusMessage = 'Finished'

        threading.Thread.__init__(self)

    def getCookData(self, key):
        return self.jobData.get('data', {}).get(key)

    def getBuildData(self, key):
        val = self.jobData.get('data', {}).get(key)
        if val is None:
            serialVersion = \
                self.jobData.get('serialVersion')
            if serialVersion == 1:
                defaults = \
                    {'autoResolve': False,
                     'maxIsoSize': '681574400',
                     'bugsUrl': 'http://issues.rpath.com/',
                     'natNetworking': False,
                     'vhdDiskType': 'dynamic',
                     'anacondaCustomTrove': '',
                     'stringArg': '',
                     'mediaTemplateTrove': '',
                     'baseFileName': '',
                     'vmSnapshots': False,
                     'swapSize': 128,
                     'betaNag': False,
                     'anacondaTemplatesTrove': '',
                     'enumArg': '2',
                     'vmMemory': 256,
                     'installLabelPath': '',
                     'intArg': 0,
                     'freespace': 250,
                     'boolArg': False,
                     'mirrorUrl': '',
                     'zisofs': True,
                     'diskAdapter': 'lsilogic',
                     'unionfs': False,
                     'showMediaCheck': False}
            else:
                defaults = {}
            val = defaults.get(key)
        return val

    def readConaryRc(self, cfg):
        conarycfgFile = os.path.join('etc', 'conaryrc')
        if os.path.exists(conarycfgFile):
            cfg.read(conarycfgFile)
        return conarycfgFile

    def write(self):
        raise NotImplementedError

    def status(self, msg = None, status = jobstatus.RUNNING):
        if msg:
            self._lastStatus = msg, status
        else:
            msg, status = self._lastStatus

        try:
            self.response().jobStatus(self.jobId, status, msg)
        except:
            print >> sys.stderr, "Error logging status to MCP:", msg

    def recordOutput(self):
        # this function runs in parent process space to record that a build had
        # output that needs to be tracked
        parent = self.parent and self.parent()
        if parent:
            parent.recordJobOutput(self.jobId, self.UUID)
        else:
            log.error("couldn't record output")

    def postOutput(self, fileList):
        # this function runs in the child process to actually post the output
        # of a build.

        # it doesn't matter what we send, just not ''
        os.write(self.parentPipe, 'post')
        self.doneStatus = jobstatus.BUILT
        self.doneStatusMessage = 'Done building image(s)'
        parent = self.parent and self.parent()
        if parent:
            parent.postJobOutput(self.jobId, self.jobData['outputQueue'],
                                 fileList)
        else:
            log.error("couldn't post output")

    def run(self):
        # use a pipe to communicate if a job posted build data
        inF, outF = os.pipe()
        self.pid = os.fork()
        if not self.pid:
            # become session leader for clean job ending.
            os.setpgid(0, 0)
            try:
                try:
                    os.close(inF)
                    self.parentPipe = outF
                    self.status('Starting job')
                    self.write()
                    try:
                        if self.workDir and os.path.exists(self.workDir):
                            util.rmtree(self.workDir)
                    except Exception, e:
                        log.error("couldn't clean up afterwards: %s" % str(e))
                    os.close(outF)
                except:
                    exc, e, bt = sys.exc_info()
                    # exceptions should *never* cross this point, so it's always
                    # an internal server error
                    self.logger.flush()
                    self.status('Internal Server Error', status = jobstatus.FAILED)
                    import traceback
                    log.error(traceback.format_exc(bt))
                    log.error(str(e))
                    log.error('Failed job: %s' % self.jobId)
                    raise
                else:
                    self.logger.flush()
                    self.status(self.doneStatusMessage,
                                status = self.doneStatus)
                    log.info('Finished job: %s' % self.jobId)
            # place exit handlers in their own exception handling layer
            # to ensure that under no circumstances can it escape
            # use os._exit to force ending now.
            except:
                os._exit(1)
            else:
                os._exit(0)
        os.close(outF)
        data = os.read(inF, 255)
        if data:
            self.recordOutput()
        else:
            util.rmtree(os.path.join(constants.finishedDir, self.UUID),
                        ignore_errors = True)
        os.close(inF)
        os.waitpid(self.pid, 0)

    def kill(self):
        if self.pid:
            try:
                # send kill signal to entire process group
                os.kill(-self.pid, signal.SIGKILL)
            except OSError, e:
                # errno 3 is "no such process"
                if e.errno != 3:
                    raise
        self.join()
        self.status('Job Killed', status = jobstatus.FAILED)
        log.error('Job killed: %s' % self.jobId)
        util.rmtree(os.path.join(constants.finishedDir, self.UUID),
                    ignore_errors = True)

class ImageGenerator(Generator):
    def __init__(self, *args, **kwargs):
        Generator.__init__(self, *args, **kwargs)
        #Figure out what group trove to use
        self.baseTrove = self.jobData['troveName']
        versionStr = self.jobData['troveVersion']
        flavorStr = self.jobData['troveFlavor']

        if 'filesystems' not in self.jobData:
            # support for legacy requests
            freeSpace = self.getBuildData("freespace") * 1048576
            swapSize = self.getBuildData("swapSize") * 1048576

            self.jobData['filesystems'] = [
                ('/', 0, freeSpace, 'ext3'),
                ('swap', 0, swapSize, 'swap'),
            ]

        self.mountDict = dict([(x[0], tuple(x[1:])) for x in self.jobData['filesystems'] if x[0]])

        #Thaw the version string
        ver = versions.ThawVersion(versionStr)
        self.baseVersion = ver.asString()

        #Thaw the flavor string
        self.baseFlavor = deps.ThawFlavor(str(flavorStr))

        try:
            self.arch = \
                self.baseFlavor.members[deps.DEP_CLASS_IS].members.keys()[0]
        except KeyError:
            self.arch = ""

        basefilename = self.getBuildData('baseFileName') or ''
        basefilename = ''.join([(x.isalnum() or x in ('-', '.')) and x or '_' \
                                for x in basefilename])
        basefilename = basefilename or \
                       "%(name)s-%(version)s-%(arch)s" % {
                           'name': self.jobData['project']['hostname'],
                           'version': ver.trailingRevision().asString().split('-')[0],
                           'arch': self.arch}

        self.basefilename = basefilename

    def _getLabelPath(self, cclient, trove):
        repos = cclient.getRepos()
        trv = repos.getTroves([trove])
        return " ".join(trv[0].getTroveInfo().labelPath)

    def writeConaryRc(self, tmpPath, cclient):
        # write the conaryrc file
        util.mkdirChain(os.path.split(tmpPath)[0])
        conaryrcFile = open(tmpPath, "w")
        ilp = self.getBuildData("installLabelPath")
        if not ilp: # allow a BuildData ILP to override the group label path
            ilp = self._getLabelPath( \
                cclient, (self.jobData['troveName'],
                          versions.VersionFromString(self.baseVersion),
                          self.baseFlavor))
        if not ilp: # fall back to a reasonable default if group trove was
                    # cooked before conary0.90 and builddata is blank
            ilp = self.jobData['projectLabel'] + " conary.rpath.com@rpl:1"
        mu = self.getBuildData('mirrorUrl')
        if mu:
            type, url = urllib.splittype(mu)
            relativeLink = ''
            if not type:
                type = 'http'
            if not url.startswith('//'):
                url = '//' + url
            if not urllib.splithost(url)[1]:
                relativeLink = '/conaryrc'
            mirrorUrl = type + ':' + url + relativeLink

        print >> conaryrcFile, "installLabelPath " + ilp
        if mu:
            print >> conaryrcFile, 'includeConfigFile ' + mirrorUrl
        print >> conaryrcFile, "pinTroves kernel.*"
        print >> conaryrcFile, "includeConfigFile /etc/conary/config.d/*"
        if self.getBuildData("autoResolve"):
            print >> conaryrcFile, "autoResolve True"
        conaryrcFile.close()

    def saveConaryRC(self, cfgPath):
        f = open(cfgPath, 'w')
        self.conarycfg.display(f)
        f.close()
        return cfgPath
<commit_msg>restore log throttling<commit_after>#
# Copyright (c) 2004-2006 rPath, Inc.
#
# All Rights Reserved
#

import os
import sys
import signal
import urllib
import weakref
import StringIO

import logging
import threading

from conary import conarycfg
from conary import conaryclient
from conary.lib import util, log
import subprocess
from conary import versions

from conary.deps import deps

from jobslave.generators import constants

from mcp import jobstatus

MSG_INTERVAL = 5

class NoConfigFile(Exception):
    def __init__(self, path = ""):
        self._path = path

    def __str__(self):
        return "Unable to access configuration file: %s" % self._path

import time
class LogHandler(logging.Handler):
    def __init__(self, jobId, response):
        self.jobId = jobId
        self.response = weakref.ref(response)
        self._msgs = ''
        self.lastSent = 0
        logging.Handler.__init__(self)

    def flush(self):
        self.lastSent = time.time()
        try:
            msgs = self._msgs
            self._msgs = ''
            self.response().jobLog(self.jobId, msgs)
        except Exception, e:
            print >> sys.stderr, "Warning: log message was lost:", self._msgs
            sys.stderr.flush()

    def emit(self, record):
        self._msgs += record.getMessage() + '\n'
        if (len(self._msgs) > 4096) or ((time.time() - self.lastSent) > 1):
            self.flush()


def logCall(cmd, ignoreErrors = False):
    log.info("+ " + cmd)
    p = subprocess.Popen(cmd, shell = True,
        stdout = subprocess.PIPE, stderr = subprocess.PIPE)
    while p.poll() is None:
        err = p.stdout.read()
        if err:
            [log.info("++ " + errLine) for errLine in err.split("\n")]
        err = p.stderr.readline()
        if err:
            [log.debug("++ " + errLine) for errLine in err.split("\n")]

    if p.returncode and not ignoreErrors:
        raise RuntimeError("Error executing command: %s (return code %d)" % (cmd, p.returncode))
    else:
        return p.returncode


class Generator(threading.Thread):
    configObject = None

    def __init__(self, jobData, parent):
        self.pid = None
        self.jobData = jobData
        self.response = weakref.ref(parent.response)
        self.parent = weakref.ref(parent)
        self.workDir = None

        # FIXME: once conary handles unicode better, remove these items
        # coerce to str if possible due to conary not handling unicode well
        for key, val in jobData.iteritems():
            if type(val) is unicode:
                jobData[key] = str(val)
        self.jobId = str(jobData['UUID'])
        #self.jobData['project']['name'] = str(self.jobData['project']['name'])
        #self.jobData['project']['label'] = str(self.jobData['project']['label'])
        #self.jobData['troveName'] = str(self.jobData['troveName'])
        # end str coercions

        self.UUID = \
            ''.join([hex(ord(os.urandom(1)))[2:] for x in range(16)]).upper()

        self.status('initializing')

        self.conarycfg = conarycfg.ConaryConfiguration(False)
        cfgData = StringIO.StringIO(jobData['project']['conaryCfg'])
        self.conarycfg.readObject(cfgData, cfgData)

        self.conarycfg.configLine('tmpDir %s' % constants.tmpDir)
        self.conarycfg.configLine('entitlementDirectory %s' % constants.entDir)

        util.mkdirChain(constants.entDir)
        for serverName, (entClass, entKey) in jobData['entitlements'].items():
            f = open(os.path.join(constants.entDir, serverName), 'w')
            f.write(conarycfg.emitEntitlement(serverName, entClass, entKey))
            f.close()

        if parent and parent.cfg.proxy:
            self.conarycfg.configLine('proxy %s' % parent.cfg.proxy)

        self.conarycfg.display() # dump conary config for debugging 
        self.cc = conaryclient.ConaryClient(self.conarycfg)
        self.nc = self.cc.getRepos()

        rootLogger = logging.getLogger('')
        for handler in rootLogger.handlers[:]:
            rootLogger.removeHandler(handler)
        self.logger = LogHandler(self.jobId, parent.response)
        rootLogger.addHandler(self.logger)
        log.setVerbosity(logging.DEBUG)

        self.doneStatus = jobstatus.FINISHED
        self.doneStatusMessage = 'Finished'

        threading.Thread.__init__(self)

    def getCookData(self, key):
        return self.jobData.get('data', {}).get(key)

    def getBuildData(self, key):
        val = self.jobData.get('data', {}).get(key)
        if val is None:
            serialVersion = \
                self.jobData.get('serialVersion')
            if serialVersion == 1:
                defaults = \
                    {'autoResolve': False,
                     'maxIsoSize': '681574400',
                     'bugsUrl': 'http://issues.rpath.com/',
                     'natNetworking': False,
                     'vhdDiskType': 'dynamic',
                     'anacondaCustomTrove': '',
                     'stringArg': '',
                     'mediaTemplateTrove': '',
                     'baseFileName': '',
                     'vmSnapshots': False,
                     'swapSize': 128,
                     'betaNag': False,
                     'anacondaTemplatesTrove': '',
                     'enumArg': '2',
                     'vmMemory': 256,
                     'installLabelPath': '',
                     'intArg': 0,
                     'freespace': 250,
                     'boolArg': False,
                     'mirrorUrl': '',
                     'zisofs': True,
                     'diskAdapter': 'lsilogic',
                     'unionfs': False,
                     'showMediaCheck': False}
            else:
                defaults = {}
            val = defaults.get(key)
        return val

    def readConaryRc(self, cfg):
        conarycfgFile = os.path.join('etc', 'conaryrc')
        if os.path.exists(conarycfgFile):
            cfg.read(conarycfgFile)
        return conarycfgFile

    def write(self):
        raise NotImplementedError

    def status(self, msg = None, status = jobstatus.RUNNING):
        if msg:
            self._lastStatus = msg, status
        else:
            msg, status = self._lastStatus

        try:
            self.response().jobStatus(self.jobId, status, msg)
        except:
            print >> sys.stderr, "Error logging status to MCP:", msg

    def recordOutput(self):
        # this function runs in parent process space to record that a build had
        # output that needs to be tracked
        parent = self.parent and self.parent()
        if parent:
            parent.recordJobOutput(self.jobId, self.UUID)
        else:
            log.error("couldn't record output")

    def postOutput(self, fileList):
        # this function runs in the child process to actually post the output
        # of a build.

        # it doesn't matter what we send, just not ''
        os.write(self.parentPipe, 'post')
        self.doneStatus = jobstatus.BUILT
        self.doneStatusMessage = 'Done building image(s)'
        parent = self.parent and self.parent()
        if parent:
            parent.postJobOutput(self.jobId, self.jobData['outputQueue'],
                                 fileList)
        else:
            log.error("couldn't post output")

    def run(self):
        # use a pipe to communicate if a job posted build data
        inF, outF = os.pipe()
        self.pid = os.fork()
        if not self.pid:
            # become session leader for clean job ending.
            os.setpgid(0, 0)
            try:
                try:
                    os.close(inF)
                    self.parentPipe = outF
                    self.status('Starting job')
                    self.write()
                    try:
                        if self.workDir and os.path.exists(self.workDir):
                            util.rmtree(self.workDir)
                    except Exception, e:
                        log.error("couldn't clean up afterwards: %s" % str(e))
                    os.close(outF)
                except:
                    exc, e, bt = sys.exc_info()
                    # exceptions should *never* cross this point, so it's always
                    # an internal server error
                    self.logger.flush()
                    self.status('Internal Server Error', status = jobstatus.FAILED)
                    import traceback
                    log.error(traceback.format_exc(bt))
                    log.error(str(e))
                    log.error('Failed job: %s' % self.jobId)
                    raise
                else:
                    self.logger.flush()
                    self.status(self.doneStatusMessage,
                                status = self.doneStatus)
                    log.info('Finished job: %s' % self.jobId)
            # place exit handlers in their own exception handling layer
            # to ensure that under no circumstances can it escape
            # use os._exit to force ending now.
            except:
                os._exit(1)
            else:
                os._exit(0)
        os.close(outF)
        data = os.read(inF, 255)
        if data:
            self.recordOutput()
        else:
            util.rmtree(os.path.join(constants.finishedDir, self.UUID),
                        ignore_errors = True)
        os.close(inF)
        os.waitpid(self.pid, 0)

    def kill(self):
        if self.pid:
            try:
                # send kill signal to entire process group
                os.kill(-self.pid, signal.SIGKILL)
            except OSError, e:
                # errno 3 is "no such process"
                if e.errno != 3:
                    raise
        self.join()
        self.status('Job Killed', status = jobstatus.FAILED)
        log.error('Job killed: %s' % self.jobId)
        util.rmtree(os.path.join(constants.finishedDir, self.UUID),
                    ignore_errors = True)

class ImageGenerator(Generator):
    def __init__(self, *args, **kwargs):
        Generator.__init__(self, *args, **kwargs)
        #Figure out what group trove to use
        self.baseTrove = self.jobData['troveName']
        versionStr = self.jobData['troveVersion']
        flavorStr = self.jobData['troveFlavor']

        if 'filesystems' not in self.jobData:
            # support for legacy requests
            freeSpace = self.getBuildData("freespace") * 1048576
            swapSize = self.getBuildData("swapSize") * 1048576

            self.jobData['filesystems'] = [
                ('/', 0, freeSpace, 'ext3'),
                ('swap', 0, swapSize, 'swap'),
            ]

        self.mountDict = dict([(x[0], tuple(x[1:])) for x in self.jobData['filesystems'] if x[0]])

        #Thaw the version string
        ver = versions.ThawVersion(versionStr)
        self.baseVersion = ver.asString()

        #Thaw the flavor string
        self.baseFlavor = deps.ThawFlavor(str(flavorStr))

        try:
            self.arch = \
                self.baseFlavor.members[deps.DEP_CLASS_IS].members.keys()[0]
        except KeyError:
            self.arch = ""

        basefilename = self.getBuildData('baseFileName') or ''
        basefilename = ''.join([(x.isalnum() or x in ('-', '.')) and x or '_' \
                                for x in basefilename])
        basefilename = basefilename or \
                       "%(name)s-%(version)s-%(arch)s" % {
                           'name': self.jobData['project']['hostname'],
                           'version': ver.trailingRevision().asString().split('-')[0],
                           'arch': self.arch}

        self.basefilename = basefilename

    def _getLabelPath(self, cclient, trove):
        repos = cclient.getRepos()
        trv = repos.getTroves([trove])
        return " ".join(trv[0].getTroveInfo().labelPath)

    def writeConaryRc(self, tmpPath, cclient):
        # write the conaryrc file
        util.mkdirChain(os.path.split(tmpPath)[0])
        conaryrcFile = open(tmpPath, "w")
        ilp = self.getBuildData("installLabelPath")
        if not ilp: # allow a BuildData ILP to override the group label path
            ilp = self._getLabelPath( \
                cclient, (self.jobData['troveName'],
                          versions.VersionFromString(self.baseVersion),
                          self.baseFlavor))
        if not ilp: # fall back to a reasonable default if group trove was
                    # cooked before conary0.90 and builddata is blank
            ilp = self.jobData['projectLabel'] + " conary.rpath.com@rpl:1"
        mu = self.getBuildData('mirrorUrl')
        if mu:
            type, url = urllib.splittype(mu)
            relativeLink = ''
            if not type:
                type = 'http'
            if not url.startswith('//'):
                url = '//' + url
            if not urllib.splithost(url)[1]:
                relativeLink = '/conaryrc'
            mirrorUrl = type + ':' + url + relativeLink

        print >> conaryrcFile, "installLabelPath " + ilp
        if mu:
            print >> conaryrcFile, 'includeConfigFile ' + mirrorUrl
        print >> conaryrcFile, "pinTroves kernel.*"
        print >> conaryrcFile, "includeConfigFile /etc/conary/config.d/*"
        if self.getBuildData("autoResolve"):
            print >> conaryrcFile, "autoResolve True"
        conaryrcFile.close()

    def saveConaryRC(self, cfgPath):
        f = open(cfgPath, 'w')
        self.conarycfg.display(f)
        f.close()
        return cfgPath

{'fraction_non_alphanumeric': 0.1111440536634177, 'fraction_numerical': 0.005670236815772894, 'mean_word_length': 1.727033855482567, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2183994', 'n_tokens_mistral': 7833, 'n_tokens_neox': 7455, 'n_words': 1819}