MDEV-14180 Automatically disable key rotation checks for file_key_managment pluginbb-10.5-MDEV-14180

Problem:
=======
- InnoDB iterates the fil_system space list to encrypt the
tablespace in case of key rotation. But it is not necessary
for any encryption plugin which doesn't do key version rotation.

Solution:
=========
- Introduce a new variable called srv_encrypt_rotate to
indicate whether encryption plugin does key rotation

fil_space_crypt_t::key_get_latest_version(): Enable the
srv_encrypt_rotate only once if current key version is higher than
innodb_encyrption_rotate_key_age

fil_crypt_must_default_encrypt(): Default encryption tables
should be added to default_encryp_tables list if
innodb_encyrption_rotate_key_age is zero and encryption
plugin doesn't do key version rotation

fil_space_create(): Add the newly created space to default_encrypt_tables
list if fil_crypt_must_default_encrypt() returns true

Removed the undetermined select from innodb-key-rotation-disable test.
By default, we add the tablespace to the rotation list and
background crypt thread does encryption of tablespace. So these
select doesn't give reliable results.

8 files changed, 101 insertions, 15 deletions

diff --git a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
index 02304fbda17..4e816bea43b 100644
--- a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
+++ b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
@@ -1,7 +1,3 @@
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
-NAME
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
-NAME
 SET GLOBAL innodb_file_per_table = ON;
 set global innodb_compression_algorithm = 1;
 create database enctests;
diff --git a/mysql-test/suite/encryption/r/key_version_rotation.result b/mysql-test/suite/encryption/r/key_version_rotation.result
new file mode 100644
index 00000000000..afc0a0dd122
--- /dev/null
+++ b/mysql-test/suite/encryption/r/key_version_rotation.result
@@ -0,0 +1,20 @@
+create table t1(f1 int not null)engine=innodb;
+create table t2(f1 int not null)engine=innodb;
+insert into t1 select * from seq_1_to_100;
+insert into t2 select * from seq_1_to_100;
+# restart: --innodb_encrypt_tables=0 --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=9
+# Enable encryption
+set global innodb_encrypt_tables=ON;
+# Create a new table and it is added to rotation list
+create table t3(f1 int not null)engine=innodb;
+insert into t3 select * from seq_1_to_100;
+# Increase the version and it should set rotation
+# variable for the encryption plugin
+set global debug_key_management_version=10;
+select  @@debug_key_management_version;
+@@debug_key_management_version
+10
+# Decrease the key version and Disable the encryption
+set global debug_key_management_version=1;
+set global innodb_encrypt_tables=off;
+DROP TABLE t1, t2, t3;
diff --git a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
index dffabaf97f1..96b62f7c05b 100644
--- a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
+++ b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
@@ -3,9 +3,6 @@
 # not embedded because of restarts
 -- source include/not_embedded.inc
 
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
-SELECT NAME FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
-
 let $encryption = `SELECT @@innodb_encrypt_tables`;
 SET GLOBAL innodb_file_per_table = ON;
 # zlib
diff --git a/mysql-test/suite/encryption/t/key_version_rotation.opt b/mysql-test/suite/encryption/t/key_version_rotation.opt
new file mode 100644
index 00000000000..d7933f0f943
--- /dev/null
+++ b/mysql-test/suite/encryption/t/key_version_rotation.opt
@@ -0,0 +1,2 @@
+--innodb-tablespaces-encryption
+--plugin-load-add=$DEBUG_KEY_MANAGEMENT_SO
diff --git a/mysql-test/suite/encryption/t/key_version_rotation.test b/mysql-test/suite/encryption/t/key_version_rotation.test
new file mode 100644
index 00000000000..d36d47251a1
--- /dev/null
+++ b/mysql-test/suite/encryption/t/key_version_rotation.test
@@ -0,0 +1,41 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_sequence.inc
+
+create table t1(f1 int not null)engine=innodb;
+create table t2(f1 int not null)engine=innodb;
+insert into t1 select * from seq_1_to_100;
+insert into t2 select * from seq_1_to_100;
+
+let $restart_parameters=--innodb_encrypt_tables=0 --innodb_encryption_threads=1 --innodb_encryption_rotate_key_age=9;
+--source include/restart_mysqld.inc
+
+--echo # Enable encryption
+
+set global innodb_encrypt_tables=ON;
+--let $tables_count= `select count(*) from information_schema.tables where engine = 'InnoDB'`
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+--source include/wait_condition.inc
+--echo # Create a new table and it is added to rotation list
+create table t3(f1 int not null)engine=innodb;
+insert into t3 select * from seq_1_to_100;
+
+--echo # Increase the version and it should set rotation
+--echo # variable for the encryption plugin
+
+set global debug_key_management_version=10;
+select  @@debug_key_management_version;
+--let $tables_count= `select count(*) from information_schema.tables where engine = 'InnoDB'`
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+--source include/wait_condition.inc
+
+--echo # Decrease the key version and Disable the encryption
+set global debug_key_management_version=1;
+set global innodb_encrypt_tables=off;
+
+--let $wait_timeout= 600
+--let $wait_condition=SELECT COUNT(*) >= $tables_count FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+--source include/wait_condition.inc
+DROP TABLE t1, t2, t3;
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 1c2634fab60..0571ad47194 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -55,6 +55,9 @@ UNIV_INTERN uint srv_n_fil_crypt_threads_started = 0;
 /** At this age or older a space/page will be rotated */
 UNIV_INTERN uint srv_fil_crypt_rotate_key_age;
 
