update IngestStats version guards to 7.6 and re-enable bwc tests (#48768)

* update IngestStats version guards to 7.6

after backporting #48485 in #48661, version guards in master can
be updated from 8.0 to 7.6 for IngestStats serialization guards

* forward port bwc tests in ingeststats

* fix null ingeststats when reading the stream

* re-enable bwc tests

Author: talevy

build.gradle

@@ -206,8 +206,8 @@ task verifyVersions {
  * after the backport of the backcompat code is complete.
  */
 
-boolean bwc_tests_enabled = false
-final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/pull/48661<Paste>" /* place a PR link here when committing bwc changes */
+boolean bwc_tests_enabled = true
+final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java

@@ -36,7 +36,6 @@
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.discovery.DiscoveryModule;
-import org.elasticsearch.ingest.IngestStats;
 import org.elasticsearch.monitor.fs.FsInfo;
 import org.elasticsearch.monitor.jvm.JvmInfo;
 import org.elasticsearch.plugins.PluginInfo;

server/src/main/java/org/elasticsearch/ingest/IngestStats.java

@@ -68,8 +68,8 @@ public IngestStats(StreamInput in) throws IOException {
             List<ProcessorStat> processorStatsPerPipeline = new ArrayList<>(processorsSize);
             for (int j = 0; j < processorsSize; j++) {
                 String processorName = in.readString();
-                String processorType = null;
-                if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+                String processorType = "_NOT_AVAILABLE";
+                if (in.getVersion().onOrAfter(Version.V_7_6_0)) {
                     processorType = in.readString();
                 }
                 Stats processorStat = new Stats(in);
@@ -93,7 +93,7 @@ public void writeTo(StreamOutput out) throws IOException {
                 out.writeVInt(processorStatsForPipeline.size());
                 for (ProcessorStat processorStat : processorStatsForPipeline) {
                     out.writeString(processorStat.getName());
-                    if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+                    if (out.getVersion().onOrAfter(Version.V_7_6_0)) {
                         out.writeString(processorStat.getType());
                     }
                     processorStat.getStats().writeTo(out);

server/src/test/java/org/elasticsearch/ingest/IngestStatsTests.java

@@ -19,10 +19,12 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -40,7 +42,24 @@ public void testSerialization() throws IOException {
         Map<String, List<IngestStats.ProcessorStat>> processorStats = createProcessorStats(pipelineStats);
         IngestStats ingestStats = new IngestStats(totalStats, pipelineStats, processorStats);
         IngestStats serializedStats = serialize(ingestStats);
-        assertIngestStats(ingestStats, serializedStats, true);
+        assertIngestStats(ingestStats, serializedStats, true, true);
+    }
+
+    public void testBWCIngestProcessorTypeStats() throws IOException {
+        IngestStats.Stats totalStats = new IngestStats.Stats(50, 100, 200, 300);
+        List<IngestStats.PipelineStat> pipelineStats = createPipelineStats();
+        Map<String, List<IngestStats.ProcessorStat>> processorStats = createProcessorStats(pipelineStats);
+        IngestStats expectedIngestStats = new IngestStats(totalStats, pipelineStats, processorStats);
+
+        //legacy output logic
+        BytesStreamOutput out = new BytesStreamOutput();
+        out.setVersion(VersionUtils.getPreviousVersion(Version.V_7_6_0));
+        expectedIngestStats.writeTo(out);
+
+        StreamInput in = out.bytes().streamInput();
+        in.setVersion(VersionUtils.getPreviousVersion(Version.V_7_6_0));
+        IngestStats serializedStats = new IngestStats(in);
+        assertIngestStats(expectedIngestStats, serializedStats, true, false);
     }
 
     private List<IngestStats.PipelineStat> createPipelineStats() {
@@ -70,7 +89,8 @@ private IngestStats serialize(IngestStats stats) throws IOException {
         return new IngestStats(in);
     }
 
-    private void assertIngestStats(IngestStats ingestStats, IngestStats serializedStats, boolean expectProcessors){
+    private void assertIngestStats(IngestStats ingestStats, IngestStats serializedStats, boolean expectProcessors,
+                                   boolean expectProcessorTypes){
         assertNotSame(ingestStats, serializedStats);
         assertNotSame(ingestStats.getTotalStats(), serializedStats.getTotalStats());
         assertNotSame(ingestStats.getPipelineStats(), serializedStats.getPipelineStats());
@@ -92,6 +112,11 @@ private void assertIngestStats(IngestStats ingestStats, IngestStats serializedSt
                     for (IngestStats.ProcessorStat serializedProcessorStat : serializedProcessorStats) {
                         IngestStats.ProcessorStat ps = it.next();
                         assertEquals(ps.getName(), serializedProcessorStat.getName());
+                        if (expectProcessorTypes) {
+                            assertEquals(ps.getType(), serializedProcessorStat.getType());
+                        } else {
+                            assertEquals("_NOT_AVAILABLE", serializedProcessorStat.getType());
+                        }
                         assertStats(ps.getStats(), serializedProcessorStat.getStats());
                     }
                     assertFalse(it.hasNext());