Merge branch 'main' into wlm/rule-sync

Signed-off-by: Kaushal Kumar <[email protected]>

Author: kaushalmahi12

CHANGELOG.md

@@ -11,9 +11,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Update API of Message in index to add the timestamp for lag calculation in ingestion polling ([#17977](https://github.com/opensearch-project/OpenSearch/pull/17977/))
 - Add composite directory factory ([#17988](https://github.com/opensearch-project/OpenSearch/pull/17988))
 - [Rule based auto-tagging] Add refresh based synchronization service for `Rule`s ([#18128](https://github.com/opensearch-project/OpenSearch/pull/18128))
+- Add pull-based ingestion error metrics and make internal queue size configurable ([#18088](https://github.com/opensearch-project/OpenSearch/pull/18088))
 
 ### Changed
-- Avoid invalid retries in multiple replicas when querying [#17370](https://github.com/opensearch-project/OpenSearch/pull/17370)
 
 ### Dependencies
 - Bump `com.google.code.gson:gson` from 2.12.1 to 2.13.0 ([#17923](https://github.com/opensearch-project/OpenSearch/pull/17923))

build.gradle

@@ -551,7 +551,6 @@ subprojects {
         includeClasses.add("org.opensearch.snapshots.SnapshotStatusApisIT")
         includeClasses.add("org.opensearch.test.rest.ClientYamlTestSuiteIT")
         includeClasses.add("org.opensearch.upgrade.DetectEsInstallationTaskTests")
-        includeClasses.add("org.opensearch.cluster.MinimumClusterManagerNodesIT")
       }
     }
   }

distribution/packages/src/common/systemd/opensearch.service

@@ -26,6 +26,9 @@ WorkingDirectory=/usr/share/opensearch
 User=opensearch
 Group=opensearch
 
+ExecStartPre=/bin/mkdir -p /dev/shm/performanceanalyzer
+ExecStartPre=/bin/chown opensearch:opensearch /dev/shm/performanceanalyzer
+
 ExecStart=/usr/share/opensearch/bin/systemd-entrypoint -p ${PID_DIR}/opensearch.pid --quiet
 
 # StandardOutput is configured to redirect to journalctl since
@@ -130,6 +133,7 @@ RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
 
 ReadWritePaths=/var/log/opensearch
 ReadWritePaths=/var/lib/opensearch
+ReadWritePaths=/dev/shm/
 ReadWritePaths=-/etc/opensearch
 ReadWritePaths=-/mnt/snapshots
 
@@ -149,17 +153,25 @@ RestrictNamespaces=true
 NoNewPrivileges=true
 
 # Memory and execution protection
-MemoryDenyWriteExecute=true           # Prevent creating writable executable memory mappings
-SystemCallArchitectures=native        # Allow only native system calls
-KeyringMode=private                   # Service does not share key material with other services
-LockPersonality=true                  # Prevent changing ABI personality
-RestrictSUIDSGID=true                 # Prevent creating SUID/SGID files
-RestrictRealtime=true                 # Prevent acquiring realtime scheduling
-ProtectHostname=true                  # Prevent changes to system hostname
-ProtectKernelLogs=true                # Prevent reading/writing kernel logs
-ProtectClock=true                     # Prevent tampering with the system clock
+
+# Allow only native system calls
+SystemCallArchitectures=native
+# Service does not share key material with other services
+KeyringMode=private
+# Prevent changing ABI personality
+LockPersonality=true
+# Prevent creating SUID/SGID files
+RestrictSUIDSGID=true
+# Prevent acquiring realtime scheduling
+RestrictRealtime=true
+# Prevent changes to system hostname
+ProtectHostname=true
+# Prevent reading/writing kernel logs
+ProtectKernelLogs=true
+# Prevent tampering with the system clock
+ProtectClock=true
 
 [Install]
 WantedBy=multi-user.target
 
-# Built for ${project.name}-${project.version} (${project.name})
+# Built for ${project.name}-${project.version} (${project.name})

libs/core/src/main/java/org/opensearch/OpenSearchException.java

@@ -302,12 +302,8 @@ public Map<String, List<String>> getHeaders() {
      * Returns the rest status code associated with this exception.
      */
     public RestStatus status() {
-        return status(this);
-    }
-
-    public static RestStatus status(Throwable t) {
-        Throwable cause = ExceptionsHelper.unwrapCause(t);
-        if (cause == t) {
+        Throwable cause = unwrapCause();
+        if (cause == this) {
             return RestStatus.INTERNAL_SERVER_ERROR;
         } else {
             return ExceptionsHelper.status(cause);

plugins/ingestion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/RemoteStoreKafkaIT.java

@@ -22,6 +22,7 @@
 import org.opensearch.index.query.BoolQueryBuilder;
 import org.opensearch.index.query.RangeQueryBuilder;
 import org.opensearch.index.query.TermQueryBuilder;
+import org.opensearch.indices.pollingingest.PollingIngestStats;
 import org.opensearch.test.InternalTestCluster;
 import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.transport.client.Requests;
@@ -135,6 +136,8 @@ public void testErrorStrategy() throws Exception {
         // malformed message
         produceData("2", "", "");
         produceData("3", "name3", "25");
+        produceData("{\"_op_type\":\"invalid\",\"_source\":{\"name\":\"name4\", \"age\": 25}}");
+        produceData("5", "name5", "25");
 
         internalCluster().startClusterManagerOnlyNode();
         final String node = internalCluster().startDataOnlyNode();
@@ -147,6 +150,7 @@ public void testErrorStrategy() throws Exception {
                 .put("ingestion_source.type", "kafka")
                 .put("ingestion_source.error_strategy", "block")
                 .put("ingestion_source.pointer.init.reset", "earliest")
+                .put("ingestion_source.internal_queue_size", "1000")
                 .put("ingestion_source.param.topic", topicName)
                 .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
                 .put("index.replication.type", "SEGMENT")
@@ -165,7 +169,15 @@ public void testErrorStrategy() throws Exception {
             .get();
         waitForState(() -> "drop".equalsIgnoreCase(getSettings(indexName, "index.ingestion_source.error_strategy")));
         resumeIngestion(indexName);
-        waitForSearchableDocs(2, Arrays.asList(node));
+        waitForSearchableDocs(3, Arrays.asList(node));
+
+        PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
+            .getPollingIngestStats();
+        assertNotNull(stats);
+        assertThat(stats.getMessageProcessorStats().totalFailedCount(), is(1L));
+        assertThat(stats.getMessageProcessorStats().totalFailuresDroppedCount(), is(1L));
+        assertThat(stats.getConsumerStats().totalConsumerErrorCount(), is(0L));
+        assertThat(stats.getConsumerStats().totalPollerMessageDroppedCount(), is(1L));
     }
 
     public void testPauseAndResumeIngestion() throws Exception {
@@ -372,6 +384,13 @@ public void testExternalVersioning() throws Exception {
             assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L));
             return true;
         });
+
+        // validate processor stats
+        PollingIngestStats stats = client().admin().indices().prepareStats(indexName).get().getIndex(indexName).getShards()[0]
+            .getPollingIngestStats();
+        assertNotNull(stats);
+        assertThat(stats.getMessageProcessorStats().totalProcessedCount(), is(11L));
+        assertThat(stats.getMessageProcessorStats().totalVersionConflictsCount(), is(3L));
     }
 
     public void testExternalVersioningWithDisabledGCDeletes() throws Exception {

plugins/transport-reactor-netty4/src/javaRestTest/java/org/opensearch/rest/ReactorNetty4StreamingStressIT.java

@@ -44,6 +44,7 @@ public void tearDown() throws Exception {
         super.tearDown();
     }
 
+    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/15840")
     public void testCloseClientStreamingRequest() throws Exception {
         final VirtualTimeScheduler scheduler = VirtualTimeScheduler.create(true);
 

qa/systemd-test/src/test/java/org/opensearch/systemdinteg/SystemdIntegTests.java

@@ -135,19 +135,6 @@ public void testFileDescriptorLimit() throws IOException, InterruptedException {
                 limits.contains("Max open files            unlimited            unlimited"));
     }
 
-    public void testSystemCallFilter() throws IOException, InterruptedException {
-        // Check if Seccomp is enabled
-        String seccomp = executeCommand("sudo su -c 'grep Seccomp /proc/" + opensearchPid + "/status'", "Failed to read Seccomp status");
-        assertFalse("Seccomp should be enabled", seccomp.contains("0"));
-        
-        // Test specific system calls that should be blocked
-        String rebootResult = executeCommand("sudo su opensearch -c 'kill -s SIGHUP 1' 2>&1 || echo 'Operation not permitted'", "Failed to test reboot system call");
-        assertTrue("Reboot system call should be blocked", rebootResult.contains("Operation not permitted"));
-        
-        String swapResult = executeCommand("sudo su opensearch -c 'swapon -a' 2>&1 || echo 'Operation not permitted'", "Failed to test swap system call");
-        assertTrue("Swap system call should be blocked", swapResult.contains("Operation not permitted"));
-    }
-
     public void testOpenSearchProcessCannotExit() throws IOException, InterruptedException {
 
         String scriptPath;

release-notes/opensearch.release-notes-3.0.0.md

@@ -25,6 +25,7 @@
 - Deprecate CamelCase PathHierarchy tokenizer name ([#10894](https://github.com/opensearch-project/OpenSearch/pull/10894))
 - Rename Class ending with Plugin to Module under modules dir ([#4042](https://github.com/opensearch-project/OpenSearch/pull/4042))
 - Remove deprecated `batch_size` parameter from `_bulk` ([#14283](https://github.com/opensearch-project/OpenSearch/issues/14283))
+- Unset discovery nodes for every transport node actions request ([#17682](https://github.com/opensearch-project/OpenSearch/pull/17682))
 
 ### Added
 - Support for HTTP/2 (server-side) ([#3847](https://github.com/opensearch-project/OpenSearch/pull/3847))
@@ -96,7 +97,6 @@
 - Introduce mapping transformer to allow transform mappings during index create/update or index template create/update ([#17635](https://github.com/opensearch-project/OpenSearch/pull/17635))
 - Introduce new dynamic cluster setting to tune maxMergeAtOnce at cluster level ([#17774](https://github.com/opensearch-project/OpenSearch/pull/17774))
 - Add multi-threaded writer support in pull-based ingestion ([#17912](https://github.com/opensearch-project/OpenSearch/pull/17912))
-- Unset discovery nodes for every transport node actions request ([#17682](https://github.com/opensearch-project/OpenSearch/pull/17682))
 - [Star Tree] Support of Boolean Queries in Aggregations ([#17941](https://github.com/opensearch-project/OpenSearch/pull/17941))
 - Enabled default throttling for all tasks submitted to cluster manager ([#17711](https://github.com/opensearch-project/OpenSearch/pull/17711))
 

server/src/internalClusterTest/java/org/opensearch/cluster/MinimumClusterManagerNodesIT.java

@@ -250,6 +250,7 @@ public void testTwoNodesNoClusterManagerBlock() throws Exception {
         }
     }
 
+    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/14289")
     public void testThreeNodesNoClusterManagerBlock() throws Exception {
         internalCluster().setBootstrapClusterManagerNodeIndex(2);
 

server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationResizeRequestIT.java

@@ -31,6 +31,7 @@
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 2)
 public class SegmentReplicationResizeRequestIT extends SegmentReplicationBaseIT {
 
+    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/17552")
     public void testCreateShrinkIndexThrowsExceptionWhenReplicasBehind() throws Exception {
 
         // create index with -1 as refresh interval as we are blocking segrep and we want to control refreshes.