gusev-p · gusev-p · Mar 26, 2025 · Apr 5, 2025 · Apr 5, 2025 · May 29, 2025
diff --git a/.gitignore b/.gitignore
@@ -35,3 +35,4 @@ compile_commands.json
 .envrc
 clang_build
 .idea/
+nuke
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -181,7 +181,7 @@ target_sources(scylla-main
     generic_server.cc
     debug.cc
     init.cc
-    keys.cc
+    keys/keys.cc
     multishard_mutation_query.cc
     mutation_query.cc
     node_ops/task_manager_module.cc

diff --git a/alternator/rmw_operation.hh b/alternator/rmw_operation.hh
@@ -15,7 +15,7 @@
 #include "consumed_capacity.hh"
 #include "executor.hh"
 #include "tracing/trace_state.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 
 namespace alternator {
 

diff --git a/alternator/serialization.hh b/alternator/serialization.hh
@@ -13,7 +13,7 @@
 #include <optional>
 #include "types/types.hh"
 #include "schema/schema_fwd.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 #include "utils/rjson.hh"
 #include "utils/big_decimal.hh"
 

diff --git a/api/storage_service.cc b/api/storage_service.cc
@@ -359,6 +359,9 @@ void set_repair(http_context& ctx, routes& r, sharded<repair_service>& repair, s
             // if the option is not sane, repair_start() throws immediately, so
             // convert the exception to an HTTP error
             throw httpd::bad_param_exception(e.what());
+        } catch (const tablets_unsupported& e) {
+            throw base_exception("Cannot repair tablet keyspace. Use /storage_service/tablets/repair to repair tablet keyspaces.",
+                    http::reply::status_type::forbidden);
         }
     });
 

