public class Ec2Util {
	
	private static final Logger logger = Logger.getLogger(Ec2Util.class);
	
	/**
	 * Returns map of instanceId:privateDnsName
	 * 
	 * @param ec2Instances
	 * @return
	 */
	public Map<String,String> getPrivateDNSNames(List<Instance> ec2Instances) {
		TreeMap<String,String> names = new TreeMap<String,String>(String.CASE_INSENSITIVE_ORDER);
		for (Instance i : ec2Instances) {
			names.put(i.getInstanceId(),i.getPrivateDnsName().toLowerCase());
		}
		return names;
	}
	
	/**
	 * Returns map of instanceId:privateIp
	 * 
	 * @param ec2Instances
	 * @return
	 */
	public Map<String,String> getPrivateIPs(List<Instance> ec2Instances) {
		TreeMap<String,String> names = new TreeMap<String,String>(String.CASE_INSENSITIVE_ORDER);
		for (Instance i : ec2Instances) {
			names.put(i.getInstanceId(),i.getPrivateIpAddress());
		}
		return names;
	}
	
	public void startInstance(AmazonEC2Client ec2Client, String instanceId) throws Exception {
		StartInstancesRequest startReq = new StartInstancesRequest();
		List<String> instanceIds = new ArrayList<String>();
		instanceIds.add(instanceId);
		startReq.setInstanceIds(instanceIds);
		logger.debug("Starting EC2 instance...." + Arrays.toString(instanceIds.toArray(new String[]{})));
		ec2Client.startInstances(startReq);
	}
	
	public void stopInstance(AmazonEC2Client ec2Client, String instanceId) throws Exception {
		StopInstancesRequest stopReq = new StopInstancesRequest();
		List<String> instanceIds = new ArrayList<String>();
		instanceIds.add(instanceId);
		stopReq.setInstanceIds(instanceIds);
		logger.debug("Stopping EC2 instance...." + Arrays.toString(instanceIds.toArray(new String[]{})));
		ec2Client.stopInstances(stopReq);
	}
	
	public void terminateEc2Instance(AmazonEC2Client ec2Client, String instanceId) throws Exception {
		try {
			TerminateInstancesRequest termReq = new TerminateInstancesRequest();
			List<String> instanceIds = new ArrayList<String>();
			instanceIds.add(instanceId);
			termReq.setInstanceIds(instanceIds);
			logger.debug("Terminating EC2 instances...." + Arrays.toString(instanceIds.toArray(new String[]{})));
			ec2Client.terminateInstances(termReq);
			
		} catch(Exception e) {
			logger.error("Unexpected error terminating: " + instanceId + " "+ e.getMessage(),e);
		}
	}

	 
	public List<Instance> launchEc2Instances(AmazonEC2Client ec2Client, Properties props) throws Exception {
		
		Integer totalExpectedWorkers = Integer.valueOf(props.getProperty("master.workers.total"));
		
		// disk size
		Collection<BlockDeviceMapping> blockDevices = new ArrayList<BlockDeviceMapping>();
		blockDevices.add(
				new BlockDeviceMapping()
					.withDeviceName(props.getProperty("master.workers.ec2.disk.deviceName"))
					.withEbs(new EbsBlockDevice()
							.withVolumeType(VolumeType.valueOf(props.getProperty("master.workers.ec2.disk.volumeType")))
							.withDeleteOnTermination(true)
						    .withVolumeSize(Integer.valueOf(props.getProperty("master.workers.ec2.disk.size.gigabytes")))));
		
		// create our run request for the total workers we expect
		RunInstancesRequest runInstancesRequest = new RunInstancesRequest();
		runInstancesRequest.withImageId(props.getProperty("master.workers.ec2.ami.id"))
					        .withInstanceType(props.getProperty("master.workers.ec2.instanceType"))
					        .withMinCount(totalExpectedWorkers)
					        .withMaxCount(totalExpectedWorkers)
					        .withBlockDeviceMappings(blockDevices)
					        .withKeyName(props.getProperty("master.workers.ec2.keyName"))
					        .withSecurityGroupIds(props.getProperty("master.workers.ec2.securityGroupId"))
					        .withInstanceInitiatedShutdownBehavior(ShutdownBehavior.valueOf(props.getProperty("master.workers.ec2.shutdownBehavior")))
					        .withSubnetId(props.getProperty("master.workers.ec2.subnetId"))
					        .withUserData(Base64.encodeAsString(readFile(props.getProperty("master.workers.ec2.userDataFile")).getBytes()));
		
		// launch
		logger.debug("Launching " + totalExpectedWorkers + " EC2 instances, " +
							"it may take few minutes for workers to come up...: \n" +
							"\tamiId:" + runInstancesRequest.getImageId() +"\n"+
							"\tsecGrpId:" + runInstancesRequest.getSecurityGroupIds().get(0) +"\n"+
							"\tsubnetId:" + runInstancesRequest.getSubnetId() +"\n"+
							"\tinstanceType:" + runInstancesRequest.getInstanceType() +"\n"+
							"\tshutdownBehavior:" + runInstancesRequest.getInstanceInitiatedShutdownBehavior() +"\n"+
							"\tkeyName:" + runInstancesRequest.getKeyName() 
							);
	

		// as the instances come up, assuming the "userData" above launches the worker we will be good
		// they will auto register w/ us the master 
		RunInstancesResult result = ec2Client.runInstances(runInstancesRequest);
		Reservation reservation = result.getReservation();
		return reservation.getInstances();
	}
	
	public InstanceStatus getInstanceStatus(AmazonEC2Client ec2Client, String instanceId) {
		List<String> instanceIds = new ArrayList<String>();
		instanceIds.add(instanceId);
		DescribeInstanceStatusRequest statusReq = new DescribeInstanceStatusRequest();
		statusReq.setInstanceIds(instanceIds);
		DescribeInstanceStatusResult result = ec2Client.describeInstanceStatus(statusReq);
		List<InstanceStatus> statuses = result.getInstanceStatuses();
		if (statuses == null || statuses.size() == 0) {
			return null;
		}
		return statuses.iterator().next();
	}
	
