Feature/spline 1255 message size info

build/parent-pom/pom.xml

@@ -46,7 +46,7 @@
         <json4s.version>3.6.7</json4s.version>
         <spring.version>5.2.2.RELEASE</spring.version>
         <springfox-swagger.version>2.9.2</springfox-swagger.version>
-        <jackson.version>2.12.1</jackson.version>
+        <jackson.version>2.12.2</jackson.version>
         <finatra.jackson.version>21.5.0</finatra.jackson.version>
     </properties>
 
@@ -293,7 +293,7 @@
             <dependency>
                 <groupId>org.apache.commons</groupId>
                 <artifactId>commons-lang3</artifactId>
-                <version>3.9</version>
+                <version>3.13.0</version>
             </dependency>
             <dependency>
                 <groupId>commons-configuration</groupId>
@@ -303,7 +303,7 @@
             <dependency>
                 <groupId>commons-io</groupId>
                 <artifactId>commons-io</artifactId>
-                <version>2.6</version>
+                <version>2.12.0</version>
             </dependency>
 
             <dependency>

kafka-gateway/pom.xml

@@ -92,9 +92,6 @@
                 <artifactId>docker-maven-plugin</artifactId>
                 <configuration>
                     <skip>false</skip>
-                    <buildArgs>
-                        <PROJECT_BUILD_FINAL_NAME>${project.build.finalName}</PROJECT_BUILD_FINAL_NAME>
-                    </buildArgs>
                 </configuration>
             </plugin>
         </plugins>

persistence/pom.xml

@@ -62,7 +62,7 @@
         <dependency>
             <groupId>com.arangodb</groupId>
             <artifactId>arangodb-java-driver</artifactId>
-            <version>6.16.1</version>
+            <version>6.25.0</version>
         </dependency>
         <dependency>
             <groupId>com.arangodb</groupId>

producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/controller/ExecutionPlansControllerMessageLengthCapturingAspect.scala

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2020 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.spline.producer.rest.controller
+
+import org.aspectj.lang.ProceedingJoinPoint
+import org.aspectj.lang.annotation.{Around, Aspect, Pointcut}
+import org.slf4s.Logging
+import org.springframework.stereotype.Component
+import org.springframework.web.context.request.{RequestContextHolder, ServletRequestAttributes}
+import za.co.absa.spline.producer.model.{ExecutionPlan, v1_1}
+import za.co.absa.spline.producer.rest.filter.MessageLengthCapturingFilter
+import za.co.absa.spline.producer.rest.filter.MessageLengthCapturingFilter.ReadOnlyCounter
+
+@Aspect
+@Component
+class ExecutionPlansControllerMessageLengthCapturingAspect extends Logging {
+
+  @Pointcut("execution(public * za.co.absa.spline.producer.rest.controller.*Controller.*(..))")
+  def publicControllerMethods(): Unit = {}
+
+  @Pointcut("execution(* *(.., za.co.absa.spline.producer.model.ExecutionPlan, ..))")
+  def acceptingExecutionPlanV1(): Unit = {}
+
+  @Pointcut("execution(* *(.., za.co.absa.spline.producer.model.v1_1.ExecutionPlan, ..))")
+  def acceptingExecutionPlan(): Unit = {}
+
+  @Around("publicControllerMethods() && (acceptingExecutionPlan() || acceptingExecutionPlanV1())")
+  def aroundAdvice(jp: ProceedingJoinPoint): AnyRef = {
+    val origArgs = jp.getArgs
+    val fixedArgs = origArgs.map {
+      case ep: ExecutionPlan =>
+        ep.copy(extraInfo = withMessageLengthInfo(ep.extraInfo))
+      case ep: v1_1.ExecutionPlan =>
+        ep.copy(extraInfo = withMessageLengthInfo(ep.extraInfo))
+      case x => x
+    }
+    jp.proceed(fixedArgs)
+  }
+
+  private def withMessageLengthInfo(m: Map[String, Any]): Map[String, Any] = {
+    val req = RequestContextHolder.getRequestAttributes.asInstanceOf[ServletRequestAttributes].getRequest
+    val counters = MessageLengthCapturingFilter.getCounters(req).toArray
+    m + ("__spline_msg_size" -> counters.map(_.count))
+  }
+}
+
+

