Removed the old blur metrics.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/e7e1a406 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/e7e1a406 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/e7e1a406 Branch: refs/heads/0.1.5 Commit: e7e1a40672ffaa989b19af264ec45da75a5f83bc Parents: 434f3a9 Author: Aaron McCurry <[email protected]> Authored: Wed Mar 27 17:12:48 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Wed Mar 27 17:12:48 2013 -0400 ---------------------------------------------------------------------- .../java/org/apache/blur/manager/IndexManager.java | 152 +++++++++------ .../indexserver/DistributedIndexServer.java | 18 +-- .../blur/thrift/ThriftBlurControllerServer.java | 9 +- .../apache/blur/thrift/ThriftBlurShardServer.java | 9 +- .../main/java/org/apache/blur/utils/BlurUtil.java | 62 ++++-- .../org/apache/blur/manager/IndexManagerTest.java | 3 - .../java/org/apache/blur/gui/HttpJettyServer.java | 7 +- .../java/org/apache/blur/gui/MetricsServlet.java | 51 ----- .../java/org/apache/blur/metrics/BlurMetrics.java | 125 ------------ 9 files changed, 149 insertions(+), 287 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java b/src/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java index f00c940..cb61aa5 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/IndexManager.java @@ -16,6 +16,8 @@ package org.apache.blur.manager; * See the License for the specific language governing permissions and * limitations under the License. */ +import static org.apache.blur.metrics.MetricsConstants.BLUR; +import static org.apache.blur.metrics.MetricsConstants.ORG_APACHE_BLUR; import static org.apache.blur.utils.BlurConstants.PRIME_DOC; import static org.apache.blur.utils.BlurConstants.RECORD_ID; import static org.apache.blur.utils.BlurConstants.ROW_ID; @@ -53,7 +55,6 @@ import org.apache.blur.manager.results.MergerBlurResultIterable; import org.apache.blur.manager.status.QueryStatus; import org.apache.blur.manager.status.QueryStatusManager; import org.apache.blur.manager.writer.BlurIndex; -import org.apache.blur.metrics.BlurMetrics; import org.apache.blur.metrics.QueryMetrics; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.thrift.BException; @@ -106,6 +107,10 @@ import org.apache.lucene.search.TopDocs; import org.apache.lucene.util.Bits; import org.apache.lucene.util.BytesRef; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.MetricName; + public class IndexManager { private static final String NOT_FOUND = "NOT_FOUND"; @@ -119,21 +124,32 @@ public class IndexManager { private boolean _closed; private BlurPartitioner<BytesWritable, Void> _blurPartitioner = new BlurPartitioner<BytesWritable, Void>(); private BlurFilterCache _filterCache = new DefaultBlurFilterCache(); - private BlurMetrics _blurMetrics; private QueryMetrics _queryMetrics; private long _defaultParallelCallTimeout = TimeUnit.MINUTES.toMillis(1); + private Meter _recordsMeter; + private Meter _rowMeter; + private Meter _queriesExternalMeter; + private Meter _queriesInternalMeter; public void setMaxClauseCount(int maxClauseCount) { BooleanQuery.setMaxClauseCount(maxClauseCount); } public void init() { + _recordsMeter = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, BLUR, "Records/s"), "Records/s", TimeUnit.SECONDS); + _rowMeter = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, BLUR, "Row/s"), "Row/s", TimeUnit.SECONDS); + _queriesExternalMeter = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, BLUR, "External Queries/s"), + "External Queries/s", TimeUnit.SECONDS); + _queriesInternalMeter = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, BLUR, "Internal Queries/s"), + "Internal Queries/s", TimeUnit.SECONDS); + _executor = Executors.newThreadPool("index-manager", _threadCount); // @TODO give the mutate it's own thread pool _mutateExecutor = Executors.newThreadPool("index-manager-mutate", _threadCount); _statusManager.init(); _queryMetrics = QueryMetrics.getInstance(); LOG.info("Init Complete"); + } public synchronized void close() { @@ -184,15 +200,13 @@ public class IndexManager { long s = System.nanoTime(); fetchRow(searcher.getIndexReader(), table, selector, fetchResult); _queryMetrics.recordDataFetch(System.nanoTime() - s, getRecordCount(fetchResult)); - if (_blurMetrics != null) { - if (fetchResult.rowResult != null) { - if (fetchResult.rowResult.row != null && fetchResult.rowResult.row.records != null) { - _blurMetrics.recordReads.addAndGet(fetchResult.rowResult.row.records.size()); - } - _blurMetrics.rowReads.incrementAndGet(); - } else if (fetchResult.recordResult != null) { - _blurMetrics.recordReads.incrementAndGet(); + if (fetchResult.rowResult != null) { + if (fetchResult.rowResult.row != null && fetchResult.rowResult.row.records != null) { + _recordsMeter.mark(fetchResult.rowResult.row.records.size()); } + _rowMeter.mark(); + } else if (fetchResult.recordResult != null) { + _recordsMeter.mark(); } } catch (Exception e) { LOG.error("Unknown error while trying to fetch row.", e); @@ -242,7 +256,8 @@ public class IndexManager { TopDocs topDocs = searcher.search(query, 1); if (topDocs.totalHits > 1) { if (selector.recordOnly) { - LOG.warn("Rowid [" + rowId + "], recordId [" + recordId + "] has more than one prime doc that is not deleted."); + LOG.warn("Rowid [" + rowId + "], recordId [" + recordId + + "] has more than one prime doc that is not deleted."); } else { LOG.warn("Rowid [" + rowId + "] has more than one prime doc that is not deleted."); } @@ -266,21 +281,25 @@ public class IndexManager { if (locationId != null) { if (recordId != null && rowId != null) { - throw new BlurException("Invalid selector locationId [" + locationId + "] and recordId [" + recordId + "] and rowId [" + rowId - + "] are set, if using locationId, then rowId and recordId are not needed.", null); + throw new BlurException("Invalid selector locationId [" + locationId + "] and recordId [" + recordId + + "] and rowId [" + rowId + "] are set, if using locationId, then rowId and recordId are not needed.", null); } else if (recordId != null) { - throw new BlurException("Invalid selector locationId [" + locationId + "] and recordId [" + recordId + "] sre set, if using locationId recordId is not needed.", null); + throw new BlurException("Invalid selector locationId [" + locationId + "] and recordId [" + recordId + + "] sre set, if using locationId recordId is not needed.", null); } else if (rowId != null) { - throw new BlurException("Invalid selector locationId [" + locationId + "] and rowId [" + rowId + "] are set, if using locationId rowId is not needed.", null); + throw new BlurException("Invalid selector locationId [" + locationId + "] and rowId [" + rowId + + "] are set, if using locationId rowId is not needed.", null); } } else { if (rowId != null && recordId != null) { if (!recordOnly) { - throw new BlurException("Invalid both rowid [" + rowId + "] and recordId [" + recordId + "] are set, and recordOnly is set to [false]. " + throw new BlurException("Invalid both rowid [" + rowId + "] and recordId [" + recordId + + "] are set, and recordOnly is set to [false]. " + "If you want entire row, then remove recordId, if you want record only set recordOnly to [true].", null); } } else if (recordId != null) { - throw new BlurException("Invalid recordId [" + recordId + "] is set but rowId is not set. If rowId is not known then a query will be required.", null); + throw new BlurException("Invalid recordId [" + recordId + + "] is set but rowId is not set. If rowId is not known then a query will be required.", null); } } } @@ -299,11 +318,12 @@ public class IndexManager { return split[0]; } - public BlurResultIterable query(final String table, final BlurQuery blurQuery, AtomicLongArray facetedCounts) throws Exception { + public BlurResultIterable query(final String table, final BlurQuery blurQuery, AtomicLongArray facetedCounts) + throws Exception { long s = System.nanoTime(); final AtomicBoolean running = new AtomicBoolean(true); final QueryStatus status = _statusManager.newQueryStatus(table, blurQuery, _threadCount, running); - _blurMetrics.queriesExternal.incrementAndGet(); + _queriesExternalMeter.mark(); try { Map<String, BlurIndex> blurIndexes; try { @@ -316,11 +336,15 @@ public class IndexManager { ParallelCall<Entry<String, BlurIndex>, BlurResultIterable> call; if (isSimpleQuery(blurQuery)) { SimpleQuery simpleQuery = blurQuery.simpleQuery; - Filter preFilter = QueryParserUtil.parseFilter(table, simpleQuery.preSuperFilter, false, analyzer, _filterCache); - Filter postFilter = QueryParserUtil.parseFilter(table, simpleQuery.postSuperFilter, true, analyzer, _filterCache); - Query userQuery = QueryParserUtil.parseQuery(simpleQuery.queryStr, simpleQuery.superQueryOn, analyzer, postFilter, preFilter, getScoreType(simpleQuery.type)); + Filter preFilter = QueryParserUtil + .parseFilter(table, simpleQuery.preSuperFilter, false, analyzer, _filterCache); + Filter postFilter = QueryParserUtil.parseFilter(table, simpleQuery.postSuperFilter, true, analyzer, + _filterCache); + Query userQuery = QueryParserUtil.parseQuery(simpleQuery.queryStr, simpleQuery.superQueryOn, analyzer, + postFilter, preFilter, getScoreType(simpleQuery.type)); Query facetedQuery = getFacetedQuery(blurQuery, userQuery, facetedCounts, analyzer); - call = new SimpleQueryParallelCall(running, table, status, _indexServer, facetedQuery, blurQuery.selector, _blurMetrics); + call = new SimpleQueryParallelCall(running, table, status, _indexServer, facetedQuery, blurQuery.selector, + _queriesInternalMeter); } else { Query query = getQuery(blurQuery.expertQuery); Filter filter = getFilter(blurQuery.expertQuery); @@ -331,7 +355,8 @@ public class IndexManager { userQuery = query; } Query facetedQuery = getFacetedQuery(blurQuery, userQuery, facetedCounts, analyzer); - call = new SimpleQueryParallelCall(running, table, status, _indexServer, facetedQuery, blurQuery.selector, _blurMetrics); + call = new SimpleQueryParallelCall(running, table, status, _indexServer, facetedQuery, blurQuery.selector, + _queriesInternalMeter); } MergerBlurResultIterable merger = new MergerBlurResultIterable(blurQuery); return ForkJoin.execute(_executor, blurIndexes.entrySet(), call, new Cancel() { @@ -361,7 +386,8 @@ public class IndexManager { return false; } - private Query getFacetedQuery(BlurQuery blurQuery, Query userQuery, AtomicLongArray counts, BlurAnalyzer analyzer) throws ParseException { + private Query getFacetedQuery(BlurQuery blurQuery, Query userQuery, AtomicLongArray counts, BlurAnalyzer analyzer) + throws ParseException { if (blurQuery.facets == null) { return userQuery; } @@ -372,7 +398,8 @@ public class IndexManager { int size = blurQuery.facets.size(); Query[] queries = new Query[size]; for (int i = 0; i < size; i++) { - queries[i] = QueryParserUtil.parseQuery(blurQuery.facets.get(i).queryStr, blurQuery.simpleQuery.superQueryOn, analyzer, null, null, ScoreType.CONSTANT); + queries[i] = QueryParserUtil.parseQuery(blurQuery.facets.get(i).queryStr, blurQuery.simpleQuery.superQueryOn, + analyzer, null, null, ScoreType.CONSTANT); } return queries; } @@ -400,7 +427,8 @@ public class IndexManager { return _statusManager.queryStatusIdList(table); } - public static void fetchRow(IndexReader reader, String table, Selector selector, FetchResult fetchResult) throws CorruptIndexException, IOException { + public static void fetchRow(IndexReader reader, String table, Selector selector, FetchResult fetchResult) + throws CorruptIndexException, IOException { fetchResult.table = table; String locationId = selector.locationId; int lastSlash = locationId.lastIndexOf('/'); @@ -504,7 +532,8 @@ public class IndexManager { this._indexServer = indexServer; } - public long recordFrequency(final String table, final String columnFamily, final String columnName, final String value) throws Exception { + public long recordFrequency(final String table, final String columnFamily, final String columnName, final String value) + throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); @@ -529,7 +558,8 @@ public class IndexManager { public Long merge(BlurExecutorCompletionService<Long> service) throws BlurException { long total = 0; while (service.getRemainingCount() > 0) { - Future<Long> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, value); + Future<Long> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, + columnFamily, columnName, value); total += service.getResultThrowException(future, table, columnFamily, columnName, value); } return total; @@ -537,7 +567,8 @@ public class IndexManager { }); } - public List<String> terms(final String table, final String columnFamily, final String columnName, final String startWith, final short size) throws Exception { + public List<String> terms(final String table, final String columnFamily, final String columnName, + final String startWith, final short size) throws Exception { Map<String, BlurIndex> blurIndexes; try { blurIndexes = _indexServer.getIndexes(table); @@ -545,24 +576,26 @@ public class IndexManager { LOG.error("Unknown error while trying to fetch index readers.", e); throw new BException(e.getMessage(), e); } - return ForkJoin.execute(_executor, blurIndexes.entrySet(), new ParallelCall<Entry<String, BlurIndex>, List<String>>() { - @Override - public List<String> call(Entry<String, BlurIndex> input) throws Exception { - BlurIndex index = input.getValue(); - IndexSearcherClosable searcher = index.getIndexReader(); - try { - return terms(searcher.getIndexReader(), columnFamily, columnName, startWith, size); - } finally { - // this will allow for closing of index - searcher.close(); - } - } - }).merge(new Merger<List<String>>() { + return ForkJoin.execute(_executor, blurIndexes.entrySet(), + new ParallelCall<Entry<String, BlurIndex>, List<String>>() { + @Override + public List<String> call(Entry<String, BlurIndex> input) throws Exception { + BlurIndex index = input.getValue(); + IndexSearcherClosable searcher = index.getIndexReader(); + try { + return terms(searcher.getIndexReader(), columnFamily, columnName, startWith, size); + } finally { + // this will allow for closing of index + searcher.close(); + } + } + }).merge(new Merger<List<String>>() { @Override public List<String> merge(BlurExecutorCompletionService<List<String>> service) throws BlurException { TreeSet<String> terms = new TreeSet<String>(); while (service.getRemainingCount() > 0) { - Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, columnFamily, columnName, startWith, size); + Future<List<String>> future = service.poll(_defaultParallelCallTimeout, TimeUnit.MILLISECONDS, true, table, + columnFamily, columnName, startWith, size); terms.addAll(service.getResultThrowException(future, table, columnFamily, columnName, startWith, size)); } return new ArrayList<String>(terms).subList(0, Math.min(size, terms.size())); @@ -570,11 +603,13 @@ public class IndexManager { }); } - public static long recordFrequency(IndexReader reader, String columnFamily, String columnName, String value) throws IOException { + public static long recordFrequency(IndexReader reader, String columnFamily, String columnName, String value) + throws IOException { return reader.docFreq(getTerm(columnFamily, columnName, value)); } - public static List<String> terms(IndexReader reader, String columnFamily, String columnName, String startWith, short size) throws IOException { + public static List<String> terms(IndexReader reader, String columnFamily, String columnName, String startWith, + short size) throws IOException { Term term = getTerm(columnFamily, columnName, startWith); List<String> terms = new ArrayList<String>(size); AtomicReader areader = BlurUtil.getAtomicReader(reader); @@ -719,7 +754,8 @@ public class IndexManager { } } - private void executeMutates(String table, String shard, Map<String, BlurIndex> indexes, List<RowMutation> mutations) throws BlurException, IOException { + private void executeMutates(String table, String shard, Map<String, BlurIndex> indexes, List<RowMutation> mutations) + throws BlurException, IOException { long s = System.nanoTime(); boolean waitToBeVisible = false; for (int i = 0; i < mutations.size(); i++) { @@ -729,7 +765,8 @@ public class IndexManager { } BlurIndex blurIndex = indexes.get(shard); if (blurIndex == null) { - throw new BlurException("Shard [" + shard + "] in table [" + table + "] is not being served by this server.", null); + throw new BlurException("Shard [" + shard + "] in table [" + table + "] is not being served by this server.", + null); } boolean waitVisiblity = false; @@ -780,7 +817,8 @@ public class IndexManager { String shard = MutationHelper.getShardName(table, mutation.rowId, getNumberOfShards(table), _blurPartitioner); BlurIndex blurIndex = indexes.get(shard); if (blurIndex == null) { - throw new BlurException("Shard [" + shard + "] in table [" + table + "] is not being served by this server.", null); + throw new BlurException("Shard [" + shard + "] in table [" + table + "] is not being served by this server.", + null); } RowMutationType type = mutation.rowMutationType; @@ -919,17 +957,18 @@ public class IndexManager { private IndexServer _indexServer; private Query _query; private Selector _selector; - private BlurMetrics _blurMetrics; private AtomicBoolean _running; + private Meter _queriesInternalMeter; - public SimpleQueryParallelCall(AtomicBoolean running, String table, QueryStatus status, IndexServer indexServer, Query query, Selector selector, BlurMetrics blurMetrics) { + public SimpleQueryParallelCall(AtomicBoolean running, String table, QueryStatus status, IndexServer indexServer, + Query query, Selector selector, Meter queriesInternalMeter) { _running = running; _table = table; _status = status; _indexServer = indexServer; _query = query; _selector = selector; - _blurMetrics = blurMetrics; + _queriesInternalMeter = queriesInternalMeter; } @Override @@ -944,9 +983,10 @@ public class IndexManager { // IndexSearcher searcher = new IndexSearcher(escapeReader); searcher.setSimilarity(_indexServer.getSimilarity(_table)); Query rewrite = searcher.rewrite((Query) _query.clone()); - return new BlurResultIterableSearcher(_running, rewrite, _table, shard, searcher, _selector, searcher.getIndexReader()); + return new BlurResultIterableSearcher(_running, rewrite, _table, shard, searcher, _selector, + searcher.getIndexReader()); } finally { - _blurMetrics.queriesInternal.incrementAndGet(); + _queriesInternalMeter.mark(); _status.deattachThread(); } } @@ -956,10 +996,6 @@ public class IndexManager { this._threadCount = threadCount; } - public void setBlurMetrics(BlurMetrics blurMetrics) { - _blurMetrics = blurMetrics; - } - public void setFilterCache(BlurFilterCache filterCache) { _filterCache = filterCache; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java b/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java index 8d96bb7..5dd1114 100644 --- a/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/manager/indexserver/DistributedIndexServer.java @@ -58,7 +58,6 @@ import org.apache.blur.manager.writer.BlurIndex; import org.apache.blur.manager.writer.BlurNRTIndex; import org.apache.blur.manager.writer.SharedMergeScheduler; import org.apache.blur.metrics.AtomicLongGauge; -import org.apache.blur.metrics.BlurMetrics; import org.apache.blur.server.IndexSearcherClosable; import org.apache.blur.server.ShardContext; import org.apache.blur.server.TableContext; @@ -109,7 +108,6 @@ public class DistributedIndexServer extends AbstractIndexServer { private String _nodeName; private int _shardOpenerThreadCount; private Cache _cache; - private BlurMetrics _blurMetrics; private ZooKeeper _zookeeper; private String _cluster; @@ -140,10 +138,10 @@ public class DistributedIndexServer extends AbstractIndexServer { public void init() throws KeeperException, InterruptedException, IOException { // evictions = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, CACHE, // EVICTION), EVICTION, TimeUnit.SECONDS); - Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, TABLE_COUNT), new AtomicLongGauge(_tableCount)); - Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_COUNT), new AtomicLongGauge(_indexCount)); - Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, SEGMENT_COUNT), new AtomicLongGauge(_segmentCount)); - Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_MEMORY_USAGE), new AtomicLongGauge(_indexMemoryUsage)); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, TABLE_COUNT, _cluster), new AtomicLongGauge(_tableCount)); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_COUNT, _cluster), new AtomicLongGauge(_indexCount)); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, SEGMENT_COUNT, _cluster), new AtomicLongGauge(_segmentCount)); + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, BLUR, INDEX_MEMORY_USAGE, _cluster), new AtomicLongGauge(_indexMemoryUsage)); BlurUtil.setupZookeeper(_zookeeper, _cluster); _openerService = Executors.newThreadPool("shard-opener", _shardOpenerThreadCount); @@ -276,7 +274,7 @@ public class DistributedIndexServer extends AbstractIndexServer { Map<String, BlurIndex> indexes = getIndexes(table); int count = indexes.size(); indexCount += count; - updateMetrics(_blurMetrics, indexes, segmentCount, indexMemoryUsage); + updateMetrics(indexes, segmentCount, indexMemoryUsage); LOG.debug("Table [{0}] has [{1}] number of shards online in this node.", table, count); } catch (IOException e) { LOG.error("Unknown error trying to warm table [{0}]", e, table); @@ -288,7 +286,7 @@ public class DistributedIndexServer extends AbstractIndexServer { } } - private void updateMetrics(BlurMetrics blurMetrics, Map<String, BlurIndex> indexes, AtomicLong segmentCount, + private void updateMetrics(Map<String, BlurIndex> indexes, AtomicLong segmentCount, AtomicLong indexMemoryUsage) throws IOException { // @TODO not sure how to do this yet // for (BlurIndex index : indexes.values()) { @@ -827,10 +825,6 @@ public class DistributedIndexServer extends AbstractIndexServer { _cache = cache; } - public void setBlurMetrics(BlurMetrics blurMetrics) { - _blurMetrics = blurMetrics; - } - public void setZookeeper(ZooKeeper zookeeper) { _zookeeper = zookeeper; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java index 8bc7b99..56289de 100644 --- a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurControllerServer.java @@ -52,11 +52,9 @@ import org.apache.blur.manager.BlurQueryChecker; import org.apache.blur.manager.clusterstatus.ZookeeperClusterStatus; import org.apache.blur.manager.indexserver.BlurServerShutDown; import org.apache.blur.manager.indexserver.BlurServerShutDown.BlurShutdown; -import org.apache.blur.metrics.BlurMetrics; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.utils.BlurUtil; import org.apache.blur.zookeeper.ZkUtils; -import org.apache.hadoop.conf.Configuration; import org.apache.zookeeper.ZooKeeper; @@ -80,9 +78,6 @@ public class ThriftBlurControllerServer extends ThriftServer { LOG.info("Shard Server using index [{0}] bind address [{1}]", serverIndex, bindAddress + ":" + bindPort); - Configuration config = new Configuration(); - BlurMetrics blurMetrics = new BlurMetrics(config); - String nodeName = ThriftBlurShardServer.getNodeName(configuration, BLUR_CONTROLLER_HOSTNAME); nodeName = nodeName + ":" + bindPort; String zkConnectionStr = isEmpty(configuration.get(BLUR_ZOOKEEPER_CONNECTION), BLUR_ZOOKEEPER_CONNECTION); @@ -120,7 +115,7 @@ public class ThriftBlurControllerServer extends ThriftServer { controllerServer.init(); - Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(blurMetrics, controllerServer, Iface.class); + Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(controllerServer, Iface.class); int threadCount = configuration.getInt(BLUR_CONTROLLER_SERVER_THRIFT_THREAD_COUNT, 32); @@ -140,7 +135,7 @@ public class ThriftBlurControllerServer extends ThriftServer { // params // without reversing the mvn dependancy and making blur-gui on top. httpServer = new HttpJettyServer(bindPort, webServerPort, configuration.getInt(BLUR_CONTROLLER_BIND_PORT, -1), configuration.getInt(BLUR_SHARD_BIND_PORT, -1), - configuration.getInt(BLUR_GUI_CONTROLLER_PORT, -1), configuration.getInt(BLUR_GUI_SHARD_PORT, -1), "controller", blurMetrics); + configuration.getInt(BLUR_GUI_CONTROLLER_PORT, -1), configuration.getInt(BLUR_GUI_SHARD_PORT, -1), "controller"); } else { httpServer = null; } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java index 2ea1599..29f4b72 100644 --- a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurShardServer.java @@ -54,7 +54,6 @@ import org.apache.blur.manager.indexserver.BlurServerShutDown.BlurShutdown; import org.apache.blur.manager.indexserver.DefaultBlurIndexWarmup; import org.apache.blur.manager.indexserver.DistributedIndexServer; import org.apache.blur.manager.writer.BlurIndexRefresher; -import org.apache.blur.metrics.BlurMetrics; import org.apache.blur.store.blockcache.BlockCache; import org.apache.blur.store.blockcache.BlockDirectory; import org.apache.blur.store.blockcache.BlockDirectoryCache; @@ -98,7 +97,7 @@ public class ThriftBlurShardServer extends ThriftServer { int bindPort = configuration.getInt(BLUR_SHARD_BIND_PORT, -1); bindPort += serverIndex; - BlurMetrics blurMetrics = new BlurMetrics(config); +// BlurMetrics blurMetrics = new BlurMetrics(config); int baseGuiPort = Integer.parseInt(configuration.get(BLUR_GUI_SHARD_PORT)); final HttpJettyServer httpServer; @@ -109,7 +108,7 @@ public class ThriftBlurShardServer extends ThriftServer { // params // without reversing the mvn dependancy and making blur-gui on top. httpServer = new HttpJettyServer(bindPort, webServerPort, configuration.getInt(BLUR_CONTROLLER_BIND_PORT, -1), configuration.getInt(BLUR_SHARD_BIND_PORT, -1), - configuration.getInt(BLUR_GUI_CONTROLLER_PORT, -1), configuration.getInt(BLUR_GUI_SHARD_PORT, -1), "shard", blurMetrics); + configuration.getInt(BLUR_GUI_CONTROLLER_PORT, -1), configuration.getInt(BLUR_GUI_SHARD_PORT, -1), "shard"); } else { httpServer = null; } @@ -173,7 +172,6 @@ public class ThriftBlurShardServer extends ThriftServer { BlurIndexWarmup indexWarmup = getIndexWarmup(configuration); final DistributedIndexServer indexServer = new DistributedIndexServer(); - indexServer.setBlurMetrics(blurMetrics); indexServer.setCache(cache); indexServer.setClusterStatus(clusterStatus); indexServer.setClusterName(configuration.get(BLUR_CLUSTER_NAME, BLUR_CLUSTER)); @@ -190,7 +188,6 @@ public class ThriftBlurShardServer extends ThriftServer { indexManager.setIndexServer(indexServer); indexManager.setMaxClauseCount(configuration.getInt(BLUR_MAX_CLAUSE_COUNT, 1024)); indexManager.setThreadCount(configuration.getInt(BLUR_INDEXMANAGER_SEARCH_THREAD_COUNT, 32)); - indexManager.setBlurMetrics(blurMetrics); indexManager.setFilterCache(filterCache); indexManager.init(); @@ -203,7 +200,7 @@ public class ThriftBlurShardServer extends ThriftServer { shardServer.setConfiguration(configuration); shardServer.init(); - Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(blurMetrics, shardServer, Iface.class); + Iface iface = BlurUtil.recordMethodCallsAndAverageTimes(shardServer, Iface.class); int threadCount = configuration.getInt(BLUR_SHARD_SERVER_THRIFT_THREAD_COUNT, 32); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java b/src/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java index 22cf6c4..12a21c4 100644 --- a/src/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java +++ b/src/blur-core/src/main/java/org/apache/blur/utils/BlurUtil.java @@ -34,8 +34,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; @@ -46,8 +48,6 @@ import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; import org.apache.blur.manager.clusterstatus.ZookeeperPathConstants; import org.apache.blur.manager.results.BlurResultIterable; -import org.apache.blur.metrics.BlurMetrics; -import org.apache.blur.metrics.BlurMetrics.MethodCall; import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurQuery; import org.apache.blur.thrift.generated.BlurResult; @@ -88,6 +88,11 @@ import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; +import com.yammer.metrics.Metrics; +import com.yammer.metrics.core.Histogram; +import com.yammer.metrics.core.MetricName; +import static org.apache.blur.metrics.MetricsConstants.*; + public class BlurUtil { private static final Object[] EMPTY_OBJECT_ARRAY = new Object[] {}; @@ -97,7 +102,13 @@ public class BlurUtil { private static final int ALL = Integer.MAX_VALUE; @SuppressWarnings("unchecked") - public static <T extends Iface> T recordMethodCallsAndAverageTimes(final BlurMetrics metrics, final T t, Class<T> clazz) { + public static <T extends Iface> T recordMethodCallsAndAverageTimes(final T t, Class<T> clazz) { + final Map<String, Histogram> histogramMap = new ConcurrentHashMap<String, Histogram>(); + Method[] declaredMethods = Iface.class.getDeclaredMethods(); + for (Method m : declaredMethods) { + String name = m.getName(); + histogramMap.put(name, Metrics.newHistogram(new MetricName(ORG_APACHE_BLUR, BLUR, name, THRIFT_CALLS))); + } InvocationHandler handler = new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { @@ -108,13 +119,9 @@ public class BlurUtil { throw e.getTargetException(); } finally { long end = System.nanoTime(); - MethodCall methodCall = metrics.methodCalls.get(method.getName()); - if (methodCall == null) { - methodCall = new MethodCall(); - metrics.methodCalls.put(method.getName(), methodCall); - } - methodCall.invokes.incrementAndGet(); - methodCall.times.addAndGet(end - start); + String name = method.getName(); + Histogram histogram = histogramMap.get(name); + histogram.update((end - start) / 1000); } } }; @@ -125,7 +132,8 @@ public class BlurUtil { setupZookeeper(zookeeper, null); } - public synchronized static void setupZookeeper(ZooKeeper zookeeper, String cluster) throws KeeperException, InterruptedException { + public synchronized static void setupZookeeper(ZooKeeper zookeeper, String cluster) throws KeeperException, + InterruptedException { BlurUtil.createIfMissing(zookeeper, ZookeeperPathConstants.getBasePath()); BlurUtil.createIfMissing(zookeeper, ZookeeperPathConstants.getOnlineControllersPath()); BlurUtil.createIfMissing(zookeeper, ZookeeperPathConstants.getClustersPath()); @@ -210,7 +218,8 @@ public class BlurUtil { return newRecordMutation(RecordMutationType.REPLACE_ENTIRE_RECORD, family, recordId, columns); } - public static RecordMutation newRecordMutation(RecordMutationType type, String family, String recordId, Column... columns) { + public static RecordMutation newRecordMutation(RecordMutationType type, String family, String recordId, + Column... columns) { Record record = new Record(); record.setRecordId(recordId); record.setFamily(family); @@ -245,7 +254,8 @@ public class BlurUtil { return newRowMutation(RowMutationType.REPLACE_ROW, table, rowId, mutations); } - public static RowMutation newRowMutation(RowMutationType type, String table, String rowId, RecordMutation... mutations) { + public static RowMutation newRowMutation(RowMutationType type, String table, String rowId, + RecordMutation... mutations) { RowMutation mutation = new RowMutation(); mutation.setRowId(rowId); mutation.setTable(table); @@ -325,8 +335,9 @@ public class BlurUtil { return new AtomicLongArray(list.size()); } - public static BlurResults convertToHits(BlurResultIterable hitsIterable, BlurQuery query, AtomicLongArray facetCounts, ExecutorService executor, Selector selector, - final Iface iface, final String table) throws InterruptedException, ExecutionException { + public static BlurResults convertToHits(BlurResultIterable hitsIterable, BlurQuery query, + AtomicLongArray facetCounts, ExecutorService executor, Selector selector, final Iface iface, final String table) + throws InterruptedException, ExecutionException { BlurResults results = new BlurResults(); results.setTotalResults(hitsIterable.getTotalResults()); results.setShardInfo(hitsIterable.getShardInfo()); @@ -395,16 +406,19 @@ public class BlurUtil { return verison.toString(); } - public static void unlockForSafeMode(ZooKeeper zookeeper, String lockPath) throws InterruptedException, KeeperException { + public static void unlockForSafeMode(ZooKeeper zookeeper, String lockPath) throws InterruptedException, + KeeperException { zookeeper.delete(lockPath, -1); LOG.info("Lock released."); } - public static String lockForSafeMode(ZooKeeper zookeeper, String nodeName, String cluster) throws KeeperException, InterruptedException { + public static String lockForSafeMode(ZooKeeper zookeeper, String nodeName, String cluster) throws KeeperException, + InterruptedException { LOG.info("Getting safe mode lock."); final Object lock = new Object(); String blurSafemodePath = ZookeeperPathConstants.getSafemodePath(cluster); - String newPath = zookeeper.create(blurSafemodePath + "/safemode-", nodeName.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); + String newPath = zookeeper.create(blurSafemodePath + "/safemode-", nodeName.getBytes(), Ids.OPEN_ACL_UNSAFE, + CreateMode.EPHEMERAL_SEQUENTIAL); Watcher watcher = new Watcher() { @Override public void process(WatchedEvent event) { @@ -485,7 +499,8 @@ public class BlurUtil { return sizeOf; } - public static void createPath(ZooKeeper zookeeper, String path, byte[] data) throws KeeperException, InterruptedException { + public static void createPath(ZooKeeper zookeeper, String path, byte[] data) throws KeeperException, + InterruptedException { zookeeper.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } @@ -522,8 +537,10 @@ public class BlurUtil { String shardIndexStr = name.substring(index + 1); int shardIndex = Integer.parseInt(shardIndexStr); if (shardIndex >= shardCount) { - LOG.error("Number of directories in table path [" + path + "] exceeds definition of [" + shardCount + "] shard count."); - throw new RuntimeException("Number of directories in table path [" + path + "] exceeds definition of [" + shardCount + "] shard count."); + LOG.error("Number of directories in table path [" + path + "] exceeds definition of [" + shardCount + + "] shard count."); + throw new RuntimeException("Number of directories in table path [" + path + "] exceeds definition of [" + + shardCount + "] shard count."); } } } @@ -647,7 +664,8 @@ public class BlurUtil { * * @throws IOException */ - public static List<Document> termSearch(IndexReader reader, Term term, ResetableDocumentStoredFieldVisitor fieldSelector) throws IOException { + public static List<Document> termSearch(IndexReader reader, Term term, + ResetableDocumentStoredFieldVisitor fieldSelector) throws IOException { IndexSearcher indexSearcher = new IndexSearcher(reader); TopDocs topDocs = indexSearcher.search(new TermQuery(term), ALL); List<Document> docs = new ArrayList<Document>(); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java b/src/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java index f6641d1..0056e36 100644 --- a/src/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java +++ b/src/blur-core/src/test/java/org/apache/blur/manager/IndexManagerTest.java @@ -47,7 +47,6 @@ import java.util.concurrent.atomic.AtomicLongArray; import org.apache.blur.manager.indexserver.LocalIndexServer; import org.apache.blur.manager.results.BlurResultIterable; -import org.apache.blur.metrics.BlurMetrics; import org.apache.blur.thrift.generated.AnalyzerDefinition; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.BlurQuery; @@ -67,7 +66,6 @@ import org.apache.blur.thrift.generated.SimpleQuery; import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.utils.BlurConstants; import org.apache.blur.utils.BlurUtil; -import org.apache.hadoop.conf.Configuration; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -106,7 +104,6 @@ public class IndexManagerTest { indexManager.setStatusCleanupTimerDelay(1000); indexManager.setIndexServer(server); indexManager.setThreadCount(1); - indexManager.setBlurMetrics(new BlurMetrics(new Configuration())); indexManager.init(); setupData(); } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java ---------------------------------------------------------------------- diff --git a/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java b/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java index d775bb8..1b271ef 100644 --- a/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java +++ b/src/blur-gui/src/main/java/org/apache/blur/gui/HttpJettyServer.java @@ -22,11 +22,12 @@ import java.util.Properties; import org.apache.blur.log.Log; import org.apache.blur.log.LogFactory; -import org.apache.blur.metrics.BlurMetrics; import org.mortbay.jetty.Server; import org.mortbay.jetty.servlet.ServletHolder; import org.mortbay.jetty.webapp.WebAppContext; +import com.yammer.metrics.reporting.MetricsServlet; + /** * Starts up a Jetty server to run the utility gui. * @@ -55,7 +56,7 @@ public class HttpJettyServer { * metrics object for using. * @throws IOException */ - public HttpJettyServer(int bindPort, int port, int baseControllerPort, int baseShardPort, int baseGuiControllerPort, int baseGuiShardPort, String base, BlurMetrics bm) + public HttpJettyServer(int bindPort, int port, int baseControllerPort, int baseShardPort, int baseGuiControllerPort, int baseGuiShardPort, String base) throws IOException { server = new Server(port); @@ -76,7 +77,7 @@ public class HttpJettyServer { context.setContextPath("/"); context.setParentLoaderPriority(true); context.addServlet(new ServletHolder(new LiveMetricsServlet()), "/livemetrics"); - context.addServlet(new ServletHolder(new MetricsServlet(bm)), "/metrics"); + context.addServlet(new ServletHolder(new MetricsServlet()), "/metrics"); context.addServlet(new ServletHolder(new LogServlet(blurLogFile)), "/logs"); LOG.info("WEB GUI coming up for resource: " + base); http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-gui/src/main/java/org/apache/blur/gui/MetricsServlet.java ---------------------------------------------------------------------- diff --git a/src/blur-gui/src/main/java/org/apache/blur/gui/MetricsServlet.java b/src/blur-gui/src/main/java/org/apache/blur/gui/MetricsServlet.java deleted file mode 100644 index c1bfd20..0000000 --- a/src/blur-gui/src/main/java/org/apache/blur/gui/MetricsServlet.java +++ /dev/null @@ -1,51 +0,0 @@ -package org.apache.blur.gui; - -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -import java.io.IOException; -import java.io.PrintWriter; - -import javax.servlet.ServletException; -import javax.servlet.http.HttpServlet; -import javax.servlet.http.HttpServletRequest; -import javax.servlet.http.HttpServletResponse; - -import org.apache.blur.metrics.BlurMetrics; -import org.codehaus.jackson.map.ObjectMapper; - - -public class MetricsServlet extends HttpServlet { - - private static final long serialVersionUID = 1L; - - private BlurMetrics bm = null; - - public MetricsServlet() { - } - - public MetricsServlet(BlurMetrics bm) { - this.bm = bm; - } - - protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException { - response.setContentType("application/json"); - PrintWriter out = response.getWriter(); - ObjectMapper mapper = new ObjectMapper(); - mapper.writeValue(out, bm); - } - -} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/e7e1a406/src/blur-util/src/main/java/org/apache/blur/metrics/BlurMetrics.java ---------------------------------------------------------------------- diff --git a/src/blur-util/src/main/java/org/apache/blur/metrics/BlurMetrics.java b/src/blur-util/src/main/java/org/apache/blur/metrics/BlurMetrics.java deleted file mode 100644 index 3fd056f..0000000 --- a/src/blur-util/src/main/java/org/apache/blur/metrics/BlurMetrics.java +++ /dev/null @@ -1,125 +0,0 @@ -package org.apache.blur.metrics; - -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -import java.util.Map; -import java.util.Map.Entry; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicLong; - -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.metrics.MetricsContext; -import org.apache.hadoop.metrics.MetricsRecord; -import org.apache.hadoop.metrics.MetricsUtil; -import org.apache.hadoop.metrics.Updater; -import org.apache.hadoop.metrics.jvm.JvmMetrics; - -public class BlurMetrics implements Updater { - - public static class MethodCall { - public AtomicLong invokes = new AtomicLong(); - public AtomicLong times = new AtomicLong(); - } - - public AtomicLong blockCacheHit = new AtomicLong(0); - public AtomicLong blockCacheMiss = new AtomicLong(0); - public AtomicLong blockCacheEviction = new AtomicLong(0); - public AtomicLong blockCacheSize = new AtomicLong(0); - public AtomicLong rowReads = new AtomicLong(0); - public AtomicLong rowWrites = new AtomicLong(0); - public AtomicLong recordReads = new AtomicLong(0); - public AtomicLong recordWrites = new AtomicLong(0); - public AtomicLong queriesExternal = new AtomicLong(0); - public AtomicLong queriesInternal = new AtomicLong(0); - public AtomicLong blurShardBuffercacheAllocate1024 = new AtomicLong(0); - public AtomicLong blurShardBuffercacheAllocate8192 = new AtomicLong(0); - public AtomicLong blurShardBuffercacheAllocateOther = new AtomicLong(0); - public AtomicLong blurShardBuffercacheLost = new AtomicLong(0); - public Map<String, MethodCall> methodCalls = new ConcurrentHashMap<String, MethodCall>(); - -// public AtomicLong tableCount = new AtomicLong(0); -// public AtomicLong rowCount = new AtomicLong(0); -// public AtomicLong recordCount = new AtomicLong(0); -// public AtomicLong indexCount = new AtomicLong(0); -// public AtomicLong indexMemoryUsage = new AtomicLong(0); -// public AtomicLong segmentCount = new AtomicLong(0); - - private MetricsRecord _metricsRecord; - private long _previous = System.nanoTime(); - - public static void main(String[] args) throws InterruptedException { - Configuration conf = new Configuration(); - BlurMetrics blurMetrics = new BlurMetrics(conf); - MethodCall methodCall = new MethodCall(); - blurMetrics.methodCalls.put("test", methodCall); - for (int i = 0; i < 100; i++) { - blurMetrics.blockCacheHit.incrementAndGet(); - blurMetrics.blockCacheMiss.incrementAndGet(); - methodCall.invokes.incrementAndGet(); - methodCall.times.addAndGet(56000000); - Thread.sleep(500); - } - } - - public BlurMetrics(Configuration conf) { - JvmMetrics.init("blur", Long.toString(System.currentTimeMillis())); - MetricsContext metricsContext = MetricsUtil.getContext("blur"); - _metricsRecord = MetricsUtil.createRecord(metricsContext, "metrics"); - metricsContext.registerUpdater(this); - } - - @Override - public void doUpdates(MetricsContext context) { - synchronized (this) { - long now = System.nanoTime(); - float seconds = (now - _previous) / 1000000000.0f; - _metricsRecord.setMetric("blockcache.hit", getPerSecond(blockCacheHit.getAndSet(0), seconds)); - _metricsRecord.setMetric("blockcache.miss", getPerSecond(blockCacheMiss.getAndSet(0), seconds)); - _metricsRecord.setMetric("blockcache.eviction", getPerSecond(blockCacheEviction.getAndSet(0), seconds)); - _metricsRecord.setMetric("blockcache.size", blockCacheSize.get()); - _metricsRecord.setMetric("row.reads", getPerSecond(rowReads.getAndSet(0), seconds)); - _metricsRecord.setMetric("row.writes", getPerSecond(rowWrites.getAndSet(0), seconds)); - _metricsRecord.setMetric("record.reads", getPerSecond(recordReads.getAndSet(0), seconds)); - _metricsRecord.setMetric("record.writes", getPerSecond(recordWrites.getAndSet(0), seconds)); - _metricsRecord.setMetric("query.external", getPerSecond(queriesExternal.getAndSet(0), seconds)); - _metricsRecord.setMetric("query.internal", getPerSecond(queriesInternal.getAndSet(0), seconds)); - for (Entry<String, MethodCall> entry : methodCalls.entrySet()) { - String key = entry.getKey(); - MethodCall value = entry.getValue(); - long invokes = value.invokes.getAndSet(0); - long times = value.times.getAndSet(0); - - float avgTimes = (times / (float) invokes) / 1000000000.0f; - _metricsRecord.setMetric("methodcalls." + key + ".count", getPerSecond(invokes, seconds)); - _metricsRecord.setMetric("methodcalls." + key + ".time", avgTimes); - } - _metricsRecord.setMetric("tables", tableCount.get()); - _metricsRecord.setMetric("rows", rowCount.get()); - _metricsRecord.setMetric("records", recordCount.get()); - _metricsRecord.setMetric("index.count", indexCount.get()); - _metricsRecord.setMetric("index.memoryusage", indexMemoryUsage.get()); - _metricsRecord.setMetric("index.segments", segmentCount.get()); - _previous = now; - } - _metricsRecord.update(); - } - - private float getPerSecond(long value, float seconds) { - return (float) (value / seconds); - } - -}