	public List<String> dumpEc2InstanceStatus(AmazonEC2Client ec2Client, List<Instance> ec2Instances) {
		try {
			List<String> instanceIds = new ArrayList<String>();
			
			for (Instance ec2node : ec2Instances) {
				instanceIds.add(ec2node.getInstanceId());
			}
			
			DescribeInstanceStatusRequest statusReq = new DescribeInstanceStatusRequest();
			statusReq.setInstanceIds(instanceIds);
			DescribeInstanceStatusResult result = ec2Client.describeInstanceStatus(statusReq);
			
			List<InstanceStatus> statuses = result.getInstanceStatuses();
			
			List<String> impairedInstances = new ArrayList<String>();
			
			StringBuffer sb = new StringBuffer("EC2 worker instance STATUS:\n");
			for (InstanceStatus status : statuses) {
				sb.append("\tid:"+status.getInstanceId() + 
						"\taz:" + status.getAvailabilityZone() + 
						"\tstate:" + status.getInstanceState().getName() + 
						"\tstatus:" + status.getInstanceStatus().getStatus() + 
						"\tsystem_status: " + status.getSystemStatus().getStatus() + "\n"); 
				
				if (status.getInstanceStatus().getStatus().equalsIgnoreCase("impaired")) {
					impairedInstances.add(status.getInstanceId());
				}
			}
			
			logger.info(sb.toString()+"\n");
			
			return impairedInstances;
			
		} catch(Exception e) {
			logger.error("Error getting instance state: " + e.getMessage(),e);
			return null;
		}
		
	}
	
