Rollup Search API : Searching both historical rollup and non-rollup data

.github/workflows/multi-node-test-workflow.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Checkout Branch
         uses: actions/checkout@v2
       - name: Run integration tests with multi node config
-        run: ./gradlew integTest -Dtests.class="org.opensearch.indexmanagement.rollup.interceptor.ResponseInterceptorIT"
+        run: ./gradlew integTest
       - name: Upload failed logs
         uses: actions/upload-artifact@v2
         if: failure()

src/main/kotlin/org/opensearch/indexmanagement/rollup/interceptor/ResponseInterceptor.kt

@@ -56,7 +56,6 @@ class ResponseInterceptor(
     val client: Client
 ) : TransportInterceptor {
     private val logger = LogManager.getLogger(javaClass)
-
     override fun interceptSender(sender: TransportInterceptor.AsyncSender): TransportInterceptor.AsyncSender {
         return CustomAsyncSender(sender)
     }
@@ -195,6 +194,7 @@ class ResponseInterceptor(
                 .indices(*rollupIndices) // add all rollup indices to this request
             var maxRolledDateResponse: SearchResponse? = null
             var latch = CountDownLatch(1)
+            logger.info("ronsax sending request to find max rollup time for index: $shardRequestIndex")
             client.search(
                 maxRolledDateRequest,
                 object : ActionListener<SearchResponse> {
@@ -230,6 +230,7 @@ class ResponseInterceptor(
 
             var minLiveDateResponse: SearchResponse? = null
             latch = CountDownLatch(1)
+            logger.info("ronsax sending request to find minLiveData for index: $shardRequestIndex")
             client.search(
                 minLiveDateRequest,
                 object : ActionListener<SearchResponse> {
@@ -256,6 +257,7 @@ class ResponseInterceptor(
                 // If intersection found on rollup index, remove overlap
                 if ((liveDataStartPoint < rollupDataEndPoint) && isShardIndexRollup) {
                     // Start at 0, end at live data
+                    logger.info("ronsax sending request to find rollup endtime for index: $shardRequestIndex")
                     val endTime = getRollupEndTime(liveDataStartPoint, rollupIndices, dateTargetField)
                     return Pair(0L, endTime)
                 }

src/main/kotlin/org/opensearch/indexmanagement/rollup/interceptor/RollupInterceptor.kt

@@ -77,7 +77,7 @@ class RollupInterceptor(
     }
     // Returns Pair<containsRollup: Boolean, rollupJob: RollupJob>
     @Suppress("SpreadOperator")
-    private fun originalSearchContainsRollup(request: ShardSearchRequest): Pair<Boolean, Rollup?> {
+    private fun originalSearchContainsRollup(request: ShardSearchRequest): Pair<Boolean, Rollup?> { // Throwing an error on data streams
         val indices = request.indices().map { it.toString() }.toTypedArray()
         val allIndices = indexNameExpressionResolver
             .concreteIndexNames(clusterService.state(), request.indicesOptions(), *indices)
@@ -169,7 +169,7 @@ class RollupInterceptor(
         }
         // Wraps all existing aggs in bucket aggregation
         // Notifies the response interceptor that was rewritten since agg name is interceptor_interval_data
-        // Edge case if User selected this as the aggregation name :/
+        // Edge case if User selected interceptor_interval_data as the aggregation name :/
         val intervalAgg = AggregationBuilders.dateHistogram("interceptor_interval_data")
             .field(dateSourceField)
             .calendarInterval(DateHistogramInterval(rollupInterval))
@@ -190,14 +190,16 @@ class RollupInterceptor(
         return object : TransportRequestHandler<T> {
             override fun messageReceived(request: T, channel: TransportChannel, task: Task) {
                 if (searchEnabled && request is ShardSearchRequest) {
-                    val (containsRollup, rollupJob) = originalSearchContainsRollup(request)
+                    val isDataStream = (request.indices().any { IndexUtils.isDataStream(it, clusterService.state()) })
                     val shardRequestIndex = request.shardId().indexName
-                    val isRollupIndex = isRollupIndex(shardRequestIndex, clusterService.state())
+                    // isRollupIndex throws an exception if the request is on a data stream
+                    val isRollupIndex = if (isDataStream) false else isRollupIndex(shardRequestIndex, clusterService.state())
+                    val (containsRollup, rollupJob) = if (isDataStream) Pair(false, null) else originalSearchContainsRollup(request)
                     // Only modifies rollup searches and avoids internal client calls
                     if (containsRollup || isRollupIndex) {
                         val (concreteRollupIndicesArray, concreteLiveIndicesArray) = getConcreteIndices(request)
                         /* Avoid infinite interceptor loop:
-                        if there is an internal client call made in the reponse interceptor there is only 1 index.
+                        if there is an internal client call made in the response interceptor there is only 1 index.
                         Therefore, conditions are not met for api to combine rollup and live data
                          */
                         val isMultiSearch = (concreteRollupIndicesArray.isNotEmpty() && concreteLiveIndicesArray.isNotEmpty())