diff --git a/cdc/cdc_partitioner.cc b/cdc/cdc_partitioner.cc
@@ -12,7 +12,7 @@
 #include "sstables/key.hh"
 #include "utils/class_registrator.hh"
 #include "cdc/generation.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 
 namespace cdc {
 

diff --git a/cdc/generation.cc b/cdc/generation.cc
@@ -16,7 +16,7 @@
 
 #include "gms/endpoint_state.hh"
 #include "gms/versioned_value.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 #include "replica/database.hh"
 #include "db/system_keyspace.hh"
 #include "db/system_distributed_keyspace.hh"

diff --git a/conf/scylla.yaml b/conf/scylla.yaml
@@ -776,6 +776,35 @@ maintenance_socket: ignore
 #       ...
 #
 
+#
+# Azure Key Vault host(s).
+#
+# The unique name of azure host/account config that can be referenced in table schema.
+#
+# host.yourdomain.com={ azure_tenant_id=<the tenant hosting your service principal>, azure_client_id=<ID of your service principal>, azure_client_secret=<secret of the service principal>, azure_client_certificate_path=<path to PEM-encoded certificate and private key of the service principal>, master_key=<vault name>/<keyname>, truststore=/path/to/truststore.pem, priority_string=<tls priority string>, key_cache_expiry=<cache expiry in ms>, key_cache_refersh=<cache refresh in ms>}:...
+#
+# Authentication can be explicit with Service Principal credentials. Either secret or certificate can be provided.
+# If both are provided, the secret will be used. If no credentials are provided, the provider will try to detect them
+# from the environment, the Azure CLI, and IMDS, in this specific order.
+#
+# master_key is a Vault key that will be used to wrap all keys used for actual encryption of scylla data.
+# This key must be pre-created and the principal must have permissions for Wrapkey and Unwrapkey operations on this key.
+#
+# azure_hosts:
+#    <name>:
+#       azure_tenant_id: <the tenant hosting your service principal> (optional)
+#       azure_client_id: <ID of your service principal> (optional)
+#       azure_client_secret: <secret of the service principal> (optional)
+#       azure_client_certificate_path: <path to PEM-encoded certificate and private key of the service principal> (optional)
+#       master_key: <vault name>/<keyname> - named Vault key for key wrapping (optional)
+#       truststore: <PEM file with CA certificates for TLS connection> (optional)
+#       priority_string: <GnuTLS priority string for TLS handshake> (optional)
+#       key_cache_expiry: <key cache expiry period (ms)> (optional)
+#       key_cache_refresh: <key cache refresh/prune period (ms)> (optional)
+#   <name>:
+#       ...
+#
+
 #
 # Server-global user information encryption settings
 #
@@ -862,3 +891,18 @@ rf_rack_valid_keyspaces: false
 # Uri for the vector store using dns name. Only http schema is supported. Port number is mandatory.
 # Default is empty, which means that the vector store is not used.
 # vector_store_uri: http://vector-store.dns.name:{port}
+
+# 
+# io-streaming rate limiting
+# When setting this value to be non-zero scylla throttles disk throughput for
+# stream (network) activities such as backup, repair, tablet migration and more.
+# This limit is useful for user queries so the network interface does 
+# not get saturated by streaming activities.
+# The recommended value is 75% of network bandwidth
+# E.g for i4i.8xlarge (https://github.com/scylladb/scylla-machine-image/tree/next/common/aws_net_params.json):
+# network: 18.75 GiB/s --> 18750 Mib/s --> 1875 MB/s (from network bits to network bytes: divide by 10, not 8)
+# Converted to disk bytes: 1875 * 1000 / 1024 = 1831 MB/s (disk wise)
+# 75% of disk bytes is: 0.75 * 1831 = 1373 megabytes/s
+# stream_io_throughput_mb_per_sec: 1373
+# 
+
diff --git a/configure.py b/configure.py
@@ -551,6 +551,7 @@ def find_ninja():
     'test/boost/sstable_conforms_to_mutation_source_test',
     'test/boost/sstable_datafile_test',
     'test/boost/sstable_generation_test',
+    'test/boost/sstable_inexact_index_test',
     'test/boost/sstable_move_test',
     'test/boost/sstable_mutation_test',
     'test/boost/sstable_partition_index_cache_test',
@@ -831,7 +832,7 @@ def find_ninja():
                 'readers/mutation_reader.cc',
                 'readers/mutation_readers.cc',
                 'mutation_query.cc',
-                'keys.cc',
+                'keys/keys.cc',
                 'counters.cc',
                 'compress.cc',
                 'sstable_dict_autotrainer.cc',
@@ -1055,6 +1056,11 @@ def find_ninja():
                 'utils/s3/credentials_providers/aws_credentials_provider_chain.cc',
                 'utils/s3/utils/manip_s3.cc',
                 'utils/advanced_rpc_compressor.cc',
+                'utils/azure/identity/credentials.cc',
+                'utils/azure/identity/service_principal_credentials.cc',
+                'utils/azure/identity/managed_identity_credentials.cc',
+                'utils/azure/identity/azure_cli_credentials.cc',
+                'utils/azure/identity/default_credentials.cc',
                 'gms/version_generator.cc',
                 'gms/versioned_value.cc',
                 'gms/gossiper.cc',
@@ -1177,6 +1183,8 @@ def find_ninja():
                 'ent/encryption/gcp_host.cc',
                 'ent/encryption/gcp_key_provider.cc',
                 'ent/encryption/utils.cc',
+                'ent/encryption/azure_host.cc',
+                'ent/encryption/azure_key_provider.cc',
                 'ent/ldap/ldap_connection.cc',
                 'multishard_mutation_query.cc',
                 'reader_concurrency_semaphore.cc',

diff --git a/cql3/expr/expr-utils.hh b/cql3/expr/expr-utils.hh
@@ -6,7 +6,7 @@
 #include "expression.hh"
 
 #include "bytes.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 #include "utils/interval.hh"
 #include "cql3/expr/restrictions.hh"
 #include "cql3/assignment_testable.hh"

diff --git a/db/cache_mutation_reader.hh b/db/cache_mutation_reader.hh
@@ -17,7 +17,7 @@
 #include "partition_snapshot_row_cursor.hh"
 #include "read_context.hh"
 #include "readers/delegating.hh"
-#include "clustering_key_filter.hh"
+#include "keys/clustering_key_filter.hh"
 
 namespace cache {
 

diff --git a/db/config.cc b/db/config.cc
@@ -819,7 +819,7 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , inter_dc_stream_throughput_outbound_megabits_per_sec(this, "inter_dc_stream_throughput_outbound_megabits_per_sec", value_status::Unused, 0,
         "Throttles all streaming file transfer between the data centers. This setting allows throttles streaming throughput betweens data centers in addition to throttling all network stream traffic as configured with stream_throughput_outbound_megabits_per_sec.")
     , stream_io_throughput_mb_per_sec(this, "stream_io_throughput_mb_per_sec", liveness::LiveUpdate, value_status::Used, 0,
-        "Throttles streaming I/O to the specified total throughput (in MiBs/s) across the entire system. Streaming I/O includes the one performed by repair and both RBNO and legacy topology operations such as adding or removing a node. Setting the value to 0 disables stream throttling.")
+        "Throttles streaming I/O to the specified total throughput (in MiBs/s) across the entire system. Streaming I/O includes the one performed by repair and both RBNO and legacy topology operations such as adding or removing a node. Setting the value to 0 disables stream throttling. It is recommended to set the value for this parameter to be 75% of network bandwidth")
     , stream_plan_ranges_fraction(this, "stream_plan_ranges_fraction", liveness::LiveUpdate, value_status::Used, 0.1,
         "Specify the fraction of ranges to stream in a single stream plan. Value is between 0 and 1.")
     , enable_file_stream(this, "enable_file_stream", liveness::LiveUpdate, value_status::Used, true, "Set true to use file based stream for tablet instead of mutation based stream")

diff --git a/db/hints/internal/hint_sender.cc b/db/hints/internal/hint_sender.cc
@@ -175,14 +175,14 @@ future<> hint_sender::stop(drain should_drain) noexcept {
             //
             // The next call for send_hints_maybe() will send the last hints to the current end point and when it is
             // done there is going to be no more pending hints and the corresponding hints directory may be removed.
-            manager_logger.trace("Draining for {}: start", end_point_key());
+            manager_logger.info("Draining for {}: start", end_point_key());
             set_draining();
             send_hints_maybe();
             _ep_manager.flush_current_hints().handle_exception([] (auto e) {
                 manager_logger.error("Failed to flush pending hints: {}. Ignoring...", e);
             }).get();
             send_hints_maybe();
-            manager_logger.trace("Draining for {}: end", end_point_key());
+            manager_logger.info("Draining for {}: end", end_point_key());
         }
         // TODO: Change this log to match the class name, but first make sure no test
         //       relies on the old one.

diff --git a/db/hints/manager.cc b/db/hints/manager.cc
@@ -539,7 +539,7 @@ future<> manager::change_host_filter(host_filter filter) {
                 "change_host_filter: cannot change the configuration because hints all hints were drained"});
     }
 
