Cat Nodes API with Protobuf

.gitignore

@@ -64,4 +64,4 @@ testfixtures_shared/
 .ci/jobs/
 
 # build files generated
-doc-tools/missing-doclet/bin/
+doc-tools/missing-doclet/bin/

libs/core/src/main/java/org/opensearch/core/common/io/stream/ProtobufWriteable.java

@@ -0,0 +1,70 @@
+/*
+* SPDX-License-Identifier: Apache-2.0
+*
+* The OpenSearch Contributors require contributions made to
+* this file be licensed under the Apache-2.0 license or a
+* compatible open source license.
+*/
+
+package org.opensearch.core.common.io.stream;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Implementers can be written to write to output and read from input using Protobuf.
+*
+* @opensearch.internal
+*/
+public interface ProtobufWriteable {
+
+    /**
+     * Write this into the stream output.
+    */
+    public void writeTo(OutputStream out) throws IOException;
+
+    /**
+     * Reference to a method that can write some object to a {@link OutputStream}.
+     * Most classes should implement {@link ProtobufWriteable} and the {@link ProtobufWriteable#writeTo(OutputStream)} method should <em>use</em>
+     * {@link OutputStream} methods directly or this indirectly:
+     * <pre><code>
+     * public void writeTo(OutputStream out) throws IOException {
+     *     out.writeVInt(someValue);
+     * }
+     * </code></pre>
+     */
+    @FunctionalInterface
+    interface Writer<V> {
+
+        /**
+         * Write {@code V}-type {@code value} to the {@code out}put stream.
+        *
+        * @param out Output to write the {@code value} too
+        * @param value The value to add
+        */
+        void write(OutputStream out, V value) throws IOException;
+
+    }
+
+    /**
+     * Reference to a method that can read some object from a stream. By convention this is a constructor that takes
+     * {@linkplain byte[]} as an argument for most classes and a static method for things like enums.
+     * <pre><code>
+     * public MyClass(final byte[] in) throws IOException {
+     *     this.someValue = in.readVInt();
+     * }
+     * </code></pre>
+     */
+    @FunctionalInterface
+    interface Reader<V> {
+
+        /**
+         * Read {@code V}-type value from a stream.
+        *
+        * @param in Input to read the value from
+        */
+        V read(byte[] in) throws IOException;
+
+    }
+
+}

libs/core/src/main/java/org/opensearch/core/transport/TransportMessage.java

@@ -32,6 +32,7 @@
 
 package org.opensearch.core.transport;
 
+import org.opensearch.core.common.io.stream.ProtobufWriteable;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.Writeable;
 import org.opensearch.core.common.transport.TransportAddress;
@@ -41,7 +42,7 @@
  *
  * @opensearch.internal
  */
