Subversion Repositories DevTools

package com.erggroup.buildtool.daemon;

import com.erggroup.buildtool.ripple.Package;

import com.erggroup.buildtool.ripple.ReleaseConfig;

import com.erggroup.buildtool.ripple.ReleaseManager;

import java.io.File;

import java.io.IOException;

import java.net.ServerSocket;

import java.sql.SQLException;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.Map;

import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger ;

import org.slf4j.LoggerFactory;

/**BuildDaemon sub component and entry point (main BuildDaemon thread)

 */

public class BuildDaemon

{

    /**hostname

     * @attribute

     */

    static String mHostname;

    /**GBE_LOG

     * @attribute

     */

    static String mGbeLog;

    /** mShutDown

	 * @attribute

	 *  Request to shutdown the build system gracefully

	 */

	static boolean mShutDown = false;

    /**Logger

     * @attribute

     */

    protected static Logger mLogger =  LoggerFactory.getLogger(BuildDaemon.class);

    /**Collection of ThreadIdentifier objects.

     * Using a ConcurrentLinkedQueue because we add and remove items from the collection

     * and it is being accessed from multiple threads.

     * @attribute

     */

    private ConcurrentLinkedQueue<ThreadIdentifier> mThreadCollection = new ConcurrentLinkedQueue<ThreadIdentifier>();

    /** Last time we did a poll for new builds

     * 

     */

    private long mLastBuildPoll = 0;

    /** Build Process poll time in seconds

     *  Modified on error to slow the poll rate

     */

    private int mPollTime = 3;

    /**Nagios

     * @attribute

     */

    ServerSocket nagiosSrv;

    NagiosThread nagiosChecker;

    /** Local class to assist in reporting Nagios Status

     * 

     */

    public static class NagiosInfo {

        int        threadCount = 0;

        int        threadAliveCount = 0;

        int        masterCount = 0;

        int        slaveCount = 0;

        ArrayList<String> extendedReason  = new ArrayList<String>();

    }

    /**mThreadCollection items

     */

    private class ThreadIdentifier

    {

        /**rcon_id associated with the thread

         * @attribute

         */

        private final int mRconId;

        /**thread identifier

         * @attribute

         */

        private final BuildThread mThread;

        /**constructor

         */

        ThreadIdentifier(int rconId, BuildThread thread)

        {

            mLogger.debug("ThreadIdentifier {}", rconId);

            mRconId = rconId;

            mThread = thread;

        }

        /**accessor

         */

        int getRconId()

        {

            mLogger.info("get_rcon_id returned {}", mRconId);

            return mRconId;

        }

        /**accessor

         */

        BuildThread getThread()

        {

            mLogger.debug("get_thread");

            return mThread;

        }

    }

    /**Exception thrown to indicate an uncorrectable error

     */

    public class BuildException extends Exception

    {

        BuildException(String msg)

        {

            mLogger.error(msg);

        }

      private static final long serialVersionUID = 1L;

    }

    /**main method for the Build Daemon program

     * instantiates a BuildDaemon object

     */

    public static void main(String[] args)

    {

        mLogger.debug("main");

        mHostname = testEnvVar("GBE_HOSTNAME");

        testEnvVar("ANT_HOME");

        testEnvVar("GBE_UNC");

        testEnvVar("GBE_DPKG_SSH_PROPERTIES");

        mGbeLog = testEnvVar("GBE_LOG");

        File gl = new File( mGbeLog );

        if ( !gl.isDirectory() )

        {

            mLogger.error("main GBE_LOG is not a directory");

            System.exit(1);

        }

        // Connection information for the database

        String connectionString = System.getenv("GBE_RM_LOCATION");

        String username = System.getenv("GBE_RM_USERNAME");

        String password = System.getenv("GBE_RM_PASSWORD");

        boolean showHelp = false;

        int consumed = 0;

        for (int optind = 0; optind < args.length; optind += 1 + consumed)

        {

            consumed = 0;

            boolean argsRemains = optind < (args.length - 1);

            if (args[optind].equals("-c") && argsRemains )

            {

                connectionString = args[optind+1];

                consumed ++;

            }

            else if (args[optind].equals("-u") && argsRemains)

            {

                username = args[optind+1];

                consumed ++;

            }

            else if (args[optind].equals("-p") && argsRemains)

            {

                password = args[optind+1];

                consumed ++;

            }

            else

            {

                showHelp = true;

            }

        }

        if (    showHelp ||

                connectionString == null ||

                connectionString.length() == 0 ||

                username == null ||

                username.length() == 0 ||

                password == null ||

                password.length() == 0)

        {

            mLogger.error("Usage: java -jar abtdD.jar -c connectionString -u username -p password");

            System.exit(1);

        }

        BuildDaemon buildDaemon = new BuildDaemon(connectionString, username, password);

        buildDaemon.cleanUp();

    }

