Remove `transport_versions` from cluster state API (#99223)

Since #99114 the cluster state API exposes per-node versions such as
their transport versions under `node_versions` so there's no need to
duplicate this information under `transport_versions`. This commit
removes the unnecessary information.

docs/changelog/99223.yaml

@@ -0,0 +1,11 @@
+pr: 99223
+summary: Remove `transport_versions` from cluster state API
+area: Infra/Core
+type: breaking
+issues: []
+breaking:
+  title: Remove `transport_versions` from cluster state API
+  area: REST API
+  details: The `transport_versions` subobject of the response to `GET _cluster/state` has been replaced by the `nodes_versions` subobject.
+  impact: If needed, retrieve the per-node transport versions from the `nodes_versions` subobject.
+  notable: false

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -629,21 +629,6 @@ public class ClusterState implements ChunkedToXContent, Diffable<ClusterState> {
                 (builder, params) -> builder.endObject()
             ),
 
-            // transportVersions - redundant with the nodes_versions section but has to stay for backwards compatibility
-            // just use NODES again, its node-related information
-            chunkedSection(
-                metrics.contains(Metric.NODES),
-                (builder, params) -> builder.startArray("transport_versions"),
-                compatibilityVersions.entrySet().iterator(),
-                e -> Iterators.single(
-                    (builder, params) -> builder.startObject()
-                        .field("node_id", e.getKey())
-                        .field("transport_version", e.getValue().transportVersion().toString())
-                        .endObject()
-                ),
-                (builder, params) -> builder.endArray()
-            ),
-
             // per-node version information
             chunkedSection(
                 metrics.contains(Metric.NODES),

server/src/test/java/org/elasticsearch/action/admin/cluster/reroute/ClusterRerouteResponseTests.java

@@ -126,12 +126,6 @@ public class ClusterRerouteResponseTests extends ESTestCase {
                             "max_index_version": %s
                           }
                         },
-                        "transport_versions": [
-                          {
-                            "node_id": "node0",
-                            "transport_version": "8000099"
-                          }
-                        ],
                         "nodes_versions": [
                           {
                             "node_id": "node0",

server/src/test/java/org/elasticsearch/cluster/ClusterStateTests.java

@@ -208,12 +208,6 @@ public class ClusterStateTests extends ESTestCase {
                               "max_index_version":%s
                             }
                           },
-                          "transport_versions" : [
-                            {
-                              "node_id" : "nodeId1",
-                              "transport_version" : "%s"
-                            }
-                          ],
                           "nodes_versions" : [
                             {
                               "node_id" : "nodeId1",
@@ -373,7 +367,6 @@ public class ClusterStateTests extends ESTestCase {
                     IndexVersion.MINIMUM_COMPATIBLE,
                     IndexVersion.current(),
                     TransportVersion.current(),
-                    TransportVersion.current(),
                     IndexVersion.current(),
                     IndexVersion.current(),
                     allocationId,
@@ -470,12 +463,6 @@ public class ClusterStateTests extends ESTestCase {
                           "max_index_version" : %s
                         }
                       },
-                      "transport_versions" : [
-                        {
-                          "node_id" : "nodeId1",
-                          "transport_version" : "%s"
-                        }
-                      ],
                       "nodes_versions" : [
                         {
                           "node_id" : "nodeId1",
@@ -631,7 +618,6 @@ public class ClusterStateTests extends ESTestCase {
                 IndexVersion.MINIMUM_COMPATIBLE,
                 IndexVersion.current(),
                 TransportVersion.current(),
-                TransportVersion.current(),
                 IndexVersion.current(),
                 IndexVersion.current(),
                 allocationId,
@@ -728,12 +714,6 @@ public class ClusterStateTests extends ESTestCase {
                           "max_index_version" : %s
                         }
                       },
-                      "transport_versions" : [
-                        {
-                          "node_id" : "nodeId1",
-                          "transport_version" : "%s"
-                        }
-                      ],
                       "nodes_versions" : [
                         {
                           "node_id" : "nodeId1",
@@ -895,7 +875,6 @@ public class ClusterStateTests extends ESTestCase {
                 IndexVersion.MINIMUM_COMPATIBLE,
                 IndexVersion.current(),
                 TransportVersion.current(),
-                TransportVersion.current(),
                 IndexVersion.current(),
                 IndexVersion.current(),
                 allocationId,
@@ -953,7 +932,6 @@ public class ClusterStateTests extends ESTestCase {
               "master_node" : null,
               "blocks" : { },
               "nodes" : { },
-              "transport_versions" : [ ],
               "nodes_versions" : [ ],
               "metadata" : {
                 "cluster_uuid" : "clusterUUID",
@@ -1185,9 +1163,9 @@ public class ClusterStateTests extends ESTestCase {
             chunkCount += 2 + clusterState.blocks().indices().size();
         }
 
-        // nodes, transport_versions, nodes_versions
+        // nodes, nodes_versions
         if (metrics.contains(ClusterState.Metric.NODES)) {
-            chunkCount += 6 + clusterState.nodes().size() + 2 * clusterState.compatibilityVersions().size();
+            chunkCount += 4 + clusterState.nodes().size() + clusterState.compatibilityVersions().size();
         }
 
         // metadata

test/framework/src/main/java/org/elasticsearch/test/rest/ObjectPath.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Holds an object and allows extraction of specific values from it, given their path
@@ -171,4 +172,14 @@ public class ObjectPath {
     public String toString() {
         return "ObjectPath[" + object + "]";
     }
+
+    public int evaluateArraySize(String path) throws IOException {
+        final List<?> list = evaluate(path);
+        return list.size();
+    }
+
+    public Set<String> evaluateMapKeys(String path) throws IOException {
+        final Map<String, ?> map = evaluate(path);
+        return map.keySet();
+    }
 }

test/yaml-rest-runner/src/test/java/org/elasticsearch/test/rest/yaml/ObjectPathTests.java

@@ -336,4 +336,43 @@ public class ObjectPathTests extends ESTestCase {
         assertThat(object, instanceOf(String.class));
         assertThat(object, equalTo("test2"));
     }
+
+    public void testEvaluateArraySize() throws Exception {
+        XContentBuilder xContentBuilder = randomXContentBuilder();
+        xContentBuilder.startObject();
+        xContentBuilder.startArray("test-array");
+        xContentBuilder.startObject();
+        xContentBuilder.field("foo", "bar");
+        xContentBuilder.endObject();
+        xContentBuilder.value(1);
+        xContentBuilder.value("test-string");
+        xContentBuilder.endArray();
+        xContentBuilder.endObject();
+
+        ObjectPath objectPath = ObjectPath.createFromXContent(
+            XContentFactory.xContent(xContentBuilder.contentType()),
+            BytesReference.bytes(xContentBuilder)
+        );
+
+        assertEquals(3, objectPath.evaluateArraySize("test-array"));
+    }
+
+    public void testEvaluateMapKeys() throws Exception {
+        XContentBuilder xContentBuilder = randomXContentBuilder();
+        xContentBuilder.startObject();
+        xContentBuilder.startObject("test-object");
+        xContentBuilder.field("key1", "bar");
+        xContentBuilder.field("key2", 42);
+        xContentBuilder.startObject("key3");
+        xContentBuilder.endObject();
+        xContentBuilder.endObject();
+        xContentBuilder.endObject();
+
+        ObjectPath objectPath = ObjectPath.createFromXContent(
+            XContentFactory.xContent(xContentBuilder.contentType()),
+            BytesReference.bytes(xContentBuilder)
+        );
+
+        assertEquals(Set.of("key1", "key2", "key3"), objectPath.evaluateMapKeys("test-object"));
+    }
 }

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/cluster/ClusterStatsMonitoringDocTests.java

@@ -768,7 +768,6 @@ public class ClusterStatsMonitoringDocTests extends BaseMonitoringDocTestCase<Cl
                     "max_index_version":%s
                   }
                 },
-                "transport_versions": [],
                 "nodes_versions": []
               },
               "cluster_settings": {

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/TransportVersionClusterStateUpgradeIT.java

@@ -12,70 +12,157 @@ import org.elasticsearch.TransportVersions;
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.test.rest.ObjectPath;
 
-import java.io.IOException;
-import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.oneOf;
 
 public class TransportVersionClusterStateUpgradeIT extends AbstractUpgradeTestCase {
 
-    public void testReadsInferredTransportVersions() throws IOException {
-        assumeTrue("TransportVersion introduced in 8.8.0", UPGRADE_FROM_VERSION.before(Version.V_8_8_0));
-        assumeTrue(
-            "This only has visible effects when upgrading beyond 8.8.0",
-            TransportVersion.current().after(TransportVersions.V_8_8_0)
-        );
-        assumeTrue("Only runs on the mixed cluster", CLUSTER_TYPE == ClusterType.MIXED);
-        // if the master is not upgraded, and the secondary node is, then the cluster info from the secondary
-        // should have inferred transport versions in it
-        // rely on randomisation to hit this case at some point in testing
-
-        Request request = new Request("GET", "/_cluster/state/nodes");
-        Map<String, Object> masterResponse = entityAsMap(client().performRequest(request));
-        assumeFalse("Master needs to not know about transport versions", masterResponse.containsKey("transport_versions"));
-
-        request = new Request("GET", "/_cluster/state/nodes?local=true");
-        Map<String, Object> localResponse = entityAsMap(client().performRequest(request));
-        // should either be empty, or using inferred versions
-        assumeTrue("Local node needs to know about transport versions", masterResponse.containsKey("transport_versions"));
-        Map<?, Version> vs = Maps.transformValues(
-            ((Map<?, ?>) localResponse.get("nodes")),
-            v -> Version.fromString(((Map<?, ?>) v).get("version").toString())
-        );
-        Map<?, TransportVersion> tvs = ((List<?>) localResponse.get("transport_versions")).stream()
-            .map(o -> (Map<?, ?>) o)
-            .collect(Collectors.toMap(m -> m.get("node_id"), m -> TransportVersion.fromString(m.get("transport_version").toString())));
-
-        for (var ver : vs.entrySet()) {
-            if (ver.getValue().after(Version.V_8_8_0)) {
-                assertThat(
-                    "Node " + ver.getKey() + " should have an inferred transport version",
-                    tvs.get(ver.getKey()),
-                    equalTo(TransportVersions.V_8_8_0)
-                );
+    private static final Version VERSION_INTRODUCING_TRANSPORT_VERSIONS = Version.V_8_8_0;
+    private static final Version VERSION_INTRODUCING_NODES_VERSIONS = Version.V_8_11_0;
+    private static final TransportVersion FIRST_TRANSPORT_VERSION = TransportVersions.V_8_8_0;
+
+    public void testReadsInferredTransportVersions() throws Exception {
+        assertEquals(VERSION_INTRODUCING_TRANSPORT_VERSIONS.id(), FIRST_TRANSPORT_VERSION.id());
+
+        // waitUntil because the versions fixup on upgrade happens in the background so may need a retry
+        assertTrue(waitUntil(() -> {
+            try {
+                // check several responses in order to sample from a selection of nodes
+                for (int i = getClusterHosts().size(); i > 0; i--) {
+                    if (runTransportVersionsTest() == false) {
+                        return false;
+                    }
+                }
+                return true;
+            } catch (Exception e) {
+                throw new AssertionError(e);
             }
-        }
+        }));
     }
 
-    public void testCompletesRealTransportVersions() throws IOException {
-        assumeTrue("TransportVersion introduced in 8.8.0", UPGRADE_FROM_VERSION.before(Version.V_8_8_0));
-        assumeTrue(
-            "This only has visible effects when upgrading beyond 8.8.0",
-            TransportVersion.current().after(TransportVersions.V_8_8_0)
+    private boolean runTransportVersionsTest() throws Exception {
+        final var clusterState = ObjectPath.createFromResponse(
+            client().performRequest(new Request("GET", "/_cluster/state" + randomFrom("", "/nodes") + randomFrom("", "?local")))
         );
-        assumeTrue("Only runs on the upgraded cluster", CLUSTER_TYPE == ClusterType.UPGRADED);
-        // once everything is upgraded, the master should fill in the real transport versions
+        final var description = clusterState.toString();
+
+        final var nodeIds = clusterState.evaluateMapKeys("nodes");
+        final Map<String, Version> versionsByNodeId = Maps.newHashMapWithExpectedSize(nodeIds.size());
+        for (final var nodeId : nodeIds) {
+            versionsByNodeId.put(nodeId, Version.fromString(clusterState.evaluate("nodes." + nodeId + ".version")));
+        }
+
+        final var hasTransportVersions = clusterState.evaluate("transport_versions") != null;
+        final var hasNodesVersions = clusterState.evaluate("nodes_versions") != null;
+        assertFalse(description, hasNodesVersions && hasTransportVersions);
+
+        switch (CLUSTER_TYPE) {
+            case OLD -> {
+                if (UPGRADE_FROM_VERSION.before(VERSION_INTRODUCING_TRANSPORT_VERSIONS)) {
+                    // Before 8.8.0 there was only DiscoveryNode#version
+                    assertFalse(description, hasTransportVersions);
+                    assertFalse(description, hasNodesVersions);
+                } else if (UPGRADE_FROM_VERSION.before(VERSION_INTRODUCING_NODES_VERSIONS)) {
+                    // In [8.8.0, 8.11.0) we exposed just transport_versions
+                    assertTrue(description, hasTransportVersions);
+                    assertFalse(description, hasNodesVersions);
+                } else {
+                    // From 8.11.0 onwards we exposed nodes_versions
+                    assertFalse(description, hasTransportVersions);
+                    assertTrue(description, hasNodesVersions);
+                }
+            }
+            case MIXED -> {
+                if (UPGRADE_FROM_VERSION.before(VERSION_INTRODUCING_TRANSPORT_VERSIONS)) {
+                    // Responding node might be <8.8.0 (so no extra versions) or >=8.11.0 (includes nodes_versions)
+                    assertFalse(description, hasTransportVersions);
+                } else if (UPGRADE_FROM_VERSION.before(VERSION_INTRODUCING_NODES_VERSIONS)) {
+                    // Responding node might be in [8.8.0, 8.11.0) (transport_versions) or >=8.11.0 (includes nodes_versions) but not both
+                    assertTrue(description, hasNodesVersions || hasTransportVersions);
+                } else {
+                    // Responding node is ≥8.11.0 so has nodes_versions for sure
+                    assertFalse(description, hasTransportVersions);
+                    assertTrue(description, hasNodesVersions);
+                }
+            }
+            case UPGRADED -> {
+                // All nodes are Version.CURRENT, ≥8.11.0, so we definitely have nodes_versions
+                assertFalse(description, hasTransportVersions);
+                assertTrue(description, hasNodesVersions);
+                assertThat(description, versionsByNodeId.values(), everyItem(equalTo(Version.CURRENT)));
+            }
+        }
+
+        if (hasTransportVersions) {
+            // Upgrading from [8.8.0, 8.11.0) and the responding node is still on the old version
+            assertThat(description, UPGRADE_FROM_VERSION, lessThan(VERSION_INTRODUCING_NODES_VERSIONS));
+            assertThat(description, UPGRADE_FROM_VERSION, greaterThanOrEqualTo(VERSION_INTRODUCING_TRANSPORT_VERSIONS));
+            assertNotEquals(description, ClusterType.UPGRADED, CLUSTER_TYPE);
 
-        Request request = new Request("GET", "/_cluster/state/nodes");
-        Map<String, Object> response = entityAsMap(client().performRequest(request));
-        Map<?, TransportVersion> tvs = ((List<?>) response.get("transport_versions")).stream()
-            .map(o -> (Map<?, ?>) o)
-            .collect(Collectors.toMap(m -> m.get("node_id"), m -> TransportVersion.fromString(m.get("transport_version").toString())));
+            // transport_versions includes the correct version for all nodes, no inference is needed
+            assertEquals(description, nodeIds.size(), clusterState.evaluateArraySize("transport_versions"));
+            for (int i = 0; i < nodeIds.size(); i++) {
+                final var path = "transport_versions." + i;
+                final String nodeId = clusterState.evaluate(path + ".node_id");
+                final var nodeDescription = nodeId + "/" + description;
+                final var transportVersion = TransportVersion.fromString(clusterState.evaluate(path + ".transport_version"));
+                final var nodeVersion = versionsByNodeId.get(nodeId);
+                assertNotNull(nodeDescription, nodeVersion);
+                if (nodeVersion.equals(Version.CURRENT)) {
+                    assertEquals(nodeDescription, TransportVersion.current(), transportVersion);
+                } else if (nodeVersion.after(VERSION_INTRODUCING_TRANSPORT_VERSIONS)) {
+                    assertThat(nodeDescription, transportVersion, greaterThan(FIRST_TRANSPORT_VERSION));
+                } else {
+                    assertEquals(nodeDescription, FIRST_TRANSPORT_VERSION, transportVersion);
+                }
+            }
+        } else if (hasNodesVersions) {
+            // Either upgrading from ≥8.11.0 (the responding node might be old or new), or from <8.8.0 (the responding node is new)
+            assertFalse(description, UPGRADE_FROM_VERSION.before(VERSION_INTRODUCING_NODES_VERSIONS) && CLUSTER_TYPE == ClusterType.OLD);
+
+            // nodes_versions includes _a_ version for all nodes; it might be correct, or it might be inferred if we're upgrading from
+            // <8.8.0 and the master is still an old node or the TransportVersionsFixupListener hasn't run yet
+            assertEquals(description, nodeIds.size(), clusterState.evaluateArraySize("nodes_versions"));
+            for (int i = 0; i < nodeIds.size(); i++) {
+                final var path = "nodes_versions." + i;
+                final String nodeId = clusterState.evaluate(path + ".node_id");
+                final var nodeDescription = nodeId + "/" + description;
+                final var transportVersion = TransportVersion.fromString(clusterState.evaluate(path + ".transport_version"));
+                final var nodeVersion = versionsByNodeId.get(nodeId);
+                assertNotNull(nodeDescription, nodeVersion);
+                if (nodeVersion.equals(Version.CURRENT)) {
+                    // Either the responding node is upgraded or the upgrade is trivial; if the responding node is upgraded but the master
+                    // is not then its transport version may be temporarily inferred as 8.8.0 until TransportVersionsFixupListener runs.
+                    assertThat(
+                        nodeDescription,
+                        transportVersion,
+                        UPGRADE_FROM_VERSION.onOrAfter(VERSION_INTRODUCING_TRANSPORT_VERSIONS)
+                            ? equalTo(TransportVersion.current())
+                            : oneOf(TransportVersion.current(), FIRST_TRANSPORT_VERSION)
+                    );
+                    if (CLUSTER_TYPE == ClusterType.UPGRADED && transportVersion.equals(FIRST_TRANSPORT_VERSION)) {
+                        // TransportVersionsFixupListener should run soon, retry
+                        logger.info("{} - not fixed up yet, retrying", nodeDescription);
+                        return false;
+                    }
+                } else if (nodeVersion.after(VERSION_INTRODUCING_TRANSPORT_VERSIONS)) {
+                    // There's no relationship between node versions and transport versions any more, although we can be sure of this:
+                    assertThat(nodeDescription, transportVersion, greaterThan(FIRST_TRANSPORT_VERSION));
+                } else {
+                    // Responding node is not upgraded, and no later than 8.8.0, so we infer its version correctly.
+                    assertEquals(nodeDescription, TransportVersion.fromId(nodeVersion.id()), transportVersion);
+                }
+            }
+        }
 
-        assertThat(tvs + " should be updated", tvs.values(), everyItem(equalTo(TransportVersion.current())));
+        return true;
     }
 }