Merge pull request #4 from lukas-vlcek/main

Skip check of initial indices if they have been rolled-over already

gradle.properties

@@ -1,2 +1,2 @@
 group = org.elasticsearch.plugin.ingest
-version = 6.8.1.0-SNAPSHOT
+version = 6.8.1.0

src/main/java/org/elasticsearch/ingest/openshift/OpenshiftIndexProcessor.java

@@ -16,8 +16,6 @@
 
 package org.elasticsearch.ingest.openshift;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.metadata.AliasOrIndex;
@@ -42,7 +40,6 @@
  * Every instance of this plugin keeps its own lookup table of latest indices/aliases.
  */
 public final class OpenshiftIndexProcessor extends AbstractProcessor {
-    private static final Logger logger = LogManager.getLogger(OpenshiftIndexProcessor.class);
 
     public static final String TYPE = "openshift-ingestion-processor";
 

src/main/java/org/elasticsearch/ingest/openshift/OpenshiftIndicesUtil.java

@@ -28,21 +28,21 @@
 public abstract class OpenshiftIndicesUtil {
 
     /**
-     * Replaces ending "-write" with "-00001".
+     * Does trim() and replaces "-write" suffix with "-00001".
      * @param aliasName assume write-alias
      * @return initial index name
      */
     public static String generateInitialIndexName(final String aliasName) {
-        return aliasName.replaceAll("-write$", "-000001");
+        return aliasName.trim().replaceAll("-write$", "-000001");
     }
 
     /**
-     * Replaces ending "-00001" with "-write".
+     * Does trim() and replaces "-00001" suffix with "-write".
      * @param index assume initialIndex
      * @return write-alias
      */
     public static String generateWriteAliasName(final String index) {
-        return index.replaceAll("-000001$", "-write");
+        return index.trim().replaceAll("-000001$", "-write");
     }
 
     public static boolean isInitialIndex(final String index) {
@@ -74,7 +74,9 @@ public static List<String> getInitialIndicesWithoutWriteAlias(final Map<String,
                             if (!isInitialIndex(imd.getIndex().getName())) return false;
                             if (!hasDataModelPrefix(imd.getIndex().getName())) return false;
                             for (ObjectCursor<AliasMetaData> md: imd.getAliases().values()) {
-                                if (md.value.writeIndex()) {
+                                // Alias metadata can be undefined because the Builder does not require all MetaData to
+                                // be specified. See <org.elasticsearch.cluster.metadata.AliasMetaData.Builder>.
+                                if (md.value.writeIndex() != null && md.value.writeIndex()) {
                                     return false;
                                 }
                             }

src/main/java/org/elasticsearch/ingest/openshift/OpenshiftIngestPlugin.java

@@ -24,6 +24,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest.AliasActions;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -123,26 +124,34 @@ Any modification of index metadata (including adding index aliases) is done on t
                     for (String index: indices) {
 
                         String writeAlias = generateWriteAliasName(index);
-                        if (!writeAlias.trim().isEmpty()) {
+                        // Initial indices that were already rolled-over will not have write alias. We need to skip them.
+                        // In other words the writeAlias already exists (perhaps pointed to "-000002" index or older).
+                        if (!writeAlias.isEmpty() && !latestAliasAndIndicesLookup.containsKey(writeAlias)) {
                             iarb.addAliasAction(AliasActions.add()
                                     .index(index)
                                     .alias(writeAlias)
                                     .writeIndex(true));
+                            logger.trace("Prepared write index alias {} request for index {}", writeAlias, index);
                         }
                     }
 
-                    client.admin().indices().aliases(iarb.request(), new ActionListener<AcknowledgedResponse>() {
-                        @Override
-                        public void onResponse(AcknowledgedResponse acknowledgedResponse) {
-                            logger.debug("Write aliases added for the following indices: {}", indices);
-                        }
-
-                        @Override
-                        public void onFailure(Exception e) {
-                            // TODO[lvlcek]: If there is an exception like "Alias already exists" then we need to ignore it
-                            logger.info("Error occurred when adding write aliases for the following indices: {}. {}", indices, e);
-                        }
-                    });
+                    IndicesAliasesRequest iar = iarb.request();
+                    if (!iar.getAliasActions().isEmpty()) {
+                        client.admin().indices().aliases(iar, new ActionListener<AcknowledgedResponse>() {
+                            @Override
+                            public void onResponse(AcknowledgedResponse acknowledgedResponse) {
+                                logger.debug("Write aliases added for the following indices: {}", indices);
+                            }
+
+                            @Override
+                            public void onFailure(Exception e) {
+                                // TODO[lvlcek]: If there is an exception like "Alias already exists" then we can ignore it.
+                                // However, if the index create request fails because the client does not have appropriate credentials
+                                // or any other serious reason then we need to escalate it.
+                                logger.warn("Error occurred when adding write aliases for the following indices: {}. {}", indices, e);
+                            }
+                        });
+                    }
                 }
             }
         }

src/test/java/org/elasticsearch/ingest/openshift/OpenshiftIndicesUtilTests.java

@@ -39,12 +39,16 @@ public void testGenerateInitialIndexName() {
         assertEquals("app-000001", generateInitialIndexName("app-write"));
         assertEquals("alias-write-000001", generateInitialIndexName("alias-write-write"));
         assertEquals("-000001", generateInitialIndexName("-write"));
+
+        assertEquals("app-000001", generateInitialIndexName("   app-write  "));
     }
 
     public void testGenerateWriteAliasName() {
         assertEquals("app-write", generateWriteAliasName("app-000001"));
         assertEquals("foo-write-write", generateWriteAliasName("foo-write-000001"));
         assertEquals("-write", generateWriteAliasName("-000001"));
+
+        assertEquals("app-write", generateWriteAliasName("   app-000001  "));
     }
 
     public void testDataModelPrefix() {
@@ -90,10 +94,17 @@ public void testIndicesWithoutWriteIndex() {
         im.put("app-index3-000003", new AliasOrIndex.Index(createIndexMetaData("app-index3-000003",
                 new AliasInfo("app-alias1", true))));
 
+        // We also test index with alias that has no metadata, just the name.
+        // Aliases without any additional metadata can lead to NPE when client is iterating over them and does
+        // not check nullity of metadata when accessing it.
+        im.put("app-index4-000001", new AliasOrIndex.Index(createIndexMetaData("app-index4-000001",
+                new AliasInfo("app-alias1"))));
+
         indices = OpenshiftIndicesUtil.getInitialIndicesWithoutWriteAlias(im);
 
-        assertEquals(indices.size(), 1);
+        assertEquals(indices.size(), 2);
         assertTrue(indices.contains("app-index1-000001"));
+        assertTrue(indices.contains("app-index4-000001"));
     }
 
     private AliasMetaData createAliasMetaData(String alias) {
@@ -110,15 +121,22 @@ private IndexMetaData createIndexMetaData(String index, AliasInfo ... alias) {
         // fails: Throwable #1: java.lang.IllegalArgumentException: must specify numberOfShards for index [___]
 
         for (AliasInfo ai : alias) {
-            imBuilder.putAlias(AliasMetaData.builder(ai.alias).writeIndex(ai.writeAlias).build());
+            AliasMetaData.Builder amdt = AliasMetaData.builder(ai.alias);
+            if (ai.writeAlias != null) {
+                amdt.writeIndex(ai.writeAlias);
+            }
+            imBuilder.putAlias(amdt.build());
         }
 
         return imBuilder.build();
     }
 
     private class AliasInfo {
-        public final String alias;
-        public final Boolean writeAlias;
+        private String alias;
+        private Boolean writeAlias;
+        AliasInfo(String alias) {
+            this.alias = alias;
+        }
         AliasInfo(String alias, Boolean writeAlias) {
             this.alias = alias;
             this.writeAlias = writeAlias;

src/test/resources/rest-api-spec/test/ingest/20_define_pipeline.yml

@@ -6,7 +6,10 @@
       cluster.put_settings:
         body:
           transient:
-            logger.org.elasticsearch.ingest.openshift: "TRACE"
+            action.auto_create_index: "-*-write,+*"
+            logger:
+              org.elasticsearch.ingest.openshift: "TRACE"
+              org.elasticsearch.action: "DEBUG"
         flat_settings: true
   - match: { acknowledged: true }
 

src/test/resources/rest-api-spec/test/ingest/30_rollover.yml

@@ -6,7 +6,10 @@
       cluster.put_settings:
         body:
           transient:
-            logger.org.elasticsearch.ingest.openshift: "TRACE"
+            action.auto_create_index: "-*-write,+*"
+            logger:
+              org.elasticsearch.ingest.openshift: "TRACE"
+              org.elasticsearch.action: "DEBUG"
         flat_settings: true
   - match: { acknowledged: true }