Merge branch 'main' of github.com:kserve/modelmesh

Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM registry.access.redhat.com/ubi8/ubi-minimal:8.6 as build_base
+FROM registry.access.redhat.com/ubi8/ubi-minimal:8.8 as build_base
 
 # https://docs.docker.com/engine/reference/builder/#automatic-platform-args-in-the-global-scope
 # don't provide "default" values (e.g. 'ARG TARGETARCH=amd64') for non-buildx environments,

pom.xml

@@ -61,11 +61,11 @@
     <netty-version>4.1.100.Final</netty-version>
     <litelinks-version>1.7.2</litelinks-version>
     <kv-utils-version>0.5.1</kv-utils-version>
-    <etcd-java-version>0.0.22</etcd-java-version>
-    <protobuf-version>3.24.4</protobuf-version>
+    <etcd-java-version>0.0.24</etcd-java-version>
+    <protobuf-version>3.25.0</protobuf-version>
     <annotation-version>9.0.82</annotation-version>
     <guava-version>32.1.3-jre</guava-version>
-    <jackson-databind-version>2.15.3</jackson-databind-version>
+    <jackson-databind-version>2.16.0</jackson-databind-version>
     <gson-version>2.10.1</gson-version>
     <thrift-version>0.19.0</thrift-version>
     <eclipse-collections-version>11.1.0</eclipse-collections-version>
@@ -75,7 +75,10 @@
          since we have some custom optimized extensions to this -->
     <prometheus-version>0.9.0</prometheus-version>
     <bouncycastle-version>1.74</bouncycastle-version>
-    <junit-version>5.10.0</junit-version>
+    <junit-version>5.10.1</junit-version>
+
+    <zookeeper-version>3.7.2</zookeeper-version>
+    <curator-version>5.3.0</curator-version>
 
     <dockerhome>${project.build.directory}/dockerhome</dockerhome>
     <skipTests>false</skipTests>
@@ -459,7 +462,7 @@
     <dependency>
       <groupId>org.apache.curator</groupId>
       <artifactId>curator-test</artifactId>