+/** Whether the encryption plugin does key rotation */
+static std::atomic<bool> srv_encrypt_rotate;
+
 /** Event to signal FROM the key rotation threads. */
 static os_event_t fil_crypt_event;
 
@@ -128,6 +131,15 @@ fil_space_crypt_t::key_get_latest_version(void)
 
 	if (is_key_found()) {
 		key_version = encryption_key_get_latest_version(key_id);
+		/* InnoDB does dirty read of srv_fil_crypt_rotate_key_age.
+		It doesn't matter because srv_encrypt_rotate
+		can set to true only once */
+		if (!srv_encrypt_rotate
+		    && key_version > srv_fil_crypt_rotate_key_age) {
+			srv_encrypt_rotate.store(
+				true, std::memory_order_relaxed);
+		}
+
 		srv_stats.n_key_requests.inc();
 		key_found = key_version;
 	}
@@ -1440,6 +1452,12 @@ static void fil_crypt_return_iops(rotate_thread_t *state, bool wake= true)
 	fil_crypt_update_total_stat(state);
 }
 
+bool fil_crypt_must_default_encrypt()
+{
+  return !srv_fil_crypt_rotate_key_age ||
+	 !srv_encrypt_rotate.load(std::memory_order_relaxed);
+}
+
 /** Acquire a tablespace reference.
 @return whether a tablespace reference was successfully acquired */
 inline bool fil_space_t::acquire_if_not_stopped()
@@ -1498,7 +1516,7 @@ inline fil_space_t *fil_system_t::default_encrypt_next(
       }
     }
   }
-  else if (it->is_stopping())
+  else if (it != end && it->is_stopping())
   {
     /* Find the next suitable default encrypt table if
     beginning of default_encrypt_tables list has stop flag
@@ -1536,7 +1554,7 @@ inline fil_space_t *fil_space_t::next(fil_space_t *space, bool recheck,
 {
   mutex_enter(&fil_system.mutex);
 
-  if (!srv_fil_crypt_rotate_key_age)
+  if (fil_crypt_must_default_encrypt())
     space= fil_system.default_encrypt_next(space, recheck, encrypt);
   else
   {
@@ -2369,7 +2387,7 @@ void fil_crypt_set_encrypt_tables(ulong val)
 
 	srv_encrypt_tables = val;
 
-	if (srv_fil_crypt_rotate_key_age == 0) {
+	if (fil_crypt_must_default_encrypt()) {
 		fil_crypt_default_encrypt_tables_fill();
 	}
 
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 329b250c94d..99cf15c7da0 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -1001,22 +1001,28 @@ fil_space_t *fil_space_t::create(const char *name, ulint id, ulint flags,
 		fil_system.max_assigned_id = id;
 	}
 
+	const bool rotate=
+		(purpose == FIL_TYPE_TABLESPACE
+		 && (mode == FIL_ENCRYPTION_ON || mode == FIL_ENCRYPTION_OFF
+		     || srv_encrypt_tables)
+		 && fil_crypt_must_default_encrypt());
+
 	/* Inform key rotation that there could be something
 	to do */
-	if (purpose == FIL_TYPE_TABLESPACE
-	    && !srv_fil_crypt_rotate_key_age && fil_crypt_threads_event &&
-	    (mode == FIL_ENCRYPTION_ON || mode == FIL_ENCRYPTION_OFF
-	     || srv_encrypt_tables)) {
+	if (rotate) {
 		/* Key rotation is not enabled, need to inform background
 		encryption threads. */
 		fil_system.default_encrypt_tables.push_back(*space);
 		space->is_in_default_encrypt = true;
 		mutex_exit(&fil_system.mutex);
-		os_event_set(fil_crypt_threads_event);
 	} else {
 		mutex_exit(&fil_system.mutex);
 	}
 
+	if (rotate && fil_crypt_threads_event) {
+		os_event_set(fil_crypt_threads_event);
+	}
+
 	return(space);
 }
 
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 49567a49a78..872053dcbcb 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -446,4 +446,10 @@ encrypted, or corrupted.
 bool fil_space_verify_crypt_checksum(const byte* page, ulint zip_size)
 	MY_ATTRIBUTE((warn_unused_result));
 
+/** Add the tablespace to the rotation list if
+innodb_encrypt_rotate_key_age is 0 or encryption plugin does
+not do key version rotation
+@return whether the tablespace should be added to rotation list */
+bool fil_crypt_must_default_encrypt();
+
 #endif /* fil0crypt_h */