public CoordinatedStack call()

in chop/webapp/src/main/java/org/apache/usergrid/chop/webapp/coordinator/SetupStackThread.java [79:203]
103 lines of code
12 McCabe index (conditional complexity)

    public CoordinatedStack call() {

        String keyFile;
        LinkedList<String> launchedInstances = new LinkedList<String>();

        providerParamsDao = InjectorFactory.getInstance( ProviderParamsDao.class );
        chopUiFig = InjectorFactory.getInstance( ChopUiFig.class );

        ProviderParams providerParams = providerParamsDao.getByUser( stack.getUser().getUsername() );

        /** Bypass the keys in AmazonFig so that it uses the ones belonging to the user */
        AmazonFig amazonFig = InjectorFactory.getInstance( AmazonFig.class );
        amazonFig.bypass( AmazonFig.AWS_ACCESS_KEY, providerParams.getAccessKey() );
        amazonFig.bypass( AmazonFig.AWS_SECRET_KEY, providerParams.getSecretKey() );

        InstanceManager instanceManager = InjectorFactory.getInstance( InstanceManager.class );
        IpRuleManager ipRuleManager = InjectorFactory.getInstance( IpRuleManager.class );

        File runnerJar = CoordinatorUtils.getRunnerJar( chopUiFig.getContextPath(), stack );

        ipRuleManager.setDataCenter( stack.getDataCenter() );
        ipRuleManager.applyIpRuleSet( stack.getIpRuleSet() );

        /** Setup clusters */
        for ( ICoordinatedCluster cluster : stack.getClusters() ) {
            LOG.info( "Starting setting up cluster {}...", cluster.getName() );
            keyFile = providerParams.getKeys().get( cluster.getInstanceSpec().getKeyName() );
            if ( keyFile == null ) {
                errorMessage = "No key found with name " + cluster.getInstanceSpec().getKeyName() +
                        " for cluster " + cluster.getName();
                LOG.warn( errorMessage + ", aborting and terminating launched instances..." );
                instanceManager.terminateInstances( launchedInstances );
                stack.setSetupState( SetupStackSignal.FAIL );
                stack.notifyAll();
                return null;
            }
            if ( !( new File( keyFile ) ).exists() ) {
                errorMessage = "Key file " + keyFile + " for cluster " + cluster.getName() + " not found";
                LOG.warn( errorMessage + ", aborting and terminating launched instances..." );
                instanceManager.terminateInstances( launchedInstances );
                stack.setSetupState( SetupStackSignal.FAIL );
                stack.notifyAll();
                return null;
            }

            LaunchResult result = instanceManager.launchCluster( stack, cluster,
                    chopUiFig.getLaunchClusterTimeout() );

            for ( Instance instance : result.getInstances() ) {
                launchedInstances.add( instance.getId() );
                cluster.add( instance );
            }

            /** Setup system properties, deploy the scripts and execute them on cluster instances */
            boolean success = false;
            try {
                success = CoordinatorUtils.executeClusterSSHCommands( cluster, runnerJar, keyFile );
            }
            catch ( Exception e ) {
                LOG.warn( "Error while executing SSH commands", e );
            }
            if ( ! success ) {
                errorMessage = "SSH commands have failed, will not continue";
                instanceManager.terminateInstances( launchedInstances );
                stack.setSetupState( SetupStackSignal.FAIL );
                stack.notifyAll();
                return null;
            }
            LOG.info( "Cluster {} is ready, moving on...", cluster.getName() );
        }

        /** Setup runners */
        LOG.info( "Starting setting up runner instances..." );
        keyFile = providerParams.getKeys().get( providerParams.getKeyName() );
        if ( keyFile == null ) {
            errorMessage = "No key found with name " + providerParams.getKeyName() + " for runners";
            LOG.warn( errorMessage + ", aborting and terminating launched instances..." );
            instanceManager.terminateInstances( launchedInstances );
            stack.setSetupState( SetupStackSignal.FAIL );
            stack.notifyAll();
            return null;
        }
        if ( ! ( new File( keyFile ) ).exists() ) {
            errorMessage = "Key file " + keyFile + " for runners not found";
            LOG.warn( errorMessage + ", aborting and terminating launched instances..." );
            instanceManager.terminateInstances( launchedInstances );
            stack.setSetupState( SetupStackSignal.FAIL );
            stack.notifyAll();
            return null;
        }
        BasicInstanceSpec runnerSpec = new BasicInstanceSpec();
        runnerSpec.setImageId( providerParams.getImageId() );
        runnerSpec.setType( providerParams.getInstanceType() );
        runnerSpec.setKeyName( providerParams.getKeyName() );

        LaunchResult result = instanceManager.launchRunners( stack, runnerSpec,
                chopUiFig.getLaunchClusterTimeout() );

        for ( Instance instance : result.getInstances() ) {
            launchedInstances.add( instance.getId() );
            stack.addRunnerInstance( instance );
        }

        /** Deploy and start runner.jar on instances */
        boolean success = false;
        try {
            success = CoordinatorUtils.executeRunnerSSHCommands( stack, runnerJar, keyFile );
        }
        catch ( Exception e ) {
            LOG.warn( "Error while executing SSH commands", e );
        }
        if ( ! success ) {
            errorMessage = "SSH commands have failed, will not continue";
            instanceManager.terminateInstances( launchedInstances );
            stack.setSetupState( SetupStackSignal.FAIL );
            stack.notifyAll();
            return null;
        }

        stack.setSetupState( SetupStackSignal.COMPLETE );
        LOG.info( "Stack {} is set up and ready...", stack.getName() );

        stack.notifyAll();
        return stack;
    }