-      <version>2.13.0</version>
+      <version>${curator-version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -524,6 +527,17 @@
         <artifactId>libthrift</artifactId>
         <version>${thrift-version}</version>
       </dependency>
+      <dependency>
+        <!-- override litelinks' versions with newer to avoid CVE -->
+        <groupId>org.apache.zookeeper</groupId>
+        <artifactId>zookeeper</artifactId>
+        <version>${zookeeper-version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.curator</groupId>
+        <artifactId>curator-recipes</artifactId>
+        <version>${curator-version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 </project>

src/main/java/com/ibm/watson/modelmesh/Metrics.java

@@ -172,7 +172,7 @@ public PrometheusMetrics(Map<String, String> params, Map<String, String> infoMet
             int port = 2112;
             boolean shortNames = true;
             boolean https = true;
-            boolean perModelMetricsEnabled = true;
+            boolean perModelMetricsEnabled = false;
             String memMetrics = "all"; // default to all
             for (Entry<String, String> ent : params.entrySet()) {
                 switch (ent.getKey()) {

src/main/java/com/ibm/watson/modelmesh/ModelMeshApi.java

@@ -725,6 +725,7 @@ public void onHalfClose() {
                 String vModelId = null;
                 String requestId = null;
                 ModelResponse response = null;
+                ByteBuf responsePayload = null;
                 try (InterruptingListener cancelListener = newInterruptingListener()) {
                     if (logHeaders != null) {
                         logHeaders.addToMDC(headers); // MDC cleared in finally block
@@ -767,18 +768,20 @@ public void onHalfClose() {
                     } finally {
                         if (payloadProcessor != null) {
                             processPayload(reqMessage.readerIndex(reqReaderIndex),
-                                    requestId, resolvedModelId, methodName, headers, null, true);
+                                    requestId, resolvedModelId, methodName, headers, null);
                         } else {
                             releaseReqMessage();
                         }
                         reqMessage = null; // ownership released or transferred
                     }
 
-                    respReaderIndex = response.data.readerIndex();
                     respSize = response.data.readableBytes();
                     call.sendHeaders(response.metadata);
+                    if (payloadProcessor != null) {
+                        responsePayload = response.data.retainedSlice();
+                    }
                     call.sendMessage(response.data);
-                    // response is released via ReleaseAfterResponse.releaseAll()
+                    // final response refcount is released via ReleaseAfterResponse.releaseAll()
                     status = OK;
                 } catch (Exception e) {
                     status = toStatus(e);
@@ -795,17 +798,13 @@ public void onHalfClose() {
                         evictMethodDescriptor(methodName);
                     }
                 } finally {
-                    final boolean releaseResponse = status != OK;
                     if (payloadProcessor != null) {
-                        ByteBuf data = null;
-                        Metadata metadata = null;
-                        if (response != null) {
-                            data = response.data.readerIndex(respReaderIndex);
-                            metadata = response.metadata;
-                        }
-                        processPayload(data, requestId, resolvedModelId, methodName, metadata, status, releaseResponse);
-                    } else if (releaseResponse && response != null) {
-                        response.release();
+                        Metadata metadata = response != null ? response.metadata : null;
+                        processPayload(responsePayload, requestId, resolvedModelId, methodName, metadata, status);
+                    }
+                    if (status != OK && response != null) {
+                        // An additional release is required if we call.sendMessage() wasn't sucessful
+                        response.data.release();
                     }
                     ReleaseAfterResponse.releaseAll();
                     clearThreadLocals();
@@ -820,23 +819,21 @@ public void onHalfClose() {
             }
 
             /**
-             * Invoke PayloadProcessor on the request/response data
+             * Invoke PayloadProcessor on the request/response data. This method takes ownership
+             * of the passed-in {@code ByteBuf}.
+             *
              * @param data the binary data
              * @param payloadId the id of the request
              * @param modelId the id of the model
              * @param methodName the name of the invoked method
              * @param metadata the method name metadata
              * @param status null for requests, non-null for responses
-             * @param takeOwnership whether the processor should take ownership
              */
             private void processPayload(ByteBuf data, String payloadId, String modelId, String methodName,
-                                        Metadata metadata, io.grpc.Status status, boolean takeOwnership) {
+                                        Metadata metadata, io.grpc.Status status) {
                 Payload payload = null;
                 try {
                     assert payloadProcessor != null;
-                    if (!takeOwnership) {
-                        ReferenceCountUtil.retain(data);
-                    }
                     payload = new Payload(payloadId, modelId, methodName, metadata, data, status);
                     if (payloadProcessor.process(payload)) {
                         data = null; // ownership transferred

src/main/scripts/start.sh

@@ -367,9 +367,6 @@ echo "SHUTDOWN_TIMEOUT_MS=$SHUTDOWN_TIMEOUT_MS"
 
 LITELINKS_ARGS="-Dlitelinks.cancel_on_client_close=true -Dlitelinks.threadcontexts=log_mdc -Dlitelinks.shutdown_timeout_ms=${SHUTDOWN_TIMEOUT_MS} -Dlitelinks.produce_pooled_bytebufs=true"
 
-# have litelinks use OpenSSL instead of JDK TLS implementation (faster)
-LL_OPENSSL_ARG="-Dlitelinks.ssl.use_jdk=false"
-
 # These two args are needed to use netty's off-the-books direct buffer allocation
 NETTY_DIRECTBUF_ARGS="-Dio.netty.tryReflectionSetAccessible=true --add-opens=java.base/java.nio=ALL-UNNAMED"
 # this defaults to equal max heap, which can result in container OOMKilled
@@ -414,7 +411,6 @@ exec $JAVA_HOME/bin/java -cp "$LL_JAR:lib/*" -XX:+UnlockExperimentalVMOptions -X
  ${JAVA_MAXDIRECT_ARG} ${NETTY_MAXDIRECT_ARG} ${NETTY_DISABLE_CHECK_ARGS} \
  ${GRPC_USE_SHARED_ALLOC_ARG} \
  ${SSL_PK_ARG} ${TRUSTSTORE_ARG} ${LITELINKS_ARGS} ${CUSTOM_JVM_ARGS} \
- $LL_OPENSSL_ARG \
  $PRIVATE_ENDPOINT_ARG \
  $LOG_CONFIG_ARG $LOG_PERF_ARGS \
  com.ibm.watson.litelinks.server.LitelinksService \