-    manager_logger.debug("change_host_filter: changing from {} to {}", _host_filter, filter);
+    manager_logger.info("change_host_filter: changing from {} to {}", _host_filter, filter);
 
     // Change the host_filter now and save the old one so that we can
     // roll back in case of failure
@@ -611,11 +611,19 @@ future<> manager::change_host_filter(host_filter filter) {
             });
         });
     } catch (...) {
+        const sstring exception_message = eptr
+                ? seastar::format("{} + {}", eptr, std::current_exception())
+                : seastar::format("{}", std::current_exception());
+
+        manager_logger.warn("Changing the host filter has failed: {}", exception_message);
+
         if (eptr) {
             std::throw_with_nested(eptr);
         }
         throw;
     }
+
+    manager_logger.info("The host filter has been changed successfully");
 }
 
 bool manager::check_dc_for(endpoint_id ep) const noexcept {
@@ -635,7 +643,7 @@ future<> manager::drain_for(endpoint_id host_id, gms::inet_address ip) noexcept
         co_return;
     }
 
-    manager_logger.trace("Draining starts for {}", host_id);
+    manager_logger.info("Draining starts for {}", host_id);
 
     const auto holder = seastar::gate::holder{_draining_eps_gate};
     // As long as we hold on to this lock, no migration of hinted handoff to host IDs
