org.opennms.netmgt.rrd

Class QueuingRrdStrategy

java.lang.Object

org.opennms.netmgt.rrd.QueuingRrdStrategy

All Implemented Interfaces:

Runnable, RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

public class QueuingRrdStrategy
extends Object
implements RrdStrategy<QueuingRrdStrategy.CreateOperation,String>, Runnable

Provides queuing implementation of RrdStrategy. In order to provide a more scalable collector. We created a queuing RrdStrategy that enabled the system to amortize the high cost of opening an round robin database across multiple updates. This RrdStrategy implementation enqueues the create and update operations on a per file basis and maintains a set of threads that process enqueued work file by file. If the I/O system can keep up with the collection threads while performing only a single update per file then eventually all the data is processed and the threads sleep until there is more work to do. If the I/O system is initially slower than than the collection threads then work will enqueue here and the write threads will get behind. As this happens each file will eventually have more than a single update enqueued and therefore the number of updates pushed thru the system will increase because more then one will be output per 'open' Eventually, the I/O system and the collection system will balance out. When this happens all data will be collected but will not be output to the rrd files until the next time the file is processed by the write threads. As another performance improving strategy. The queue distinguishes between files with significant vs insignificant updates. Files with only insignificant updates are put at the lowest priority and are only written when the highest priority updates have been written This implementation delegates all the actual writing to another RrdStrategy implementation. System properties effecting the operation: org.opennms.rrd.queuing.writethreads: (default 2) The number of rrd write threads that process the queue org.opennms.rrd.queuing.queueCreates: (default false) indicates whether rrd file creates should be queued or processed synchronously org.opennms.rrd.queuing.maxInsigUpdateSeconds: (default 0) the number of seconds over which all files with significant updates only should be promoted onto the significant less. This is to ensure they don't stay unprocessed forever. Zero means not promotion. org.opennms.rrd.queuing.modulus: (default 10000) the number of updates the get enqueued between statistics output org.opennms.rrd.queuing.category: (default "queued") the log routing prefix to place the statistics output in TODO: Promote files when ZeroUpdate operations can't be merged. This may be a collection miss which we want to push thru. It should also help with memory. TODO: Set an upper bound on enqueued operations TODO: Provide an event that will write data for a particular file... Say right before we try to graph it.

Version:

$Id: $

Author:

ranger

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
class	QueuingRrdStrategy.CreateOperation This class represents an operation to create an rrd file
class	QueuingRrdStrategy.UpdateOperation Represents an update to a rrd file.
class	QueuingRrdStrategy.ZeroUpdateOperation Represents an update whose value is 0.

Constructor Summary

Constructors

Constructor and Description
QueuingRrdStrategy(RrdStrategy<Object,Object> delegate) Constructor for QueuingRrdStrategy.

Method Summary

Methods

