Waiting for index range calculation before switching deflector alias.

graylog2-server/src/main/java/org/graylog2/Configuration.java

@@ -156,6 +156,9 @@ public class Configuration extends BaseConfiguration {
     @Parameter(value = "index_ranges_cleanup_interval", validator = PositiveDurationValidator.class)
     private Duration indexRangesCleanupInterval = Duration.hours(1L);
 
+    @Parameter(value = "deflector_index_read_only_timeout", validator = PositiveDurationValidator.class)
+    private Duration deflectorIndexReadOnlyTimeout = Duration.hours(1L);
+
     @Parameter(value = "trusted_proxies", converter = IPSubnetConverter.class)
     private Set<IpSubnet> trustedProxies = Collections.emptySet();
 

graylog2-server/src/main/java/org/graylog2/indexer/Deflector.java

@@ -16,10 +16,13 @@
  */
 package org.graylog2.indexer;
 
+import com.github.joschi.jadconfig.util.Duration;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.indices.InvalidAliasNameException;
 import org.graylog2.indexer.indices.Indices;
 import org.graylog2.indexer.ranges.CreateNewSingleIndexRangeJob;
+import org.graylog2.indexer.ranges.IndexRange;
+import org.graylog2.indexer.ranges.IndexRangeService;
 import org.graylog2.shared.system.activities.Activity;
 import org.graylog2.shared.system.activities.ActivityWriter;
 import org.graylog2.system.jobs.SystemJob;
@@ -34,7 +37,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -58,6 +63,8 @@ public class Deflector { // extends Ablenkblech
     private final SystemJobManager systemJobManager;
     private final ActivityWriter activityWriter;
     private final CreateNewSingleIndexRangeJob.Factory createNewSingleIndexRangeJobFactory;
+    private final IndexRangeService indexRangeService;
+    private final Duration deflectorIndexReadOnlyTimeout;
     private final String indexPrefix;
     private final String deflectorName;
     private final Indices indices;
@@ -71,13 +78,17 @@ public Deflector(final SystemJobManager systemJobManager,
                      final ActivityWriter activityWriter,
                      final SetIndexReadOnlyJob.Factory indexReadOnlyJobFactory,
                      final CreateNewSingleIndexRangeJob.Factory createNewSingleIndexRangeJobFactory,
-                     final Indices indices) {
+                     final Indices indices,
+                     final IndexRangeService indexRangeService,
+                     @Named("deflector_index_read_only_timeout") final Duration deflectorIndexReadOnlyTimeout) {
         this.indexPrefix = indexPrefix;
 
         this.systemJobManager = systemJobManager;
         this.activityWriter = activityWriter;
         this.indexReadOnlyJobFactory = indexReadOnlyJobFactory;
         this.createNewSingleIndexRangeJobFactory = createNewSingleIndexRangeJobFactory;
+        this.indexRangeService = indexRangeService;
+        this.deflectorIndexReadOnlyTimeout = deflectorIndexReadOnlyTimeout;
 
         this.deflectorName = buildName(indexPrefix);
         this.indices = indices;
@@ -144,6 +155,7 @@ public void cycle() {
         ClusterHealthStatus healthStatus = indices.waitForRecovery(newTarget);
         LOG.debug("Health status of index <{}>: {}", newTarget, healthStatus);
 
+        addDeflectorIndexRange(newTarget);
         LOG.info("Done!");
 
         // Point deflector to new index.
@@ -156,29 +168,36 @@ public void cycle() {
             activity.setMessage("Cycled deflector from <none> to <" + newTarget + ">");
         } else {
             // Re-pointing from existing old index to the new one.
+            LOG.debug("Now switching over deflector alias.");
             pointTo(newTarget, oldTarget);
-            addSingleIndexRanges(oldTarget);
 
             // perform these steps after a delay, so we don't race with indexing into the alias
             // it can happen that an index request still writes to the old deflector target, while we cycled it above.
             // setting the index to readOnly would result in ClusterBlockExceptions in the indexing request.
             // waiting 30 seconds to perform the background task should completely get rid of these errors.
             final SystemJob makeReadOnlyJob = indexReadOnlyJobFactory.create(oldTarget);
             try {
-                systemJobManager.submitWithDelay(makeReadOnlyJob, 30, TimeUnit.SECONDS);
-            } catch (SystemJobConcurrencyException e) {
+                final SystemJobManager.ScheduleResult scheduleResult = systemJobManager.submitWithDelayForResult(makeReadOnlyJob, 30, TimeUnit.SECONDS);
+                if (scheduleResult != null) {
+                    scheduleResult.getFuture().get(deflectorIndexReadOnlyTimeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+                }
+            } catch (SystemJobConcurrencyException | InterruptedException | ExecutionException | TimeoutException e) {
                 LOG.error("Cannot set index <" + oldTarget + "> to read only. It won't be optimized.", e);
             }
+            addSingleIndexRanges(oldTarget);
             activity.setMessage("Cycled deflector from <" + oldTarget + "> to <" + newTarget + ">");
         }
 
-        addSingleIndexRanges(newTarget);
-
         LOG.info("Done!");
 
         activityWriter.write(activity);
     }
 
+    private void addDeflectorIndexRange(String newTarget) {
+        final IndexRange deflectorRange = indexRangeService.createForDeflector(newTarget);
+        indexRangeService.save(deflectorRange);
+    }
+
     public int getNewestTargetNumber() throws NoTargetIndexException {
         final Set<String> indexNames = indices.getIndexNamesAndAliases(getDeflectorWildcard()).keySet();
 
@@ -267,14 +286,16 @@ public void pointTo(final String newIndex) {
         indices.cycleAlias(getName(), newIndex);
     }
 
-    private void addSingleIndexRanges(String indexName) {
+    @Nullable
+    private SystemJobManager.ScheduleResult addSingleIndexRanges(String indexName) {
         try {
-            systemJobManager.submit(createNewSingleIndexRangeJobFactory.create(this, indexName));
+            return systemJobManager.submitForResult(createNewSingleIndexRangeJobFactory.create(this, indexName));
         } catch (SystemJobConcurrencyException e) {
             final String msg = "Could not calculate index ranges for index " + indexName + " after cycling deflector: Maximum concurrency of job is reached.";
             activityWriter.write(new Activity(msg, Deflector.class));
             LOG.error(msg, e);
         }
+        return null;
     }
 
     @Nullable

graylog2-server/src/main/java/org/graylog2/indexer/IndexHelper.java

@@ -106,17 +106,6 @@ public static Set<IndexRange> determineAffectedIndicesWithRanges(IndexRangeServi
             indices.add(indexRange);
         }
 
-        // Always include the deflector target
-        final String targetIndex = deflector.getCurrentActualTargetIndex();
-        if (targetIndex != null) {
-            try {
-                final IndexRange deflectorIndexRange = indexRangeService.get(targetIndex);
-                indices.add(deflectorIndexRange);
-            } catch (NotFoundException e) {
-                LOG.warn("Couldn't find latest deflector target index", e);
-            }
-        }
-
         return indices.build();
     }
 }

graylog2-server/src/main/java/org/graylog2/indexer/ranges/EsIndexRangeService.java

@@ -178,6 +178,11 @@ public IndexRange calculateRange(String index) {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public IndexRange createForDeflector(String index) {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public void save(IndexRange indexRange) {
         throw new UnsupportedOperationException();

graylog2-server/src/main/java/org/graylog2/indexer/ranges/IndexRangeService.java

@@ -31,4 +31,5 @@ public interface IndexRangeService {
     void save(IndexRange indexRange);
 
     IndexRange calculateRange(String index);
+    IndexRange createForDeflector(String index);
 }

graylog2-server/src/main/java/org/graylog2/indexer/ranges/LegacyMongoIndexRangeService.java

@@ -108,6 +108,11 @@ public IndexRange calculateRange(String index) {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public IndexRange createForDeflector(String index) {
+        throw new UnsupportedOperationException();
+    }
+
     public int delete(String index) {
         return destroy(new BasicDBObject(FIELD_INDEX, index), COLLECTION_NAME);
     }

graylog2-server/src/main/java/org/graylog2/indexer/ranges/MongoIndexRangeService.java

@@ -66,27 +66,27 @@ public MongoIndexRangeService(MongoConnection mongoConnection,
                                   EventBus eventBus) {
         this.indices = indices;
         this.collection = JacksonDBCollection.wrap(
-                mongoConnection.getDatabase().getCollection(COLLECTION_NAME),
-                MongoIndexRange.class,
-                ObjectId.class,
-                objectMapperProvider.get());
+            mongoConnection.getDatabase().getCollection(COLLECTION_NAME),
+            MongoIndexRange.class,
+            ObjectId.class,
+            objectMapperProvider.get());
 
         // This sucks. We need to bridge Elasticsearch's and our own Guice injector.
         IndexChangeMonitor.setEventBus(eventBus);
         eventBus.register(this);
 
         collection.createIndex(new BasicDBObject(MongoIndexRange.FIELD_INDEX_NAME, 1));
         collection.createIndex(BasicDBObjectBuilder.start()
-                .add(MongoIndexRange.FIELD_BEGIN, 1)
-                .add(MongoIndexRange.FIELD_END, 1)
-                .get());
+            .add(MongoIndexRange.FIELD_BEGIN, 1)
+            .add(MongoIndexRange.FIELD_END, 1)
+            .get());
     }
 
     @Override
     public IndexRange get(String index) throws NotFoundException {
         final DBQuery.Query query = DBQuery.and(
-                DBQuery.notExists("start"),
-                DBQuery.is(IndexRange.FIELD_INDEX_NAME, index));
+            DBQuery.notExists("start"),
+            DBQuery.is(IndexRange.FIELD_INDEX_NAME, index));
         final MongoIndexRange indexRange = collection.findOne(query);
         if (indexRange == null) {
             throw new NotFoundException("Index range for index <" + index + "> not found.");
@@ -98,11 +98,18 @@ public IndexRange get(String index) throws NotFoundException {
     @Override
     public SortedSet<IndexRange> find(DateTime begin, DateTime end) {
         final DBCursor<MongoIndexRange> indexRanges = collection.find(
+            DBQuery.or(
                 DBQuery.and(
-                        DBQuery.notExists("start"),  // "start" has been used by the old index ranges in MongoDB
-                        DBQuery.lessThanEquals(IndexRange.FIELD_BEGIN, end.getMillis()),
-                        DBQuery.greaterThanEquals(IndexRange.FIELD_END, begin.getMillis())
+                    DBQuery.notExists("start"),  // "start" has been used by the old index ranges in MongoDB
+                    DBQuery.lessThanEquals(IndexRange.FIELD_BEGIN, end.getMillis()),
+                    DBQuery.greaterThanEquals(IndexRange.FIELD_END, begin.getMillis())
+                ),
+                DBQuery.and(
+                    DBQuery.notExists("start"),  // "start" has been used by the old index ranges in MongoDB
+                    DBQuery.lessThanEquals(IndexRange.FIELD_BEGIN, 0L),
+                    DBQuery.greaterThanEquals(IndexRange.FIELD_END, 0L)
                 )
+            )
         );
 
         return ImmutableSortedSet.copyOf(IndexRange.COMPARATOR, (Iterator<? extends IndexRange>) indexRanges);
@@ -125,6 +132,11 @@ public IndexRange calculateRange(String index) {
         return MongoIndexRange.create(index, stats.min(), stats.max(), now, duration);
     }
 
+    @Override
+    public IndexRange createForDeflector(String index) {
+        return MongoIndexRange.create(index, new DateTime(0L), new DateTime(0L), DateTime.now(), 0);
+    }
+
     @Override
     public void save(IndexRange indexRange) {
         collection.remove(DBQuery.in(IndexRange.FIELD_INDEX_NAME, indexRange.indexName()));
@@ -158,10 +170,10 @@ public void handleIndexReopening(IndicesReopenedEvent event) {
             indices.waitForRecovery(index);
 
             final Retryer<IndexRange> retryer = RetryerBuilder.<IndexRange>newBuilder()
-                    .retryIfException(input -> !(input instanceof IndexClosedException))
-                    .withWaitStrategy(WaitStrategies.exponentialWait())
-                    .withStopStrategy(StopStrategies.stopAfterDelay(5, TimeUnit.MINUTES))
-                    .build();
+                .retryIfException(input -> !(input instanceof IndexClosedException))
+                .withWaitStrategy(WaitStrategies.exponentialWait())
+                .withStopStrategy(StopStrategies.stopAfterDelay(5, TimeUnit.MINUTES))
+                .build();
 
             final IndexRange indexRange;
             try {

graylog2-server/src/main/java/org/graylog2/indexer/ranges/RebuildIndexRangesJob.java

@@ -26,6 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 
 public class RebuildIndexRangesJob extends SystemJob {
@@ -85,7 +86,11 @@ public void execute() {
         indicesToCalculate = indices.length;
 
         Stopwatch sw = Stopwatch.createStarted();
+        final String deflectorIndexName = deflector.getName();
         for (String index : indices) {
+            if (deflectorIndexName.equals(index)) {
+                continue;
+            }
             if (cancelRequested) {
                 info("Stop requested. Not calculating next index range, not updating ranges.");
                 sw.stop();

graylog2-server/src/main/java/org/graylog2/system/jobs/SystemJobManager.java

@@ -26,17 +26,39 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
 import javax.inject.Inject;
+import javax.validation.constraints.Null;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
 import static com.codahale.metrics.MetricRegistry.name;
 
 public class SystemJobManager {
+
+    public class ScheduleResult {
+        private final String jobId;
+        private final ScheduledFuture<?> future;
+
+        ScheduleResult(final String jobId, final ScheduledFuture<?> future) {
+            this.jobId = jobId;
+            this.future = future;
+        }
+
+        public String getJobId() {
+            return jobId;
+        }
+
+        public ScheduledFuture<?> getFuture() {
+            return future;
+        }
+    }
+
     private static final Logger LOG = LoggerFactory.getLogger(SystemJobManager.class);
     private static final int THREAD_POOL_SIZE = 15;
 
@@ -60,10 +82,18 @@ private ScheduledExecutorService executorService(final MetricRegistry metricRegi
     }
 
     public String submit(final SystemJob job) throws SystemJobConcurrencyException {
-        return submitWithDelay(job, 0, TimeUnit.SECONDS);
+        return submitForResult(job).getJobId();
+    }
+
+    public ScheduleResult submitForResult(final SystemJob job) throws SystemJobConcurrencyException {
+        return submitWithDelayForResult(job, 0, TimeUnit.SECONDS);
     }
 
     public String submitWithDelay(final SystemJob job, final long delay, TimeUnit timeUnit) throws SystemJobConcurrencyException {
+        return submitWithDelayForResult(job, delay, timeUnit).getJobId();
+    }
+
+    public ScheduleResult submitWithDelayForResult(final SystemJob job, final long delay, TimeUnit timeUnit) throws SystemJobConcurrencyException {
         // for immediate jobs, check allowed concurrency right now
         if (delay == 0) {
             checkAllowedConcurrency(job);
@@ -74,7 +104,7 @@ public String submitWithDelay(final SystemJob job, final long delay, TimeUnit ti
         job.setId(new UUID().toString());
         jobs.put(job.getId(), job);
 
-        executor.schedule(new Runnable() {
+        final ScheduledFuture<?> future = executor.schedule(new Runnable() {
             @Override
             public void run() {
                 try {
@@ -89,7 +119,7 @@ public void run() {
                     x.stop();
 
                     final String msg = "SystemJob <" + job.getId() + "> [" + jobClass + "] finished in " + x.elapsed(
-                            TimeUnit.MILLISECONDS) + "ms.";
+                        TimeUnit.MILLISECONDS) + "ms.";
                     LOG.info(msg);
                     activityWriter.write(new Activity(msg, SystemJobManager.class));
                 } catch (SystemJobConcurrencyException ignored) {
@@ -102,7 +132,7 @@ public void run() {
         }, delay, timeUnit);
 
         LOG.info("Submitted SystemJob <{}> [{}]", job.getId(), jobClass);
-        return job.getId();
+        return new ScheduleResult(job.getId(), future);
     }
 
     protected void checkAllowedConcurrency(SystemJob job) throws SystemJobConcurrencyException {