Fix bugs with tests that have no cluster state

Author: craigtaverner

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestsDataLoader.java

@@ -665,7 +665,10 @@ private static void deleteView(RestClient client, String viewName, Logger logger
         try {
             client.performRequest(new Request("DELETE", "/_query/view/" + viewName));
         } catch (ResponseException e) {
-            if (e.getResponse().getStatusLine().getStatusCode() != 404) {
+            logger.info("View delete error: {}", e.getMessage());
+            int code = e.getResponse().getStatusLine().getStatusCode();
+            // On older servers the view listing succeeds when it should, so we get here when we should not, hence the 400
+            if (code != 404 && code != 400) {
                 throw e;
             }
         }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlPlugin.java

@@ -82,6 +82,7 @@
 import org.elasticsearch.xpack.esql.querydsl.query.SingleValueQuery;
 import org.elasticsearch.xpack.esql.querylog.EsqlQueryLog;
 import org.elasticsearch.xpack.esql.session.IndexResolver;
+import org.elasticsearch.xpack.esql.view.ClusterViewService;
 import org.elasticsearch.xpack.esql.view.DeleteViewAction;
 import org.elasticsearch.xpack.esql.view.GetViewAction;
 import org.elasticsearch.xpack.esql.view.ListViewsAction;
@@ -233,7 +234,7 @@ public Collection<?> createComponents(PluginServices services) {
         BigArrays bigArrays = services.indicesService().getBigArrays().withCircuitBreaking();
         var blockFactoryProvider = blockFactoryProvider(circuitBreaker, bigArrays, maxPrimitiveArrayBlockSize);
         EsqlFunctionRegistry functionRegistry = new EsqlFunctionRegistry();
-        ViewService viewService = new ViewService(services.clusterService(), functionRegistry);
+        ViewService viewService = new ClusterViewService(functionRegistry, services.clusterService());
         setupSharedSecrets();
         List<BiConsumer<LogicalPlan, Failures>> extraCheckers = extraCheckerProviders.stream()
             .flatMap(p -> p.checkers(services.projectResolver(), services.clusterService()).stream())

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/TransportEsqlQueryAction.java

@@ -55,7 +55,7 @@
 import org.elasticsearch.xpack.esql.session.Configuration;
 import org.elasticsearch.xpack.esql.session.EsqlSession.PlanRunner;
 import org.elasticsearch.xpack.esql.session.Result;
-import org.elasticsearch.xpack.esql.view.ViewService;
+import org.elasticsearch.xpack.esql.view.ClusterViewService;
 
 import java.io.IOException;
 import java.time.ZoneOffset;
@@ -80,7 +80,7 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
     private final ClusterService clusterService;
     private final Executor requestExecutor;
     private final EnrichPolicyResolver enrichPolicyResolver;
-    private final ViewService viewService;
+    private final ClusterViewService viewService;
     private final EnrichLookupService enrichLookupService;
     private final LookupFromIndexService lookupFromIndexService;
     private final AsyncTaskManagementService<EsqlQueryRequest, EsqlQueryResponse, EsqlQueryTask> asyncTaskManagementService;
@@ -102,7 +102,7 @@ public TransportEsqlQueryAction(
         SearchService searchService,
         ExchangeService exchangeService,
         ClusterService clusterService,
-        ViewService viewService,
+        ClusterViewService viewService,
         ProjectResolver projectResolver,
         ThreadPool threadPool,
         BigArrays bigArrays,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/ClusterViewService.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.view;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateUpdateTask;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
+
+import java.util.Map;
+import java.util.function.Function;
+
+/**
+ * Implementation of {@link ViewService} that keeps the views in the cluster state.
+ */
+public class ClusterViewService extends ViewService {
+    private final ClusterService clusterService;
+
+    public ClusterViewService(EsqlFunctionRegistry functionRegistry, ClusterService clusterService) {
+        super(functionRegistry);
+        this.clusterService = clusterService;
+    }
+
+    @Override
+    protected ViewMetadata getMetadata() {
+        return clusterService.state().metadata().custom(ViewMetadata.TYPE, ViewMetadata.EMPTY);
+    }
+
+    @Override
+    protected void updateViewMetadata(ActionListener<Void> callback, Function<ViewMetadata, Map<String, View>> function) {
+        submitUnbatchedTask("update-esql-view-metadata", new ClusterStateUpdateTask() {
+            @Override
+            public ClusterState execute(ClusterState currentState) {
+                var views = currentState.metadata().custom(ViewMetadata.TYPE, ViewMetadata.EMPTY);
+                Map<String, View> policies = function.apply(views);
+                Metadata metadata = Metadata.builder(currentState.metadata())
+                    .putCustom(ViewMetadata.TYPE, new ViewMetadata(policies))
+                    .build();
+                return ClusterState.builder(currentState).metadata(metadata).build();
+            }
+
+            @Override
+            public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
+                callback.onResponse(null);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                callback.onFailure(e);
+            }
+        });
+    }
+
+    @SuppressForbidden(reason = "legacy usage of unbatched task") // TODO add support for batching here
+    private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String source, ClusterStateUpdateTask task) {
+        clusterService.submitUnbatchedStateUpdateTask(source, task);
+    }
+
+    @Override
+    protected void assertMasterNode() {
+        assert clusterService.localNode().isMasterNode();
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/InMemoryViewService.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.view;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.xpack.esql.expression.function.EsqlFunctionRegistry;
+
+import java.util.Map;
+import java.util.function.Function;
+
+/**
+ * Simple implementation of {@link ClusterViewService} that keeps the views in memory.
+ * This is useful for testing.
+ */
+public class InMemoryViewService extends ViewService {
+
+    private ViewMetadata metadata;
+
+    public InMemoryViewService(EsqlFunctionRegistry functionRegistry) {
+        super(functionRegistry);
+        this.metadata = ViewMetadata.EMPTY;
+    }
+
+    @Override
+    protected ViewMetadata getMetadata() {
+        return metadata;
+    }
+
+    @Override
+    protected void updateViewMetadata(ActionListener<Void> callback, Function<ViewMetadata, Map<String, View>> function) {
+        Map<String, View> updated = function.apply(metadata);
+        this.metadata = new ViewMetadata(updated);
+    }
+
+    @Override
+    protected void assertMasterNode() {
+        // no-op
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/TransportDeleteViewAction.java

@@ -21,15 +21,15 @@
 import org.elasticsearch.transport.TransportService;
 
 public class TransportDeleteViewAction extends AcknowledgedTransportMasterNodeAction<DeleteViewAction.Request> {
-    private final ViewService viewService;
+    private final ClusterViewService viewService;
 
     @Inject
     public TransportDeleteViewAction(
         TransportService transportService,
         ClusterService clusterService,
         ThreadPool threadPool,
         ActionFilters actionFilters,
-        ViewService viewService
+        ClusterViewService viewService
     ) {
         super(
             DeleteViewAction.NAME,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/TransportGetViewAction.java

@@ -18,10 +18,10 @@
 
 public class TransportGetViewAction extends HandledTransportAction<GetViewAction.Request, GetViewAction.Response> {
     public static final ActionType<RemoteInfoResponse> TYPE = new ActionType<>(GetViewAction.NAME);
-    private final ViewService viewService;
+    private final ClusterViewService viewService;
 
     @Inject
-    public TransportGetViewAction(TransportService transportService, ActionFilters actionFilters, ViewService viewService) {
+    public TransportGetViewAction(TransportService transportService, ActionFilters actionFilters, ClusterViewService viewService) {
         super(GetViewAction.NAME, transportService, actionFilters, GetViewAction.Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.viewService = viewService;
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/TransportListViewsAction.java

@@ -21,10 +21,10 @@
 
 public class TransportListViewsAction extends HandledTransportAction<ListViewsAction.Request, ListViewsAction.Response> {
     public static final ActionType<RemoteInfoResponse> TYPE = new ActionType<>(ListViewsAction.NAME);
-    private final ViewService viewService;
+    private final ClusterViewService viewService;
 
     @Inject
-    public TransportListViewsAction(TransportService transportService, ActionFilters actionFilters, ViewService viewService) {
+    public TransportListViewsAction(TransportService transportService, ActionFilters actionFilters, ClusterViewService viewService) {
         super(ListViewsAction.NAME, transportService, actionFilters, ListViewsAction.Request::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.viewService = viewService;
     }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/TransportPutViewAction.java

@@ -28,15 +28,15 @@
 import java.util.Map;
 
 public class TransportPutViewAction extends AcknowledgedTransportMasterNodeAction<PutViewAction.Request> {
-    private final ViewService viewService;
+    private final ClusterViewService viewService;
 
     @Inject
     public TransportPutViewAction(
         TransportService transportService,
         ClusterService clusterService,
         ThreadPool threadPool,
         ActionFilters actionFilters,
-        ViewService viewService
+        ClusterViewService viewService
     ) {
         super(
             PutViewAction.NAME,

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/view/ViewMetadata.java

@@ -26,7 +26,7 @@
 import java.util.Objects;
 
 /**
- * Encapsulates enrich policies as custom metadata inside cluster state.
+ * Encapsulates view definitions as custom metadata inside cluster state.
  */
 public final class ViewMetadata extends AbstractNamedDiffable<Metadata.ClusterCustom> implements Metadata.ClusterCustom {
     public static final String TYPE = "esql_view";