Modifier and Type	Method and Description
void	closeFile(String rrd) closeFile
QueuingRrdStrategy.CreateOperation	createDefinition(String creator, String directory, String rrdName, int step, List<RrdDataSource> dataSources, List<String> rraList) Create a round robin database definition from the supplied parameters.
void	createFile(QueuingRrdStrategy.CreateOperation op, Map<String,String> attributeMappings) createFile
InputStream	createGraph(String command, File workDir) Creates an InputStream representing the bytes of a graph created from round robin data.
RrdGraphDetails	createGraphReturnDetails(String command, File workDir) Creates an RrdGraphDetails object representing the graph created from round robin data.
Double	fetchLastValue(String rrdFile, String ds, int interval) Fetches the last value from the round robin database with the given name.
Double	fetchLastValue(String rrdFile, String ds, String consolidationFunction, int interval) Fetches the last value from the round robin database with the given name.
Double	fetchLastValueInRange(String rrdFile, String ds, int interval, int range) Fetches the last value from the round robin database with the given name within a time range.
String	getCategory() getCategory
Properties	getConfigurationProperties() getConfigurationProperties
long	getCreatesCompleted() getCreatesCompleted
String	getDefaultFileExtension() getDefaultFileExtension
long	getDequeuedItems() getDequeuedItems
long	getDequeuedOperations() getDequeuedOperations
long	getEnqueuedOperations() getEnqueuedOperations
long	getErrors() getErrors
int	getGraphLeftOffset() getGraphLeftOffset
int	getGraphRightOffset() getGraphRightOffset
int	getGraphTopOffsetWithText() getGraphTopOffsetWithText
long	getInSigHighWaterMark() getInSigHighWaterMark
long	getMaxInsigUpdateSeconds() getMaxInsigUpdateSeconds
long	getModulus() getModulus
long	getPromotionCount() getPromotionCount
long	getQueueHighWaterMark() getQueueHighWaterMark
long	getSigHighWaterMark() getSigHighWaterMark
long	getSignificantOpsCompleted() getSignificantOpsCompleted
long	getSignificantOpsDequeued() getSignificantOpsDequeued
long	getSignificantOpsEnqueued() getSignificantOpsEnqueued
long	getStartTime() getStartTime
String	getStats() Print queue statistics.
long	getTotalOperationsPending() getTotalOperationsPending
long	getUpdatesCompleted() getUpdatesCompleted
long	getWriteThreadExitDelay() getWriteThreadExitDelay
int	getWriteThreads() getWriteThreads
long	getWriteThreadSleepTime() getWriteThreadSleepTime
String	openFile(String fileName) Opens the round robin database with the supplied name.
boolean	prioritizeSignificantUpdates() prioritizeSignificantUpdates
void	promoteEnqueuedFiles(Collection<String> rrdFiles) In the event that this is a queuing implementation of the RrdStrategy.
boolean	queueCreates() queueCreates
void	run() run
void	setCategory(String category) setCategory
void	setConfigurationProperties(Properties configurationParameters) setConfigurationProperties
void	setCreatesCompleted(long createsCompleted) setCreatesCompleted
void	setDequeuedItems(long dequeuedItems) setDequeuedItems
void	setDequeuedOperations(long dequeuedOperations) setDequeuedOperations
void	setEnqueuedOperations(long enqueuedOperations) setEnqueuedOperations
void	setErrors(long errors) setErrors
void	setInSigHighWaterMark(long inSigHighWaterMark) setInSigHighWaterMark
void	setMaxInsigUpdateSeconds(long maxInsigUpdateSeconds) setMaxInsigUpdateSeconds
void	setModulus(long modulus) setModulus
void	setPrioritizeSignificantUpdates(boolean prioritizeSignificantUpdates) setPrioritizeSignificantUpdates
void	setPromotionCount(long promotionCount) setPromotionCount
void	setQueueCreates(boolean queueCreates) setQueueCreates
void	setQueueHighWaterMark(long queueHighWaterMark) setQueueHighWaterMark
void	setSigHighWaterMark(long sigHighWaterMark) setSigHighWaterMark
void	setSignificantOpsCompleted(long significantOpsCompleted) setSignificantOpsCompleted
void	setSignificantOpsDequeued(long significantOpsDequeued) setSignificantOpsDequeued
void	setSignificantOpsEnqueued(long significantOpsEnqueued) setSignificantOpsEnqueued
void	setStartTime(long updateStart) setStartTime
void	setTotalOperationsPending(long totalOperationsPending) setTotalOperationsPending
void	setUpdatesCompleted(long updatesCompleted) setUpdatesCompleted
void	setWriteThreadExitDelay(long writeThreadExitDelay) setWriteThreadExitDelay
void	setWriteThreads(int writeThreads) setWriteThreads
void	setWriteThreadSleepTime(long writeThreadSleepTime) setWriteThreadSleepTime
void	updateFile(String rrdFile, String owner, String data) Updates the supplied round robin database with the given timestamp:value point

Methods inherited from class java.lang.Object

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

Constructor Detail

QueuingRrdStrategy

public QueuingRrdStrategy(RrdStrategy<Object,Object> delegate)

Constructor for QueuingRrdStrategy.

Parameters:

delegate - a RrdStrategy object.

Method Detail

getConfigurationProperties

public Properties getConfigurationProperties()

getConfigurationProperties

Returns:

a Properties object.

setConfigurationProperties

public void setConfigurationProperties(Properties configurationParameters)

setConfigurationProperties

Specified by:

setConfigurationProperties in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

