rename audit index and create an alias for accessing it

Author: Hendrik Muhs

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/transforms/persistence/TransformInternalIndexConstants.java

@@ -29,9 +29,10 @@ public final class TransformInternalIndexConstants {
     public static final String INDEX_NAME_PATTERN = INDEX_PATTERN + "*";
 
     // audit index
-    public static final String AUDIT_TEMPLATE_VERSION = "1";
+    public static final String AUDIT_TEMPLATE_VERSION = "000001";
     public static final String AUDIT_INDEX_PREFIX = ".transform-notifications-";
     public static final String AUDIT_INDEX_PATTERN = AUDIT_INDEX_PREFIX + "*";
+    public static final String AUDIT_INDEX_DEPRECATED = ".data-frame-notifications-1";
     public static final String AUDIT_INDEX_PATTERN_DEPRECATED = ".data-frame-notifications-*";
 
     public static final String AUDIT_INDEX_READ_ALIAS = ".transform-notifications-read";

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/Transform.java

@@ -222,7 +222,8 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
                                                                       transformConfigManager.get(),
                                                                       transformAuditor.get()));
 
-        return Arrays.asList(transformConfigManager.get(), transformAuditor.get(), transformCheckpointService.get());
+        return Arrays.asList(transformConfigManager.get(), transformAuditor.get(), transformCheckpointService.get(),
+                new TransformClusterStateListener(clusterService, client));
     }
 
     @Override

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/TransformClusterStateListener.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.transform;
+
+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.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.gateway.GatewayService;
+import org.elasticsearch.xpack.core.transform.transforms.persistence.TransformInternalIndexConstants;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.elasticsearch.xpack.core.ClientHelper.TRANSFORM_ORIGIN;
+import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
+
+class TransformClusterStateListener implements ClusterStateListener {
+
+    private static final Logger logger = LogManager.getLogger(TransformClusterStateListener.class);
+
+    private final Client client;
+    private final AtomicBoolean isIndexCreationInProgress = new AtomicBoolean(false);
+
+    TransformClusterStateListener(ClusterService clusterService, Client client) {
+        this.client = client;
+        clusterService.addListener(this);
+        logger.debug("Created TransformClusterStateListener");
+    }
+
+    @Override
+    public void clusterChanged(ClusterChangedEvent event) {
+        if (event.state().blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
+            // Wait until the gateway has recovered from disk.
+            return;
+        }
+
+        // The atomic flag prevents multiple simultaneous attempts to run alias creation
+        // if there is a flurry of cluster state updates in quick succession
+        if (event.localNodeMaster() && isIndexCreationInProgress.compareAndSet(false, true)) {
+            createAuditAliasForDataFrameBWC(event.state(), client, ActionListener.wrap(
+                r -> {
+                    isIndexCreationInProgress.set(false);
+                    if (r) {
+                        logger.info("Created alias for deprecated data frame notifications index");
+                    } else {
+                        logger.debug("Skipped creating alias for deprecated data frame notifications index");
+                    }
+                },
+                e -> {
+                    isIndexCreationInProgress.set(false);
+                    logger.error("Error creating alias for deprecated data frame notifications index", e);
+                }));
+        }
+    }
+
+    private static void createAuditAliasForDataFrameBWC(ClusterState state, Client client, final ActionListener<Boolean> finalListener) {
+
+        // check if alias already exists
+        if (state.getMetaData().getAliasAndIndexLookup().containsKey(TransformInternalIndexConstants.AUDIT_INDEX_READ_ALIAS)) {
+            finalListener.onResponse(false);
+            return;
+        }
+
+        // check if old audit index exists, no need to create the alias if it does not
+        if (state.getMetaData().getAliasAndIndexLookup().containsKey(TransformInternalIndexConstants.AUDIT_INDEX_DEPRECATED)) {
+            finalListener.onResponse(false);
+            return;
+        }
+
+        final IndicesAliasesRequest request = client.admin().indices().prepareAliases()
+                .addAlias(TransformInternalIndexConstants.AUDIT_INDEX_DEPRECATED, TransformInternalIndexConstants.AUDIT_INDEX_READ_ALIAS)
+                .request();
+
+        executeAsyncWithOrigin(client.threadPool().getThreadContext(), TRANSFORM_ORIGIN, request,
+                ActionListener.<AcknowledgedResponse>wrap(r -> finalListener.onResponse(r.isAcknowledged()), finalListener::onFailure),
+                client.admin().indices()::aliases);
+    }
+
+}

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/persistence/TransformInternalIndex.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.AliasMetaData;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexTemplateMetaData;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -93,6 +94,7 @@ public static IndexTemplateMetaData getAuditIndexTemplateMetaData() throws IOExc
                 .put(IndexMetaData.SETTING_NUMBER_OF_SHARDS, 1)
                 .put(IndexMetaData.SETTING_AUTO_EXPAND_REPLICAS, "0-1"))
             .putMapping(MapperService.SINGLE_MAPPING_NAME, Strings.toString(auditMappings()))
+            .putAlias(AliasMetaData.builder(TransformInternalIndexConstants.AUDIT_INDEX_READ_ALIAS))
             .build();
         return transformTemplate;
     }