-public abstract class TransportMessage implements Writeable {
+public abstract class TransportMessage implements Writeable, ProtobufWriteable {
 
     private TransportAddress remoteAddress;
 
@@ -63,4 +64,10 @@ public TransportMessage() {}
      * currently a no-op
      */
     public TransportMessage(StreamInput in) {}
+
+    /**
+    * Constructs a new transport message with the data from the {@link byte[]}. This is
+    * currently a no-op
+    */
+    public TransportMessage(byte[] in) {}
 }

libs/core/src/main/java/org/opensearch/core/transport/TransportResponse.java

@@ -36,6 +36,7 @@
 import org.opensearch.core.common.io.stream.StreamOutput;
 
 import java.io.IOException;
+import java.io.OutputStream;
 
 /**
  * Response over the transport interface
@@ -58,6 +59,21 @@ public TransportResponse(StreamInput in) throws IOException {
         super(in);
     }
 
+    /**
+     * Constructs a new transport response with the data from the {@link StreamInput}. This is
+    * currently a no-op. However, this exists to allow extenders to call <code>super(in)</code>
+    * so that reading can mirror writing where we often call <code>super.writeTo(out)</code>.
+    */
+    public TransportResponse(byte[] in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public void writeTo(OutputStream out) throws IOException {}
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {}
+
     /**
      * Empty transport response
      *
@@ -73,5 +89,8 @@ public String toString() {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {}
+
+        @Override
+        public void writeTo(OutputStream out) throws IOException {}
     }
 }

modules/transport-netty4/src/main/java/org/opensearch/transport/netty4/Netty4MessageChannelHandler.java

@@ -77,7 +77,8 @@ final class Netty4MessageChannelHandler extends ChannelDuplexHandler {
             threadPool::relativeTimeInMillis,
             transport.getInflightBreaker(),
             requestHandlers::getHandler,
-            transport::inboundMessage
+            transport::inboundMessage,
+            transport::inboundMessageProtobuf
         );
     }
 

plugins/events-correlation-engine/src/test/java/org/opensearch/plugin/correlation/core/index/mapper/CorrelationVectorFieldMapperTests.java

@@ -190,7 +190,7 @@ public void testTypeParser_parse_fromCorrelationParamsContext_InvalidVectorSimil
 
     /**
      * test parseCreateField in CorrelationVectorFieldMapper
-     * @throws IOException
+     * @throws IOException io exception
      */
     public void testCorrelationVectorFieldMapper_parseCreateField() throws IOException {
         String fieldName = "test-field-name";

plugins/events-correlation-engine/src/test/java/org/opensearch/plugin/correlation/core/index/query/CorrelationQueryBuilderTests.java

@@ -203,7 +203,7 @@ public void testDoToQueryInvalidFieldType() {
 
     /**
      * test serialization of Correlation Query Builder
-     * @throws Exception
+     * @throws Exception exception
      */
     public void testSerialization() throws Exception {
         assertSerialization(Optional.empty());

plugins/transport-nio/src/main/java/org/opensearch/transport/nio/TcpReadWriteHandler.java

@@ -68,7 +68,8 @@ public TcpReadWriteHandler(NioTcpChannel channel, PageCacheRecycler recycler, Tc
             threadPool::relativeTimeInMillis,
             breaker,
             requestHandlers::getHandler,
-            transport::inboundMessage
+            transport::inboundMessage,
+            transport::inboundMessageProtobuf
         );
     }
 

qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/IndexingIT.java

@@ -248,7 +248,7 @@ public void testIndexing() throws IOException, ParseException {
      * This test verifies that during rolling upgrades the segment replication does not break when replica shards can
      * be running on older codec versions.
      *
-     * @throws Exception
+     * @throws Exception exception
      */
     @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/8322")
     public void testIndexingWithSegRep() throws Exception {

server/build.gradle

@@ -394,6 +394,8 @@ tasks.named("licenseHeaders").configure {
     excludes << 'org/opensearch/client/documentation/placeholder.txt'
     // Ignore for protobuf generated code
     excludes << 'org/opensearch/extensions/proto/*'
+    excludes << 'org/opensearch/tasks/proto/*'
+    excludes << 'org/opensearch/server/proto/*'
 }
 
 tasks.test {

server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/CancellableTasksIT.java

@@ -544,6 +544,12 @@ public String executor() {
                                 public TestResponse read(StreamInput in) throws IOException {
                                     return new TestResponse(in);
                                 }
+
+                                @Override
+                                public TestResponse read(byte[] in) throws IOException {
+                                    // TODO Auto-generated method stub
+                                    throw new UnsupportedOperationException("Unimplemented method 'read'");
+                                }
                             }
                         );
                     }

server/src/main/java/org/opensearch/action/ActionListenerResponseHandler.java

@@ -88,4 +88,10 @@ public Response read(StreamInput in) throws IOException {
     public String toString() {
         return super.toString() + "/" + listener;
     }
+
+    @Override
+    public Response read(byte[] in) throws IOException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'read'");
+    }
 }