IGNITE-29165 [ducktests] Add reproducers of a full disk volume under a server node

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/DiskFullLoadApplication.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.ignite.internal.ducktest.tests.disk_full;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.client.ClientCache;
+import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
+
+/**
+ * Puts values under growing keys, with a new binary type every 1000 puts, so the partitions, the WAL and the binary
+ * metadata keep growing. Retries failed puts and records {@code progress-><puts> <errors><-} once a second.
+ */
+public class DiskFullLoadApplication extends IgniteAwareApplication {
+    /** Value size, in bytes. */
+    private static final int VALUE_SIZE = 1024;
+
+    /** Puts per new binary type. */
+    private static final int BINARY_TYPE_EVERY = 1000;
+
+    /** Successful puts after which the application reports it is initialized. */
+    private static final int WARM_UP_PUTS = 1000;
+
+    /** Pause after a failed put, in milliseconds. */
+    private static final long ERROR_PAUSE = 200;
+
+    /** {@inheritDoc} */
+    @Override protected void run(JsonNode jsonNode) throws Exception {
+        ClientCache<Long, Object> cache = client.getOrCreateCache(jsonNode.get("cacheName").asText());
+
+        byte[] val = new byte[VALUE_SIZE];
+
+        long key = 0;
+        long puts = 0;
+        long errs = 0;
+        long nextReport = System.currentTimeMillis() + 1000;
+
+        while (active()) {
+            try {
+                if (key % BINARY_TYPE_EVERY == 0)
+                    cache.put(key, client.binary().builder("DiskFullType" + key).setField("val", val).build());
+                else
+                    cache.put(key, val);
+
+                key++;
+                puts++;
+
+                // The first record comes before the marker, so the test can read the counters once start() returns.
+                if (puts == WARM_UP_PUTS) {
+                    recordResult("progress", puts + " " + errs);
+
+                    markInitialized();
+                }
+            }
+            catch (Exception e) {
+                errs++;
+
+                log.warn("Put failed [key=" + key + ", err=" + e + ']');
+
+                Thread.sleep(ERROR_PAUSE);
+            }
+
+            long now = System.currentTimeMillis();
+
+            if (now >= nextReport) {
+                recordResult("progress", puts + " " + errs);
+
+                nextReport = now + 1000;
+            }
+        }
+
+        markFinished();
+    }
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodApplication.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.ignite.internal.ducktest.tests.disk_full;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
+
+/**
+ * Deploys {@link StdoutFloodServiceImpl} on every server node, records {@code lines-><n><-}, the lines the service
+ * wrote, once a second, and cancels the service on stop.
+ */
+public class StdoutFloodApplication extends IgniteAwareApplication {
+    /** Service name. */
+    private static final String SVC_NAME = "stdout-flood";
+
+    /** {@inheritDoc} */
+    @Override protected void run(JsonNode jsonNode) throws Exception {
+        ignite.services(ignite.cluster().forServers()).deployNodeSingleton(SVC_NAME, new StdoutFloodServiceImpl());
+
+        StdoutFloodService svc = ignite.services().serviceProxy(SVC_NAME, StdoutFloodService.class, false);
+
+        recordResult("lines", svc.lines());
+
+        markInitialized();
+
+        while (active()) {
+            Thread.sleep(1000);
+
+            recordResult("lines", svc.lines());
+        }
+
+        ignite.services().cancel(SVC_NAME);
+
+        markFinished();
+    }
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodService.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.ignite.internal.ducktest.tests.disk_full;
+
+/**
+ * Writes lines to {@code System.out} without end, as user code left with debug output does.
+ */
+public interface StdoutFloodService {
+    /** @return Lines written so far. */
+    public long lines();
+}

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodServiceImpl.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.ignite.internal.ducktest.tests.disk_full;
+
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.ignite.resources.ServiceContextResource;
+import org.apache.ignite.services.Service;
+import org.apache.ignite.services.ServiceContext;
+
+/**
+ * Writes 1 KB lines to {@code System.out} at about 10 MB/s until cancelled.
+ */
+public class StdoutFloodServiceImpl implements Service, StdoutFloodService {
+    /** Serial version UID. */
+    private static final long serialVersionUID = 0L;
+
+    /** Line, 1 KB with its line separator. */
+    private static final String LINE = "x".repeat(1023);
+
+    /** Lines written per millisecond. */
+    private static final int LINES_PER_MS = 10;
+
+    /** Service context. */
+    @ServiceContextResource
+    private ServiceContext ctx;
+
+    /** Lines written. */
+    private final AtomicLong lines = new AtomicLong();
+
+    /** {@inheritDoc} */
+    @Override public void execute() throws Exception {
+        while (!ctx.isCancelled()) {
+            for (int i = 0; i < LINES_PER_MS; i++) {
+                System.out.println(LINE);
+
+                lines.incrementAndGet();
+            }
+
+            Thread.sleep(1);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public long lines() {
+        return lines.get();
+    }
+}

modules/ducktests/tests/checks/utils/check_ignite_spec.py

@@ -17,13 +17,14 @@
 Checks Spec class that describes config and command line to start Ignite-aware service.
 """
 import os
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
 
 from ignitetest.services.utils import IgniteServiceType
 from ignitetest.services.utils.gc_params import CLIENT_ROLE, SERVER_ROLE
-from ignitetest.services.utils.ignite_spec import IgniteApplicationSpec, IgniteNodeSpec, service_role
+from ignitetest.services.utils.ignite_spec import IgniteApplicationSpec, IgniteNodeSpec, IgniteNodeSpecFileOutput, \
+    service_role
 from ignitetest.services.utils.jvm_utils import GC_PROFILES, GC_G1, MultipleGcSelectedError
 from ignitetest.utils.ignite_test import JFR_ENABLED
 
@@ -287,3 +288,18 @@ def check_heap_override__from_jvm_opts__still_wins(service):
     assert "-Xms4G" in spec.jvm_opts
     assert "-Xmx1G" not in spec.jvm_opts
     assert "-Xms1G" not in spec.jvm_opts
+
+
+"""
+Checks the shell redirection of a node's output.
+"""
+
+
+def check_node_spec_file_output__appends_stdout_and_stderr_to_console_log(service):
+    service.log_dir = "/logs"
+
+    with patch.object(IgniteNodeSpecFileOutput, "envs", return_value={}):
+        cmd = IgniteNodeSpecFileOutput(service).command(None)
+
+    assert cmd.endswith(f" >> {os.path.join(service.log_dir, 'console.log')} 2>&1 &")
+    assert "tee" not in cmd

modules/ducktests/tests/checks/utils/check_parametrized.py

@@ -24,7 +24,7 @@
 from ducktape.mark import parametrized, parametrize, matrix, ignore
 from ducktape.mark.mark_expander import MarkedFunctionExpander
 
-from ignitetest.utils import ignite_versions, ignore_if
+from ignitetest.utils import ignite_versions, ignore_if, reproducer
 from ignitetest.utils._mark import IgniteVersionParametrize
 from ignitetest.utils.version import IgniteVersion, V_2_8_0, V_2_8_1, V_2_7_6, DEV_BRANCH
 
@@ -267,3 +267,31 @@ def function_3(ignite_version):
 
         assert next(filter(lambda x: x.injected_args['ver'] == str(V_2_8_0), context_list)).ignore
         assert not next(filter(lambda x: x.injected_args['ver'] == str(DEV_BRANCH), context_list)).ignore
+
+
+class CheckReproducer:
+    """
+    Checks @reproducer.
+    """
+    @pytest.mark.parametrize(['global_args', 'ignored'], [({}, True), ({'my_tests': False}, True),
+                                                          ({'my_tests': True}, False)])
+    def check_ignored_unless_global_is_set(self, global_args, ignored):
+        """
+        Check that a test is ignored unless its global is set, with and without @ignite_versions.
+        """
+        @reproducer('my_tests')
+        @ignite_versions(str(DEV_BRANCH), str(V_2_8_0))
+        def versioned(ignite_version):
+            return IgniteVersion(ignite_version)
+
+        @reproducer('my_tests')
+        def plain():
+            pass
+
+        sess_ctx = mock_session_ctx(global_args=global_args)
+
+        context_list = expand_function(func=versioned, sess_ctx=sess_ctx)
+        context_list += MarkedFunctionExpander(session_context=sess_ctx, function=plain).expand()
+
+        assert len(context_list) == 3
+        assert all(ctx.ignore == ignored for ctx in context_list)

modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py

@@ -48,6 +48,9 @@ class IgniteAwareService(BackgroundThreadService, IgnitePathAware, JvmProcessMix
     """
     The base class to build services aware of Ignite.
     """
+    # Mount source of the volumes mount_tmpfs makes: clean_node finds them by it.
+    TMPFS_SOURCE = "ignitetest-tmpfs"
+
     @constructible
     class NetPart(IntEnum):
         """
@@ -79,6 +82,8 @@ def __init__(self, context, config, num_nodes, startup_timeout_sec, shutdown_tim
 
         self.disconnected_nodes = []
 
+        self.tmpfs_volumes = {}
+
     @property
     def product(self):
         return str(self.config.version)
@@ -109,7 +114,15 @@ def await_started(self, nodes=None):
 
         self.await_event("Topology snapshot", self.startup_timeout_sec, nodes=nodes, from_the_beginning=True)
 
+    def mount_tmpfs(self, path, size_mb):
+        """
+        Mounts a tmpfs of size_mb over the path on every node when it starts, so writes there fail once it is full.
+        """
+        self.tmpfs_volumes[path] = size_mb
+
     def start_node(self, node, **kwargs):
+        self.__mount_tmpfs_volumes(node)
+
         self.init_shared(node)
         self.init_persistent(node)
 
@@ -186,8 +199,26 @@ def clean(self, **kwargs):
     def clean_node(self, node, **kwargs):
         super().clean_node(node, **kwargs)
 
+        self.__unmount_tmpfs_volumes(node)
+
         node.account.ssh("rm -rf -- %s" % self.persistent_root, allow_fail=False)
 
+    def __mount_tmpfs_volumes(self, node):
+        """
+        Mounts the volumes registered by mount_tmpfs, unless they are still mounted from the node's previous start.
+        """
+        for path, size_mb in self.tmpfs_volumes.items():
+            node.account.ssh(f"findmnt -rn -S {self.TMPFS_SOURCE} -M {path} > /dev/null || (mkdir -p {path} && "
+                             f"sudo mount -t tmpfs -o size={size_mb}m,uid=$(id -u),gid=$(id -g) {self.TMPFS_SOURCE} "
+                             f"{path})", allow_fail=False)
+
+    def __unmount_tmpfs_volumes(self, node):
+        """
+        Unmounts every volume mount_tmpfs made on the node, also one left by an aborted run.
+        """
+        node.account.ssh(f"findmnt -rn -S {self.TMPFS_SOURCE} -o TARGET | xargs -r -n1 sudo umount -l",
+                         allow_fail=False)
+
     def init_persistent(self, node):
         """
         Init persistent directory.

modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/data_storage.py

@@ -42,6 +42,7 @@ class DataStorageConfiguration(NamedTuple):
     """
     Ignite DataStorage configuration
     """
+    cdc_wal_directory_max_size: int = None
     cdc_wal_path: str = None
     checkpoint_frequency: int = None
     default: DataRegionConfiguration = DataRegionConfiguration()
@@ -50,6 +51,7 @@ class DataStorageConfiguration(NamedTuple):
     metrics_rate_time_interval: int = None
     page_size: int = None
     regions: list = []
+    storage_path: str = None
     wal_archive_path: str = None
     wal_buffer_size: int = None
     wal_compaction_enabled: bool = None