    /**

     * Test a named EnvVar to ensure that it has been set

     * Call after the mLogger has been setup

     *  @param varName - Name of string to examine

     *  @return - String value of the parameter

     */

    private static String testEnvVar(String varName) {

        String envVar = System.getenv(varName);

        if ( envVar == null )

        {

            mLogger.error("main {} not set", varName);

            System.exit(1);

        }

        return envVar;

    }

    /**constructor, implements the sequence diagram spawn thread

     */

    public BuildDaemon(String connectionString, String username, String password)

    {

        this(new ReleaseManager(connectionString, username + "[release_manager]", password));

    }

    public BuildDaemon(ReleaseManager releaseManager)

    {

        String utf = null;

        mLogger.warn("BuildDaemon");

        try

        {

            // Flag UTF in progress

            if ( releaseManager.mConnectionString.compareTo("unit test spawn thread") == 0 )

            {

                utf = releaseManager.mConnectionString;

            }

            if ( Package.mGenericMachtype == null )

            {

                throw new BuildException("run GBE_MACHTYPE not set");

            }

            if ( Package.mGbeDpkg == null )

            {

                throw new BuildException("run GBE_DPKG not set");

            }

            //	Set the default handler invoked when a thread abruptly terminates due to an

            //	uncaught exception, and no other handler has been defined for that thread.

            //

            Thread.setDefaultUncaughtExceptionHandler(new BuildDaemonUncaughtExceptionHandler());

            //

            //  Start the Nagios Interface Thread

            //    Unless performing a unit test

            //

            if ( utf == null )

            {

                startNagiosServer();

            }

            //	Discover new build daemons to be started on the current host

            //

            while (!mShutDown)

            {

                mPollTime = 3;

                // Poll for slave build requests

                lookForBuildRequests(releaseManager, utf);

                // Poll for new builds every few (5) minutes

                if ( System.currentTimeMillis() - mLastBuildPoll > 5 * 60 * 1000) {

                    lookForNewBuildDaemons(releaseManager, utf);

                    mLastBuildPoll = System.currentTimeMillis();

                }

                //  In UTF mode we only execute the loop once

                if ( utf != null ) {

                    break;

                }

                // Wait 3 seconds before polling again

                daemonSleepSecs(mPollTime);

            }

            mLogger.error("BuildDaemon daemon spawning shutdown");

            if (mShutDown)

            {

                notifyAllBuildThreads();

                cleanUp();

            }

        }

        catch( BuildException e )

        {

            mLogger.error("BuildDaemon caught Exception");

        }

    }

    /**

     * Used by the Polling Thread to notify slaves of an active build request

     * The slaves will 'wait' for notification in order to proceed

     * Thus the polling is done once, rather than each slave thread polling individually

     * @param rconId

     */

    public void notifyBuildThread(Integer rconId) {

        for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext(); )