configurationParameters - a Properties object.

getWriteThreads

public int getWriteThreads()

getWriteThreads

Returns:

a int.

setWriteThreads

public void setWriteThreads(int writeThreads)

setWriteThreads

Parameters:

writeThreads - a int.

queueCreates

public boolean queueCreates()

queueCreates

Returns:

a boolean.

setQueueCreates

public void setQueueCreates(boolean queueCreates)

setQueueCreates

Parameters:

queueCreates - a boolean.

prioritizeSignificantUpdates

public boolean prioritizeSignificantUpdates()

prioritizeSignificantUpdates

Returns:

a boolean.

setPrioritizeSignificantUpdates

public void setPrioritizeSignificantUpdates(boolean prioritizeSignificantUpdates)

setPrioritizeSignificantUpdates

Parameters:

prioritizeSignificantUpdates - a boolean.

getInSigHighWaterMark

public long getInSigHighWaterMark()

getInSigHighWaterMark

Returns:

a long.

setInSigHighWaterMark

public void setInSigHighWaterMark(long inSigHighWaterMark)

setInSigHighWaterMark

Parameters:

inSigHighWaterMark - a long.

getSigHighWaterMark

public long getSigHighWaterMark()

getSigHighWaterMark

Returns:

a long.

setSigHighWaterMark

public void setSigHighWaterMark(long sigHighWaterMark)

setSigHighWaterMark

Parameters:

sigHighWaterMark - a long.

getQueueHighWaterMark

public long getQueueHighWaterMark()

getQueueHighWaterMark

Returns:

a long.

setQueueHighWaterMark

public void setQueueHighWaterMark(long queueHighWaterMark)

setQueueHighWaterMark

Parameters:

queueHighWaterMark - a long.

getModulus

public long getModulus()

getModulus

Returns:

a long.

setModulus

public void setModulus(long modulus)

setModulus

Parameters:

modulus - a long.

getCategory

public String getCategory()

getCategory

Returns:

a String object.

setCategory

public void setCategory(String category)

setCategory

Parameters:

category - a String object.

getMaxInsigUpdateSeconds

public long getMaxInsigUpdateSeconds()

getMaxInsigUpdateSeconds

Returns:

a long.

setMaxInsigUpdateSeconds

public void setMaxInsigUpdateSeconds(long maxInsigUpdateSeconds)

setMaxInsigUpdateSeconds

Parameters:

maxInsigUpdateSeconds - a long.

getWriteThreadSleepTime

public long getWriteThreadSleepTime()

getWriteThreadSleepTime

Returns:

a long.

setWriteThreadSleepTime

public void setWriteThreadSleepTime(long writeThreadSleepTime)

setWriteThreadSleepTime

Parameters:

writeThreadSleepTime - a long.

getWriteThreadExitDelay

public long getWriteThreadExitDelay()

getWriteThreadExitDelay

Returns:

a long.

setWriteThreadExitDelay

public void setWriteThreadExitDelay(long writeThreadExitDelay)

setWriteThreadExitDelay

Parameters:

writeThreadExitDelay - a long.

promoteEnqueuedFiles

public void promoteEnqueuedFiles(Collection<String> rrdFiles)

In the event that this is a queuing implementation of the RrdStrategy. This method causes all queued but not yet written data to be to the rrd files as soon as possible.

Specified by:

promoteEnqueuedFiles in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

rrdFiles - a Collection object.

closeFile

public void closeFile(String rrd)
               throws Exception

closeFile

Specified by:

closeFile in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

rrd - a String object.

Throws:

Exception - if any.

createDefinition

public QueuingRrdStrategy.CreateOperation createDefinition(String creator,
                                                  String directory,
                                                  String rrdName,
                                                  int step,
                                                  List<RrdDataSource> dataSources,
                                                  List<String> rraList)
                                                    throws Exception

Create a round robin database definition from the supplied parameters. This definition is used in the createFile call to create the actual file.

Specified by:

createDefinition in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

creator - - A string representing who is creating this file for use in log msgs

directory - - The directory to create the file in

rrdName - - The name to use for the round robin database

step - - the step for the database

dataSources - - the data sources to use for round robin database

