Updated Branches: refs/heads/0.2-dev 0270f9a09 -> 80751ebab
Cleaned up more 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/80751eba Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/80751eba Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/80751eba Branch: refs/heads/0.2-dev Commit: 80751ebab0ccdf630784580fbefee2b81c644413 Parents: 0270f9a Author: Aaron McCurry <[email protected]> Authored: Wed Mar 6 10:03:20 2013 -0500 Committer: Aaron McCurry <[email protected]> Committed: Wed Mar 6 10:03:20 2013 -0500 ---------------------------------------------------------------------- .../indexserver/DistributedIndexServer.java | 19 +++-- .../java/org/apache/blur/server/BlurServer.java | 25 ++++-- .../org/apache/blur/thrift/ThriftBlurServer.java | 1 + .../store/refcounter/DirectoryReferenceFileGC.java | 5 +- .../org/apache/blur/store/hdfs/HdfsDirectory.java | 69 +++++++++++--- 5 files changed, 89 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/80751eba/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 b6009dc..3c0984b 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 @@ -56,6 +56,8 @@ import org.apache.blur.metrics.BlurMetrics; import org.apache.blur.server.ShardContext; import org.apache.blur.server.TableContext; import org.apache.blur.server.ZooKeeperTypeManager; +import org.apache.blur.store.blockcache.BlockDirectory; +import org.apache.blur.store.blockcache.Cache; import org.apache.blur.store.hdfs.BlurLockFactory; import org.apache.blur.thrift.generated.TableDescriptor; import org.apache.blur.utils.BlurConstants; @@ -109,6 +111,7 @@ public class DistributedIndexServer extends AbstractIndexServer { private WatchChildren _watchOnlineShards; private SharedMergeScheduler _mergeScheduler; private IndexInputCloser _closer; + private Cache _cache; public static interface ReleaseReader { void release() throws IOException; @@ -452,13 +455,12 @@ public class DistributedIndexServer extends AbstractIndexServer { Path hdfsDirPath = shardContext.getHdfsDirPath(); BlurLockFactory lockFactory = new BlurLockFactory(_configuration, hdfsDirPath, _nodeName, BlurConstants.getPid()); - // Directory directory = shardContext.getDirectory(); - // directory.setLockFactory(lockFactory); - // Directory dir = new BlockDirectory(table + "_" + shard, directory, - // _cache); + Directory directory = shardContext.getDirectory(); + directory.setLockFactory(lockFactory); + Directory dir = new BlockDirectory(table + "_" + shard, directory, _cache); - Directory dir = shardContext.getDirectory(); - dir.setLockFactory(lockFactory); +// Directory dir = shardContext.getDirectory(); +// dir.setLockFactory(lockFactory); BlurIndex index; if (tableDescriptor.isReadOnly()) { @@ -697,4 +699,9 @@ public class DistributedIndexServer extends AbstractIndexServer { _cluster = cluster; } + public void setCache(Cache cache) { + _cache = cache; + } + + } http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/80751eba/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java b/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java index 9dfda86..c0c6e57 100644 --- a/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/server/BlurServer.java @@ -76,9 +76,18 @@ import org.apache.thrift.TException; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.MetricName; public class BlurServer extends TableAdmin implements Iface { + private static final String REQUESTS = "Requests"; + private static final String ORG_APACHE_BLUR = "org.apache.blur"; + private static final String DELETE = "delete"; + private static final String UPDATE = "update"; + private static final String ADD = "add"; + private static final String SEARCH = "search"; + private static final String REMOTE = "remote"; + private static final String LOCAL = "local"; private static final Log LOG = LogFactory.getLog(BlurServer.class); private IndexServer _indexServer; private boolean _closed; @@ -138,14 +147,14 @@ public class BlurServer extends TableAdmin implements Iface { _sessionManager = new SessionManager(); _sessionManager.init(); - localSearches = Metrics.newMeter(BlurServer.class, "searches", "local", TimeUnit.SECONDS); - remoteSearches = Metrics.newMeter(BlurServer.class, "searches", "remote", TimeUnit.SECONDS); - localAddRequests = Metrics.newMeter(BlurServer.class, "addRequests", "local", TimeUnit.SECONDS); - remoteAddRequests = Metrics.newMeter(BlurServer.class, "addRequests", "remote", TimeUnit.SECONDS); - localUpdateRequests = Metrics.newMeter(BlurServer.class, "updateRequests", "local", TimeUnit.SECONDS); - remoteUpdateRequests = Metrics.newMeter(BlurServer.class, "updateRequests", "remote", TimeUnit.SECONDS); - localDeleteRequests = Metrics.newMeter(BlurServer.class, "deleteRequests", "local", TimeUnit.SECONDS); - remoteDeleteRequests = Metrics.newMeter(BlurServer.class, "deleteRequests", "remote", TimeUnit.SECONDS); + localSearches = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, SEARCH), LOCAL, TimeUnit.SECONDS); + remoteSearches = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, SEARCH), REMOTE, TimeUnit.SECONDS); + localAddRequests = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, ADD), LOCAL, TimeUnit.SECONDS); + remoteAddRequests = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, ADD), REMOTE, TimeUnit.SECONDS); + localUpdateRequests = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, UPDATE), LOCAL, TimeUnit.SECONDS); + remoteUpdateRequests = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, UPDATE), REMOTE, TimeUnit.SECONDS); + localDeleteRequests = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, DELETE), LOCAL, TimeUnit.SECONDS); + remoteDeleteRequests = Metrics.newMeter(new MetricName(ORG_APACHE_BLUR, REQUESTS, DELETE), REMOTE, TimeUnit.SECONDS); } @Override http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/80751eba/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java ---------------------------------------------------------------------- diff --git a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java index 587b3dc..8f32a9f 100644 --- a/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java +++ b/src/blur-core/src/main/java/org/apache/blur/thrift/ThriftBlurServer.java @@ -206,6 +206,7 @@ public class ThriftBlurServer extends AbstractThriftServer { indexServer.setZookeeper(zooKeeper); indexServer.setFilterCache(filterCache); indexServer.setSafeModeDelay(configuration.getLong(BLUR_SHARD_SAFEMODEDELAY, 60000)); + indexServer.setCache(cache); indexServer.init(); TableLayout layout = new TableLayout() { http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/80751eba/src/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java ---------------------------------------------------------------------- diff --git a/src/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java b/src/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java index b8a9301..834dc38 100644 --- a/src/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java +++ b/src/blur-store/src/main/java/org/apache/blur/lucene/store/refcounter/DirectoryReferenceFileGC.java @@ -30,9 +30,12 @@ import org.apache.lucene.store.Directory; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Gauge; +import com.yammer.metrics.core.MetricName; public class DirectoryReferenceFileGC extends TimerTask { + private static final String ORG_APACHE_BLUR = "org.apache.blur"; + private static final Log LOG = LogFactory.getLog(DirectoryReferenceFileGC.class); private Timer _timer; @@ -69,7 +72,7 @@ public class DirectoryReferenceFileGC extends TimerTask { _timer = new Timer("Blur-File-GC", true); _timer.scheduleAtFixedRate(this, _delay, _delay); _queue = new LinkedBlockingQueue<Value>(); - Metrics.newGauge(getClass(), "filesToBeDeleted", new Gauge<Integer>() { + Metrics.newGauge(new MetricName(ORG_APACHE_BLUR, "Lucene", "Files in Queue to be Deleted"), new Gauge<Integer>() { @Override public Integer value() { return numberOfFilesToBeDeleted; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/80751eba/src/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java ---------------------------------------------------------------------- diff --git a/src/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java b/src/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java index fdc5a52..9881dd2 100644 --- a/src/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java +++ b/src/blur-store/src/main/java/org/apache/blur/store/hdfs/HdfsDirectory.java @@ -19,9 +19,11 @@ package org.apache.blur.store.hdfs; import java.io.FileNotFoundException; import java.io.IOException; +import java.net.URI; import java.util.Collection; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; +import java.util.WeakHashMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import org.apache.blur.log.Log; @@ -44,6 +46,8 @@ import org.apache.lucene.store.NoLockFactory; import com.yammer.metrics.Metrics; import com.yammer.metrics.core.Histogram; +import com.yammer.metrics.core.Meter; +import com.yammer.metrics.core.MetricName; public class HdfsDirectory extends Directory { @@ -84,23 +88,53 @@ public class HdfsDirectory extends Directory { private final Path path; private final FileSystem fileSystem; + private final MetricsGroup metricsGroup; - private final Histogram readAccess; - private static Map<String, Histogram> readAccessMap = new ConcurrentHashMap<String, Histogram>(); + static class MetricsGroup { + final Histogram readAccess; + final Histogram writeAccess; + final Meter writeThroughput; + final Meter readThroughput; + + MetricsGroup(Histogram readAccess, Histogram writeAccess, Meter readThroughput, Meter writeThroughput) { + this.readAccess = readAccess; + this.writeAccess = writeAccess; + this.readThroughput = readThroughput; + this.writeThroughput = writeThroughput; + } + } + + public static void main(String[] args) { + System.out.println(new MetricName(HdfsDirectory.class,"name","scope")); + System.out.println(new MetricName("group","type","name","scope")); + } + + /** + * We keep the metrics separate per filesystem. + */ + private static Map<URI, MetricsGroup> metricsGroupMap = new WeakHashMap<URI, MetricsGroup>(); public HdfsDirectory(Configuration configuration, Path path) throws IOException { this.path = path; fileSystem = path.getFileSystem(configuration); setLockFactory(NoLockFactory.getNoLockFactory()); - synchronized (readAccessMap) { - Histogram histogram = readAccessMap.get(fileSystem.toString()); - if (histogram == null) { - histogram = Metrics.newHistogram(HdfsDirectory.class, "Read Access [" + fileSystem.getUri() + "]"); - readAccessMap.put(fileSystem.toString(), histogram); + synchronized (metricsGroupMap) { + URI uri = fileSystem.getUri(); + MetricsGroup metricsGroup = metricsGroupMap.get(uri); + if (metricsGroup == null) { + String group = "org.apache.blur"; + String type = "HDFS"; + String scope = uri.toString(); + + Histogram readAccess = Metrics.newHistogram(new MetricName(group, type, "Read Latency in \u00B5s", scope)); + Histogram writeAccess = Metrics.newHistogram(new MetricName(group, type, "Write Latency in \u00B5s", scope)); + Meter readThroughput = Metrics.newMeter(new MetricName(group, type, "Read Throughput", scope), "Read Bytes", TimeUnit.SECONDS); + Meter writeThroughput = Metrics.newMeter(new MetricName(group, type, "Write Throughput", scope), "Write Bytes", TimeUnit.SECONDS); + metricsGroup = new MetricsGroup(readAccess, writeAccess, readThroughput, writeThroughput); + metricsGroupMap.put(uri, metricsGroup); } - readAccess = histogram; + this.metricsGroup = metricsGroup; } - } @Override @@ -113,14 +147,14 @@ public class HdfsDirectory extends Directory { private final long len; private FSDataInputStream inputStream; private boolean isClone; - private final Histogram readAccess; + private final MetricsGroup metricsGroup; - public HdfsIndexInput(FileSystem fileSystem, Path filePath, Histogram readAccess) throws IOException { + public HdfsIndexInput(FileSystem fileSystem, Path filePath, MetricsGroup metricsGroup) throws IOException { super(filePath.toString()); inputStream = fileSystem.open(filePath); FileStatus fileStatus = fileSystem.getFileStatus(filePath); len = fileStatus.getLen(); - this.readAccess = readAccess; + this.metricsGroup = metricsGroup; } @Override @@ -140,7 +174,8 @@ public class HdfsDirectory extends Directory { inputStream.seek(getFilePointer()); inputStream.readFully(b, offset, length); long end = System.nanoTime(); - readAccess.update((end - start) / 1000); + metricsGroup.readAccess.update((end - start) / 1000); + metricsGroup.readThroughput.mark(length); } } @@ -175,7 +210,11 @@ public class HdfsDirectory extends Directory { @Override protected void flushBuffer(byte[] b, int offset, int len) throws IOException { + long start = System.nanoTime(); outputStream.write(b, offset, len); + long end = System.nanoTime(); + metricsGroup.writeAccess.update((end - start) / 1000); + metricsGroup.writeThroughput.mark(len); } @Override @@ -197,7 +236,7 @@ public class HdfsDirectory extends Directory { throw new FileNotFoundException("File [" + name + "] not found."); } Path filePath = getPath(name); - return new HdfsIndexInput(fileSystem, filePath, readAccess); + return new HdfsIndexInput(fileSystem, filePath, metricsGroup); } @Override