rest-gateway/src/main/scala/za/co/absa/spline/gateway/rest/filter/GzipFilter.scala → producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/filter/GzipFilter.scala

@@ -1,6 +1,5 @@
 /*
- * Copyright 2020 ABSA Group Limited
- *
+ * Copyright 2023 ABSA Group Limited
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -14,13 +13,15 @@
  * limitations under the License.
  */
 
-package za.co.absa.spline.gateway.rest.filter
+package za.co.absa.spline.producer.rest.filter
 
-import javax.servlet._
-import javax.servlet.http.HttpServletRequest
 import org.springframework.http.HttpHeaders
 import za.co.absa.spline.producer.rest.HttpConstants.Encoding
 
+import java.util.zip.GZIPInputStream
+import javax.servlet._
+import javax.servlet.http.HttpServletRequest
+
 /**
  * Filter for decompressing gziped Http requests
  *
@@ -30,7 +31,7 @@ class GzipFilter extends Filter {
   override def doFilter(request: ServletRequest, response: ServletResponse, chain: FilterChain): Unit = {
 
     val newRequest = request match {
-      case r: HttpServletRequest if isCompressed(r) => new GZIPRequestWrapper(r)
+      case r: HttpServletRequest if isCompressed(r) => new HttpRequestWrapper(r, new GZIPInputStream(r.getInputStream))
       case _ => request
     }
 

rest-gateway/src/main/scala/za/co/absa/spline/gateway/rest/filter/GZIPRequestWrapper.scala → producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/filter/HttpRequestWrapper.scala

@@ -1,6 +1,5 @@
 /*
- * Copyright 2020 ABSA Group Limited
- *
+ * Copyright 2023 ABSA Group Limited
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -14,20 +13,19 @@
  * limitations under the License.
  */
 
-package za.co.absa.spline.gateway.rest.filter
-
-import java.io.{BufferedReader, InputStreamReader}
+package za.co.absa.spline.producer.rest.filter
 
+import java.io.{BufferedReader, InputStream, InputStreamReader}
 import javax.servlet.ServletInputStream
 import javax.servlet.http.{HttpServletRequest, HttpServletRequestWrapper}
 
 
-final class GZIPRequestWrapper(val request: HttpServletRequest) extends HttpServletRequestWrapper(request) {
+final class HttpRequestWrapper(request: HttpServletRequest, stream: InputStream)
+  extends HttpServletRequestWrapper(request) {
 
-  val stream = new GZIPServletInputStream(request.getInputStream)
-  val reader = new BufferedReader(new InputStreamReader(stream))
+  private val reader = new BufferedReader(new InputStreamReader(stream))
 
+  override def getInputStream: ServletInputStream = new ServletInputStreamAdapter(stream)
 
-  override def getInputStream: ServletInputStream = stream
   override def getReader: BufferedReader = reader
-}
+}

producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/filter/MessageLengthCapturingFilter.scala

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2020 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.spline.producer.rest.filter
+
+import za.co.absa.spline.producer.rest.filter.MessageLengthCapturingFilter.{LengthCountingInputStreamWrapper, getCounters}
+
+import java.io.InputStream
+import javax.servlet._
+import javax.servlet.http.HttpServletRequest
+import scala.collection.mutable
+
+class MessageLengthCapturingFilter extends Filter {
+  override def doFilter(request: ServletRequest, response: ServletResponse, chain: FilterChain): Unit = {
+    val newRequest = request match {
+      case r: HttpServletRequest =>
+        val inputStreamWrapper = new LengthCountingInputStreamWrapper(r.getInputStream)
+        getCounters(r) += inputStreamWrapper.lengthCounter
+        new HttpRequestWrapper(r, inputStreamWrapper)
+      case _ => request
+    }
+    chain.doFilter(newRequest, response)
+  }
+
+  override def init(config: FilterConfig): Unit = {
+    // nothing to do here
+  }
+
+  override def destroy(): Unit = {
+    // nothing to do here
+  }
+}
+
+object MessageLengthCapturingFilter {
+  private val CountersRequestAttributeKey: String = s"${classOf[MessageLengthCapturingFilter].getName}.counters"
+
+  def getCounters(r: ServletRequest): mutable.Buffer[ReadOnlyCounter] = {
+    val countersAttrOrNull = r.getAttribute(CountersRequestAttributeKey).asInstanceOf[mutable.Buffer[ReadOnlyCounter]]
+    val counters = Option(countersAttrOrNull).getOrElse(mutable.Buffer.empty[ReadOnlyCounter])
+    if (counters.isEmpty) r.setAttribute(CountersRequestAttributeKey, counters)
+    counters
+  }
+
+  trait ReadOnlyCounter {
+    def count: Int
+  }
+
+  class LengthCountingInputStreamWrapper(r: InputStream) extends InputStream {
+    private var _bytesReadCount: Int = 0
+
+    val lengthCounter: ReadOnlyCounter = new ReadOnlyCounter {
+      override def count: Int = _bytesReadCount
+    }
+
+    override def read(): Int = {
+      _bytesReadCount += 1
+      r.read()
+    }
+  }
+}

