Add stats collector

main
Ankit Gupta 11 years ago
parent 403636842a
commit 143ee5a1c1
  1. 86
      java/org/rocksdb/StatisticsCollector.java
  2. 27
      java/org/rocksdb/StatisticsCollectorCallback.java
  3. 39
      java/org/rocksdb/test/StatisticsCollectorTest.java
  4. 22
      java/org/rocksdb/test/StatsCallbackMock.java

@ -0,0 +1,86 @@
// Copyright (c) 2014, Facebook, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.
package org.rocksdb;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Helper class to collect DB statistics periodically at a period specified in
* constructor. Callback function (provided in constructor) is called with
* every statistics collection.
*
* Caller should call start() to start statistics collection. Shutdown() should
* be called to stop stats collection and should be called before statistics (
* provided in constructor) reference has been disposed.
*/
public class StatisticsCollector {
private final Statistics _statistics;
private final ThreadPoolExecutor _threadPoolExecutor;
private final int _statsCollectionInterval;
private final StatisticsCollectorCallback _statsCallback;
private volatile boolean _isRunning = true;
public StatisticsCollector(Statistics statistics,
int statsCollectionIntervalInMilliSeconds,
StatisticsCollectorCallback statsCallback) {
_statistics = statistics;
_statsCollectionInterval = statsCollectionIntervalInMilliSeconds;
_statsCallback = statsCallback;
_threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(1));
}
public void start() {
_threadPoolExecutor.submit(collectStatistics());
}
public void shutDown() throws InterruptedException {
_isRunning = false;
_threadPoolExecutor.shutdown();
// Wait for collectStatistics runnable to finish so that disposal of
// statistics does not cause any exceptions to be thrown.
_threadPoolExecutor.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
}
private Runnable collectStatistics() {
return new Runnable() {
@Override
public void run() {
while (_isRunning) {
try {
// Collect ticker data
for(TickerType ticker : TickerType.values()) {
long tickerValue = _statistics.getTickerCount(ticker);
_statsCallback.tickerCallback(ticker, tickerValue);
}
// Collect histogram data
for(HistogramType histogramType : HistogramType.values()) {
HistogramData histogramData =
_statistics.geHistogramData(histogramType);
_statsCallback.histogramCallback(histogramType, histogramData);
}
Thread.sleep(_statsCollectionInterval);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Thread got interrupted!", e);
}
catch (Exception e) {
throw new RuntimeException("Error while calculating statistics", e);
}
}
}
};
}
}

@ -0,0 +1,27 @@
// Copyright (c) 2014, Facebook, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.
package org.rocksdb;
/**
* Callback interface provided to StatisticsCollector.
* @param tickerType
* @param tickerCount
*/
public interface StatisticsCollectorCallback {
/**
* Callback function to get ticker values.
* @param tickerType Ticker type.
* @param tickerCount Value of ticker type.
*/
void tickerCallback(TickerType tickerType, long tickerCount);
/**
* Callback function to get histogram values.
* @param histType Histogram type.
* @param histData Histogram data.
*/
void histogramCallback(HistogramType histType, HistogramData histData);
}

@ -0,0 +1,39 @@
// Copyright (c) 2014, Facebook, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.
package org.rocksdb.test;
import org.rocksdb.*;
public class StatisticsCollectorTest {
static final String db_path = "/tmp/backupablejni_db";
static {
RocksDB.loadLibrary();
}
public static void main(String[] args) throws InterruptedException, RocksDBException {
Options opt = new Options().createStatistics().setCreateIfMissing(true);
Statistics stats = opt.statisticsPtr();
RocksDB db = RocksDB.open(db_path);
StatsCallbackMock callback = new StatsCallbackMock();
StatisticsCollector statsCollector = new StatisticsCollector(stats, 100,
callback);
statsCollector.start();
Thread.sleep(1000);
assert(callback.tickerCallbackCount > 0);
assert(callback.histCallbackCount > 0);
statsCollector.shutDown();
db.close();
opt.dispose();
System.out.println("Stats collector test passed.!");
}
}

@ -0,0 +1,22 @@
// Copyright (c) 2014, Facebook, Inc. All rights reserved.
// This source code is licensed under the BSD-style license found in the
// LICENSE file in the root directory of this source tree. An additional grant
// of patent rights can be found in the PATENTS file in the same directory.
package org.rocksdb.test;
import org.rocksdb.*;
public class StatsCallbackMock implements StatisticsCollectorCallback {
public int tickerCallbackCount = 0;
public int histCallbackCount = 0;
public void tickerCallback(TickerType tickerType, long tickerCount) {
tickerCallbackCount++;
}
public void histogramCallback(HistogramType histType,
HistogramData histData) {
histCallbackCount++;
}
}
Loading…
Cancel
Save