        {

            ThreadIdentifier threadIdentifier = it.next();

            if (threadIdentifier.mThread.isAlive())

            {

                BuildThread bt = threadIdentifier.mThread;

                if (bt.mRconId == rconId ) {

                    mLogger.warn("Notify RtagId:{}, bmlId:{}", bt.mRtagId, bt.mRconId );

                    synchronized (bt.mActiveBuildMonitor) {

                        bt.mActiveBuildMonitor.notifyAll();

                    }

                }

            }

        }

    }

    /**

     * Notify all build threads of some activity

     * This will release all threads waiting on the ActiveBuildMonitor

     */

    public void notifyAllBuildThreads() {

        mLogger.error("notifyAllBuildThreads");

        for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext(); )

        {

            ThreadIdentifier threadIdentifier = it.next();

            if (threadIdentifier.mThread.isAlive())

            {

                BuildThread bt = threadIdentifier.mThread;

                synchronized (bt.mActiveBuildMonitor) {

                    bt.mActiveBuildMonitor.notifyAll();

                }

            }

        }

    }

    /** Look for changes in the build daemons required

     *  Startup new daemons

     *  Shutdown ones that are no longer required

     * 

     *  @param  releaseManager - The release Manager instance

     *  @param  utf            - Unit Test information 

     */

    private void lookForNewBuildDaemons (ReleaseManager releaseManager, String utf) {

        mLogger.error("lookForNewBuildDaemons as {}", mHostname );

        try

        {

            // Create a list of all machines that are configured to run on this machine

            //

            releaseManager.queryReleaseConfig(mHostname);

            //

            //  Iterate over all the configured machines

            //  Start up new build threads for new machines

            //

            for (Iterator<ReleaseConfig> it = releaseManager.mReleaseConfigCollection.mReleaseConfig.iterator(); it.hasNext(); )

            {

                ReleaseConfig rc = it.next();

                if (!isActive(rc.getRconId()))

                {

                    //

                    //  Ensure the Release is configured with exactly one master

                    //

                    int masterCount = releaseManager.queryMasterCount( rc.getRtagId() );

                    if ( masterCount != 1 )

                    {

                        mLogger.error("BuildDaemon activating. Invalid Masters {} MasterCount: {}", rc.getRtagId(),  masterCount );

                        continue;

                    }

                    mLogger.warn("BuildDaemon activating {} {} {}",rc.getRtagId(), rc.getRconId(), rc.getDaemonMode());

                    //

                    //    Clone the BuildDaemons ReleaseManager thread

                    //    Done so that we can perform unit testing by extending the ReleaseManager class

                    //    Need to be able to have the Master/Slave threads process through the overridden

                    //    methods within the extended ReleaseManager class

                    //

                    ReleaseManager threadReleaseManager = (ReleaseManager) releaseManager.clone();

                    // spawn and run the BuildThread

                    if (rc.getDaemonMode() == 'M')

                    {

                        MasterThread thread = new MasterThread(rc.getRtagId(), rc.getRconId(), threadReleaseManager, utf);

                        ThreadIdentifier threadIdentifier =  new ThreadIdentifier(rc.getRconId(), thread);

                        mThreadCollection.add(threadIdentifier);

                        thread.start();

                    }

                    else if (rc.getDaemonMode() == 'S')

                    {

                        SlaveThread thread = new SlaveThread(rc.getRtagId(), rc.getRconId(), threadReleaseManager, utf);

                        ThreadIdentifier threadIdentifier = new ThreadIdentifier(rc.getRconId(), thread);

                        mThreadCollection.add(threadIdentifier);

                        thread.start();

                    }

                }

            }

            //

            //  Clean out terminated threads from the thread collection

            //

            cleanupTerminatedThreads();

        }

        catch (SQLException e)

        {

            mLogger.warn("BuildDaemon caught SQLException");

        }

        catch (InterruptedException e)

        {

            mLogger.warn("BuildDaemon caught InterruptedException");

            Thread.currentThread().interrupt();

        }

        catch (Exception e)

        {

            mLogger.warn("BuildDaemon caught Exception");

        }

    }

    /** Look for new build requests for slaves on this machine

     *  It is better to have one thread poll every 3 seconds, than have all the slave

     *  threads poll every 3 seconds 

     * 

     * @param releaseManager

     * @param utf

     */

    private void lookForBuildRequests(ReleaseManager releaseManager, String utf) {

        if (utf != null) {

            return;

        }

        mLogger.debug("Poll Cycle");

        try {

            //  Generate a list of all rconIds that are currently active on this machine

            //  Will be used to limit the query

            //

            StringBuilder rconIdList = new StringBuilder();

            String joiner = null;

            for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext(); )

            {

                ThreadIdentifier threadIdentifier = it.next();

                if (threadIdentifier.mThread.isAlive())

                {

                    BuildThread bt = threadIdentifier.mThread;

                    if( joiner != null) {

                        rconIdList.append(joiner);

                    }

                    rconIdList.append(bt.mRconId);

                    joiner = ",";

                }

            }

            if (joiner != null) {

                //

                //  Locate slaves that have an outstanding build request and master that have forced poll requests

                //  Notify the threads of this condition

                ArrayList<Integer> rv = releaseManager.queryActivatedBuilds(mHostname, rconIdList.toString());

                for (Integer rconId : rv) {

                    mLogger.warn("Activate:{}", rconId);

                    notifyBuildThread(rconId);

                }

            }

        } catch (Exception e) {

            mLogger.error("lookForBuildRequests Exception: {}", e.getMessage());

            mPollTime = 30;

        }

    }

    /**

     * Start up the Nagios server

     * @throws BuildException

     */

    private void startNagiosServer() throws BuildException {

        try {

            nagiosSrv = new ServerSocket(1111);

            nagiosChecker = new NagiosThread(nagiosSrv, this);

            nagiosChecker.start();

        } catch ( IOException e ) {

            throw new BuildException("Nagios port in use");

        }

    }

    /**

     * Clean out terminated threads from the thread collection

     * Examine all the build threads and get rid of those that have been terminated

     */

    private void cleanupTerminatedThreads()

    {

        for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext(); )

        {

            ThreadIdentifier threadIdentifier = it.next();

            if (!threadIdentifier.getThread().isAlive())

            {

                BuildThread bt = threadIdentifier.getThread();

                mLogger.warn("BuildDaemon removal {} {} {}", bt.mRtagId, bt.mRconId, bt.getMode());

                it.remove();

            }

        }

    }

    /**calls isAlive on the Thread object associated with the rcon_id

     */

    public boolean isActive(final int rcon_id)

    {

        mLogger.debug("isActive {}", rcon_id);

        boolean retVal = false;

        boolean found = false;

        for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext(); )

        {

            ThreadIdentifier threadIdentifier = it.next();

            if (threadIdentifier.getRconId() == rcon_id)

            {

                found = true;

                if (threadIdentifier.getThread().isAlive())

                {

                    retVal = true;

                    break;

                }

                else

                {

                    mLogger.warn("isActive found dead thread {}", rcon_id );

                }

            }

        }

        if ( !found )

        {

            mLogger.warn("isActive thread not found {}", rcon_id);

        }

        mLogger.debug("isActive returned {}", retVal);

        return retVal;

    }

    /**

     *  Nagios interface

     *          Must have one active thread

     *          Examine all threads - for logging

     */

    void checkThreads( NagiosInfo nagInfo)

    {

        mLogger.info("checkThreads called");

        if (!mThreadCollection.isEmpty())

        {

            for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext();) 

            {

                ThreadIdentifier threadIdentifier = it.next();

                nagInfo.threadCount ++;

                if ( threadIdentifier.getThread().isAlive()) {

                    nagInfo.threadAliveCount ++;

                }

                threadIdentifier.getThread().checkThread(nagInfo);

            }

        }

        else

        {

            nagInfo.extendedReason.add("No Build Threads configured");

        }

    }

    /**

     *  Nagios interface

     *      Provide extended information on all threads

     * @param estatus 

     */

    public void extendedStatus(Map<String, Object> estatus)

    {

        NagiosInfo nagInfo = new NagiosInfo();

        for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext();) 

        {

            ThreadIdentifier threadIdentifier = it.next();

            threadIdentifier.getThread().extendedStatus(nagInfo, estatus);

        }

    }

    /**

     * daemonSleepSecs

     * Sleep for a specified number of seconds allowing for

     * a termination request

     * 

     * The function will sleep for 5 seconds at a time, 

     * then check for a termination request

     * 

     * @param sleepSecs The number of seconds to sleep

     * 

     * @return True :    Termination requested

     *         

     *         False:    No Termination requested

     */

    public static boolean daemonSleepSecs(int sleepSecs)

    {

        while (sleepSecs > 0)

        {

            if (mShutDown)

            {

                mLogger.error("daemonSleepSecs detected termiantion request");

                return true;

            }

            int sleepTime = sleepSecs;

            if (sleepTime > 5)

                sleepTime = 5;

            try

            {

                Thread.sleep(sleepTime * 1000L);

            }

            catch (InterruptedException e)

            {

                mLogger.warn("daemonSleepSecs sleep caught InterruptedException");

                Thread.currentThread().interrupt();

                break;

            }

            sleepSecs -= sleepTime;

        }

        return false;

    }

    /**terminates all BuildThreads

     */

    public void cleanUp()

    {

        mLogger.warn("cleanUp");

        for (Iterator<ThreadIdentifier> it = mThreadCollection.iterator(); it.hasNext(); )

        {

            ThreadIdentifier threadIdentifier = it.next();

            if (threadIdentifier.getThread().isAlive())

            {

                try

                {

                    threadIdentifier.getThread().interrupt();

                    threadIdentifier.getThread().join();

                }

                catch( InterruptedException e )

                {

                    Thread.currentThread().interrupt();

                }

            }

        }

        if ( nagiosChecker != null )

        {

            nagiosChecker.terminate();

        }

        //  If a shutdown has been requested, then exit the program

        //  The only thing left running should be a timer task

        if (mShutDown)

        {

            System.exit(6);

        }

    }

}

class BuildDaemonUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler

{

    private static final Logger mLogger = LoggerFactory.getLogger(BuildDaemonUncaughtExceptionHandler.class);

    //Implements Thread.UncaughtExceptionHandler.uncaughtException()

    public void uncaughtException(Thread th, Throwable ex)

    {

        System.out.println("You crashed thread " + th.getName());

        System.out.println("Exception was: " + ex.toString());

        mLogger.error("UncaughtException ThreadName: {}", th.getName());

        mLogger.error("UncaughtException was: {}", ex);

    }

}