diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/DiskFullLoadApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/DiskFullLoadApplication.java new file mode 100644 index 0000000000000..875b17ca12231 --- /dev/null +++ b/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-> <-} 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 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(); + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodApplication.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodApplication.java new file mode 100644 index 0000000000000..1272054346e39 --- /dev/null +++ b/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-><-}, 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(); + } +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodService.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodService.java new file mode 100644 index 0000000000000..e3b277cbd9a63 --- /dev/null +++ b/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(); +} diff --git a/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodServiceImpl.java b/modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/disk_full/StdoutFloodServiceImpl.java new file mode 100644 index 0000000000000..3904e8a840984 --- /dev/null +++ b/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(); + } +} diff --git a/modules/ducktests/tests/checks/utils/check_ignite_spec.py b/modules/ducktests/tests/checks/utils/check_ignite_spec.py index 1e2cc4d3920a1..41ff5909202a7 100644 --- a/modules/ducktests/tests/checks/utils/check_ignite_spec.py +++ b/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 diff --git a/modules/ducktests/tests/checks/utils/check_parametrized.py b/modules/ducktests/tests/checks/utils/check_parametrized.py index 7c814fec45780..3bc84d161fd53 100644 --- a/modules/ducktests/tests/checks/utils/check_parametrized.py +++ b/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) diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py index e0f81ab35b785..822d422fad9a5 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py +++ b/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. diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/data_storage.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/data_storage.py index 54e5be421668e..6dfa2739ee783 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/data_storage.py +++ b/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 diff --git a/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py b/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py index ae7036df48c2e..891c56619fce1 100644 --- a/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py +++ b/modules/ducktests/tests/ignitetest/services/utils/ignite_spec.py @@ -373,6 +373,12 @@ def _jvm_opts(self): opts = ["-J%s" % o for o in self.jvm_opts] return " ".join(opts) + def _output_redirect(self): + """ + :return: shell redirection of the service's stdout and stderr: through tee, appended to console.log. + """ + return "2>&1 | tee -a %s" % os.path.join(self.service.log_dir, "console.log") + def _runcmd(self, cmd): self.service.logger.debug(cmd) proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) @@ -388,16 +394,26 @@ class IgniteNodeSpec(IgniteSpec): """ def command(self, node): - cmd = "%s %s %s %s 2>&1 | tee -a %s &" % \ + cmd = "%s %s %s %s %s &" % \ (envs_to_exports(self.envs()), self.service.script("ignite.sh"), self._jvm_opts(), self.config_file_path(), - os.path.join(self.service.log_dir, "console.log")) + self._output_redirect()) return cmd +class IgniteNodeSpecFileOutput(IgniteNodeSpec): + """ + Spec to run ignite node with stdout and stderr appended to console.log by the shell, as nohup does, so the JVM + writes them itself instead of a pipe to tee. + """ + + def _output_redirect(self): + return ">> %s 2>&1" % os.path.join(self.service.log_dir, "console.log") + + class IgniteApplicationSpec(IgniteSpec): """ Spec to run ignite application @@ -422,12 +438,12 @@ def command(self, node): str(base64.b64encode(json.dumps(self.service.params).encode('utf-8')), 'utf-8') ] - cmd = "%s %s %s %s 2>&1 | tee -a %s &" % \ + cmd = "%s %s %s %s %s &" % \ (envs_to_exports(self.envs()), self.service.script("ignite.sh"), self._jvm_opts(), ",".join(args), - os.path.join(self.service.log_dir, "console.log")) + self._output_redirect()) return cmd diff --git a/modules/ducktests/tests/ignitetest/tests/disk_full_test.py b/modules/ducktests/tests/ignitetest/tests/disk_full_test.py new file mode 100644 index 0000000000000..100b567e63b42 --- /dev/null +++ b/modules/ducktests/tests/ignitetest/tests/disk_full_test.py @@ -0,0 +1,465 @@ +# 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. + +""" +Fills one volume of a persistent server node under load and checks whether the node survives. + +The tests reproduce what Ignite does today and guard no behavior, so they run only with +--globals '{"disk_full_tests": true}'. +""" +import os +import time + +from ducktape.mark import matrix +from ducktape.utils.util import wait_until + +from ignitetest.services.ignite import IgniteService +from ignitetest.services.ignite_app import IgniteApplicationService +from ignitetest.services.utils.control_utility import ControlUtility +from ignitetest.services.utils.ignite_configuration import IgniteConfiguration, IgniteThinClientConfiguration, \ + DataStorageConfiguration +from ignitetest.services.utils.ignite_configuration.data_storage import DataRegionConfiguration +from ignitetest.services.utils.ignite_configuration.discovery import from_ignite_cluster +from ignitetest.services.utils.ignite_spec import IgniteNodeSpecFileOutput +from ignitetest.services.utils.jvm_utils import merge_jvm_settings +from ignitetest.services.utils.ssl.client_connector_configuration import ClientConnectorConfiguration +from ignitetest.utils import cluster, ignite_versions, reproducer +from ignitetest.utils.ignite_test import IgniteTest +from ignitetest.utils.version import DEV_BRANCH, IgniteVersion + +MB = 1024 * 1024 + +# A full tmpfs still takes writes into the rest of a file's last page. +PAGE = 4096 + +FILLER = "filler" + +# Failure type and failed writer the node logs when the volume fills. +FAILURES = { + "wal_archive": ("SYSTEM_WORKER_TERMINATION", "Failed to archive WAL segment"), + "partitions": ("CRITICAL_ERROR", "StorageException"), + "work": ("CRITICAL_ERROR", "BinaryMetadataFileStore") +} + + +class DiskFullTest(IgniteTest): + """ + Each test puts a size-limited tmpfs over a directory of the node, fills it and records what the node does today. + """ + LOAD_APP = "org.apache.ignite.internal.ducktest.tests.disk_full.DiskFullLoadApplication" + + FLOOD_APP = "org.apache.ignite.internal.ducktest.tests.disk_full.StdoutFloodApplication" + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_node_survives_full_log_volume(self, ignite_version): + """ + The log volume holds the log4j2 log, console.log and gc.log. + """ + ignite = self.__ignite(ignite_version) + ignite.mount_tmpfs(ignite.log_dir, 32) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + fill_volume(ignite, ignite.log_dir) + + check_serves_load_for(ignite, load, seconds=15) + assert ControlUtility(ignite).cluster_state().state == "ACTIVE" + + for log in log_files(ignite): + check_write_failed(ignite, log) + + free_volume(ignite, ignite.log_dir) + + for log in log_files(ignite): + check_grows(ignite, log) + + load.stop() + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_node_restarts_on_full_log_volume(self, ignite_version): + """ + The node restarts while its log volume is full. + """ + ignite = self.__ignite(ignite_version) + ignite.mount_tmpfs(ignite.log_dir, 32) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + fill_volume(ignite, ignite.log_dir) + + ignite.stop() + + # Waits for a failed put, so the count of puts read next is final until the node is back. + _, errors = progress(load) + wait_until(lambda: progress(load)[1] > errors, timeout_sec=30, err_msg="The load still reaches the node") + + puts, _ = progress(load) + + # start() awaits a line in the node's new log, which the full volume refuses. + ignite.start_async(clean=False) + + wait_until(lambda: progress(load)[0] > puts, timeout_sec=60, err_msg="No puts after the restart") + + check_serves_load_for(ignite, load, seconds=15) + assert ControlUtility(ignite).cluster_state().state == "ACTIVE" + assert file_size(ignite, ignite.nodes[0].log_file) == 0, "The restarted node wrote its log" + + free_volume(ignite, ignite.log_dir) + + for log in log_files(ignite): + check_grows(ignite, log) + + load.stop() + + @cluster(num_nodes=3) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_node_survives_log_volume_filled_by_stdout(self, ignite_version): + """ + A service on the node writes to System.out until its output fills the log volume. + """ + ignite = self.__ignite(ignite_version) + ignite.mount_tmpfs(ignite.log_dir, 32) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + flood = self.__start_flood_service(ignite) + + wait_until(lambda: free_space(ignite, ignite.log_dir) == 0, timeout_sec=60, + err_msg="System.out did not fill the log volume") + + lines = printed_lines(flood) + + check_serves_load_for(ignite, load, seconds=15) + assert ControlUtility(ignite).cluster_state().state == "ACTIVE" + assert printed_lines(flood) > lines, "The service stopped writing to System.out" + + for log in log_files(ignite): + check_write_failed(ignite, log) + + flood.stop() + + enlarge_volume(ignite, ignite.log_dir, 128) + + for log in log_files(ignite): + check_grows(ignite, log) + + load.stop() + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_jvm_exits_on_full_jfr_volume(self, ignite_version): + """ + JFR ends the JVM once it can't write its recording. + """ + ignite = self.__ignite(ignite_version) + repository = os.path.join(ignite.persistent_root, "jfr_repository") + jfr_opts = ["-XX:StartFlightRecording", f"-XX:FlightRecorderOptions=repository={repository}"] + ignite.spec.jvm_opts = merge_jvm_settings(ignite.spec.jvm_opts, jfr_opts) + ignite.mount_tmpfs(repository, 32) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + fill_volume(ignite, repository) + + ignite.await_event("An irrecoverable error in Jfr", 60, from_the_beginning=True, log_file="console.log") + check_stops(ignite) + + load.stop() + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_node_stops_on_full_volume_shared_by_logs_and_data(self, ignite_version): + """ + One volume holds the logs and all the data, as in Ignite's default layout. + """ + ignite = self.__ignite(ignite_version) + ignite.mount_tmpfs(ignite.persistent_root, 256) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + fill_volume(ignite, ignite.persistent_root) + + check_stops(ignite) + + load.stop() + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_node_survives_full_wal_volume(self, ignite_version): + """ + WAL segments are preallocated, so puts go on into a full WAL volume. + """ + ignite = self.__ignite(ignite_version) + wal = os.path.join(ignite.persistent_root, "wal") + ignite.mount_tmpfs(wal, 64) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + segments = ignite.exec_command(ignite.nodes[0], f"ls {wal}/*/*.wal 2>/dev/null | wc -l") + assert int(segments) > 0, "No WAL segments on the volume" + + fill_volume(ignite, wal) + + check_serves_load_for(ignite, load, seconds=15) + + load.stop() + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + @matrix(volume=["wal_archive", "partitions", "work"]) + def test_node_stops_on_full_volume(self, ignite_version, volume): + """ + The node stops once it can't write the WAL archive, the partitions or the binary metadata (in work). + """ + ignite = self.__ignite(ignite_version) + path = os.path.join(ignite.persistent_root, volume) + ignite.mount_tmpfs(path, 64) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + fill_volume(ignite, path) + + check_stops(ignite) + check_failure(ignite, *FAILURES[volume]) + + load.stop() + + @cluster(num_nodes=2) + @reproducer("disk_full_tests") + @ignite_versions(str(DEV_BRANCH)) + def test_node_stops_when_cdc_fills_wal_archive_volume(self, ignite_version): + """ + CDC with no consumer and no cdcWalDirectoryMaxSize keeps every archived segment. + """ + ignite = self.__ignite(ignite_version) + archive = os.path.join(ignite.persistent_root, "wal_archive") + data_storage = ignite.config.data_storage + + # CDC hard-links archived segments, so its directory must be on the archive's volume. + ignite.config = ignite.config._replace(data_storage=data_storage._replace( + cdc_wal_path=os.path.join(archive, "cdc"), default=data_storage.default._replace(cdc_enabled=True))) + ignite.mount_tmpfs(archive, 256) + + ignite.start() + ControlUtility(ignite).activate() + + load = self.__start_load_client(ignite) + + check_stops(ignite, timeout_sec=120) + check_failure(ignite, *FAILURES["wal_archive"]) + + load.stop() + + def __ignite(self, ignite_version): + """ + :return: Persistent server node that writes its stdout and stderr into console.log itself, with the + partitions, the WAL and its archive in directories of their own. + """ + ignite = IgniteService(self.test_context, IgniteConfiguration( + version=IgniteVersion(ignite_version), + client_connector_configuration=ClientConnectorConfiguration(), + metrics_log_frequency=1000, + data_storage=DataStorageConfiguration( + checkpoint_frequency=1000, + wal_segment_size=4 * MB, + max_wal_archive_size=32 * MB, + default=DataRegionConfiguration(persistence_enabled=True, max_size=256 * MB))), num_nodes=1) + + ignite.spec = ignite.spec.rebuild_as(IgniteNodeSpecFileOutput) + + root = ignite.persistent_root + + ignite.config = ignite.config._replace(data_storage=ignite.config.data_storage._replace( + storage_path=os.path.join(root, "partitions"), + wal_path=os.path.join(root, "wal"), + wal_archive_path=os.path.join(root, "wal_archive"))) + + return ignite + + def __start_load_client(self, ignite): + """ + Starts a load that records its puts and errors once a second. + :return: Load application service. + """ + address = f"{ignite.nodes[0].account.hostname}:{ignite.config.client_connector_configuration.port}" + + load = IgniteApplicationService( + self.test_context, + IgniteThinClientConfiguration(addresses=[address], version=ignite.config.version), + java_class_name=self.LOAD_APP, + params={"cacheName": "disk-full"}) + + load.start() + + return load + + def __start_flood_service(self, ignite): + """ + Starts the application that deploys the System.out flood service on the server. + :return: Flood application service. + """ + flood = IgniteApplicationService( + self.test_context, + IgniteConfiguration(version=ignite.config.version, client_mode=True, + discovery_spi=from_ignite_cluster(ignite)), + java_class_name=self.FLOOD_APP) + + flood.start() + + return flood + + +def log_files(ignite): + """ + :return: The node's log4j2 log, console.log and gc.log, all on the log volume. + """ + return [ignite.nodes[0].log_file] + [os.path.join(ignite.log_dir, log) for log in ("console.log", "gc.log")] + + +def fill_volume(ignite, path): + """ + Writes a file into the path until its volume refuses the write. + """ + out = ignite.exec_command(ignite.nodes[0], f"dd if=/dev/zero of={os.path.join(path, FILLER)} bs=1M 2>&1") + + assert "No space left on device" in out, f"{path} did not fill up: {out}" + + +def free_volume(ignite, path): + """ + Removes the file fill_volume wrote. + """ + ignite.nodes[0].account.ssh(f"rm -f {os.path.join(path, FILLER)}", allow_fail=False) + + +def enlarge_volume(ignite, path, size_mb): + """ + Grows the tmpfs mounted over the path, keeping its files and what they hold, so writes there go on. + """ + ignite.nodes[0].account.ssh(f"sudo mount -o remount,size={size_mb}m {path}", allow_fail=False) + + +def free_space(ignite, path): + """ + :return: Bytes free on the volume of the path. + """ + return int(ignite.exec_command(ignite.nodes[0], f"df --output=avail -B1 {path} | tail -1")) + + +def printed_lines(flood): + """ + :return: Lines the service wrote to System.out, as the flood application recorded last. + """ + return int(flood.extract_results("lines")[-1]) + + +def file_size(ignite, path): + """ + :return: Bytes written into the file on the node. + """ + return int(ignite.exec_command(ignite.nodes[0], f"stat -c %s {path}")) + + +def check_write_failed(ignite, path): + """ + Checks that the file ends on a page boundary: a write the full tmpfs refused filled its last page. + """ + assert file_size(ignite, path) % PAGE == 0, f"No write into {path} hit the full volume" + + +def check_grows(ignite, path): + """ + Checks that the file grows within 15 s. + """ + size = file_size(ignite, path) + + wait_until(lambda: file_size(ignite, path) > size, timeout_sec=15, err_msg=f"{path} did not grow") + + +def check_serves_load_for(ignite, load, seconds): + """ + Waits the given seconds and checks that the node is alive and the puts went on without errors. + """ + puts, errors = progress(load) + + time.sleep(seconds) + + assert ignite.alive(ignite.nodes[0]), "The node stopped" + + puts_after, errors_after = progress(load) + + assert puts_after > puts, f"No puts in {seconds} s" + assert errors_after == errors, f"{errors_after - errors} puts failed in {seconds} s" + + +def check_stops(ignite, timeout_sec=60): + """ + Checks that the node's process ends. + """ + wait_until(lambda: not ignite.alive(ignite.nodes[0]), timeout_sec=timeout_sec, + err_msg=f"The node is still up {timeout_sec} s after its volume filled up") + + +def check_failure(ignite, failure_type, cause): + """ + Checks that the node logged a FailureContext of the type, caused by the writer and "No space left on device". + """ + node = ignite.nodes[0] + + failure = ignite.exec_command(node, f"grep -h -m1 -F -A60 'FailureContext [type={failure_type}' " + f"{node.log_file} || true") + + assert failure, f"No {failure_type} failure in the node's log" + assert cause in failure, f"The {failure_type} failure is not caused by '{cause}': {failure[:2000]}" + assert "No space left on device" in failure, f"The {failure_type} failure is not ENOSPC: {failure[:2000]}" + + +def progress(load): + """ + :return: Puts and failed puts the load application recorded last. + """ + puts, errors = load.extract_results("progress")[-1].split() + + return int(puts), int(errors) diff --git a/modules/ducktests/tests/ignitetest/utils/__init__.py b/modules/ducktests/tests/ignitetest/utils/__init__.py index 6d6da0d14ed12..2cc208c44baf0 100644 --- a/modules/ducktests/tests/ignitetest/utils/__init__.py +++ b/modules/ducktests/tests/ignitetest/utils/__init__.py @@ -17,6 +17,6 @@ This module contains convenient utils for test. """ -from ._mark import ignore_if, ignite_versions, cluster +from ._mark import ignore_if, ignite_versions, cluster, reproducer -__all__ = ['ignore_if', 'ignite_versions', 'cluster'] +__all__ = ['ignore_if', 'ignite_versions', 'cluster', 'reproducer'] diff --git a/modules/ducktests/tests/ignitetest/utils/_mark.py b/modules/ducktests/tests/ignitetest/utils/_mark.py index 9d0a06df4edc2..5964312fcc3c2 100644 --- a/modules/ducktests/tests/ignitetest/utils/_mark.py +++ b/modules/ducktests/tests/ignitetest/utils/_mark.py @@ -51,6 +51,26 @@ def __eq__(self, other): return super().__eq__(other) and self.condition == other.condition +class Reproducer(Ignore): + """ + Ignore test unless the global that enables it is set. + """ + def __init__(self, global_name): + super().__init__() + self.global_name = global_name + + def apply(self, seed_context, context_list): + assert len(context_list) > 0, "reproducer decorator is not being applied to any test cases" + + for ctx in context_list: + ctx.ignore = ctx.ignore or not ctx.globals.get(self.global_name) + + return context_list + + def __eq__(self, other): + return super().__eq__(other) and self.global_name == other.global_name + + class IgniteVersionParametrize(Mark): """ Parametrize function with ignite_version @@ -219,6 +239,20 @@ def ignorer(func): return ignorer +def reproducer(global_name): + """ + Mark decorated test method as IGNORE unless the global is set, e.g. --globals '{"": true}'. + For tests that reproduce a behavior on demand rather than guard it on every run. + + :param global_name: name of the global that enables the test. + """ + def ignorer(func): + Mark.mark(func, Reproducer(global_name)) + return func + + return ignorer + + def cluster(**kwargs): """Test method decorator used to provide hints about how the test will use the given cluster.