rraList - - a List of the round robin archives to create in the database. defines after which time the data is condensed to a defined lower step

Returns:

an object representing the definition of an round robin database. Can be null if the database exists already.

Throws:

Exception - If an error occurs while creating the definition

createFile

public void createFile(QueuingRrdStrategy.CreateOperation op,
              Map<String,String> attributeMappings)
                throws Exception

createFile

Specified by:

createFile in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

op - a QueuingRrdStrategy.Operation object.

attributeMappings - a Map that represents the mapping of attributeId to rrd track names. default there is only one track per datasource, but it is possible to store multiple tracks in one datasource

Throws:

Exception - if any.

openFile

public String openFile(String fileName)
                throws Exception

Opens the round robin database with the supplied name. It is assumed the name refers to a round robin database appropriate for this strategy implementation

Specified by:

openFile in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

fileName - the name of the associated rrd file

Returns:

an open rrd reference that can by used in calls to updateFile and closeFile

Throws:

Exception - if an error occurs opening the file

updateFile

public void updateFile(String rrdFile,
              String owner,
              String data)
                throws Exception

Updates the supplied round robin database with the given timestamp:value point

Specified by:

updateFile in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

rrdFile - an rrd object created using openFile

owner - the owner of the rrd

data - a string of the form :

Throws:

Exception - if an error occurs updating the file

fetchLastValue

public Double fetchLastValue(String rrdFile,
                    String ds,
                    int interval)
                      throws NumberFormatException,
                             RrdException

Fetches the last value from the round robin database with the given name. The interval passed in should be the interval associated with the round robin database.

Specified by:

fetchLastValue in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

rrdFile - a name the represents a round robin database

ds - a name the represents a data source to be used

interval - a step interval of the round robin database

Returns:

The last value as a Double (if the last value didn't exist returns a Double.NaN)

Throws:

NumberFormatException - if any.

RrdException - if any.

fetchLastValue

public Double fetchLastValue(String rrdFile,
                    String ds,
                    String consolidationFunction,
                    int interval)
                      throws NumberFormatException,
                             RrdException

Fetches the last value from the round robin database with the given name. The interval passed in should be the interval associated with the round robin database.

Specified by:

fetchLastValue in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

rrdFile - a name the represents a round robin database

ds - a name the represents a data source to be used

consolidationFunction - a String object.

interval - a step interval of the round robin database

Returns:

The last value as a Double (if the last value didn't exist returns a Double.NaN)

Throws:

NumberFormatException - if any.

RrdException - if any.

fetchLastValueInRange

public Double fetchLastValueInRange(String rrdFile,
                           String ds,
                           int interval,
                           int range)
                             throws NumberFormatException,
                                    RrdException

Fetches the last value from the round robin database with the given name within a time range. The interval passed in should be the interval associated with the round robin database. The range should be the amount of "lag" acceptable for an update to be considered valid. Range must be a multiple of the RRD interval.

Specified by:

fetchLastValueInRange in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

rrdFile - a name the represents a round robin database

ds - a name the represents a data source to be used

interval - a step interval of the round robin database

range - an acceptable range for which the last value will be returned

Returns:

The last value as a Double (if the last value didn't exist returns a Double.NaN)

Throws:

NumberFormatException - if any.

RrdException - if any.

createGraph

public InputStream createGraph(String command,
                      File workDir)
                        throws IOException,
                               RrdException

Creates an InputStream representing the bytes of a graph created from round robin data. It accepts an rrdtool graph command. The underlying implementation converts this command to a format appropriate for it .

Specified by:

createGraph in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

command - the command needed to create the graph

workDir - the directory that all referenced files are relative to

Returns:

an input stream representing the bytes of a graph image as a PNG file

Throws:

IOException - if an IOError occurs

RrdException - if an RRD error occurs

run

public void run()

run

Specified by:

run in interface Runnable

getStats

public String getStats()

Print queue statistics.

Specified by:

getStats in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Returns:

a String object.

getGraphLeftOffset

public int getGraphLeftOffset()

getGraphLeftOffset

Specified by:

getGraphLeftOffset in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Returns:

a int.

getGraphRightOffset

public int getGraphRightOffset()

getGraphRightOffset

Specified by:

getGraphRightOffset in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Returns:

a int.

getGraphTopOffsetWithText

public int getGraphTopOffsetWithText()

getGraphTopOffsetWithText

Specified by:

getGraphTopOffsetWithText in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Returns:

a int.

getDefaultFileExtension

public String getDefaultFileExtension()

getDefaultFileExtension

Specified by:

getDefaultFileExtension in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Returns:

a String object.

createGraphReturnDetails

public RrdGraphDetails createGraphReturnDetails(String command,
                                       File workDir)
                                         throws IOException,
                                                RrdException

Creates an RrdGraphDetails object representing the graph created from round robin data. It accepts an rrdtool graph command. The underlying implementation converts this command to a format appropriate for it .

Specified by:

createGraphReturnDetails in interface RrdStrategy<QueuingRrdStrategy.CreateOperation,String>

Parameters:

command - the command needed to create the graph

workDir - the directory that all referenced files are relative to

Returns:

details for the graph including an InputStream, any PRINTed lines, and graph dimensions.

Throws:

IOException - if an IOError occurs

RrdException - if an RRD error occurs

getTotalOperationsPending

public long getTotalOperationsPending()

getTotalOperationsPending

Returns:

a long.

setTotalOperationsPending

public void setTotalOperationsPending(long totalOperationsPending)

setTotalOperationsPending

Parameters:

totalOperationsPending - a long.

getCreatesCompleted

public long getCreatesCompleted()

getCreatesCompleted

Returns:

a long.

setCreatesCompleted

public void setCreatesCompleted(long createsCompleted)

setCreatesCompleted

Parameters:

createsCompleted - a long.

getUpdatesCompleted

public long getUpdatesCompleted()

getUpdatesCompleted

Returns:

a long.

setUpdatesCompleted

public void setUpdatesCompleted(long updatesCompleted)

setUpdatesCompleted

Parameters:

updatesCompleted - a long.

getErrors

public long getErrors()

getErrors

Returns:

a long.

setErrors

public void setErrors(long errors)

setErrors

Parameters:

errors - a long.

getPromotionCount

public long getPromotionCount()

getPromotionCount

Returns:

a long.

setPromotionCount

public void setPromotionCount(long promotionCount)

setPromotionCount

Parameters:

promotionCount - a long.

getSignificantOpsEnqueued

public long getSignificantOpsEnqueued()

getSignificantOpsEnqueued

Returns:

a long.

setSignificantOpsEnqueued

public void setSignificantOpsEnqueued(long significantOpsEnqueued)

setSignificantOpsEnqueued

Parameters:

significantOpsEnqueued - a long.

getSignificantOpsDequeued

public long getSignificantOpsDequeued()

getSignificantOpsDequeued

Returns:

a long.

setSignificantOpsDequeued

public void setSignificantOpsDequeued(long significantOpsDequeued)

setSignificantOpsDequeued

Parameters:

significantOpsDequeued - a long.

getEnqueuedOperations

public long getEnqueuedOperations()

getEnqueuedOperations

Returns:

a long.

setEnqueuedOperations

public void setEnqueuedOperations(long enqueuedOperations)

setEnqueuedOperations

Parameters:

enqueuedOperations - a long.

getDequeuedOperations

public long getDequeuedOperations()

getDequeuedOperations

Returns:

a long.

setDequeuedOperations

public void setDequeuedOperations(long dequeuedOperations)

setDequeuedOperations

Parameters:

dequeuedOperations - a long.

getDequeuedItems

public long getDequeuedItems()

getDequeuedItems

Returns:

a long.

setDequeuedItems

public void setDequeuedItems(long dequeuedItems)

setDequeuedItems

Parameters:

dequeuedItems - a long.

getSignificantOpsCompleted

public long getSignificantOpsCompleted()

getSignificantOpsCompleted

Returns:

a long.

setSignificantOpsCompleted

public void setSignificantOpsCompleted(long significantOpsCompleted)

setSignificantOpsCompleted

Parameters:

significantOpsCompleted - a long.

getStartTime

public long getStartTime()

getStartTime

Returns:

a long.

setStartTime

public void setStartTime(long updateStart)

setStartTime

Parameters:

updateStart - a long.