rest-gateway/src/main/scala/za/co/absa/spline/gateway/rest/filter/GZIPServletInputStream.scala → producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/filter/ServletInputStreamAdapter.scala

@@ -1,6 +1,5 @@
 /*
- * Copyright 2020 ABSA Group Limited
- *
+ * Copyright 2023 ABSA Group Limited
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
@@ -14,16 +13,12 @@
  * limitations under the License.
  */
 
-package za.co.absa.spline.gateway.rest.filter
-
-import java.util.zip.GZIPInputStream
+package za.co.absa.spline.producer.rest.filter
 
+import java.io.InputStream
 import javax.servlet.{ReadListener, ServletInputStream}
 
-final class GZIPServletInputStream(val inputStream: ServletInputStream) extends ServletInputStream {
-
-  val gzipStream = new GZIPInputStream(inputStream)
-
+final class ServletInputStreamAdapter(val gzipStream: InputStream) extends ServletInputStream {
 
   override def read: Int = gzipStream.read
 

rest-gateway/pom.xml

@@ -169,9 +169,6 @@
                 <artifactId>docker-maven-plugin</artifactId>
                 <configuration>
                     <skip>false</skip>
-                    <buildArgs>
-                        <PROJECT_BUILD_FINAL_NAME>${project.build.finalName}</PROJECT_BUILD_FINAL_NAME>
-                    </buildArgs>
                 </configuration>
             </plugin>
         </plugins>

rest-gateway/src/main/scala/za/co/absa/spline/gateway/rest/AppInitializer.scala

@@ -24,9 +24,9 @@ import za.co.absa.spline.common.webmvc.cors.PermissiveCorsFilter
 import za.co.absa.spline.common.webmvc.diagnostics.{DiagnosticsRESTConfig, RootWebContextConfig}
 import za.co.absa.spline.consumer.rest.ConsumerRESTConfig
 import za.co.absa.spline.consumer.service.ConsumerServicesConfig
-import za.co.absa.spline.gateway.rest.filter.GzipFilter
 import za.co.absa.spline.persistence.ArangoRepoConfig
 import za.co.absa.spline.producer.rest.ProducerRESTConfig
+import za.co.absa.spline.producer.rest.filter.{GzipFilter, MessageLengthCapturingFilter}
 import za.co.absa.spline.producer.service.ProducerServicesConfig
 
 import javax.servlet.ServletContext
@@ -43,7 +43,9 @@ object AppInitializer extends WebApplicationInitializer {
       }))
 
     registerFilter[PermissiveCorsFilter](container, "CORSFilter", "/*")
+    registerFilter[MessageLengthCapturingFilter](container, "MessageSizeCapturingFilter_before_gzip", "/*")
     registerFilter[GzipFilter](container, "GzipFilter", "/*")
+    registerFilter[MessageLengthCapturingFilter](container, "MessageSizeCapturingFilter_after_gzip", "/*")
 
     registerRESTDispatcher[ConsumerRESTConfig](container, "consumer")
     registerRESTDispatcher[ProducerRESTConfig](container, "producer")