Merge pull request #23362 from WillemKauf/backport-pr-23337-v24.1.x-651

[v24.1.x] `admin`: add `cloud_storage_cache_size` check to `config_multi_property_validation()` (manual backport)

src/v/cloud_storage/cache_service.cc

@@ -1968,4 +1968,27 @@ ss::future<> cache::sync_access_time_tracker(
  }
 }
 
+std::optional<ss::sstring>
+cache::validate_cache_config(const config::configuration& conf) {
+ const auto& cloud_storage_cache_size = conf.cloud_storage_cache_size;
+ const auto& cloud_storage_cache_size_pct
+ = conf.cloud_storage_cache_size_percent;
+
+ // If not set, cloud cache uses default value of 0.0
+ auto cache_size_pct = cloud_storage_cache_size_pct().value_or(0.0);
+
+ using cache_size_pct_type = double;
+ static constexpr auto epsilon
+ = std::numeric_limits<cache_size_pct_type>::epsilon();
+
+ if ((cache_size_pct < epsilon) && (cloud_storage_cache_size() == 0)) {
+ return ss::format(
+ "Cannot set both {} and {} to 0.",
+ cloud_storage_cache_size.name(),
+ cloud_storage_cache_size_pct.name());
+ }
+
+ return std::nullopt;
+}
+
 } // namespace cloud_storage

src/v/cloud_storage/cache_service.h

@@ -15,6 +15,7 @@
 #include "cloud_storage/access_time_tracker.h"
 #include "cloud_storage/cache_probe.h"
 #include "cloud_storage/recursive_directory_walker.h"
+#include "config/configuration.h"
 #include "config/property.h"
 #include "resource_mgmt/io_priority.h"
 #include "resource_mgmt/storage.h"
@@ -197,6 +198,16 @@ class cache : public ss::peering_sharded_service<cache> {
  return _cache_dir / key;
  }
 
+ // Checks if a cluster configuration is valid for the properties
+ // `cloud_storage_cache_size` and `cloud_storage_cache_size_percent`.
+ // Two cases are invalid: 1. the case in which both are 0, 2. the case in
+ // which `cache_size` is 0 while `cache_size_percent` is `std::nullopt`.
+ //
+ // Returns `std::nullopt` if the passed configuration is valid, or an
+ // `ss::sstring` explaining the misconfiguration otherwise.
+ static std::optional<ss::sstring>
+ validate_cache_config(const config::configuration& conf);
+
 private:
  /// Load access time tracker from file
  ss::future<> load_access_time_tracker();

src/v/redpanda/admin/server.cc

@@ -1634,6 +1634,14 @@ void config_multi_property_validation(
  errors[ss::sstring(name)] = ssx::sformat(
  "{} requires schema_registry to be enabled in redpanda.yaml", name);
  }
+
+ // cloud_storage_cache_size/size_percent validation
+ if (auto invalid_cache = cloud_storage::cache::validate_cache_config(
+ updated_config);
+ invalid_cache.has_value()) {
+ auto name = ss::sstring(updated_config.cloud_storage_cache_size.name());
+ errors[name] = invalid_cache.value();
+ }
 }
 } // namespace
 

tests/rptest/tests/cluster_config_test.py

@@ -1491,6 +1491,41 @@ def test_status_read_after_write_consistency(self):
  if s['node_id'] == self.redpanda.idx(controller_node))
  assert local_status['config_version'] == config_version
 
+ # None for pct_value is std::nullopt, which defaults to 0.0 in the cloud cache.
+ @cluster(num_nodes=1)
+ def test_validate_cloud_storage_cache_size_config(self):
+ CloudCacheConf = namedtuple('CloudCacheConf',
+ ['size_value', 'pct_value', 'valid'])
+ test_cases = [
+ CloudCacheConf(size_value=0, pct_value=None, valid=False),
+ CloudCacheConf(size_value=0, pct_value=0.0, valid=False),
+ CloudCacheConf(size_value=0, pct_value=-1.0, valid=False),
+ CloudCacheConf(size_value=0, pct_value=101.0, valid=False),
+ CloudCacheConf(size_value=-1, pct_value=None, valid=False),
+ CloudCacheConf(size_value=1024, pct_value=None, valid=True),
+ CloudCacheConf(size_value=10, pct_value=50.0, valid=True),
+ CloudCacheConf(size_value=0, pct_value=0.1, valid=True)
+ ]
+
+ for size_value, pct_value, valid in test_cases:
+ upsert = {}
+ upsert["cloud_storage_cache_size"] = size_value
+ upsert["cloud_storage_cache_size_percent"] = pct_value
+
+ if valid:
+ patch_result = self.admin.patch_cluster_config(upsert=upsert)
+ new_version = patch_result['config_version']
+ wait_for_version_status_sync(self.admin, self.redpanda,
+ new_version)
+ updated_config = self.admin.get_cluster_config()
+ assert updated_config["cloud_storage_cache_size"] == size_value
+ assert updated_config[
+ "cloud_storage_cache_size_percent"] == pct_value
+ else:
+ with expect_exception(requests.exceptions.HTTPError,
+ lambda e: e.response.status_code == 400):
+ self.admin.patch_cluster_config(upsert=upsert)
+
 
 """
 PropertyAliasData: