[Transform] Fix stats can return old state information if security is…

… enabled (elastic#51732) do index refresh of the internal transform index with the system user instead of using the calling user which does not have sufficient rights if security is enabled fixes elastic#51728
hendrikmuhs · Jan 31, 2020 · c112d5c · c112d5c
1 parent ec0964f
commit c112d5c
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 10 deletions.
diff --git a/.../src/main/java/org/elasticsearch/xpack/transform/action/TransportStopTransformAction.java b/.../src/main/java/org/elasticsearch/xpack/transform/action/TransportStopTransformAction.java
@@ -43,7 +43,6 @@
 import org.elasticsearch.xpack.core.transform.transforms.TransformState;
 import org.elasticsearch.xpack.core.transform.transforms.TransformTaskParams;
 import org.elasticsearch.xpack.core.transform.transforms.TransformTaskState;
-import org.elasticsearch.xpack.core.transform.transforms.persistence.TransformInternalIndexConstants;
 import org.elasticsearch.xpack.transform.TransformServices;
 import org.elasticsearch.xpack.transform.persistence.TransformConfigManager;
 import org.elasticsearch.xpack.transform.transforms.TransformTask;
@@ -67,7 +66,6 @@ public class TransportStopTransformAction extends TransportTasksAction<Transform
     private final ThreadPool threadPool;
     private final TransformConfigManager transformConfigManager;
     private final PersistentTasksService persistentTasksService;
-    private final Client client;
 
     @Inject
     public TransportStopTransformAction(
@@ -105,7 +103,6 @@ protected TransportStopTransformAction(
         this.threadPool = threadPool;
         this.transformConfigManager = transformServices.getConfigManager();
         this.persistentTasksService = persistentTasksService;
-        this.client = client;
     }
 
     static void validateTaskState(ClusterState state, List<String> transformIds, boolean isForce) {
@@ -274,13 +271,10 @@ protected StopTransformAction.Response newResponse(
     private ActionListener<Response> waitForStopListener(Request request, ActionListener<Response> listener) {
 
         ActionListener<Response> onStopListener = ActionListener.wrap(
-            waitResponse -> client.admin()
-                .indices()
-                .prepareRefresh(TransformInternalIndexConstants.LATEST_INDEX_NAME)
-                .execute(ActionListener.wrap(r -> listener.onResponse(waitResponse), e -> {
-                    logger.info("Failed to refresh internal index after delete", e);
-                    listener.onResponse(waitResponse);
-                })),
+            waitResponse -> transformConfigManager.refresh(ActionListener.wrap(r -> listener.onResponse(waitResponse), e -> {
+                logger.warn("Could not refresh state, state information might be outdated", e);
+                listener.onResponse(waitResponse);
+            })),
             listener::onFailure
         );
         return ActionListener.wrap(

diff --git a/.../java/org/elasticsearch/xpack/transform/persistence/IndexBasedTransformConfigManager.java b/.../java/org/elasticsearch/xpack/transform/persistence/IndexBasedTransformConfigManager.java
@@ -14,6 +14,8 @@
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.index.IndexAction;
 import org.elasticsearch.action.index.IndexRequest;
@@ -634,6 +636,17 @@ public void getTransformStoredDocs(Collection<String> transformIds, ActionListen
         );
     }
 
+    @Override
+    public void refresh(ActionListener<Boolean> listener) {
+        executeAsyncWithOrigin(
+            client.threadPool().getThreadContext(),
+            TRANSFORM_ORIGIN,
+            new RefreshRequest(TransformInternalIndexConstants.LATEST_INDEX_NAME),
+            ActionListener.<RefreshResponse>wrap(r -> listener.onResponse(true), listener::onFailure),
+            client.admin().indices()::refresh
+        );
+    }
+
     private void parseTransformLenientlyFromSource(
         BytesReference source,
         String transformId,

diff --git a/...m/src/main/java/org/elasticsearch/xpack/transform/persistence/TransformConfigManager.java b/...m/src/main/java/org/elasticsearch/xpack/transform/persistence/TransformConfigManager.java
@@ -149,4 +149,5 @@ void putOrUpdateTransformStoredDoc(
 
     void getTransformStoredDocs(Collection<String> transformIds, ActionListener<List<TransformStoredDoc>> listener);
 
+    void refresh(ActionListener<Boolean> listener);
 }
diff --git a/...st/java/org/elasticsearch/xpack/transform/persistence/InMemoryTransformConfigManager.java b/...st/java/org/elasticsearch/xpack/transform/persistence/InMemoryTransformConfigManager.java
@@ -214,4 +214,9 @@ public void getTransformStoredDocs(Collection<String> transformIds, ActionListen
         listener.onResponse(docs);
     }
 
+    @Override
+    public void refresh(ActionListener<Boolean> listener) {
+        listener.onResponse(true);
+    }
+
 }