org.apache.hadoop.chukwa.extraction.demux

Class DemuxManager

java.lang.Object

org.apache.hadoop.chukwa.extraction.demux.DemuxManager

All Implemented Interfaces:

CHUKWA_CONSTANT

public class DemuxManager
extends Object
implements CHUKWA_CONSTANT

Field Summary

Fields

Modifier and Type	Field and Description
protected ChukwaConfiguration	conf
protected SimpleDateFormat	dayTextFormat
protected int	DEFAULT_MAX_ERROR_COUNT
protected int	DEFAULT_MAX_FILES_PER_DEMUX
protected int	DEFAULT_REDUCER_COUNT
protected int	demuxReducerCount
protected int	ERROR_SLEEP_TIME
protected org.apache.hadoop.fs.FileSystem	fs
protected boolean	isRunning
protected int	maxPermittedErrorCount
protected int	NO_DATASINK_SLEEP_TIME
protected int	reprocess
protected boolean	sendAlert

Fields inherited from interface org.apache.hadoop.chukwa.extraction.CHUKWA_CONSTANT

ARCHIVES_IN_ERROR_DIR_NAME, ARCHIVES_MR_INPUT_DIR_NAME, ARCHIVES_MR_OUTPUT_DIR_NAME, ARCHIVES_PROCESSING_DIR_NAME, CHUKWA_ARCHIVE_DIR_FIELD, CHUKWA_ARCHIVE_MAX_ERROR_COUNT_FIELD, CHUKWA_DATA_SINK_DIR_FIELD, CHUKWA_DEMUX_MAX_ERROR_COUNT_FIELD, CHUKWA_DEMUX_REDUCER_COUNT_FIELD, CHUKWA_NAGIOS_HOST_FIELD, CHUKWA_NAGIOS_PORT_FIELD, CHUKWA_POST_PROCESS_DIR_FIELD, CHUKWA_POSTPROCESS_IN_ERROR_DIR_FIELD, CHUKWA_POSTPROCESS_MAX_ERROR_COUNT_FIELD, CHUKWA_REPORTING_HOST_FIELD, CHUKWA_ROOT_DIR_FIELD, CHUKWA_ROOT_REPOS_DIR_FIELD, DEFAULT_CHUKWA_DATASINK_DIR_NAME, DEFAULT_CHUKWA_LOGS_DIR_NAME, DEFAULT_CHUKWA_POSTPROCESS_DIR_NAME, DEFAULT_CHUKWA_ROOT_DIR_NAME, DEFAULT_DEMUX_IN_ERROR_DIR_NAME, DEFAULT_DEMUX_MR_INPUT_DIR_NAME, DEFAULT_DEMUX_MR_OUTPUT_DIR_NAME, DEFAULT_DEMUX_PROCESSING_DIR_NAME, DEFAULT_FINAL_ARCHIVES, DEFAULT_POSTPROCESS_IN_ERROR_DIR_NAME, DEFAULT_REPOS_DIR_NAME, HDFS_DEFAULT_NAME_FIELD, HIERARCHY_CONNECTOR, INCLUDE_KEY_IN_PARTITIONER, POST_DEMUX_DATA_LOADER, POST_DEMUX_SUCCESS_ACTION, WRITER_HDFS_FILESYSTEM_FIELD

Constructor Summary

Constructors

Constructor and Description
DemuxManager()
DemuxManager(ChukwaConfiguration conf)

Method Summary

Modifier and Type	Method and Description
protected boolean	checkDemuxInputDir(String demuxInputDir) Test if demuxInputDir exists
protected boolean	checkDemuxOutputDir(String demuxOutputDir) Test if demuxOutputDir exists
protected boolean	deleteDemuxOutputDir(String demuxOutputDir) Delete DemuxOutput directory
protected boolean	dirExists(String directory) Check if source exists and if source is a directory
int	getReprocess()
protected void	init()
static void	main(String[] args)
protected boolean	moveDataSinkFilesToArchiveDirectory(String demuxInputDir, String archiveDirectory) Move sourceFolder inside destFolder
protected boolean	moveDataSinkFilesToDemuxErrorDirectory(String dataSinkDir, String demuxErrorDir) Move sourceFolder inside destFolder
protected boolean	moveDataSinkFilesToDemuxInputDirectory(String dataSinkDir, String demuxInputDir) Move dataSink files to Demux input directory
protected boolean	moveDemuxOutputDirToPostProcessDirectory(String demuxOutputDir, String postProcessDirectory) Move sourceFolder inside destFolder
protected boolean	moveFolder(String srcDir, String destDir, String prefix) Move sourceFolder inside destFolder
protected boolean	processData(String dataSinkDir, String demuxInputDir, String demuxOutputDir, String postProcessDir, String archiveDir) Process Data, i.e.
protected boolean	runDemux(String demuxInputDir, String demuxOutputDir) Submit and Run demux Job
protected void	sendDemuxStatusToNagios(String nagiosHost, int nagiosPort, String reportingHost, String demuxInErrorDir, boolean demuxStatus, String demuxException) Send NSCA status to Nagios
protected void	setup(org.apache.hadoop.fs.Path directory) Create directory if !exists
void	shutdown()
void	start() Start the Demux Manager daemon

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

ERROR_SLEEP_TIME

protected int ERROR_SLEEP_TIME

NO_DATASINK_SLEEP_TIME

protected int NO_DATASINK_SLEEP_TIME

DEFAULT_MAX_ERROR_COUNT

protected int DEFAULT_MAX_ERROR_COUNT

