Merge branch 'opensearch-project:main' into xcontentstatuscode

CHANGELOG.md

@@ -93,6 +93,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Migrate client transports to Apache HttpClient / Core 5.x ([#4459](https://github.com/opensearch-project/OpenSearch/pull/4459))
 - Changed http code on create index API with bad input raising NotXContentException to 400([#4773](https://github.com/opensearch-project/OpenSearch/pull/4773))
 - Refactored BalancedAllocator.Balancer to LocalShardsBalancer ([#4761](https://github.com/opensearch-project/OpenSearch/pull/4761))
+- Fail weight update when decommission ongoing and fail decommission when attribute not weighed away ([#4839](https://github.com/opensearch-project/OpenSearch/pull/4839))
 ### Deprecated
 ### Removed
 - Remove deprecated code to add node name into log pattern of log4j property file ([#4568](https://github.com/opensearch-project/OpenSearch/pull/4568))
@@ -102,6 +103,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Always auto release the flood stage block ([#4703](https://github.com/opensearch-project/OpenSearch/pull/4703))
 - Remove LegacyESVersion.V_7_4_ and V_7_5_ Constants ([#4704](https://github.com/opensearch-project/OpenSearch/pull/4704))
 - Remove Legacy Version support from Snapshot/Restore Service ([#4728](https://github.com/opensearch-project/OpenSearch/pull/4728))
+- Remove deprecated serialization logic from pipeline aggs ([#4847](https://github.com/opensearch-project/OpenSearch/pull/4847))
 
 ### Fixed
 - `opensearch-service.bat start` and `opensearch-service.bat manager` failing to run ([#4289](https://github.com/opensearch-project/OpenSearch/pull/4289))

server/src/internalClusterTest/java/org/opensearch/cluster/coordination/AwarenessAttributeDecommissionIT.java

@@ -20,11 +20,15 @@
 import org.opensearch.action.admin.cluster.decommission.awareness.put.DecommissionAction;
 import org.opensearch.action.admin.cluster.decommission.awareness.put.DecommissionRequest;
 import org.opensearch.action.admin.cluster.decommission.awareness.put.DecommissionResponse;
+import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.opensearch.action.admin.cluster.shards.routing.weighted.put.ClusterPutWeightedRoutingResponse;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.decommission.DecommissionAttribute;
 import org.opensearch.cluster.decommission.DecommissionStatus;
+import org.opensearch.cluster.decommission.DecommissioningFailedException;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodeRole;
+import org.opensearch.cluster.routing.WeightedRouting;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
 import org.opensearch.common.settings.Settings;
@@ -37,6 +41,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutionException;
 
 import static org.opensearch.test.NodeRoles.onlyRole;
@@ -102,6 +107,17 @@ public void testDecommissionStatusUpdatePublishedToAllNodes() throws ExecutionEx
 
         ensureStableCluster(6);
 
+        logger.info("--> setting shard routing weights for weighted round robin");
+        Map<String, Double> weights = Map.of("a", 1.0, "b", 1.0, "c", 0.0);
+        WeightedRouting weightedRouting = new WeightedRouting("zone", weights);
+
+        ClusterPutWeightedRoutingResponse weightedRoutingResponse = client().admin()
+            .cluster()
+            .prepareWeightedRouting()
+            .setWeightedRouting(weightedRouting)
+            .get();
+        assertTrue(weightedRoutingResponse.isAcknowledged());
+
         logger.info("--> starting decommissioning nodes in zone {}", 'c');
         DecommissionAttribute decommissionAttribute = new DecommissionAttribute("zone", "c");
         DecommissionRequest decommissionRequest = new DecommissionRequest(decommissionAttribute);
@@ -162,4 +178,57 @@ public void testDecommissionStatusUpdatePublishedToAllNodes() throws ExecutionEx
         // as by then all nodes should have joined the cluster
         ensureStableCluster(6, TimeValue.timeValueMinutes(2));
     }
+
+    public void testDecommissionFailedWhenAttributeNotWeighedAway() throws Exception {
+        Settings commonSettings = Settings.builder()
+            .put("cluster.routing.allocation.awareness.attributes", "zone")
+            .put("cluster.routing.allocation.awareness.force.zone.values", "a,b,c")
+            .build();
+        // Start 3 cluster manager eligible nodes
+        internalCluster().startClusterManagerOnlyNodes(3, Settings.builder().put(commonSettings).build());
+        // start 3 data nodes
+        internalCluster().startDataOnlyNodes(3, Settings.builder().put(commonSettings).build());
+        ensureStableCluster(6);
+        ClusterHealthResponse health = client().admin()
+            .cluster()
+            .prepareHealth()
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForGreenStatus()
+            .setWaitForNodes(Integer.toString(6))
+            .execute()
+            .actionGet();
+        assertFalse(health.isTimedOut());
+
+        DecommissionAttribute decommissionAttribute = new DecommissionAttribute("zone", "c");
+        DecommissionRequest decommissionRequest = new DecommissionRequest(decommissionAttribute);
+        assertBusy(() -> {
+            DecommissioningFailedException ex = expectThrows(
+                DecommissioningFailedException.class,
+                () -> client().execute(DecommissionAction.INSTANCE, decommissionRequest).actionGet()
+            );
+            assertTrue(
+                ex.getMessage()
+                    .contains("no weights are set to the attribute. Please set appropriate weights before triggering decommission action")
+            );
+        });
+
+        logger.info("--> setting shard routing weights for weighted round robin");
+        Map<String, Double> weights = Map.of("a", 1.0, "b", 1.0, "c", 1.0);
+        WeightedRouting weightedRouting = new WeightedRouting("zone", weights);
+
+        ClusterPutWeightedRoutingResponse weightedRoutingResponse = client().admin()
+            .cluster()
+            .prepareWeightedRouting()
+            .setWeightedRouting(weightedRouting)
+            .get();
+        assertTrue(weightedRoutingResponse.isAcknowledged());
+
+        assertBusy(() -> {
+            DecommissioningFailedException ex = expectThrows(
+                DecommissioningFailedException.class,
+                () -> client().execute(DecommissionAction.INSTANCE, decommissionRequest).actionGet()
+            );
+            assertTrue(ex.getMessage().contains("weight for decommissioned attribute is expected to be [0.0] but found [1.0]"));
+        });
+    }
 }

server/src/main/java/org/opensearch/cluster/decommission/DecommissionService.java

@@ -20,7 +20,9 @@
 import org.opensearch.cluster.ClusterStateUpdateTask;
 import org.opensearch.cluster.NotClusterManagerException;
 import org.opensearch.cluster.metadata.Metadata;
+import org.opensearch.cluster.metadata.WeightedRoutingMetadata;
 import org.opensearch.cluster.node.DiscoveryNode;
+import org.opensearch.cluster.routing.WeightedRouting;
 import org.opensearch.cluster.routing.allocation.AllocationService;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -129,6 +131,8 @@ public ClusterState execute(ClusterState currentState) {
                 DecommissionAttributeMetadata decommissionAttributeMetadata = currentState.metadata().decommissionAttributeMetadata();
                 // check that request is eligible to proceed
                 ensureEligibleRequest(decommissionAttributeMetadata, decommissionAttribute);
+                // ensure attribute is weighed away
+                ensureToBeDecommissionedAttributeWeighedAway(currentState, decommissionAttribute);
                 decommissionAttributeMetadata = new DecommissionAttributeMetadata(decommissionAttribute);
                 logger.info("registering decommission metadata [{}] to execute action", decommissionAttributeMetadata.toString());
                 return ClusterState.builder(currentState)
@@ -413,6 +417,30 @@ private static void validateAwarenessAttribute(
         }
     }
 
+    private static void ensureToBeDecommissionedAttributeWeighedAway(ClusterState state, DecommissionAttribute decommissionAttribute) {
+        WeightedRoutingMetadata weightedRoutingMetadata = state.metadata().weightedRoutingMetadata();
+        if (weightedRoutingMetadata == null) {
+            throw new DecommissioningFailedException(
+                decommissionAttribute,
+                "no weights are set to the attribute. Please set appropriate weights before triggering decommission action"
+            );
+        }
+        WeightedRouting weightedRouting = weightedRoutingMetadata.getWeightedRouting();
+        if (weightedRouting.attributeName().equals(decommissionAttribute.attributeName()) == false) {
+            throw new DecommissioningFailedException(
+                decommissionAttribute,
+                "no weights are specified to attribute [" + decommissionAttribute.attributeName() + "]"
+            );
+        }
+        Double attributeValueWeight = weightedRouting.weights().get(decommissionAttribute.attributeValue());
+        if (attributeValueWeight == null || attributeValueWeight.equals(0.0) == false) {
+            throw new DecommissioningFailedException(
+                decommissionAttribute,
+                "weight for decommissioned attribute is expected to be [0.0] but found [" + attributeValueWeight + "]"
+            );
+        }
+    }
+
     private static void ensureEligibleRequest(
         DecommissionAttributeMetadata decommissionAttributeMetadata,
         DecommissionAttribute requestedDecommissionAttribute

server/src/main/java/org/opensearch/cluster/routing/WeightedRoutingService.java

@@ -19,6 +19,8 @@
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.ClusterStateUpdateTask;
 import org.opensearch.cluster.ack.ClusterStateUpdateResponse;
+import org.opensearch.cluster.decommission.DecommissionAttributeMetadata;
+import org.opensearch.cluster.decommission.DecommissionStatus;
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.metadata.WeightedRoutingMetadata;
 import org.opensearch.cluster.routing.allocation.decider.AwarenessAllocationDecider;
@@ -68,6 +70,8 @@ public void registerWeightedRoutingMetadata(
         clusterService.submitStateUpdateTask("update_weighted_routing", new ClusterStateUpdateTask(Priority.URGENT) {
             @Override
             public ClusterState execute(ClusterState currentState) {
+                // verify currently no decommission action is ongoing
+                ensureNoOngoingDecommissionAction(currentState);
                 Metadata metadata = currentState.metadata();
                 Metadata.Builder mdBuilder = Metadata.builder(currentState.metadata());
                 WeightedRoutingMetadata weightedRoutingMetadata = metadata.custom(WeightedRoutingMetadata.TYPE);
@@ -154,4 +158,15 @@ public void verifyAwarenessAttribute(String attributeName) {
             throw validationException;
         }
     }
+
+    public void ensureNoOngoingDecommissionAction(ClusterState state) {
+        DecommissionAttributeMetadata decommissionAttributeMetadata = state.metadata().decommissionAttributeMetadata();
+        if (decommissionAttributeMetadata != null && decommissionAttributeMetadata.status().equals(DecommissionStatus.FAILED) == false) {
+            throw new IllegalStateException(
+                "a decommission action is ongoing with status ["
+                    + decommissionAttributeMetadata.status().status()
+                    + "], cannot update weight during this state"
+            );
+        }
+    }
 }

server/src/main/java/org/opensearch/plugins/SearchPlugin.java

@@ -621,52 +621,6 @@ class PipelineAggregationSpec extends SearchExtensionSpec<
         PipelineAggregationBuilder,
         ContextParser<String, ? extends PipelineAggregationBuilder>> {
         private final Map<String, Writeable.Reader<? extends InternalAggregation>> resultReaders = new TreeMap<>();
-        /**
-         * Read the aggregator from a stream.
-         * @deprecated Pipelines implemented after 7.8.0 do not need to be sent across the wire
-         */
-        @Deprecated
-        private final Writeable.Reader<? extends PipelineAggregator> aggregatorReader;
-
-        /**
-         * Specification of a {@link PipelineAggregator}.
-         *
-         * @param name holds the names by which this aggregation might be parsed. The {@link ParseField#getPreferredName()} is special as it
-         *        is the name by under which the readers are registered. So it is the name that the {@link PipelineAggregationBuilder} and
-         *        {@link PipelineAggregator} should return from {@link NamedWriteable#getWriteableName()}. It is an error if
-         *        {@link ParseField#getPreferredName()} conflicts with another registered name, including names from other plugins.
-         * @param builderReader the reader registered for this aggregation's builder. Typically, a reference to a constructor that takes a
-         *        {@link StreamInput}
-         * @param parser reads the aggregation builder from XContent
-         */
-        public PipelineAggregationSpec(
-            ParseField name,
-            Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
-            ContextParser<String, ? extends PipelineAggregationBuilder> parser
-        ) {
-            super(name, builderReader, parser);
-            this.aggregatorReader = null;
-        }
-
-        /**
-         * Specification of a {@link PipelineAggregator}.
-         *
-         * @param name name by which this aggregation might be parsed or deserialized. Make sure it is the name that the
-         *        {@link PipelineAggregationBuilder} and {@link PipelineAggregator} should return from
-         *        {@link NamedWriteable#getWriteableName()}.  It is an error if this name conflicts with another registered name, including
-         *        names from other plugins.
-         * @param builderReader the reader registered for this aggregation's builder. Typically, a reference to a constructor that takes a
-         *        {@link StreamInput}
-         * @param parser reads the aggregation builder from XContent
-         */
-        public PipelineAggregationSpec(
-            String name,
-            Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
-            ContextParser<String, ? extends PipelineAggregationBuilder> parser
-        ) {
-            super(name, builderReader, parser);
-            this.aggregatorReader = null;
-        }
 
         /**
          * Specification of a {@link PipelineAggregator}.
@@ -677,20 +631,14 @@ public PipelineAggregationSpec(
          *        {@link ParseField#getPreferredName()} conflicts with another registered name, including names from other plugins.
          * @param builderReader the reader registered for this aggregation's builder. Typically, a reference to a constructor that takes a
          *        {@link StreamInput}
-         * @param aggregatorReader reads the {@link PipelineAggregator} from a stream
          * @param parser reads the aggregation builder from XContent
-         * @deprecated Use {@link PipelineAggregationSpec#PipelineAggregationSpec(ParseField, Writeable.Reader, ContextParser)} for
-         *             pipelines implemented after 7.8.0
          */
-        @Deprecated
         public PipelineAggregationSpec(
             ParseField name,
             Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
-            Writeable.Reader<? extends PipelineAggregator> aggregatorReader,
             ContextParser<String, ? extends PipelineAggregationBuilder> parser
         ) {
             super(name, builderReader, parser);
-            this.aggregatorReader = aggregatorReader;
         }
 
         /**
@@ -702,20 +650,15 @@ public PipelineAggregationSpec(
          *        names from other plugins.
          * @param builderReader the reader registered for this aggregation's builder. Typically, a reference to a constructor that takes a
          *        {@link StreamInput}
-         * @param aggregatorReader reads the {@link PipelineAggregator} from a stream
          * @param parser reads the aggregation builder from XContent
-         * @deprecated Use {@link PipelineAggregationSpec#PipelineAggregationSpec(String, Writeable.Reader, ContextParser)} for pipelines
-         *             implemented after 7.8.0
+
          */
-        @Deprecated
         public PipelineAggregationSpec(
             String name,
             Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
-            Writeable.Reader<? extends PipelineAggregator> aggregatorReader,
             ContextParser<String, ? extends PipelineAggregationBuilder> parser
         ) {
             super(name, builderReader, parser);
-            this.aggregatorReader = aggregatorReader;
         }
 
         /**
@@ -727,19 +670,16 @@ public PipelineAggregationSpec(
          *        {@link ParseField#getPreferredName()} conflicts with another registered name, including names from other plugins.
          * @param builderReader the reader registered for this aggregation's builder. Typically, a reference to a constructor that takes a
          *        {@link StreamInput}
-         * @param aggregatorReader reads the {@link PipelineAggregator} from a stream
          * @param parser reads the aggregation builder from XContent
          * @deprecated prefer the ctor that takes a {@link ContextParser}
          */
         @Deprecated
         public PipelineAggregationSpec(
             ParseField name,
             Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
-            Writeable.Reader<? extends PipelineAggregator> aggregatorReader,
             PipelineAggregator.Parser parser
         ) {
             super(name, builderReader, (p, n) -> parser.parse(n, p));
-            this.aggregatorReader = aggregatorReader;
         }
 
         /**
@@ -751,18 +691,15 @@ public PipelineAggregationSpec(
          *        names from other plugins.
          * @param builderReader the reader registered for this aggregation's builder. Typically, a reference to a constructor that takes a
          *        {@link StreamInput}
-         * @param aggregatorReader reads the {@link PipelineAggregator} from a stream
          * @deprecated prefer the ctor that takes a {@link ContextParser}
          */
         @Deprecated
         public PipelineAggregationSpec(
             String name,
             Writeable.Reader<? extends PipelineAggregationBuilder> builderReader,
-            Writeable.Reader<? extends PipelineAggregator> aggregatorReader,
             PipelineAggregator.Parser parser
         ) {
             super(name, builderReader, (p, n) -> parser.parse(n, p));
-            this.aggregatorReader = aggregatorReader;
         }
 
         /**
@@ -781,15 +718,6 @@ public PipelineAggregationSpec addResultReader(String writeableName, Writeable.R
             return this;
         }
 
-        /**
-         * Read the aggregator from a stream.
-         * @deprecated Pipelines implemented after 7.8.0 do not need to be sent across the wire
-         */
-        @Deprecated
-        public Writeable.Reader<? extends PipelineAggregator> getAggregatorReader() {
-            return aggregatorReader;
-        }
-
         /**
          * Get the readers that must be registered for this aggregation's results.
          */