The advanced version - too advanced really.

Author: henningandersen

server/src/main/java/org/elasticsearch/index/IndexModule.java

@@ -57,6 +57,9 @@
 import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.index.similarity.SimilarityService;
 import org.elasticsearch.index.store.FsDirectoryFactory;
+import org.elasticsearch.index.store.MetricHandler;
+import org.elasticsearch.index.store.StoreMetrics;
+import org.elasticsearch.index.store.ThreadLocalMetricHandler;
 import org.elasticsearch.indices.IndicesQueryCache;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.indices.fielddata.cache.IndicesFieldDataCache;
@@ -182,6 +185,7 @@ public interface DirectoryWrapper {
     private final IndexingStatsSettings indexingStatsSettings;
     private final SearchStatsSettings searchStatsSettings;
     private final MergeMetrics mergeMetrics;
+    private MetricHandler<StoreMetrics> metricHandler;
 
     /**
      * Construct the index module for the index with the specified index settings. The index module contains extension points for plugins
@@ -206,7 +210,7 @@ public IndexModule(
         final List<SearchOperationListener> searchOperationListeners,
         final IndexingStatsSettings indexingStatsSettings,
         final SearchStatsSettings searchStatsSettings,
-        final MergeMetrics mergeMetrics
+        final MergeMetrics mergeMetrics, MetricHandler<StoreMetrics> metricHandler
     ) {
         this.indexSettings = indexSettings;
         this.analysisRegistry = analysisRegistry;
@@ -224,6 +228,7 @@ public IndexModule(
         this.indexingStatsSettings = indexingStatsSettings;
         this.searchStatsSettings = searchStatsSettings;
         this.mergeMetrics = mergeMetrics;
+        this.metricHandler = metricHandler;
     }
 
     /**
@@ -560,7 +565,8 @@ public IndexService newIndexService(
                 mapperMetrics,
                 indexingStatsSettings,
                 searchStatsSettings,
-                mergeMetrics
+                mergeMetrics,
+                metricHandler
             );
             success = true;
             return indexService;

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -82,7 +82,9 @@
 import org.elasticsearch.index.shard.ShardNotInPrimaryModeException;
 import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.index.similarity.SimilarityService;
+import org.elasticsearch.index.store.MetricHandler;
 import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.store.StoreMetrics;
 import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.indices.cluster.IndexRemovalReason;
@@ -172,6 +174,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final IndexingStatsSettings indexingStatsSettings;
     private final SearchStatsSettings searchStatsSettings;
     private final MergeMetrics mergeMetrics;
+    private MetricHandler<StoreMetrics> metricHandler;
 
     @SuppressWarnings("this-escape")
     public IndexService(
@@ -210,7 +213,8 @@ public IndexService(
         MapperMetrics mapperMetrics,
         IndexingStatsSettings indexingStatsSettings,
         SearchStatsSettings searchStatsSettings,
-        MergeMetrics mergeMetrics
+        MergeMetrics mergeMetrics,
+        MetricHandler<StoreMetrics> metricHandler
     ) {
         super(indexSettings);
         assert indexCreationContext != IndexCreationContext.RELOAD_ANALYZERS
@@ -297,6 +301,7 @@ public IndexService(
         this.indexingStatsSettings = indexingStatsSettings;
         this.searchStatsSettings = searchStatsSettings;
         this.mergeMetrics = mergeMetrics;
+        this.metricHandler = metricHandler;
         updateFsyncTaskIfNecessary();
     }
 
@@ -559,7 +564,8 @@ public synchronized IndexShard createShard(
                 directory,
                 lock,
                 new StoreCloseListener(shardId, () -> eventListener.onStoreClosed(shardId)),
-                this.indexSettings.getIndexSortConfig().hasIndexSort()
+                this.indexSettings.getIndexSortConfig().hasIndexSort(),
+                metricHandler
             );
             eventListener.onStoreCreated(shardId);
             indexShard = new IndexShard(

server/src/main/java/org/elasticsearch/index/store/Metric.java

@@ -0,0 +1,13 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+public class Metric {
+}

server/src/main/java/org/elasticsearch/index/store/MetricHandler.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+import java.util.function.Function;
+
+public interface MetricHandler<M extends MetricHandler.Metrics> {
+    static <M extends MetricHandler.Metrics> MetricHandler<M> noop() {
+        return new MetricHandler<>() {
+            @Override
+            public void add(long amount, Function<M, LongMetricBase> metric) {
+
+            }
+
+            @Override
+            public MetricHandler<M> singleThreaded() {
+                return this;
+            }
+        };
+    }
+
+    interface LongMetricBase {
+        void add(long amount);
+    }
+    interface Metrics {
+
+    }
+    void add(long amount, Function<M, LongMetricBase> metric);
+
+
+    MetricHandler<M> singleThreaded();
+}

server/src/main/java/org/elasticsearch/index/store/SingleThreadMetricHandler.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+import java.util.function.Function;
+
+public class SingleThreadMetricHandler<M extends MetricHandler.Metrics> implements MetricHandler<M> {
+    private final ThreadLocalMetricHandler<M> delegate;
+    private Thread owner;
+    private M current;
+
+    public SingleThreadMetricHandler(ThreadLocalMetricHandler<M> delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void add(long amount, Function<M, LongMetricBase> metric) {
+        metric.apply(instance()).add(amount);
+    }
+
+    @Override
+    public MetricHandler<M> singleThreaded() {
+        return this;
+    }
+
+    private M instance() {
+        Thread thread = Thread.currentThread();
+        if (owner == thread) {
+            return current;
+        } else {
+            current = delegate.instance();
+            owner = thread;
+            return current;
+        }
+    }
+}

server/src/main/java/org/elasticsearch/index/store/Store.java

@@ -183,10 +183,22 @@ public Store(
         ShardLock shardLock,
         OnClose onClose,
         boolean hasIndexSort
+    ) {
+        this(shardId, indexSettings, directory, shardLock, onClose, hasIndexSort, new ThreadLocalMetricHandler<>(StoreMetrics::new));
+    }
+
+    public Store(
+        ShardId shardId,
+        IndexSettings indexSettings,
+        Directory directory,
+        ShardLock shardLock,
+        OnClose onClose,
+        boolean hasIndexSort,
+        MetricHandler<StoreMetrics> metricHandler
     ) {
         super(shardId, indexSettings);
         this.directory = new StoreDirectory(
-            byteSizeDirectory(directory, indexSettings, logger),
+            byteSizeDirectory(new StoreMetricsDirectory(directory, metricHandler), indexSettings, logger),
             Loggers.getLogger("index.store.deletes", shardId)
         );
         this.shardLock = shardLock;

server/src/main/java/org/elasticsearch/index/store/StoreMetricIndexInput.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+import org.apache.lucene.store.FilterIndexInput;
+import org.apache.lucene.store.IndexInput;
+
+import java.io.IOException;
+
+public class StoreMetricIndexInput extends FilterIndexInput {
+    private final MetricHandler<? extends StoreMetrics> metricHandler;
+
+    public StoreMetricIndexInput(String resourceDescription, IndexInput in, MetricHandler<? extends StoreMetrics> metricHandler) {
+        super(resourceDescription, in);
+        this.metricHandler = metricHandler;
+    }
+
+    @Override
+    public byte readByte() throws IOException {
+        byte result = super.readByte();
+        metricHandler.add(1, StoreMetrics::getBytesReadMetric);
+        return result;
+    }
+    @Override
+    public void readBytes(byte[] b, int offset, int len) throws IOException {
+        super.readBytes(b, offset, len);
+        metricHandler.add(len, StoreMetrics::getBytesReadMetric);
+    }
+}

server/src/main/java/org/elasticsearch/index/store/StoreMetrics.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+public class StoreMetrics implements MetricHandler.Metrics {
+    private long bytesRead;
+    private final MetricHandler.LongMetricBase bytesReadMetric = new MetricHandler.LongMetricBase() {
+        @Override
+        public void add(long amount) {
+            bytesRead += amount;
+        }
+    };
+
+    public MetricHandler.LongMetricBase getBytesReadMetric() {
+      return bytesReadMetric;
+    }
+
+    public long getBytesRead() {
+        return bytesRead;
+    }
+
+    public StoreMetrics snapshot() {
+        StoreMetrics metrics = new StoreMetrics();
+        metrics.bytesRead = bytesRead;
+        return metrics;
+    }
+}

server/src/main/java/org/elasticsearch/index/store/StoreMetricsDirectory.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.FilterDirectory;
+import org.apache.lucene.store.IOContext;
+import org.apache.lucene.store.IndexInput;
+
+import java.io.IOException;
+
+public class StoreMetricsDirectory extends FilterDirectory {
+    private final MetricHandler<? extends StoreMetrics> metricHandler;
+
+    public StoreMetricsDirectory(Directory in, MetricHandler<? extends StoreMetrics> metricHandler) {
+        super(in);
+        this.metricHandler = metricHandler;
+    }
+
+    @Override
+    public IndexInput openInput(String name, IOContext context) throws IOException {
+        return new StoreMetricIndexInput(name, super.openInput(name, context), metricHandler.singleThreaded());
+    }
+}

server/src/main/java/org/elasticsearch/index/store/ThreadLocalMetricHandler.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.store;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+public class ThreadLocalMetricHandler<M extends MetricHandler.Metrics>  implements MetricHandler<M> {
+    private final Supplier<? extends M> metricsSupplier;
+    private final ThreadLocal<M>  threadLocal = new ThreadLocal<>() {
+        @Override
+        protected M initialValue() {
+            return metricsSupplier.get();
+        }
+    };
+
+    public ThreadLocalMetricHandler(Supplier<? extends M> metricsSupplier) {
+        this.metricsSupplier = metricsSupplier;
+    }
+
+    @Override
+    public void add(long amount, Function<M, LongMetricBase> metric) {
+        metric.apply(instance()).add(amount);
+    }
+
+    @Override
+    public MetricHandler<M> singleThreaded() {
+        return new SingleThreadMetricHandler<>(this);
+    }
+
+    M instance() {
+        return threadLocal.get();
+    }
+}