Batch translog sync/upload per x ms for remote-backed indexes

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -59,6 +59,7 @@
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.xcontent.ToXContent;
 import org.opensearch.common.xcontent.ToXContentFragment;
 import org.opensearch.common.xcontent.XContentBuilder;
@@ -301,6 +302,8 @@ public Iterator<Setting<?>> settings() {
 
  public static final String SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY = "index.remote_store.translog.repository";
 
+ public static final String SETTING_REMOTE_TRANSLOG_BUFFER_INTERVAL = "index.remote_store.translog.buffer_interval";
+
  /**
  * Used to specify if the index data should be persisted in the remote store.
  */
@@ -446,6 +449,45 @@ public Iterator<Setting<?>> settings() {
  Property.Final
  );
 
+ public static final Setting<TimeValue> INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING = Setting.timeSetting(
+ SETTING_REMOTE_TRANSLOG_BUFFER_INTERVAL,
+ TimeValue.timeValueMillis(100),
+ TimeValue.timeValueMillis(50),
+ new Setting.Validator<>() {
+
+ @Override
+ public void validate(final TimeValue value) {}
+
+ @Override
+ public void validate(final TimeValue value, final Map<Setting<?>, Object> settings) {
+ if (value == null) {
+ throw new IllegalArgumentException(
+ "Setting " + SETTING_REMOTE_TRANSLOG_BUFFER_INTERVAL + " should be provided with a valid time value"
+ );
+ } else {
+ final Boolean isRemoteTranslogStoreEnabled = (Boolean) settings.get(INDEX_REMOTE_TRANSLOG_STORE_ENABLED_SETTING);
+ if (isRemoteTranslogStoreEnabled == null || isRemoteTranslogStoreEnabled == false) {
+ throw new IllegalArgumentException(
+ "Setting "
+ + SETTING_REMOTE_TRANSLOG_BUFFER_INTERVAL
+ + " can only be set when "
+ + SETTING_REMOTE_TRANSLOG_STORE_ENABLED
+ + " is set to true"
+ );
+ }
+ }
+ }
+
+ @Override
+ public Iterator<Setting<?>> settings() {
+ final List<Setting<?>> settings = Collections.singletonList(INDEX_REMOTE_TRANSLOG_STORE_ENABLED_SETTING);
+ return settings.iterator();
+ }
+ },
+ Property.IndexScope,
+ Property.Final
+ );
+
  public static final String SETTING_AUTO_EXPAND_REPLICAS = "index.auto_expand_replicas";
  public static final Setting<AutoExpandReplicas> INDEX_AUTO_EXPAND_REPLICAS_SETTING = AutoExpandReplicas.SETTING;
 

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -227,7 +227,8 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
  IndexMetadata.INDEX_REMOTE_STORE_ENABLED_SETTING,
  IndexMetadata.INDEX_REMOTE_STORE_REPOSITORY_SETTING,
  IndexMetadata.INDEX_REMOTE_TRANSLOG_STORE_ENABLED_SETTING,
- IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING
+ IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING,
+ IndexMetadata.INDEX_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING
  ),
  FeatureFlags.SEARCHABLE_SNAPSHOT,
  List.of(

server/src/main/java/org/opensearch/common/settings/Setting.java

@@ -2173,6 +2173,23 @@ public static Setting<TimeValue> timeSetting(String key, Setting<TimeValue> fall
  return new Setting<>(key, fallbackSetting, (s) -> TimeValue.parseTimeValue(s, key), properties);
  }
 
+ public static Setting<TimeValue> timeSetting(
+ String key,
+ TimeValue defaultValue,
+ TimeValue minValue,
+ Validator<TimeValue> validator,
+ Property... properties
+ ) {
+ final SimpleKey simpleKey = new SimpleKey(key);
+ return new Setting<>(
+ simpleKey,
+ s -> defaultValue.getStringRep(),
+ minTimeValueParser(key, minValue, isFiltered(properties)),
+ validator,
+ properties
+ );
+ }
+
  public static Setting<TimeValue> timeSetting(
  String key,
  Setting<TimeValue> fallBackSetting,

server/src/main/java/org/opensearch/common/util/concurrent/AsyncIOProcessor.java

@@ -57,6 +57,7 @@ public abstract class AsyncIOProcessor<Item> {
  private final ArrayBlockingQueue<Tuple<Item, Consumer<Exception>>> queue;
  private final ThreadContext threadContext;
  private final Semaphore promiseSemaphore = new Semaphore(1);
+ private long lastRunStartTimeInNs;
 
  protected AsyncIOProcessor(Logger logger, int queueSize, ThreadContext threadContext) {
  this.logger = logger;
@@ -67,7 +68,7 @@ protected AsyncIOProcessor(Logger logger, int queueSize, ThreadContext threadCon
  /**
  * Adds the given item to the queue. The listener is notified once the item is processed
  */
- public final void put(Item item, Consumer<Exception> listener) {
+ public void put(Item item, Consumer<Exception> listener) {
  Objects.requireNonNull(item, "item must not be null");
  Objects.requireNonNull(listener, "listener must not be null");
  // the algorithm here tires to reduce the load on each individual caller.
@@ -78,12 +79,7 @@ public final void put(Item item, Consumer<Exception> listener) {
  final boolean promised = promiseSemaphore.tryAcquire();
  if (promised == false) {
  // in this case we are not responsible and can just block until there is space
- try {
- queue.put(new Tuple<>(item, preserveContext(listener)));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- listener.accept(e);
- }
+ addToQueue(item, listener);
  }
 
  // here we have to try to make the promise again otherwise there is a race when a thread puts an entry without making the promise
@@ -104,7 +100,17 @@ public final void put(Item item, Consumer<Exception> listener) {
  }
  }
 
- private void drainAndProcessAndRelease(List<Tuple<Item, Consumer<Exception>>> candidates) {
+ void addToQueue(Item item, Consumer<Exception> listener) {
+ try {
+ queue.put(new Tuple<>(item, preserveContext(listener)));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ listener.accept(e);
+ }
+ }
+
+ void drainAndProcessAndRelease(List<Tuple<Item, Consumer<Exception>>> candidates) {
+ lastRunStartTimeInNs = System.nanoTime();
  Exception exception;
  try {
  queue.drainTo(candidates);
@@ -130,7 +136,7 @@ private Exception processList(List<Tuple<Item, Consumer<Exception>>> candidates)
  return exception;
  }
 
- private void notifyList(List<Tuple<Item, Consumer<Exception>>> candidates, Exception exception) {
+ void notifyList(List<Tuple<Item, Consumer<Exception>>> candidates, Exception exception) {
  for (Tuple<Item, Consumer<Exception>> tuple : candidates) {
  Consumer<Exception> consumer = tuple.v2();
  try {
@@ -141,7 +147,7 @@ private void notifyList(List<Tuple<Item, Consumer<Exception>>> candidates, Excep
  }
  }
 
- private Consumer<Exception> preserveContext(Consumer<Exception> consumer) {
+ Consumer<Exception> preserveContext(Consumer<Exception> consumer) {
  Supplier<ThreadContext.StoredContext> restorableContext = threadContext.newRestorableContext(false);
  return e -> {
  try (ThreadContext.StoredContext ignore = restorableContext.get()) {
@@ -154,4 +160,20 @@ private Consumer<Exception> preserveContext(Consumer<Exception> consumer) {
  * Writes or processes the items out or to disk.
  */
  protected abstract void write(List<Tuple<Item, Consumer<Exception>>> candidates) throws IOException;
+
+ Logger getLogger() {
+ return logger;
+ }
+
+ Semaphore getPromiseSemaphore() {
+ return promiseSemaphore;
+ }
+
+ long getLastRunStartTimeInNs() {
+ return lastRunStartTimeInNs;
+ }
+
+ ArrayBlockingQueue<Tuple<Item, Consumer<Exception>>> getQueue() {
+ return queue;
+ }
 }

server/src/main/java/org/opensearch/common/util/concurrent/BufferedAsyncIOProcessor.java

@@ -0,0 +1,93 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.common.util.concurrent;
+
+import org.apache.logging.log4j.Logger;
+import org.opensearch.common.collect.Tuple;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.threadpool.ThreadPool;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+/**
+ * A variant of {@link AsyncIOProcessor} that allows to batch and buffer processing items at every
+ * {@link BufferedAsyncIOProcessor#bufferInterval} in a separate threadpool.
+ * <p>
+ * Requests are buffered till processor thread calls @{link drainAndProcessAndRelease} after bufferInterval.
+ * If more requests are enqueued between invocations of drainAndProcessAndRelease, another processor thread
+ * gets scheduled. Subsequent requests will get buffered till drainAndProcessAndRelease gets called in this new
+ * processor thread.
+ *
+ * @opensearch.internal
+ */
+public abstract class BufferedAsyncIOProcessor<Item> extends AsyncIOProcessor<Item> {
+
+ private final ThreadPool threadpool;
+ private final TimeValue bufferInterval;
+
+ protected BufferedAsyncIOProcessor(
+ Logger logger,
+ int queueSize,
+ ThreadContext threadContext,
+ ThreadPool threadpool,
+ TimeValue bufferInterval
+ ) {
+ super(logger, queueSize, threadContext);
+ this.threadpool = threadpool;
+ this.bufferInterval = bufferInterval;
+ }
+
+ @Override
+ public void put(Item item, Consumer<Exception> listener) {
+ Objects.requireNonNull(item, "item must not be null");
+ Objects.requireNonNull(listener, "listener must not be null");
+ addToQueue(item, listener);
+ scheduleProcess();
+ }
+
+ private void scheduleProcess() {
+ if (getQueue().isEmpty() == false && getPromiseSemaphore().tryAcquire()) {
+ try {
+ threadpool.schedule(this::process, getBufferInterval(), getBufferRefreshThreadPoolName());
+ } catch (Exception e) {
+ getLogger().error("failed to schedule process");
+ processSchedulingFailure(e);
+ getPromiseSemaphore().release();
+ // This is to make sure that any new items that are added to the queue between processSchedulingFailure
+ // and releasing the semaphore is handled by a subsequent refresh and not starved.
+ scheduleProcess();
+ }
+ }
+ }
+
+ private void processSchedulingFailure(Exception e) {
+ List<Tuple<Item, Consumer<Exception>>> candidates = new ArrayList<>();
+ getQueue().drainTo(candidates);
+ notifyList(candidates, e);
+ }
+
+ private void process() {
+ drainAndProcessAndRelease(new ArrayList<>());
+ scheduleProcess();
+ }
+
+ private TimeValue getBufferInterval() {
+ long timeSinceLastRunStartInNS = System.nanoTime() - getLastRunStartTimeInNs();
+ if (timeSinceLastRunStartInNS >= bufferInterval.getNanos()) {
+ return TimeValue.ZERO;
+ }
+ return TimeValue.timeValueNanos(bufferInterval.getNanos() - timeSinceLastRunStartInNS);
+ }
+
+ protected abstract String getBufferRefreshThreadPoolName();
+
+}

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -586,6 +586,7 @@ public final class IndexSettings {
  private final ReplicationType replicationType;
  private final boolean isRemoteStoreEnabled;
  private final boolean isRemoteTranslogStoreEnabled;
+ private final TimeValue remoteTranslogUploadBufferInterval;
  private final String remoteStoreTranslogRepository;
  private final String remoteStoreRepository;
  private final boolean isRemoteSnapshot;
@@ -753,6 +754,10 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
  isRemoteStoreEnabled = settings.getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false);
  isRemoteTranslogStoreEnabled = settings.getAsBoolean(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_ENABLED, false);
  remoteStoreTranslogRepository = settings.get(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY);
+ remoteTranslogUploadBufferInterval = settings.getAsTime(
+ IndexMetadata.SETTING_REMOTE_TRANSLOG_BUFFER_INTERVAL,
+ TimeValue.timeValueMillis(100)
+ );
  remoteStoreRepository = settings.get(IndexMetadata.SETTING_REMOTE_STORE_REPOSITORY);
  isRemoteSnapshot = IndexModule.Type.REMOTE_SNAPSHOT.match(this.settings);
 
@@ -1135,6 +1140,15 @@ public void setTranslogSyncInterval(TimeValue translogSyncInterval) {
  this.syncInterval = translogSyncInterval;
  }
 
+ /**
+ * Returns the translog sync/upload buffer interval when remote translog store is enabled and index setting
+ * {@code index.translog.durability} is set as {@code request}.
+ * @return the buffer interval.
+ */
+ public TimeValue getRemoteTranslogUploadBufferInterval() {
+ return remoteTranslogUploadBufferInterval;
+ }
+
  /**
  * Returns this interval in which the shards of this index are asynchronously refreshed. {@code -1} means async refresh is disabled.
  */

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -95,6 +95,7 @@
 import org.opensearch.common.util.BigArrays;
 import org.opensearch.common.util.concurrent.AbstractRunnable;
 import org.opensearch.common.util.concurrent.AsyncIOProcessor;
+import org.opensearch.common.util.concurrent.BufferedAsyncIOProcessor;
 import org.opensearch.common.util.concurrent.RunOnce;
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.common.util.set.Sets;
@@ -364,7 +365,13 @@ public IndexShard(
  this.indexSortSupplier = indexSortSupplier;
  this.indexEventListener = indexEventListener;
  this.threadPool = threadPool;
- this.translogSyncProcessor = createTranslogSyncProcessor(logger, threadPool.getThreadContext(), this::getEngine);
+ this.translogSyncProcessor = createTranslogSyncProcessor(
+ logger,
+ threadPool,
+ this::getEngine,
+ indexSettings.isRemoteTranslogStoreEnabled(),
+ indexSettings.getRemoteTranslogUploadBufferInterval()
+ );
  this.mapperService = mapperService;
  this.indexCache = indexCache;
  this.internalIndexingStats = new InternalIndexingStats();
@@ -3808,21 +3815,41 @@ public List<String> getActiveOperations() {
 
  private static AsyncIOProcessor<Translog.Location> createTranslogSyncProcessor(
  Logger logger,
- ThreadContext threadContext,
- Supplier<Engine> engineSupplier
+ ThreadPool threadPool,
+ Supplier<Engine> engineSupplier,
+ boolean bufferAsyncIoProcessor,
+ TimeValue bufferInterval
  ) {
- return new AsyncIOProcessor<Translog.Location>(logger, 1024, threadContext) {
+ ThreadContext threadContext = threadPool.getThreadContext();
+ CheckedConsumer<List<Tuple<Translog.Location, Consumer<Exception>>>, IOException> writeConsumer = candidates -> {
+ try {
+ engineSupplier.get().translogManager().ensureTranslogSynced(candidates.stream().map(Tuple::v1));
+ } catch (AlreadyClosedException ex) {
+ // that's fine since we already synced everything on engine close - this also is conform with the methods
+ // documentation
+ } catch (IOException ex) { // if this fails we are in deep shit - fail the request
+ logger.debug("failed to sync translog", ex);
+ throw ex;
+ }
+ };
+ if (bufferAsyncIoProcessor) {
+ return new BufferedAsyncIOProcessor<>(logger, 102400, threadContext, threadPool, bufferInterval) {
+ @Override
+ protected void write(List<Tuple<Translog.Location, Consumer<Exception>>> candidates) throws IOException {
+ writeConsumer.accept(candidates);
+ }
+
+ @Override
+ protected String getBufferRefreshThreadPoolName() {
+ return ThreadPool.Names.TRANSLOG_SYNC;
+ }
+ };
+ }
+
+ return new AsyncIOProcessor<>(logger, 1024, threadContext) {
  @Override
  protected void write(List<Tuple<Translog.Location, Consumer<Exception>>> candidates) throws IOException {
- try {
- engineSupplier.get().translogManager().ensureTranslogSynced(candidates.stream().map(Tuple::v1));
- } catch (AlreadyClosedException ex) {
- // that's fine since we already synced everything on engine close - this also is conform with the methods
- // documentation
- } catch (IOException ex) { // if this fails we are in deep shit - fail the request
- logger.debug("failed to sync translog", ex);
- throw ex;
- }
+ writeConsumer.accept(candidates);
  }
  };
  }