@@ -661,7 +669,7 @@ future<> manager::drain_for(endpoint_id host_id, gms::inet_address ip) noexcept
 
             return ep_man.with_file_update_mutex([&ep_man] -> future<> {
                 return remove_file(ep_man.hints_dir().native()).then([&ep_man] {
-                    manager_logger.debug("Removed hint directory for {}", ep_man.end_point_key());
+                    manager_logger.info("Removed hint directory for {}", ep_man.end_point_key());
                 });
             });
         });
@@ -722,7 +730,7 @@ future<> manager::drain_for(endpoint_id host_id, gms::inet_address ip) noexcept
         manager_logger.error("Exception when draining {}: {}", host_id, eptr);
     }
 
-    manager_logger.trace("drain_for: finished draining {}", host_id);
+    manager_logger.info("drain_for: finished draining {}", host_id);
 }
 
 void manager::update_backlog(size_t backlog, size_t max_backlog) {

diff --git a/db/row_cache.cc b/db/row_cache.cc
@@ -25,7 +25,7 @@
 #include "readers/nonforwardable.hh"
 #include "cache_mutation_reader.hh"
 #include "partition_snapshot_reader.hh"
-#include "clustering_key_filter.hh"
+#include "keys/clustering_key_filter.hh"
 #include "utils/assert.hh"
 #include "utils/updateable_value.hh"
 #include "utils/labels.hh"

diff --git a/db/size_estimates_virtual_reader.cc b/db/size_estimates_virtual_reader.cc
@@ -12,7 +12,7 @@
 
 #include "utils/assert.hh"
 
-#include "clustering_bounds_comparator.hh"
+#include "keys/clustering_bounds_comparator.hh"
 #include "replica/database_fwd.hh"
 #include "db/system_keyspace.hh"
 #include "dht/i_partitioner.hh"

diff --git a/db/view/row_locking.hh b/db/view/row_locking.hh
@@ -30,7 +30,7 @@
 #include "dht/decorated_key.hh"
 #include "utils/estimated_histogram.hh"
 #include "utils/latency.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 
 class row_locker {
 public:

diff --git a/db/view/view.cc b/db/view/view.cc
@@ -25,7 +25,7 @@
 
 #include "db/view/base_info.hh"
 #include "replica/database.hh"
-#include "clustering_bounds_comparator.hh"
+#include "keys/clustering_bounds_comparator.hh"
 #include "cql3/statements/select_statement.hh"
 #include "cql3/util.hh"
 #include "cql3/restrictions/statement_restrictions.hh"
@@ -45,7 +45,7 @@
 #include "dht/sharder.hh"
 #include "gms/inet_address.hh"
 #include "gms/feature_service.hh"
-#include "keys.hh"
+#include "keys/keys.hh"
 #include "locator/abstract_replication_strategy.hh"
 #include "locator/network_topology_strategy.hh"
 #include "mutation/mutation.hh"

diff --git a/dht/auto_refreshing_sharder.hh b/dht/auto_refreshing_sharder.hh
@@ -41,7 +41,7 @@ public:
 
     virtual ~auto_refreshing_sharder();
 
-    virtual unsigned shard_for_reads(const token& t) const override;
+    virtual std::optional<unsigned> try_get_shard_for_reads(const token& t) const override;
 
     virtual dht::shard_replica_set shard_for_writes(const token& t, std::optional<write_replica_set_selector> sel) const override;
 

diff --git a/dht/decorated_key.hh b/dht/decorated_key.hh
@@ -9,7 +9,7 @@
 
 #pragma once
 
-#include "keys.hh"
+#include "keys/keys.hh"
 #include "schema/schema_fwd.hh"
 #include "dht/token.hh"
 #include "dht/i_partitioner_fwd.hh"

diff --git a/dht/i_partitioner.cc b/dht/i_partitioner.cc
@@ -43,8 +43,8 @@ static_sharder::shard_of(const token& t) const {
     return dht::shard_of(_shard_count, _sharding_ignore_msb_bits, t);
 }
 
-unsigned
-static_sharder::shard_for_reads(const token& t) const {
+std::optional<unsigned>
+static_sharder::try_get_shard_for_reads(const token& t) const {
     return shard_of(t);
 }
 
@@ -513,8 +513,8 @@ auto_refreshing_sharder::refresh() {
     });
 }
 
-unsigned auto_refreshing_sharder::shard_for_reads(const token& t) const {
-    return _sharder->shard_for_reads(t);
+std::optional<unsigned> auto_refreshing_sharder::try_get_shard_for_reads(const token& t) const {
+    return _sharder->try_get_shard_for_reads(t);
 }
 
 dht::shard_replica_set

diff --git a/dht/i_partitioner.hh b/dht/i_partitioner.hh
@@ -10,7 +10,7 @@
 #pragma once
 
 #include <seastar/core/sstring.hh>
-#include "keys.hh"
+#include "keys/keys.hh"
 #include <memory>
 #include <utility>
 #include "dht/token.hh"

diff --git a/dht/ring_position.hh b/dht/ring_position.hh
@@ -9,7 +9,7 @@
 
 #pragma once
 
-#include "keys.hh"
+#include "keys/keys.hh"
 #include "dht/token.hh"
 #include "dht/decorated_key.hh"
 

diff --git a/dht/token-sharding.hh b/dht/token-sharding.hh
@@ -54,6 +54,12 @@ public:
     sharder(unsigned shard_count = smp::count, unsigned sharding_ignore_msb_bits = 0);
     virtual ~sharder() = default;
 
+    /**
+    * Returns the shard that handles a particular token for reads, or empty if this
+    * node doesn't contain data for it.
+    */
+    virtual std::optional<unsigned> try_get_shard_for_reads(const token& t) const = 0;
+
     /**
      * Returns the shard that handles a particular token for reads.
      * Use shard_for_writes() to determine the set of shards that should receive writes.
@@ -67,7 +73,12 @@ public:
      *   }
      *
      */
-    virtual unsigned shard_for_reads(const token& t) const = 0;
+    unsigned shard_for_reads(const token& t) const {
+        // FIXME: Consider throwing when there is no owning shard on the current host rather than returning 0.
+        // It's a coordination mistake to route requests to non-owners. Topology coordinator should synchronize
+        // with request coordinators before moving the shard away.
+        return try_get_shard_for_reads(t).value_or(0);
+    }
 
     /**
      * Returns the set of shards which should receive a write to token t.
@@ -134,7 +145,7 @@ public:
     virtual std::optional<shard_and_token> next_shard(const token& t) const;
     virtual token token_for_next_shard(const token& t, shard_id shard, unsigned spans = 1) const;
 
-    virtual unsigned shard_for_reads(const token& t) const override;
+    virtual std::optional<unsigned> try_get_shard_for_reads(const token& t) const override;
     virtual shard_replica_set shard_for_writes(const token& t, std::optional<write_replica_set_selector> sel) const override;
     virtual token token_for_next_shard_for_reads(const token& t, shard_id shard, unsigned spans = 1) const override;
     virtual std::optional<shard_and_token> next_shard_for_reads(const token& t) const override;
-Original file line number
+Diff line change
@@ Expand Up / @@ -35,3 +35,4 @@ compile_commands.json @@
     .envrc
     clang_build
     .idea/
+    nuke