	public static String readFile(String path) throws IOException {
		File file = new File(path);
		FileInputStream fis = new FileInputStream(file);
		byte[] data = new byte[(int)file.length()];
		fis.read(data);
		fis.close();
		return new String(data, "UTF-8");
	}
	
} public class S3Util {
	
	private static final Logger logger = Logger.getLogger(S3Util.class);

	public void uploadToS3(AmazonS3Client s3Client, 
						  String bucketName, 
						  String s3LogBucketFolderRoot, 
						  String host, 
						  List<String> filePathsToUpload) {
		
		try {
			
			for (String file : filePathsToUpload) {
			
				String key = null;
				try {
					File item = new File(file.trim());
					
					if (!item.exists()) {
						logger.error("uploadToS3() cannot upload item, does not exist! " + item.getAbsolutePath());
						continue;
					}
					
					// default to the one file
					File[] allFiles = new File[]{item};
					
					if (item.isDirectory()) {
						allFiles = item.listFiles();
					}
					
					for (File toUpload : allFiles) {
						
						if (!toUpload.exists() || toUpload.getName().startsWith(".") || toUpload.isDirectory()) {
							logger.error("uploadToS3() cannot upload, does not exist, starts w/ . or is a directory: " + toUpload.getAbsolutePath());
							continue;
						}
					
						key = s3LogBucketFolderRoot + "/" + host + "/" + toUpload.getName();
						 
						PutObjectRequest req = new PutObjectRequest(bucketName, key, toUpload);
						req.setStorageClass(StorageClass.ReducedRedundancy);
						ObjectMetadata objectMetadata = new ObjectMetadata();
						objectMetadata.setContentType("text/plain");
						objectMetadata.setSSEAlgorithm(ObjectMetadata.AES_256_SERVER_SIDE_ENCRYPTION);     
						req.setMetadata(objectMetadata);
						
						s3Client.putObject(req);
					}
					
				} catch(Exception e) {
					logger.error("uploadToS3() unexpected error uploading logs to: " +bucketName + " key:"+ key + " for " +file);
				}
				
			}
			
			
		} catch(Exception e) {
			logger.error("uploadToS3() error uploading logs to S3: " + e.getMessage(),e);
		}
		
	}

} public class CmdResult {
	
	private int exitCode;
	private String stdOut;
	private String stdErr;
	
	public CmdResult(int exitCode, String stdOut, String stdErr) {
		super();
		this.exitCode = exitCode;
		this.stdOut = stdOut;
		this.stdErr = stdErr;
	}
	
	public int getExitCode() {
		return exitCode;
	}
	public void setExitCode(int exitCode) {
		this.exitCode = exitCode;
	}
	public String getStdOut() {
		return stdOut;
	}
	public void setStdOut(String stdOut) {
		this.stdOut = stdOut;
	}
	public String getStdErr() {
		return stdErr;
	}
	public void setStdErr(String stdErr) {
		this.stdErr = stdErr;
	}

} public class CommandExecutor {

	private static final Logger logger = Logger.getLogger(CommandExecutor.class);

	public CmdResult execute(CommandLine cmdLine, int maxAttempts) {
		
		CmdResult lastCmdResult = null;
		
		int attempts = 0;
		while(attempts < maxAttempts) {
		
			attempts++;
			
			final StringWriter stdOut = new StringWriter();
			final StringWriter stdErr = new StringWriter();
			
			try {
			
				DefaultExecutor executor = new DefaultExecutor();
				executor.setStreamHandler(new ExecuteStreamHandler() {
						public void setProcessOutputStream(InputStream is) throws IOException {IOUtils.copy(is, stdOut, "UTF-8");}
						public void setProcessErrorStream(InputStream is) throws IOException {IOUtils.copy(is, stdErr, "UTF-8");}
						public void stop() throws IOException {}
						public void start() throws IOException {}
						public void setProcessInputStream(OutputStream os) throws IOException {}
					});
					
				logger.trace("Executing: attempt:" + attempts + " " + cmdLine.toString());
					
				int exitValue = executor.execute(cmdLine);
				if (exitValue > 0) {
					logger.error("ERROR: attempt #: " + attempts+ " exitCode: "+exitValue+" cmd=" + cmdLine.toString());
				}
				
				//System.out.println("STDOUT:"+stdOut);
				//System.out.println("STDERR:"+stdErr);
				
				lastCmdResult = new CmdResult(exitValue,stdOut.toString(),stdErr.toString());
				
				// if successful exit loop immediately...
				if (exitValue == 0) {
					logger.trace("SUCCESS! exitCode = 0: " + cmdLine.toString());
					break;
				}
				
			} catch(Exception e) {
				logger.error("execute() attempt #: " + attempts+ " cmd:"+cmdLine.toString() + " exception:"+e.getMessage(),e);
				lastCmdResult = new CmdResult(9999, stdOut.toString(), "attempt #: " + attempts+ " exception: " + e.getMessage() + " stdErr: " + stdErr.toString());
			}
		}
		
		return lastCmdResult;
	}
	

} public enum CCMode {
	
	INITIALIZED,
	PREPARE,
	WRITE,
	IDLE,
	VALIDATE,
	REPORT_ERRORS
	
} public class CCPayload {

	public boolean fromMaster = false;
	public String sourceHostId = null;
	public String sourceHostIP = null;
	public String onlyForHostIdOrIP = null;
	public CCPayloadType type = null;
	public Object value = null;

} public interface CCPayloadHandler {

	public void handlePayload(CCPayload payload) throws Exception;
	
} public enum CCPayloadType {
	
	MASTER_CURRENT_MODE,  // current mode of the master
	WORKER_CURRENT_MODE,  // current mode of workers
	
	WORKER_WRITES_CURRENT_SUMMARY, 		// sent periodically during write mode stating number of successful/failed so far
	WORKER_VALIDATIONS_CURRENT_SUMMARY, 		// sent periodically during write mode stating number of successful/failed so far

	WORKER_WRITES_FINISHED_SUMMARY,  	// sent by workers when idle and report total WRITE mode messages processed
	WORKER_VALIDATIONS_FINISHED_SUMMARY, // sent by workers when idle and report total VALIDATE mode messages processed
	
	WORKER_ERROR_REPORT_DETAILS,  // sent by workers when REPORT_ERRORS mode is switched on
	
	CMD_WORKER_SHUTDOWN   // sent when master to tell worker to shutdown

} public class ControlChannel implements Runnable {
	
	private static final Logger logger = Logger.getLogger(ControlChannel.class);
	
	private AmazonSNSClient snsClient = null;
	private AmazonSQSClient sqsClient = null;
	
	private String snsTopicARN = null;
	private String snsSubscriptionARN = null;
	private String sqsQueueUrl = null;
	private String sqsQueueARN = null;
	
	private Gson gson = new Gson();
	
	private CCPayloadHandler ccPayloadHandler = null;
	private String mySourceIdentifier = null;
	private String mySourceIp = null;
	private Thread consumerThread = null;
	
	private boolean canDestroyTopic = false;
	
	private boolean running = true;
	private String snsControlTopicName = null;
	
	private String uuid = UUID.randomUUID().toString().replace("-", "").substring(0,4);;
	
	public ControlChannel(boolean callerIsMaster,
						  String awsAccessKey, String awsSecretKey, String snsControlTopicName, 
						  String userAccountPrincipalId, 
						  String userARN, 
						  CCPayloadHandler ccPayloadHandler) throws Exception {
		super();
		
		try {
			this.mySourceIp = InetAddress.getLocalHost().getHostAddress();
		} catch(Exception e) {
			logger.error("Error getting local inet address: " + e.getMessage());
		}
		
		mySourceIdentifier = determineHostName() + "-" +uuid;
		this.snsControlTopicName = snsControlTopicName;
		
		if (callerIsMaster) {
			canDestroyTopic = true;
			this.snsControlTopicName += "-" + mySourceIdentifier;
		}
		
		this.ccPayloadHandler = ccPayloadHandler;
		
		sqsClient = new AmazonSQSClient(new BasicAWSCredentials(awsAccessKey, awsSecretKey));
		snsClient = new AmazonSNSClient(new BasicAWSCredentials(awsAccessKey, awsSecretKey));
		
		
		this.connectToTopic(callerIsMaster, 1000, userAccountPrincipalId, userARN);
		
	

	}

	
	public void connectToTopic(boolean callerIsMaster, int maxAttempts, String userAccountPrincipalId, String userARN) throws Exception {
		
		
		// try up to max attempts to connect to pre-existing topic
		for (int i=0; i<maxAttempts; i++) {
			
			logger.debug("connectToTopic() attempt: " + (i+1));
			
			ListTopicsResult listResult = snsClient.listTopics();
			List<Topic> topics = listResult.getTopics();
			
			while(topics != null) {
				
				for (Topic topic : topics) {
	
					// note we do index of match....
					if (topic.getTopicArn().indexOf(snsControlTopicName) != -1) {
						snsTopicARN = topic.getTopicArn();
						logger.info("Found existing SNS topic by name: "+snsControlTopicName + " @ " + snsTopicARN);
						break;
					}
				}
	
				String nextToken = listResult.getNextToken();
				
				if (nextToken != null && snsTopicARN == null) {
					listResult = snsClient.listTopics(nextToken);
					topics = listResult.getTopics();
					
				} else {
					break;
				}
			}
			
			// if consumer, retry, otherwise is master, so just exit quick to create...
			if (snsTopicARN == null && !callerIsMaster) {
				Thread.currentThread().sleep(1000);
				continue;
			} else {
				break; // exit;
			}
		}
		
		
		
		// if master only he can create...
		if (snsTopicARN == null && callerIsMaster) {
			this.snsControlTopicName = this.snsControlTopicName.substring(0,(snsControlTopicName.length() > 80 ? 80 : this.snsControlTopicName.length()));
			
			logger.info("Attempting to create new SNS control channel topic by name: "+this.snsControlTopicName);
			
			CreateTopicResult createTopicResult = snsClient.createTopic(this.snsControlTopicName);
			snsTopicARN = createTopicResult.getTopicArn();
			snsClient.addPermission(snsTopicARN, "Permit_SNSAdd", 
									Arrays.asList(new String[]{userARN}), 
									Arrays.asList(new String[]{"Publish","Subscribe","Receive"}));
			logger.info("Created new SNS control channel topic by name: "+this.snsControlTopicName + " @ " + snsTopicARN);
			
		} else if (snsTopicARN == null) {
			throw new Exception("Worker() cannot start, snsControlTopicName has yet to be created by master?: " + this.snsControlTopicName);
		}
		
		// http://www.jorgjanke.com/2013/01/aws-sns-topic-subscriptions-with-sqs.html
		
		// create SQS queue to get SNS notifications (max 80 len)
		String prefix =  ("s3bktLoaderCC_" + mySourceIdentifier);
		String sqsQueueName = prefix.substring(0,(prefix.length() > 80 ? 80 : prefix.length()));
		
		CreateQueueResult createQueueResult = sqsClient.createQueue(sqsQueueName);
		this.sqsQueueUrl = createQueueResult.getQueueUrl();
		this.sqsQueueARN = sqsClient.getQueueAttributes(sqsQueueUrl, Arrays.asList(new String[]{"QueueArn"})).getAttributes().get("QueueArn");

		Statement statement = new Statement(Effect.Allow)
								.withActions(SQSActions.SendMessage)
								 .withPrincipals(new Principal("*"))
								 .withConditions(ConditionFactory.newSourceArnCondition(snsTopicARN))
								 .withResources(new Resource(sqsQueueARN));
		Policy policy = new Policy("SubscriptionPermission").withStatements(statement);

		HashMap<String, String> attributes = new HashMap<String, String>();
		attributes.put("Policy", policy.toJson());
		SetQueueAttributesRequest request = new SetQueueAttributesRequest(sqsQueueUrl, attributes);
		sqsClient.setQueueAttributes(request);

		logger.info("Created SQS queue: " + sqsQueueARN + " @ " + sqsQueueUrl);
		
		// subscribe our SQS queue to the SNS:s3MountTest topic
		SubscribeResult subscribeResult = snsClient.subscribe(snsTopicARN,"sqs",sqsQueueARN);
		snsSubscriptionARN = subscribeResult.getSubscriptionArn();
		logger.info("Subscribed for messages from SNS control channel:" + snsTopicARN + " ----> SQS: "+sqsQueueARN);
		logger.info("Subscription ARN: " + snsSubscriptionARN);
		
		this.consumerThread = new Thread(this,"ControlChannel msg consumer thread");
		this.consumerThread.start();

		logger.info("\n-------------------------------------------\n" +
					"CONTROL CHANNEL: ALL SNS/SQS resources hooked up OK\n" +
					"-------------------------------------------\n");
	}
		

	public void send(boolean fromMaster, CCPayloadType type, Object value) throws Exception {
		this.send(fromMaster,type,null,value);
	}
	
	public void send(boolean fromMaster, CCPayloadType type, String onlyForHostOrIp, Object value) throws Exception {
		CCPayload payload = new CCPayload();
		payload.fromMaster = fromMaster;
		payload.type = type;
		payload.value = value;
		payload.onlyForHostIdOrIP = onlyForHostOrIp;
		payload.sourceHostId = this.mySourceIdentifier.trim();
		payload.sourceHostIP = this.mySourceIp;
		
		logger.debug("Sending: " + type + "="+value);
		
		// send!
		this.snsClient.publish(this.snsTopicARN, gson.toJson(payload));
	}

	public void run() {
		
		while(running) {
			
			try {
				
				ReceiveMessageRequest req = new ReceiveMessageRequest();
				req.setWaitTimeSeconds(10);
				req.setMaxNumberOfMessages(10);
				req.setVisibilityTimeout(300);
				req.setQueueUrl(sqsQueueUrl);
				
				ReceiveMessageResult msgResult = sqsClient.receiveMessage(req);
				List<Message> messages = msgResult.getMessages();
				
				for (Message msg : messages) {
					
					CCPayload payload = null;

					try {
						Map<String,String> body = gson.fromJson(msg.getBody(), new TypeToken<Map<String, String>>(){}.getType()); 
						payload = gson.fromJson(body.get("Message"), CCPayload.class);
					
					} catch(Exception e) {
						logger.error("ERROR: unexpected error converting SQS Message " +
								"body (json -> CCPayload) body= " + msg.getBody() + " error="+e.getMessage());
						
						// delete the message we just analyzed
						sqsClient.deleteMessage(sqsQueueUrl, msg.getReceiptHandle());
						
						continue;
					}
				
					// if NOT equal to THIS MACHINE... defer tp payload handler
					if (!payload.sourceHostId.equalsIgnoreCase(mySourceIdentifier))  {
						this.ccPayloadHandler.handlePayload(payload);
					}
					
					// delete the message we just analyzed
					sqsClient.deleteMessage(sqsQueueUrl, msg.getReceiptHandle());
				}

			} catch(Exception e) {
				logger.error("run() " + e.getMessage(),e);
			}
			
			try {
				Thread.currentThread().sleep(500);
			} catch(Exception ignore) {}
			
		}
	}


	
	public void destroy() throws Exception {
		
		this.running = false;
		
		Thread.currentThread().sleep(30000);

		if (snsClient != null) {
			try {
				logger.debug("destroy() unsubscribe " + this.snsSubscriptionARN);
				snsClient.unsubscribe(snsSubscriptionARN);
			} catch(Exception e) {
				logger.debug("destroy() error: " + e.getMessage());
			}
		}
		
		if (sqsClient != null) {
			try {
				logger.debug("destroy() " + this.sqsQueueUrl);
				sqsClient.deleteQueue(sqsQueueUrl);
			} catch(Exception e) {
				logger.debug("destroy() error: " + e.getMessage());
			}
		}
		
		if (canDestroyTopic && snsClient != null) {
			try {
				logger.debug("destroy() " + this.snsTopicARN);
				snsClient.deleteTopic(this.snsTopicARN);
			} catch(Exception e) {
				logger.debug("destroy() error: " + e.getMessage());
			}
		}

	}
	
	public AmazonSNSClient getSnsClient() {
		return snsClient;
	}

	public void setSnsClient(AmazonSNSClient snsClient) {
		this.snsClient = snsClient;
	}
	
	private static String determineHostName() throws Exception {

		InetAddress addr = InetAddress.getLocalHost();
		
		// Get IP Address
		byte[] ipAddr = addr.getAddress();
		// Get sourceHost
		String tmpHost = addr.getHostName();

		// we only care about the HOST portion, strip everything else
		// as some boxes report a fully qualified sourceHost such as
		// host.domainname.com

		int firstDot = tmpHost.indexOf('.');
		if (firstDot != -1) {
			tmpHost = tmpHost.substring(0,firstDot);
		}
		return tmpHost;

	}

	public static Logger getLogger() {
		return logger;
	}

	public AmazonSQSClient getSqsClient() {
		return sqsClient;
	}

	public String getSnsTopicARN() {
		return snsTopicARN;
	}

	public String getSnsSubscriptionARN() {
		return snsSubscriptionARN;
	}

	public String getSqsQueueUrl() {
		return sqsQueueUrl;
	}

	public String getSqsQueueARN() {
		return sqsQueueARN;
	}

	public Gson getGson() {
		return gson;
	}

	public CCPayloadHandler getCcPayloadHandler() {
		return ccPayloadHandler;
	}

	public String getMySourceIdentifier() {
		return mySourceIdentifier;
	}

	public Thread getConsumerThread() {
		return consumerThread;
	}


	public String getMySourceIp() {
		return mySourceIp;
	}
	

} public class DirectoryCrawler implements SourceTOCGenerator, Runnable {
	
	private static final Logger logger = Logger.getLogger(DirectoryCrawler.class);

	private File rootDir = null;
	private boolean running = true;
	private int tocInfosGenerated = 0;
	private long lastModifiedAtGreaterThanFilter = -1;
	
	public DirectoryCrawler() {}
	
	public DirectoryCrawler(File rootDir) {
		this.rootDir = rootDir;
	}
	
	public void setRootDir(File rootDir) {
		this.rootDir = rootDir;
		if (!rootDir.exists()) {
			throw new RuntimeException("DirectoryCrawler invalid rootDir: " + rootDir.getAbsolutePath());
		}
	}
	
	public Set<TocInfo> generateTOC(Queue<TocInfo> tocQueue) throws Exception {
		Thread loggingThread = new Thread(this);
		
		Set<TocInfo> toc = new HashSet<TocInfo>();
		loggingThread.start();
		
		scanNode(this.rootDir,toc,tocQueue);
		
		this.running = false; // stop logging
		return toc;
	}
	
	public void run() {
		while (running) {
			try {
				logger.info("Generated TOC current size: " + tocInfosGenerated);
				Thread.currentThread().sleep(30000);
				
			} catch(Exception ignore){}
		}
	}
	
	private void scanNode(File node, Set<TocInfo> toc, Queue<TocInfo> tocQueue) throws Exception {
	
		try {
		
			if (node.exists() && !node.getName().startsWith(".")) {
				
				String adjustedPath = node.getAbsolutePath().replace(this.rootDir.getAbsolutePath(), "");

				// skip root dir
				if (adjustedPath.trim().length() > 0) {
					
					if (node.isFile()) {
						if (this.lastModifiedAtGreaterThanFilter > 0) {
							if (node.lastModified() < this.lastModifiedAtGreaterThanFilter) {
								return; // do nothing, file is older than our filter...
							}
						}
					}
	
					TocInfo finfo = new TocInfo(adjustedPath, (node.isFile() ? node.length() : 0));
					finfo.setIsDirectory(node.isDirectory());
					toc.add(finfo);
					tocQueue.add(finfo);
					tocInfosGenerated++; // increment for logging
				}
			}
			
			if (node.exists() && !node.getName().startsWith(".") && node.isDirectory()) {
				for (File n : node.listFiles()) {
					scanNode(n,toc,tocQueue);
				}
			}
			
		} catch(Throwable e) {
			logger.error("Permission issue? scanNode(node:"+(node != null? (" path:"+node.getAbsolutePath() +" isDirectory():"+node.isDirectory() + " exists:"+node.exists()): "NULL") + 
								  " toc: " + (toc != null? toc.size(): "NULL") + 
								  " tocQueue:"+(tocQueue != null ? tocQueue.size() : " NULL"));
			
			throw new Exception("scanNode() " + e.getMessage(),e);
		}
	}

	public long getLastModifiedAtGreaterThanFilter() {
		return lastModifiedAtGreaterThanFilter;
	}

	public void setLastModifiedAtGreaterThanFilter(long lastModifiedAtGreaterThanFilter) {
		this.lastModifiedAtGreaterThanFilter = lastModifiedAtGreaterThanFilter;
	}

} public class S3BucketObjectLister implements Runnable, SourceTOCGenerator, ProgressListener {
	
	private static final Logger logger = Logger.getLogger(S3BucketObjectLister.class);
	
	private AmazonS3Client s3Client = null;
	private String s3BucketName = null;
	private boolean running = true;
	private int tocInfosGenerated = 0;
	

	public Set<TocInfo> generateTOC(Queue<TocInfo> tocQueue) throws Exception {
		Thread loggingThread = new Thread(this);
		
		Set<TocInfo> toc = new HashSet<TocInfo>();
		loggingThread.start();
		
		scanBucket(toc,tocQueue);
		
		this.running = false; // stop logging
		return toc;
	}
	
	private void scanBucket(Set<TocInfo> toc, Queue<TocInfo> tocQueue) throws Exception {
		
		ListObjectsRequest listRequest = new ListObjectsRequest();
		listRequest.setBucketName(s3BucketName);
		// listRequest.setGeneralProgressListener(this);
		listRequest.setMaxKeys(1000);
		
		String nextMarker = null;
		ObjectListing objectListing = null;
		
		while(true) {
			
			objectListing = s3Client.listObjects(listRequest);
			
			List<S3ObjectSummary> objectSummaries = objectListing.getObjectSummaries();
			
			for (S3ObjectSummary objSummary : objectSummaries) {
				String key = objSummary.getKey();
				
				TocInfo tocInfo = new TocInfo(key, objSummary.getSize());
				
				// is it a "dir/" ?
				if (key.lastIndexOf("/") == (key.length() - 1)) {
					tocInfo.isDirectory = true;
				} else {
					tocInfo.isDirectory = false;
				}
				
				toc.add(tocInfo);
				tocQueue.add(tocInfo);
				tocInfosGenerated++; // increment for logging
	
			}
			
			// for pagination
			nextMarker = objectListing.getNextMarker();
			if (nextMarker == null) {
				break;
			} else {
				listRequest.setMarker(nextMarker);
				logger.debug("scanBucket() nextMarker we will request listing for => " + nextMarker);
			}
		}
		
	}



	@Override
	public void progressChanged(ProgressEvent progressEvent) {
		logger.debug("progressChanged() " +progressEvent.getEventType() + 
				       " bytes:" + progressEvent.getBytes() + 
				      " bytesTransferred: " + progressEvent.getBytesTransferred());
	}
	
	public void run() {
		while (running) {
			try {
				logger.info("\nGenerated TOC current size: " + tocInfosGenerated + "\n");
				Thread.currentThread().sleep(15000);
				
			} catch(Exception ignore){}
		}
	}

	public AmazonS3Client getS3Client() {
		return s3Client;
	}

	public void setS3Client(AmazonS3Client s3Client) {
		this.s3Client = s3Client;
	}

	public String getS3BucketName() {
		return s3BucketName;
	}

	public void setS3BucketName(String s3BucketName) {
		this.s3BucketName = s3BucketName;
	}


} public interface SourceTOCGenerator {

	/**
	 * Generate the TOC list. Implementor should write live/realtime
	 * to the passed tocQueue's as well as return a static Set when complete.
	 * 
	 * @param tocQueue 
	 * @param largeFileMinSize
	 * @param largeFileTocQueue
	 * @return
	 * @throws Exception
	 */
	public Set<TocInfo> generateTOC(Queue<TocInfo> tocQueue) throws Exception;
	
} public class TOCManifestBasedGenerator implements SourceTOCGenerator, Runnable {

	private static final Logger logger = Logger.getLogger(DirectoryCrawler.class);

	private File rootDir = null;
	private File manifestFile = null;
	private boolean running = true;
	private int tocInfosGenerated = 0;
	
	public TOCManifestBasedGenerator() {}
	
	public TOCManifestBasedGenerator(File rootDir) {
		this.rootDir = rootDir;
	}
	
	public void setRootDir(File rootDir) {
		this.rootDir = rootDir;
		if (!rootDir.exists()) {
			throw new RuntimeException("TOCManifestBasedGenerator invalid rootDir: " + rootDir.getAbsolutePath());
		}
	}
	
	public void setManifestFile(File manifestFile) {
		this.manifestFile = manifestFile;
		if (!manifestFile.exists()) {
			throw new RuntimeException("TOCManifestBasedGenerator invalid manifestFile: " + manifestFile.getAbsolutePath());
		}
	}
	
	public Set<TocInfo> generateTOC(Queue<TocInfo> tocQueue) throws Exception {
		Thread loggingThread = new Thread(this);
		
		Set<TocInfo> toc = new HashSet<TocInfo>();
		loggingThread.start();
		
		BufferedReader reader = new BufferedReader(new FileReader(manifestFile));
		String line = null;
		while ((line = reader.readLine()) != null) {
			
			if (line.trim().isEmpty()) {
				continue;
			}
			
			File tocEntry = new File(rootDir.getAbsolutePath() + line.trim());
			
			if (tocEntry.exists()) {
				
				String adjustedPath = tocEntry.getAbsolutePath().replace(this.rootDir.getAbsolutePath(), "");
				TocInfo finfo = new TocInfo(adjustedPath, (tocEntry.isFile() ? tocEntry.length() : 0));
				finfo.setIsDirectory(tocEntry.isDirectory());
				toc.add(finfo);
				tocQueue.add(finfo);
				tocInfosGenerated++; // increment for logging
				
			} else {
				logger.warn("generateTOC() file referenced in manifest file: " + tocEntry.getAbsolutePath() + " does not exist!");
			}
		}

		reader.close();
		
		this.running = false; // stop logging
		return toc;
	}
	
	public void run() {
		while (running) {
			try {
				logger.info("Generated TOC current size: " + tocInfosGenerated);
				Thread.currentThread().sleep(30000);
				
			} catch(Exception ignore){}
		}
	}

} public static enum MODE {WRITE, VALIDATE} public class TOCQueue implements Runnable {
	
	private static final Logger logger = Logger.getLogger(TOCQueue.class);
	
	private AmazonSQSClient sqsClient = null;
	private String tocQueueUrl = null;
	
	private Thread consumerThread = null;
	
	private Gson gson = new Gson();
	
	private TOCPayloadHandler tocPayloadHandler = null;
	private String mySourceIdentifier = null;
	private boolean canDestroyQueue = false;
	private String sqsQueueName = null;
	private String myId = UUID.randomUUID().toString().replaceAll("-", "").substring(0,6);
	private boolean running = true;
	private boolean paused = false;
	
	private long lastSQSMessageReceivedMS = -1;
	private int totalMessagesProcessed = 0;
	private int totalMessageRequestsMade = 0;
	private boolean currentlyProcessingMessage = false;
	
	public TOCQueue(boolean isConsumer, String awsAccessKey, String awsSecretKey, String sqsQueueName, TOCPayloadHandler tocPayloadHandler) throws Exception {
		super();

		mySourceIdentifier = determineHostName() + "-" + UUID.randomUUID().toString().replace("-", "").substring(0,4);
		this.sqsQueueName = sqsQueueName;
		this.tocPayloadHandler = tocPayloadHandler;
		
		if (!isConsumer) { // then I am the master...
			canDestroyQueue = true; 
			this.sqsQueueName += "-" + mySourceIdentifier;
		}
		

		sqsClient = new AmazonSQSClient(new BasicAWSCredentials(awsAccessKey, awsSecretKey));

		
		connectToQueue(isConsumer, 1000);
		
		if (isConsumer) {
			this.consumerThread = new Thread(this,"TOCQueue["+myId+"] msg consumer thread");
		}
		
		logger.info("\n-------------------------------------------\n" +
					"TOC Queue["+myId+"]: ALL SQS resources hooked up OK: "+this.tocQueueUrl+"\n" +
					"-------------------------------------------\n");
	}
	
	public Long getLastMsgReceivedTimeMS() {
		return this.lastSQSMessageReceivedMS;
	}
	
	public void start() {
		this.consumerThread.start();
		this.paused = false;
	}
	
	public void pauseConsuming() {
		this.paused = true;
	}
	
	public void resumeConsuming() {
		this.lastSQSMessageReceivedMS = System.currentTimeMillis(); // set to now.
		this.totalMessageRequestsMade = 0; // reset to zero
		this.paused = false;
	}
	
	public boolean isPaused() {
		return this.paused;
	}
	
	/**
	 * Note here we attempt to the TOCQueue which may take some time to be shown as available
	 * @param isConsumer
	 * @param maxAttempts
	 * @throws Exception
	 */
	public void connectToQueue(boolean isConsumer, int maxAttempts) throws Exception{
		
		for (int i=0; i<maxAttempts; i++) {
			
			logger.debug("connectToQueue() attempt: " + (i+1));
			
			ListQueuesResult queuesResult = sqsClient.listQueues();
			if (queuesResult != null) {
				for (String queueUrl : queuesResult.getQueueUrls()) {
					if (queueUrl.indexOf(sqsQueueName) != -1) {
						tocQueueUrl = queueUrl;
						break;
					}
				}
			}
			
			// if consumer, retry, otherwise is master, so just exit quick to create...
			if (tocQueueUrl == null && isConsumer) {
				Thread.currentThread().sleep(1000);
				continue;
			} else {
				break; // exit;
			}
		}
		
		if (tocQueueUrl == null && !isConsumer) {
			CreateQueueResult createQueueResult = sqsClient.createQueue(sqsQueueName);
			this.tocQueueUrl = createQueueResult.getQueueUrl();
			
		} else if (tocQueueUrl == null) {
			throw new Exception("TOCQueue() isConsumer:"+ isConsumer+ " cannot start, sqsQueueName has yet to be created by master?: " + sqsQueueName);
		}
	}
	
	public void send(TocInfo fileInfo, MODE mode) throws Exception {
		TOCPayload payload = new TOCPayload();
		payload.tocInfo = fileInfo;
		payload.mode = mode;
		
		// send!
		this.sqsClient.sendMessage(this.tocQueueUrl, gson.toJson(payload));
	}
	
	public int emptyTOCQueue() {
		try {
			logger.trace("emptyTOCQueue() attempting to purge all messages from the TOCQueue!");
			
			int totalRemoved = 0;
			
			// to empty it we will just consume all the messages
			ReceiveMessageRequest req = new ReceiveMessageRequest();
			req.setQueueUrl(this.tocQueueUrl);
			req.setVisibilityTimeout(900); // 15 minutes it will be invisible to other consumers
			req.setMaxNumberOfMessages(10);
			
			ReceiveMessageResult msgResult = sqsClient.receiveMessage(req);
			List<Message> messages = msgResult.getMessages();
			
			DeleteMessageBatchRequest deleteRequest = new DeleteMessageBatchRequest();
			deleteRequest.setQueueUrl(tocQueueUrl);
			Collection<DeleteMessageBatchRequestEntry> entries = new ArrayList<DeleteMessageBatchRequestEntry>();
			for (Message msg : messages) {
				entries.add(new DeleteMessageBatchRequestEntry()
									.withId(msg.getReceiptHandle())
									.withReceiptHandle(msg.getReceiptHandle()));
			}

			if (entries.size() > 0) {
				// delete batch
				deleteRequest.setEntries(entries);
				sqsClient.deleteMessageBatch(deleteRequest);
			}
			
			// if ok inc, total removed
			totalRemoved += entries.size();
			
			logger.trace("emptyTOCQueue() purging completed!");
			return totalRemoved;
			
		} catch(AmazonClientException e) {
			logger.error("Error in emptyTOCQueue() (aws error): " + e.getMessage());
			
		} catch(Exception e) {
			logger.error("Error in emptyTOCQueue(): " + e.getMessage(),e);
		}
		
		return 0;
	}
	
	public void run() {
		
		Random rand = new Random();
		
		while(this.running) {
			
			if (!this.paused) {
				try {
					this.totalMessageRequestsMade++;
					
					ReceiveMessageRequest req = new ReceiveMessageRequest();
					req.setWaitTimeSeconds(10);
					req.setQueueUrl(this.tocQueueUrl);
					
					// 30 minutes it will be invisible to other consumers
					// this should be enought time for the tocPayloadHandler to
					// complete and then we delete the message
					req.setVisibilityTimeout(600*3); 
					req.setMaxNumberOfMessages(1); // only one at a time..
					
					ReceiveMessageResult msgResult = sqsClient.receiveMessage(req);
					List<Message> messages = msgResult.getMessages();
	
					for (Message msg : messages) {
						
						this.currentlyProcessingMessage = true;
						this.lastSQSMessageReceivedMS = System.currentTimeMillis();
						this.totalMessagesProcessed++;
						
						logger.debug("TOCQueue["+myId+"] Received SQS Message " +
								"body (json -> TOCPayload) body= " + msg.getBody());
						TOCPayload payload = null;
	
						try {
							payload = gson.fromJson(msg.getBody(), TOCPayload.class);
						
						} catch(Exception e) {
							logger.error("TOCQueue["+myId+"] ERROR: unexpected error converting SQS Message " +
									"body (json -> TOCPayload) body= " + msg.getBody()+ " error="+e.getMessage());
							
							// delete the message we just analyzed
							sqsClient.deleteMessage(tocQueueUrl, msg.getReceiptHandle());
							
							continue;
						}
						
						logger.debug("TOCQueue["+myId+"] TOCPayload received: filePath:" + payload.tocInfo.getPath());
	
						// handle
						this.tocPayloadHandler.handlePayload(payload);
						
						// delete the message, got here no exception
						sqsClient.deleteMessage(tocQueueUrl, msg.getReceiptHandle());
						
						// set to false, we are done processing message
						this.currentlyProcessingMessage = false;
					
					}
	
				} catch(Exception e) {
					logger.error("TOCQueue["+myId+"] run() unexpected error in handling TOCPayload: " + e.getMessage(),e);
					
					// set to false, we are done processing message
					this.currentlyProcessingMessage = false;
				}
			}
			
			try {
				Thread.currentThread().sleep(rand.nextInt(1000));
			} catch(Exception ignore) {}
			
		}
	}

	public void stopConsuming() {
		this.running = false;
	}
	
	public void destroy() throws Exception {

		Thread.currentThread().sleep(30000);
		
		try {
			if (canDestroyQueue) {
				logger.debug("TOCQueue["+myId+"] destroy() " + this.tocQueueUrl);
				this.sqsClient.deleteQueue(this.tocQueueUrl);
			}
		} catch(Exception e) {
			logger.error("TOCQueue["+myId+"] destroy() error deleting TOCQueue: " + e.getMessage(),e);
		}
	}
	
	public AmazonSQSClient getSqsClient() {
		return sqsClient;
	}
	public void setSqsClient(AmazonSQSClient sqsClient) {
		this.sqsClient = sqsClient;
	}
	public String getTocQueueUrl() {
		return tocQueueUrl;
	}
	public void setTocQueueUrl(String tocQueueUrl) {
		this.tocQueueUrl = tocQueueUrl;
	}
	
	public void sendMessage(String messageBody) {
		this.sqsClient.sendMessage(this.tocQueueUrl, messageBody);
	}
	
	private static String determineHostName() throws Exception {

		InetAddress addr = InetAddress.getLocalHost();
		
		// Get IP Address
		byte[] ipAddr = addr.getAddress();
		// Get sourceHost
		String tmpHost = addr.getHostName();

		// we only care about the HOST portion, strip everything else
		// as some boxes report a fully qualified sourceHost such as
		// host.domainname.com

		int firstDot = tmpHost.indexOf('.');
		if (firstDot != -1) {
			tmpHost = tmpHost.substring(0,firstDot);
		}
		return tmpHost;

	}

	public int getTotalMessagesProcessed() {
		return totalMessagesProcessed;
	}

	public boolean isCurrentlyProcessingMessage() {
		return currentlyProcessingMessage;
	}

	public int getTotalMessageRequestsMade() {
		return this.totalMessageRequestsMade;
	}

} public class TocInfo {

	public String path = null;
	public boolean isDirectory = false;
	public long size = 0;
	
	public String getPath() {
		return path;
	}
	public void setPath(String path) {
		this.path = path;
	}
	public long getSize() {
		return size;
	}
	public void setSize(long size) {
		this.size = size;
	}
	
	public TocInfo(String filePath, long size) {
		super();
		this.path = filePath;
		this.size = size;
	}
	
	public boolean isDirectory() {
		return isDirectory;
	}
	
	public void setIsDirectory(boolean isDirectory) {
		this.isDirectory = isDirectory;
	}

} public class CompressUtil {

	public static char[] decompressAndB64DecodeUTF8Bytes(byte[] b64EncodedCompressedBytes) throws Exception {

		byte[] input = Base64.decode(b64EncodedCompressedBytes);
		
		// Compressor with highest level of compression
	    Inflater inflater = new Inflater();
	    
	    // Give the compressor the data to compress
	    inflater.setInput(input);
	    
	    ByteArrayOutputStream stream = new ByteArrayOutputStream();
	    byte[] buf = new byte[32];
	    while (!inflater.finished()) {
	        int count = inflater.inflate(buf);
	        stream.write(buf, 0, count);
	    }
	    return new String(stream.toByteArray(),"UTF-8").toCharArray();
	}
	
	public static String compressAndB64EncodeUTF8Bytes(byte[] bytes) throws Exception{
		
		byte[] input = bytes;
		
		// Compressor with highest level of compression
	    Deflater compressor = new Deflater();
	    compressor.setLevel(Deflater.BEST_COMPRESSION);
	    
	    // Give the compressor the data to compress
	    compressor.setInput(input);
	    compressor.finish();
	    
	    // Create an expandable byte array to hold the compressed data.
	    // It is not necessary that the compressed data will be smaller than
	    // the uncompressed data.
	    ByteArrayOutputStream bos = new ByteArrayOutputStream(input.length);
	    
	    // Compress the data
	    byte[] buf = new byte[32];
	    while (!compressor.finished()) {
	        int count = compressor.deflate(buf);
	        bos.write(buf, 0, count);
	    }
	    try {
	        bos.close();
	    } catch (IOException e) {
	    }
	    
	    // Get the compressed data
	    byte[] compressedData = bos.toByteArray();
	    
	    return new String(Base64.encode(compressedData),"UTF-8");
	}
}
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.Writer;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.commons.exec.CommandLine;
import org.apache.log4j.Logger;
import org.bitsofinfo.ec2.Ec2Util;
import org.bitsofinfo.s3.S3Util;
import org.bitsofinfo.s3.cmd.CmdResult;
import org.bitsofinfo.s3.cmd.CommandExecutor;
import org.bitsofinfo.s3.control.CCMode;
import org.bitsofinfo.s3.control.CCPayload;
import org.bitsofinfo.s3.control.CCPayloadHandler;
import org.bitsofinfo.s3.control.CCPayloadType;
import org.bitsofinfo.s3.control.ControlChannel;
import org.bitsofinfo.s3.toc.DirectoryCrawler;
import org.bitsofinfo.s3.toc.S3BucketObjectLister;
import org.bitsofinfo.s3.toc.SourceTOCGenerator;
import org.bitsofinfo.s3.toc.TOCManifestBasedGenerator;
import org.bitsofinfo.s3.toc.TOCPayload.MODE;
import org.bitsofinfo.s3.toc.TOCQueue;
import org.bitsofinfo.s3.toc.TocInfo;
import org.bitsofinfo.s3.util.CompressUtil;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.ec2.AmazonEC2Client;
import com.amazonaws.services.ec2.model.CreateTagsRequest;
import com.amazonaws.services.ec2.model.Instance;
import com.amazonaws.services.ec2.model.InstanceStatus;
import com.amazonaws.services.ec2.model.Tag;
import com.amazonaws.services.s3.AmazonS3Client;
import com.google.gson.Gson;
			controlChannel = new ControlChannel(true, awsAccessKey, awsSecretKey, snsControlTopicName, userAccountPrincipalId, userARN, this);
			
			totalExpectedWorkers = Integer.valueOf(props.getProperty("master.workers.total"));
			
			workersEc2Managed = Boolean.valueOf(props.getProperty("master.workers.ec2.managed"));

		} catch(Exception e) {
			logger.error("Master() unexpected error: " + e.getMessage(),e);
			try {
				destroy();
			} catch(Exception ignore){}
		}
	}
	
	// called by TOCGeneratorAndSender when TOC generation is completed
	public void tocGenerationComplete(Collection<TocInfo> generatedTOC) {
		this.toc = generatedTOC;
	}
	
	public void tocGenerationError(String msg, Exception e) {
		logger.error("TOCGeneration threw an error, destroying myself....: " + msg, e);
		try {
			this.destroy();
		} catch(Exception e2) {
			// ignore;
		}
	}
	
	private void spawnEC2() throws Exception {
		
		if (!workersEc2Managed) {
			return;
		}

		this.ec2Instances = ec2util.launchEc2Instances(ec2Client, props);
		
		// collect some info
		StringBuffer sb = new StringBuffer("EC2 instance request created, reservation details:\n");
		for (Instance instance : ec2Instances) {
			sb.append("\tid:" + instance.getInstanceId() + "\tip:" + instance.getPrivateIpAddress() + "\thostname:"+instance.getPrivateDnsName()+"\n");
		}
		logger.info(sb.toString()+"\n");
		
		Thread.currentThread().sleep(60000);
		
		// tag instances
		CreateTagsRequest tagRequest = new CreateTagsRequest();
		String dateStr = ec2TagSimpleDateFormat.format(new Date());
		
		for (Instance instance : this.ec2Instances) {
			List<String> amiIds = new ArrayList<String>();
			List<Tag> tags = new ArrayList<Tag>();
			
			amiIds.add(instance.getInstanceId());
			tags.add(new Tag("Name","s3BktLdr-wkr-"+dateStr));
			
			tagRequest.setResources(amiIds);
			tagRequest.setTags(tags);
			
			ec2Client.createTags(tagRequest);
		}

		

	}
	
	
	public void start() throws Exception {

		
		// seed start
		this.masterStartAt = new Date();
		
		/**
		 * START the source for our TOC... if specified
		 */
		if (this.sourceEc2StartStopInstanceId != null) {
			
			// already running?
			InstanceStatus alreadyRunning = ec2util.getInstanceStatus(this.ec2Client, this.sourceEc2StartStopInstanceId);
			if (alreadyRunning == null || !alreadyRunning.getInstanceState().getName().equalsIgnoreCase("running")) {
				this.ec2util.startInstance(this.ec2Client, this.sourceEc2StartStopInstanceId);
			}
			
			// check status
			boolean sourceEc2InstanceStarted = false;
			while(!sourceEc2InstanceStarted) {
				InstanceStatus status = ec2util.getInstanceStatus(this.ec2Client, this.sourceEc2StartStopInstanceId);
				if (status == null || !status.getInstanceState().getName().equalsIgnoreCase("running")) {
					sourceEc2InstanceStarted = false;
					logger.debug("Still waiting for source host instance ["+this.sourceEc2StartStopInstanceId+"] to be running....");
					Thread.currentThread().sleep(2000);
				} else {
					logger.debug("Up and RUNNING! source host instance ["+this.sourceEc2StartStopInstanceId+"]");
					sourceEc2InstanceStarted = true;
					break;
				}
			}

			
			if (this.sourceEc2PostStartCmd != null) {
				
				Thread.currentThread().sleep(30000);
				
				// run the 'master.source.host.post.start.cmd' command
				CmdResult postStartResult = new CommandExecutor().execute(CommandLine.parse(this.sourceEc2PostStartCmd), 3);
				if (postStartResult.getExitCode() > 0) {
					throw new Exception("Source host post start cmd failed: " + this.sourceEc2PostStartCmd);
				}
			}
			
		}

		
		if (props.getProperty("tocGenerator.source.dir") != null && !new File(props.getProperty("tocGenerator.source.dir")).exists()) {
			throw new Exception("'tocGenerator.source.dir' does not exist! " + props.getProperty("tocGenerator.source.dir"));
		}
		
		// initialize workers 
		this.currentMode = CCMode.INITIALIZED;
		this.controlChannel.send(true, CCPayloadType.MASTER_CURRENT_MODE, this.currentMode);