DEFAULT_MAX_FILES_PER_DEMUX

protected int DEFAULT_MAX_FILES_PER_DEMUX

DEFAULT_REDUCER_COUNT

protected int DEFAULT_REDUCER_COUNT

maxPermittedErrorCount

protected int maxPermittedErrorCount

demuxReducerCount

protected int demuxReducerCount

conf

protected ChukwaConfiguration conf

fs

protected org.apache.hadoop.fs.FileSystem fs

reprocess

protected int reprocess

sendAlert

protected boolean sendAlert

dayTextFormat

protected SimpleDateFormat dayTextFormat

isRunning

protected volatile boolean isRunning

Constructor Detail

DemuxManager

public DemuxManager()
             throws Exception

Throws:

Exception

DemuxManager

public DemuxManager(ChukwaConfiguration conf)
             throws Exception

Throws:

Exception

Method Detail

main

public static void main(String[] args)
                 throws Exception

Throws:

Exception

init

protected void init()
             throws IOException,
                    URISyntaxException

Throws:

IOException

URISyntaxException

shutdown

public void shutdown()

getReprocess

public int getReprocess()

start

public void start()
           throws Exception

Start the Demux Manager daemon

Throws:

Exception

sendDemuxStatusToNagios

protected void sendDemuxStatusToNagios(String nagiosHost,
                                       int nagiosPort,
                                       String reportingHost,
                                       String demuxInErrorDir,
                                       boolean demuxStatus,
                                       String demuxException)

Send NSCA status to Nagios

Parameters:

nagiosHost -

nagiosPort -

reportingHost -

demuxInErrorDir -

demuxStatus -

exception -

processData

protected boolean processData(String dataSinkDir,
                              String demuxInputDir,
                              String demuxOutputDir,
                              String postProcessDir,
                              String archiveDir)
                       throws IOException

Process Data, i.e. - run demux - move demux output to postProcessDir - move dataSink file to archiveDir

Parameters:

dataSinkDir -

demuxInputDir -

demuxOutputDir -

postProcessDir -

archiveDir -

Returns:

True iff succeed

Throws:

IOException

runDemux

protected boolean runDemux(String demuxInputDir,
                           String demuxOutputDir)

Submit and Run demux Job

Parameters:

demuxInputDir -

demuxOutputDir -

Returns:

true id Demux succeed

moveDataSinkFilesToDemuxInputDirectory

protected boolean moveDataSinkFilesToDemuxInputDirectory(String dataSinkDir,
                                                         String demuxInputDir)
                                                  throws IOException

Move dataSink files to Demux input directory

Parameters:

dataSinkDir -

demuxInputDir -

Returns:

true if there's any dataSink files ready to be processed

Throws:

IOException

moveDataSinkFilesToDemuxErrorDirectory

protected boolean moveDataSinkFilesToDemuxErrorDirectory(String dataSinkDir,
                                                         String demuxErrorDir)
                                                  throws IOException

Move sourceFolder inside destFolder

Parameters:

dataSinkDir - : ex chukwa/demux/inputDir

demuxErrorDir - : ex /chukwa/demux/inError

Returns:

true if able to move chukwa/demux/inputDir to /chukwa/demux/inError//demuxInputDirXXX

Throws:

IOException

moveDataSinkFilesToArchiveDirectory

protected boolean moveDataSinkFilesToArchiveDirectory(String demuxInputDir,
                                                      String archiveDirectory)
                                               throws IOException

Move sourceFolder inside destFolder

Parameters:

demuxInputDir - ex chukwa/demux/inputDir

archiveDirectory - ex /chukwa/archives

Returns:

true if able to move chukwa/demux/inputDir to /chukwa/archives/raw//dataSinkDirXXX

Throws:

IOException

moveDemuxOutputDirToPostProcessDirectory

protected boolean moveDemuxOutputDirToPostProcessDirectory(String demuxOutputDir,
                                                           String postProcessDirectory)
                                                    throws IOException

Move sourceFolder inside destFolder

Parameters:

demuxOutputDir - ex chukwa/demux/outputDir

postProcessDirectory - ex /chukwa/postProcess

Returns:

true if able to move chukwa/demux/outputDir to /chukwa/postProcess/demuxOutputDirXXX

Throws:

IOException

checkDemuxInputDir

protected boolean checkDemuxInputDir(String demuxInputDir)
                              throws IOException

Test if demuxInputDir exists

Parameters:

demuxInputDir -

Returns:

true if demuxInputDir exists

Throws:

IOException

checkDemuxOutputDir

protected boolean checkDemuxOutputDir(String demuxOutputDir)
                               throws IOException

Test if demuxOutputDir exists

Parameters:

demuxOutputDir -

Returns:

true if demuxOutputDir exists

Throws:

IOException

deleteDemuxOutputDir

protected boolean deleteDemuxOutputDir(String demuxOutputDir)
                                throws IOException

Delete DemuxOutput directory

Parameters:

demuxOutputDir -

Returns:

true if succeed

Throws:

IOException

setup

protected void setup(org.apache.hadoop.fs.Path directory)
              throws IOException

Create directory if !exists

Parameters:

directory -

Throws:

IOException

dirExists

protected boolean dirExists(String directory)
                     throws IOException

Check if source exists and if source is a directory

Parameters:

f - source file

Throws:

IOException

moveFolder

protected boolean moveFolder(String srcDir,
                             String destDir,
                             String prefix)
                      throws IOException

Move sourceFolder inside destFolder

Parameters:

srcDir -

destDir -

Returns:

Throws:

IOException