MDEV-11738: Mariadb uses 100% of several of my 8 cpus doing nothing10.1-MDEV-11738

MDEV-11581: Mariadb starts innodb encryption threads
when key has not changed or data scrubbing turned off

Introduce a new configuration variable innodb-encryption-keyrotation
that is by default ON (key rotation is done periodially). If set
OFF there is no periodical key rotation. If key rotation is
disabled we do not allow changing innodb-encrypt-tables as
it would not have any effect.

When a new tables are created they are added to list of tablespaces
needing key rotation and encryption threads are informed this
by setting an event. Encryption threads do not periodically
iterate all tablespaces on fil_system, instead they wait unil
event is received.

When event is received only tablespaces on key rotation list
are iterated and if necessary encryption for them is started.

Similarly scrubbing is done only if scrubbing is enabled.

fil0crypt.cc: Introduced a vector where new tablespaces
requiring encryption are added. New functions
fil_crypt_add_space_to_keyrotation will add tablespace
to vector and fil_crypt_get_space_from_keyrotaion()
will get tablespace from vector if any are present.
fil_crypt_set_keyrotation() is new function to set
value for global configuration variable.

fil_crypt_find_space_to_rotate(): if key rotation
is enabled look tablespaces from fil_system and if not
look only from key rotation vector.

fil0fil.cc: fil_create_new_single_table_tablespace()
if table requires key rotation add tablespace
to vector.

ha_innodb.cc: add innodb-encryption-keyrotation
global configuration parameter and method to
update its value. Issue error if innodb-encrypt-tables
is tried to change when key rotation is disabled.

srv0srv.h, srv0srv.cc: Add status variable to contain the
key rotation vector length.

18 files changed, 553 insertions, 18 deletions

diff --git a/mysql-test/suite/encryption/r/debug_key_management.result b/mysql-test/suite/encryption/r/debug_key_management.result
index 8793e6ba363..c8cca4067c3 100644
--- a/mysql-test/suite/encryption/r/debug_key_management.result
+++ b/mysql-test/suite/encryption/r/debug_key_management.result
@@ -3,6 +3,7 @@ show variables like 'innodb_encrypt%';
 Variable_name	Value
 innodb_encrypt_log	ON
 innodb_encrypt_tables	ON
+innodb_encryption_keyrotation	ON
 innodb_encryption_rotate_key_age	2
 innodb_encryption_rotation_iops	100
 innodb_encryption_threads	4
diff --git a/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
new file mode 100644
index 00000000000..2d8e6d3f73c
--- /dev/null
+++ b/mysql-test/suite/encryption/r/innodb-key-rotation-disable.result
@@ -0,0 +1,59 @@
+SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+COUNT(*) = 0
+1
+SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+COUNT(*) = 0
+0
+create database enctests;
+use enctests;
+create table t1(a int not null primary key, b char(200)) engine=innodb;
+create table t2(a int not null primary key, b char(200)) engine=innodb row_format=compressed;
+create table t3(a int not null primary key, b char(200)) engine=innodb page_compressed=yes;
+create table t4(a int not null primary key, b char(200)) engine=innodb encrypted=yes;
+create table t5(a int not null primary key, b char(200)) engine=innodb encrypted=yes row_format=compressed;
+create table t6(a int not null primary key, b char(200)) engine=innodb encrypted=yes page_compressed=yes;
+create table t7(a int not null primary key, b char(200)) engine=innodb encrypted=no;
+create table t8(a int not null primary key, b char(200)) engine=innodb encrypted=no row_format=compressed;
+create table t9(a int not null primary key, b char(200)) engine=innodb encrypted=no page_compressed=yes;
+insert into t1 values (1, 'secredmessage');
+insert into t2 values (1, 'secredmessage');
+insert into t3 values (1, 'secredmessagecompressedaaaaaaaaabbbbbbbbbbbbbbccccccccccccccc');
+insert into t4 values (1, 'secredmessage');
+insert into t5 values (1, 'secredmessage');
+insert into t6 values (1, 'secredmessagecompressedaaaaaaaaabbbbbbbbbbbbbbccccccccccccccc');
+insert into t7 values (1, 'publicmessage');
+insert into t8 values (1, 'publicmessage');
+insert into t9 values (1, 'pugliccompressedaaaaaaaaabbbbbbbbbbbbbbccccccccccccccc');
+# should list tables t1-t6
+SELECT NAME,ENCRYPTION_SCHEME,CURRENT_KEY_ID FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0 AND NAME LIKE 'enctests%';
+NAME	ENCRYPTION_SCHEME	CURRENT_KEY_ID
+enctests/t1	1	1
+enctests/t2	1	1
+enctests/t3	1	1
+enctests/t4	1	1
+enctests/t5	1	1
+enctests/t6	1	1
+# should list tables t7-t9
+SELECT NAME,ENCRYPTION_SCHEME,CURRENT_KEY_ID FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0 and NAME LIKE 'enctests%';
+NAME	ENCRYPTION_SCHEME	CURRENT_KEY_ID
+enctests/t7	0	1
+enctests/t8	0	1
+enctests/t9	0	1
+SET GLOBAL innodb_encrypt_tables=OFF;
+ERROR 42000: Variable 'innodb_encrypt_tables' can't be set to the value of 'OFF'
+SET GLOBAL innodb_encrypt_tables=ON;
+ERROR 42000: Variable 'innodb_encrypt_tables' can't be set to the value of 'ON'
+# t1 default on expecting NOT FOUND
+NOT FOUND /secred/ in t1.ibd
+# t2 default on expecting NOT FOUND
+NOT FOUND /secred/ in t2.ibd
+# t3 default on expecting NOT FOUND
+NOT FOUND /secred/ in t3.ibd
+# t4 on expecting NOT FOUND
+NOT FOUND /secred/ in t4.ibd
+# t5 on expecting NOT FOUND
+NOT FOUND /secred/ in t5.ibd
+# t6 on expecting NOT FOUND
+NOT FOUND /secred/ in t6.ibd
+use test;
+drop database enctests;
diff --git a/mysql-test/suite/encryption/r/innodb_encryption.result b/mysql-test/suite/encryption/r/innodb_encryption.result
index 9b762bbba11..3b12316b2b6 100644
--- a/mysql-test/suite/encryption/r/innodb_encryption.result
+++ b/mysql-test/suite/encryption/r/innodb_encryption.result
@@ -3,6 +3,7 @@ SHOW VARIABLES LIKE 'innodb_encrypt%';
 Variable_name	Value
 innodb_encrypt_log	ON
 innodb_encrypt_tables	ON
+innodb_encryption_keyrotation	ON
 innodb_encryption_rotate_key_age	15
 innodb_encryption_rotation_iops	100
 innodb_encryption_threads	4
@@ -47,6 +48,7 @@ SHOW VARIABLES LIKE 'innodb_encrypt%';
 Variable_name	Value
 innodb_encrypt_log	ON
 innodb_encrypt_tables	OFF
+innodb_encryption_keyrotation	ON
 innodb_encryption_rotate_key_age	15
 innodb_encryption_rotation_iops	100
 innodb_encryption_threads	0
diff --git a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.opt b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.opt
new file mode 100644
index 00000000000..5bac7b8701f
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.opt
@@ -0,0 +1,6 @@
+--innodb-encryption-keyrotation=OFF
+--innodb-encrypt-tables
+--innodb-encrypt-log
+--innodb-encryption-rotate-key-age=15
+--innodb-encryption-threads=4
+--innodb-tablespaces-encryption
diff --git a/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
new file mode 100644
index 00000000000..3ad80c8f87d
--- /dev/null
+++ b/mysql-test/suite/encryption/t/innodb-key-rotation-disable.test
@@ -0,0 +1,95 @@
+-- source include/have_innodb.inc
+-- source include/have_file_key_management_plugin.inc
+
+SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0;
+SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0;
+
+--disable_query_log
+--disable_warnings
+let $innodb_compression_algorithm_orig=`SELECT @@innodb_compression_algorithm`;
+let $innodb_file_format_orig = `SELECT @@innodb_file_format`;
+let $innodb_file_per_table_orig = `SELECT @@innodb_file_per_table`;
+let $encryption = `SELECT @@innodb_encrypt_tables`;
+SET GLOBAL innodb_file_format = `Barracuda`;
+SET GLOBAL innodb_file_per_table = ON;
+# zlib
+set global innodb_compression_algorithm = 1;
+--enable_warnings
+--enable_query_log
+
+create database enctests;
+use enctests;
+create table t1(a int not null primary key, b char(200)) engine=innodb;
+create table t2(a int not null primary key, b char(200)) engine=innodb row_format=compressed;
+create table t3(a int not null primary key, b char(200)) engine=innodb page_compressed=yes;
+create table t4(a int not null primary key, b char(200)) engine=innodb encrypted=yes;
+create table t5(a int not null primary key, b char(200)) engine=innodb encrypted=yes row_format=compressed;
+create table t6(a int not null primary key, b char(200)) engine=innodb encrypted=yes page_compressed=yes;
+create table t7(a int not null primary key, b char(200)) engine=innodb encrypted=no;
+create table t8(a int not null primary key, b char(200)) engine=innodb encrypted=no row_format=compressed;
+create table t9(a int not null primary key, b char(200)) engine=innodb encrypted=no page_compressed=yes;
+
+insert into t1 values (1, 'secredmessage');
+insert into t2 values (1, 'secredmessage');
+insert into t3 values (1, 'secredmessagecompressedaaaaaaaaabbbbbbbbbbbbbbccccccccccccccc');
+insert into t4 values (1, 'secredmessage');
+insert into t5 values (1, 'secredmessage');
+insert into t6 values (1, 'secredmessagecompressedaaaaaaaaabbbbbbbbbbbbbbccccccccccccccc');
+insert into t7 values (1, 'publicmessage');
+insert into t8 values (1, 'publicmessage');
+insert into t9 values (1, 'pugliccompressedaaaaaaaaabbbbbbbbbbbbbbccccccccccccccc');
+
+--echo # should list tables t1-t6
+SELECT NAME,ENCRYPTION_SCHEME,CURRENT_KEY_ID FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION <> 0 AND NAME LIKE 'enctests%';
+--echo # should list tables t7-t9
+SELECT NAME,ENCRYPTION_SCHEME,CURRENT_KEY_ID FROM INFORMATION_SCHEMA.INNODB_TABLESPACES_ENCRYPTION WHERE MIN_KEY_VERSION = 0 and NAME LIKE 'enctests%';
+
+--error 1231
+SET GLOBAL innodb_encrypt_tables=OFF;
+--error 1231
+SET GLOBAL innodb_encrypt_tables=ON;
+
+--let $MYSQLD_DATADIR=`select @@datadir`
+--let t1_IBD = $MYSQLD_DATADIR/enctests/t1.ibd
+--let t2_IBD = $MYSQLD_DATADIR/enctests/t2.ibd
+--let t3_IBD = $MYSQLD_DATADIR/enctests/t3.ibd
+--let t4_IBD = $MYSQLD_DATADIR/enctests/t4.ibd
+--let t5_IBD = $MYSQLD_DATADIR/enctests/t5.ibd
+--let t6_IBD = $MYSQLD_DATADIR/enctests/t6.ibd
+--let t7_IBD = $MYSQLD_DATADIR/enctests/t7.ibd
+--let t8_IBD = $MYSQLD_DATADIR/enctests/t8.ibd
+--let t9_IBD = $MYSQLD_DATADIR/enctests/t9.ibd
+--let SEARCH_RANGE = 10000000
+--let SEARCH_PATTERN=secred
+--echo # t1 default on expecting NOT FOUND
+-- let SEARCH_FILE=$t1_IBD
+-- source include/search_pattern_in_file.inc
+--echo # t2 default on expecting NOT FOUND
+-- let SEARCH_FILE=$t2_IBD
+-- source include/search_pattern_in_file.inc
+--echo # t3 default on expecting NOT FOUND
+-- let SEARCH_FILE=$t3_IBD
+-- source include/search_pattern_in_file.inc
+--echo # t4 on expecting NOT FOUND
+-- let SEARCH_FILE=$t4_IBD
+-- source include/search_pattern_in_file.inc
+--echo # t5 on expecting NOT FOUND
+-- let SEARCH_FILE=$t5_IBD
+-- source include/search_pattern_in_file.inc
+--echo # t6 on expecting NOT FOUND
+-- let SEARCH_FILE=$t6_IBD
+-- source include/search_pattern_in_file.inc
+--let SEARCH_PATTERN=public
+
+use test;
+drop database enctests;
+# reset system
+
+--disable_query_log
+--disable_warnings
+EVAL SET GLOBAL innodb_compression_algorithm = $innodb_compression_algorithm_orig;
+EVAL SET GLOBAL innodb_file_per_table = $innodb_file_per_table_orig;
+EVAL SET GLOBAL innodb_file_format = $innodb_file_format_orig;
+set global innodb_compression_algorithm = DEFAULT;
+--enable_warnings
+--enable_query_log
diff --git a/mysql-test/suite/sys_vars/r/sysvars_innodb.result b/mysql-test/suite/sys_vars/r/sysvars_innodb.result
index afa152b0c7b..3306fd2bb8f 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_innodb.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_innodb.result
@@ -775,6 +775,20 @@ NUMERIC_BLOCK_SIZE	0
 ENUM_VALUE_LIST	NULL
 READ_ONLY	YES
 COMMAND_LINE_ARGUMENT	OPTIONAL
+VARIABLE_NAME	INNODB_ENCRYPTION_KEYROTATION
+SESSION_VALUE	NULL
+GLOBAL_VALUE	ON
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	ON
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	BOOLEAN
+VARIABLE_COMMENT	Enable encryption key rotation (default ON)
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	OFF,ON
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	INNODB_ENCRYPTION_ROTATE_KEY_AGE
 SESSION_VALUE	NULL
 GLOBAL_VALUE	1
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 2999bea2765..321519936d2 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -37,7 +37,7 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 #include "fsp0fsp.h"
 #include "fil0pagecompress.h"
 #include "ha_prototypes.h" // IB_LOG_
-
+#include <vector>
 #include <my_crypt.h>
 
 /** Mutex for keys */
@@ -104,6 +104,28 @@ static mysql_pfs_key_t fil_crypt_stat_mutex_key;
 UNIV_INTERN mysql_pfs_key_t fil_crypt_data_mutex_key;
 #endif
 
+/** Mutex protecting vector of space_id's that require
+key rotation. */
+UNIV_INTERN ib_mutex_t fil_crypt_rotate_list_mutex;
+
+/** Key for rotate list mutex */
+#ifdef UNIV_PFS_MUTEX
+static mysql_pfs_key_t fil_crypt_rotate_list_mutex_key;
+#endif
+
+/** Vector type holding space_id's that require key rotation */
+typedef std::vector<ulint> fil_crypt_rotate_spaces_t;
+
+/** Vector holding space_id's that require key rotation */
+UNIV_INTERN fil_crypt_rotate_spaces_t fil_crypt_rotate_list;
+
+/** is keyrotation enabled */
+UNIV_INTERN my_bool	innodb_encryption_keyrotation;
+
+/** Is background scrubbing enabled, defined on btr0scrub.cc */
+extern my_bool srv_background_scrub_data_uncompressed;
+extern my_bool srv_background_scrub_data_compressed;
+
 static bool
 fil_crypt_needs_rotation(
 /*=====================*/
@@ -127,6 +149,10 @@ fil_space_crypt_init()
 
 	mutex_create(fil_crypt_stat_mutex_key,
 		     &crypt_stat_mutex, SYNC_NO_ORDER_CHECK);
+
+	mutex_create(fil_crypt_rotate_list_mutex_key,
+		     &fil_crypt_rotate_list_mutex, SYNC_NO_ORDER_CHECK);
+
 	memset(&crypt_stat, 0, sizeof(crypt_stat));
 }
 
@@ -140,9 +166,53 @@ fil_space_crypt_cleanup()
 	os_event_free(fil_crypt_throttle_sleep_event);
 	mutex_free(&fil_crypt_key_mutex);
 	mutex_free(&crypt_stat_mutex);
+	fil_crypt_rotate_list.clear();
+	mutex_free(&fil_crypt_rotate_list_mutex);
 }
 
 /**
+Add space to key rotation list for encryption.
+@param[in]	space_id	space to add to rotation */
+UNIV_INTERN
+void
+fil_crypt_add_space_to_keyrotation(
+	ulint	space_id)
+{
+	mutex_enter(&fil_crypt_rotate_list_mutex);
+	fil_crypt_rotate_list.push_back(space_id);
+	srv_stats.key_rotation_list_length.inc();
+	mutex_exit(&fil_crypt_rotate_list_mutex);
+	os_event_set(fil_crypt_threads_event);
+
+}
+
+/**
+Get space from key rotation list
+@return space_id or ULINT_UNDEFINED if list is empty */
+static
+ulint
+fil_crypt_get_space_from_keyrotation(void)
+{
+	ulint space_id = ULINT_UNDEFINED;
+
+	mutex_enter(&fil_crypt_rotate_list_mutex);
+
+	/* If key rotation list is empty, there is nothing
+	to do. */
+	if (fil_crypt_rotate_list.empty()) {
+		goto exit;
+	}
+
+	space_id = fil_crypt_rotate_list.back();
+	fil_crypt_rotate_list.pop_back();
+	srv_stats.key_rotation_list_length.dec();
+
+exit:
+	mutex_exit(&fil_crypt_rotate_list_mutex);
+
+	return (space_id);
+}
+/**
 Get latest key version from encryption plugin.
 @return key version or ENCRYPTION_KEY_VERSION_INVALID */
 uint
@@ -1644,11 +1714,18 @@ fil_crypt_find_space_to_rotate(
 		return false;
 	}
 
-	if (state->first) {
-		state->first = false;
-		state->space = fil_get_first_space_safe();
+	/* If key rotation is enabled (default) we iterate all tablespaces.
+	If key rotation is not enabled we iterate only the tablespaces
+	added to keyrotation list. */
+	if (innodb_encryption_keyrotation) {
+		if (state->first) {
+			state->first = false;
+			state->space = fil_get_first_space_safe();
+		} else {
+			state->space = fil_get_next_space_safe(state->space);
+		}
 	} else {
-		state->space = fil_get_next_space_safe(state->space);
+		state->space = fil_crypt_get_space_from_keyrotation();
 	}
 
 	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
@@ -1664,7 +1741,11 @@ fil_crypt_find_space_to_rotate(
 			}
 		}
 
-		state->space = fil_get_next_space_safe(state->space);
+		if (innodb_encryption_keyrotation) {
+			state->space = fil_get_next_space_safe(state->space);
+		} else {
+			state->space = fil_crypt_get_space_from_keyrotation();
+		}
 	}
 
 	/* if we didn't find any space return iops */
@@ -2305,7 +2386,13 @@ DECLARE_THREAD(fil_crypt_thread)(
 			* i.e either new key version of change or
 			* new rotate_key_age */
 			os_event_reset(fil_crypt_threads_event);
-			if (os_event_wait_time(fil_crypt_threads_event, 1000000) == 0) {
+
+			if(innodb_encryption_keyrotation) {
+				if (os_event_wait_time(fil_crypt_threads_event, 1000000) == 0) {
+					break;
+				}
+			} else {
+				os_event_wait(fil_crypt_threads_event);
 				break;
 			}
 
@@ -2318,7 +2405,11 @@ DECLARE_THREAD(fil_crypt_thread)(
 
 			time_t waited = time(0) - wait_start;
 
-			if (waited >= srv_background_scrub_data_check_interval) {
+			/* Break if we have waited the background scrub
+			internal and background scrubbing is enabled */
+			if (waited >= srv_background_scrub_data_check_interval
+			    && (srv_background_scrub_data_uncompressed
+			        || srv_background_scrub_data_compressed)) {
 				break;
 			}
 		}
@@ -2447,6 +2538,23 @@ fil_crypt_set_encrypt_tables(
        os_event_set(fil_crypt_threads_event);
 }
 
+/**
+Adjust keyrotation
+@param	val		value to be set */
+UNIV_INTERN
+void
+fil_crypt_set_keyrotation(
+	my_bool	val)
+{
+	innodb_encryption_keyrotation = val;
+
+	/* If key rotation is enabled inform encryption threads
+	if they exists. */
+	if (innodb_encryption_keyrotation && srv_n_fil_crypt_threads) {
+		os_event_set(fil_crypt_threads_event);
+	}
+}
+
 /*********************************************************************
 Init threads for key rotation */
 UNIV_INTERN
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index fe6a2922b35..5ce11e66b92 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -3902,6 +3902,15 @@ fil_create_new_single_table_tablespace(
 		goto error_exit_1;
 	}
 
+	if (success) {
+		/* Inform key rotation that there could be something
+		to do */
+		if (mode == FIL_SPACE_ENCRYPTION_ON || mode == FIL_SPACE_ENCRYPTION_OFF ||
+			srv_encrypt_tables) {
+			fil_crypt_add_space_to_keyrotation(space_id);
+		}
+	}
+
 #ifndef UNIV_HOTBACKUP
 	{
 		mtr_t		mtr;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 350438e0495..dab7ca9b5c8 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -1084,6 +1084,9 @@ static SHOW_VAR innodb_status_variables[]= {
   {"encryption_rotation_estimated_iops",
   (char*) &export_vars.innodb_encryption_rotation_estimated_iops,
    SHOW_LONG},
+  {"encryption_key_rotation_list_length",
+  (char*)&export_vars.innodb_key_rotation_list_length,
+   SHOW_LONGLONG},
 
   /* scrubing */
   {"scrub_background_page_reorganizations",
@@ -18440,6 +18443,22 @@ innodb_encrypt_tables_update(
 	fil_crypt_set_encrypt_tables(*static_cast<const ulong*>(save));
 }
 
+/******************************************************************
+Update the system variable innodb_encryption_keyrotation */
+static
+void
+innodb_encryption_keyrotation_update(
+	THD*                            thd,    /*!< in: thread handle */
+	struct st_mysql_sys_var*        var,    /*!< in: pointer to
+						system variable */
+	void*                           var_ptr,/*!< out: where the
+						formal string goes */
+	const void*                     save)   /*!< in: immediate result
+						from check function */
+{
+	fil_crypt_set_keyrotation(*static_cast<const my_bool*>(save));
+}
+
 static SHOW_VAR innodb_status_variables_export[]= {
 	{"Innodb", (char*) &show_innodb_vars, SHOW_FUNC},
 	{NullS, NullS, SHOW_LONG}
@@ -19833,6 +19852,13 @@ static MYSQL_SYSVAR_UINT(encryption_rotation_iops, srv_n_fil_crypt_iops,
 			 innodb_encryption_rotation_iops_update,
 			 srv_n_fil_crypt_iops, 0, UINT_MAX32, 0);
 
+static MYSQL_SYSVAR_BOOL(encryption_keyrotation, innodb_encryption_keyrotation,
+  PLUGIN_VAR_RQCMDARG,
+  "Enable encryption key rotation (default ON)",
+  0,
+  innodb_encryption_keyrotation_update,
+  TRUE);
+
 static MYSQL_SYSVAR_BOOL(scrub_log, srv_scrub_log,
   PLUGIN_VAR_OPCMDARG | PLUGIN_VAR_READONLY,
   "Enable background redo log (ib_logfile0, ib_logfile1...) scrubbing",
@@ -20101,6 +20127,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(scrub_log_speed),
   MYSQL_SYSVAR(encrypt_log),
   MYSQL_SYSVAR(default_encryption_key_id),
+  MYSQL_SYSVAR(encryption_keyrotation),
   /* Scrubing feature */
   MYSQL_SYSVAR(immediate_scrub_data_uncompressed),
   MYSQL_SYSVAR(background_scrub_data_uncompressed),
@@ -20741,8 +20768,9 @@ innodb_encrypt_tables_validate(
 						for update function */
 	struct st_mysql_value*		value)	/*!< in: incoming string */
 {
-	if (check_sysvar_enum(thd, var, save, value))
+	if (check_sysvar_enum(thd, var, save, value)) {
 		return 1;
+	}
 
 	ulong encrypt_tables = *(ulong*)save;
 
@@ -20754,6 +20782,16 @@ innodb_encrypt_tables_validate(
 		                    "encryption plugin is not available");
 		return 1;
 	}
+
+	if (!innodb_encryption_keyrotation) {
+		const char *msg = (encrypt_tables ? "enable" : "disable");
+		push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+				    HA_ERR_UNSUPPORTED,
+				    "InnoDB: cannot %s encryption, "
+				    "innodb-encryption-keyrotation disabled", msg);
+		return 1;
+	}
+
 	return 0;
 }
 
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 42cdafde4d0..827d63bc670 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -530,6 +530,21 @@ fil_crypt_calculate_checksum(
 /*=========================*/
 	ulint	zip_size,	/*!< in: zip_size or 0 */
 	byte*	dst_frame);	/*!< in: page where to calculate */
+/**
+Add space to key rotation list for encryption.
+@param[in]	space_id	space to add to rotation */
+UNIV_INTERN
+void
+fil_crypt_add_space_to_keyrotation(
+	ulint	space_id);
+
+/**
+Adjust keyrotation
+@param	val		value to be set */
+UNIV_INTERN
+void
+fil_crypt_set_keyrotation(
+	my_bool	val);
 
 #ifndef UNIV_NONINL
 #include "fil0crypt.ic"
diff --git a/storage/innobase/include/srv0srv.h b/storage/innobase/include/srv0srv.h
index 9dd5b3efa9a..fe8961bbb36 100644
--- a/storage/innobase/include/srv0srv.h
+++ b/storage/innobase/include/srv0srv.h
@@ -185,6 +185,9 @@ struct srv_stats_t {
 
 	/** Number of encryption_get_latest_key_version calls */
 	ulint_ctr_64_t		n_key_requests;
+
+	/** Number of spaces in keyrotation list */
+	ulint_ctr_64_t		key_rotation_list_length;
 };
 
 extern const char*	srv_main_thread_op_info;
@@ -561,6 +564,9 @@ extern my_bool	srv_cmp_per_index_enabled;
 /* is encryption enabled */
 extern ulong	srv_encrypt_tables;
 
+/* is keyrotation enabled */
+extern my_bool	innodb_encryption_keyrotation;
+
 /** Status variables to be passed to MySQL */
 extern struct export_var_t export_vars;
 
@@ -1039,6 +1045,7 @@ struct export_var_t{
 	ulint innodb_encryption_rotation_pages_flushed;
 	ulint innodb_encryption_rotation_estimated_iops;
 	ib_int64_t innodb_encryption_key_requests;
+	ib_int64_t innodb_key_rotation_list_length;
 
 	ulint innodb_scrub_page_reorganizations;
 	ulint innodb_scrub_page_splits;
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index 147279dd0d1..aa086543bc2 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -1664,6 +1664,8 @@ srv_export_innodb_status(void)
 		crypt_stat.estimated_iops;
 	export_vars.innodb_encryption_key_requests =
 		srv_stats.n_key_requests;
+	export_vars.innodb_key_rotation_list_length =
+		srv_stats.key_rotation_list_length;
 
 	export_vars.innodb_scrub_page_reorganizations =
 		scrub_stat.page_reorganizations;
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index 2999bea2765..321519936d2 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -37,7 +37,7 @@ Modified           Jan Lindström jan.lindstrom@mariadb.com
 #include "fsp0fsp.h"
 #include "fil0pagecompress.h"
 #include "ha_prototypes.h" // IB_LOG_
-
+#include <vector>
 #include <my_crypt.h>
 
 /** Mutex for keys */
@@ -104,6 +104,28 @@ static mysql_pfs_key_t fil_crypt_stat_mutex_key;
 UNIV_INTERN mysql_pfs_key_t fil_crypt_data_mutex_key;
 #endif
 
+/** Mutex protecting vector of space_id's that require
+key rotation. */
+UNIV_INTERN ib_mutex_t fil_crypt_rotate_list_mutex;
+
+/** Key for rotate list mutex */
+#ifdef UNIV_PFS_MUTEX
+static mysql_pfs_key_t fil_crypt_rotate_list_mutex_key;
+#endif
+
+/** Vector type holding space_id's that require key rotation */
+typedef std::vector<ulint> fil_crypt_rotate_spaces_t;
+
+/** Vector holding space_id's that require key rotation */
+UNIV_INTERN fil_crypt_rotate_spaces_t fil_crypt_rotate_list;
+
+/** is keyrotation enabled */
+UNIV_INTERN my_bool	innodb_encryption_keyrotation;
+
+/** Is background scrubbing enabled, defined on btr0scrub.cc */
+extern my_bool srv_background_scrub_data_uncompressed;
+extern my_bool srv_background_scrub_data_compressed;
+
 static bool
 fil_crypt_needs_rotation(
 /*=====================*/
@@ -127,6 +149,10 @@ fil_space_crypt_init()
 
 	mutex_create(fil_crypt_stat_mutex_key,
 		     &crypt_stat_mutex, SYNC_NO_ORDER_CHECK);
+
+	mutex_create(fil_crypt_rotate_list_mutex_key,
+		     &fil_crypt_rotate_list_mutex, SYNC_NO_ORDER_CHECK);
+
 	memset(&crypt_stat, 0, sizeof(crypt_stat));
 }
 
@@ -140,9 +166,53 @@ fil_space_crypt_cleanup()
 	os_event_free(fil_crypt_throttle_sleep_event);
 	mutex_free(&fil_crypt_key_mutex);
 	mutex_free(&crypt_stat_mutex);
+	fil_crypt_rotate_list.clear();
+	mutex_free(&fil_crypt_rotate_list_mutex);
 }
 
 /**
+Add space to key rotation list for encryption.
+@param[in]	space_id	space to add to rotation */
+UNIV_INTERN
+void
+fil_crypt_add_space_to_keyrotation(
+	ulint	space_id)
+{
+	mutex_enter(&fil_crypt_rotate_list_mutex);
+	fil_crypt_rotate_list.push_back(space_id);
+	srv_stats.key_rotation_list_length.inc();
+	mutex_exit(&fil_crypt_rotate_list_mutex);
+	os_event_set(fil_crypt_threads_event);
+
+}
+
+/**
+Get space from key rotation list
+@return space_id or ULINT_UNDEFINED if list is empty */
+static
+ulint
+fil_crypt_get_space_from_keyrotation(void)
+{
+	ulint space_id = ULINT_UNDEFINED;
+
+	mutex_enter(&fil_crypt_rotate_list_mutex);
+
+	/* If key rotation list is empty, there is nothing
+	to do. */
+	if (fil_crypt_rotate_list.empty()) {
+		goto exit;
+	}
+
+	space_id = fil_crypt_rotate_list.back();
+	fil_crypt_rotate_list.pop_back();
+	srv_stats.key_rotation_list_length.dec();
+
+exit:
+	mutex_exit(&fil_crypt_rotate_list_mutex);
+
+	return (space_id);
+}
+/**
 Get latest key version from encryption plugin.
 @return key version or ENCRYPTION_KEY_VERSION_INVALID */
 uint
@@ -1644,11 +1714,18 @@ fil_crypt_find_space_to_rotate(
 		return false;
 	}
 
-	if (state->first) {
-		state->first = false;
-		state->space = fil_get_first_space_safe();
+	/* If key rotation is enabled (default) we iterate all tablespaces.
+	If key rotation is not enabled we iterate only the tablespaces
+	added to keyrotation list. */
+	if (innodb_encryption_keyrotation) {
+		if (state->first) {
+			state->first = false;
+			state->space = fil_get_first_space_safe();
+		} else {
+			state->space = fil_get_next_space_safe(state->space);
+		}
 	} else {
-		state->space = fil_get_next_space_safe(state->space);
+		state->space = fil_crypt_get_space_from_keyrotation();
 	}
 
 	while (!state->should_shutdown() && state->space != ULINT_UNDEFINED) {
@@ -1664,7 +1741,11 @@ fil_crypt_find_space_to_rotate(
 			}
 		}
 
-		state->space = fil_get_next_space_safe(state->space);
+		if (innodb_encryption_keyrotation) {
+			state->space = fil_get_next_space_safe(state->space);
+		} else {
+			state->space = fil_crypt_get_space_from_keyrotation();
+		}
 	}
 
 	/* if we didn't find any space return iops */
@@ -2305,7 +2386,13 @@ DECLARE_THREAD(fil_crypt_thread)(
 			* i.e either new key version of change or
 			* new rotate_key_age */
 			os_event_reset(fil_crypt_threads_event);
-			if (os_event_wait_time(fil_crypt_threads_event, 1000000) == 0) {
+
+			if(innodb_encryption_keyrotation) {
+				if (os_event_wait_time(fil_crypt_threads_event, 1000000) == 0) {
+					break;
+				}
+			} else {
+				os_event_wait(fil_crypt_threads_event);
 				break;
 			}
 
@@ -2318,7 +2405,11 @@ DECLARE_THREAD(fil_crypt_thread)(
 
 			time_t waited = time(0) - wait_start;
 
-			if (waited >= srv_background_scrub_data_check_interval) {
+			/* Break if we have waited the background scrub
+			internal and background scrubbing is enabled */
+			if (waited >= srv_background_scrub_data_check_interval
+			    && (srv_background_scrub_data_uncompressed
+			        || srv_background_scrub_data_compressed)) {
 				break;
 			}
 		}
@@ -2447,6 +2538,23 @@ fil_crypt_set_encrypt_tables(
        os_event_set(fil_crypt_threads_event);
 }
 
+/**
+Adjust keyrotation
+@param	val		value to be set */
+UNIV_INTERN
+void
+fil_crypt_set_keyrotation(
+	my_bool	val)
+{
+	innodb_encryption_keyrotation = val;
+
+	/* If key rotation is enabled inform encryption threads
+	if they exists. */
+	if (innodb_encryption_keyrotation && srv_n_fil_crypt_threads) {
+		os_event_set(fil_crypt_threads_event);
+	}
+}
+
 /*********************************************************************
 Init threads for key rotation */
 UNIV_INTERN
diff --git a/storage/xtradb/fil/fil0fil.cc b/storage/xtradb/fil/fil0fil.cc
index f4301d47028..d90232279a3 100644
--- a/storage/xtradb/fil/fil0fil.cc
+++ b/storage/xtradb/fil/fil0fil.cc
@@ -3943,6 +3943,15 @@ fil_create_new_single_table_tablespace(
 		goto error_exit_1;
 	}
 
+	if (success) {
+		/* Inform key rotation that there could be something
+		to do */
+		if (mode == FIL_SPACE_ENCRYPTION_ON || mode == FIL_SPACE_ENCRYPTION_OFF ||
+			srv_encrypt_tables) {
+			fil_crypt_add_space_to_keyrotation(space_id);
+		}
+	}
+
 #ifndef UNIV_HOTBACKUP
 	{
 		mtr_t		mtr;
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index bef54a910e6..59572798bfc 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -1290,6 +1290,9 @@ static SHOW_VAR innodb_status_variables[]= {
   {"encryption_rotation_estimated_iops",
   (char*) &export_vars.innodb_encryption_rotation_estimated_iops,
    SHOW_LONG},
+  {"encryption_key_rotation_list_length",
+  (char*)&export_vars.innodb_key_rotation_list_length,
+   SHOW_LONGLONG},
 
   /* Scrubing feature */
   {"scrub_background_page_reorganizations",
@@ -19603,6 +19606,22 @@ innodb_encrypt_tables_update(
 	fil_crypt_set_encrypt_tables(*static_cast<const ulong*>(save));
 }
 
+/******************************************************************
+Update the system variable innodb_encryption_keyrotation */
+static
+void
+innodb_encryption_keyrotation_update(
+	THD*                            thd,    /*!< in: thread handle */
+	struct st_mysql_sys_var*        var,    /*!< in: pointer to
+						system variable */
+	void*                           var_ptr,/*!< out: where the
+						formal string goes */
+	const void*                     save)   /*!< in: immediate result
+						from check function */
+{
+	fil_crypt_set_keyrotation(*static_cast<const my_bool*>(save));
+}
+
 static SHOW_VAR innodb_status_variables_export[]= {
 	{"Innodb", (char*) &show_innodb_vars, SHOW_FUNC},
 	{NullS, NullS, SHOW_LONG}
@@ -21315,6 +21334,13 @@ static MYSQL_SYSVAR_UINT(encryption_rotation_iops, srv_n_fil_crypt_iops,
 			 innodb_encryption_rotation_iops_update,
 			 srv_n_fil_crypt_iops, 0, UINT_MAX32, 0);
 
+static MYSQL_SYSVAR_BOOL(encryption_keyrotation, innodb_encryption_keyrotation,
+  PLUGIN_VAR_RQCMDARG,
+  "Enable encryption key rotation (default ON)",
+  0,
+  innodb_encryption_keyrotation_update,
+  TRUE);
+
 static MYSQL_SYSVAR_BOOL(scrub_log, srv_scrub_log,
   PLUGIN_VAR_OPCMDARG | PLUGIN_VAR_READONLY,
   "Enable background redo log (ib_logfile0, ib_logfile1...) scrubbing",
@@ -21620,6 +21646,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(scrub_log_speed),
   MYSQL_SYSVAR(encrypt_log),
   MYSQL_SYSVAR(default_encryption_key_id),
+  MYSQL_SYSVAR(encryption_keyrotation),
   /* Scrubing feature */
   MYSQL_SYSVAR(immediate_scrub_data_uncompressed),
   MYSQL_SYSVAR(background_scrub_data_uncompressed),
@@ -22269,8 +22296,9 @@ innodb_encrypt_tables_validate(
 						for update function */
 	struct st_mysql_value*		value)	/*!< in: incoming string */
 {
-	if (check_sysvar_enum(thd, var, save, value))
+	if (check_sysvar_enum(thd, var, save, value)) {
 		return 1;
+	}
 
 	ulong encrypt_tables = *(ulong*)save;
 
@@ -22282,6 +22310,16 @@ innodb_encrypt_tables_validate(
 		                    "encryption plugin is not available");
 		return 1;
 	}
+
+	if (!innodb_encryption_keyrotation) {
+		const char *msg = (encrypt_tables ? "enable" : "disable");
+		push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+				    HA_ERR_UNSUPPORTED,
+				    "InnoDB: cannot %s encryption, "
+				    "innodb-encryption-keyrotation disabled", msg);
+		return 1;
+	}
+
 	return 0;
 }
 
diff --git a/storage/xtradb/include/fil0crypt.h b/storage/xtradb/include/fil0crypt.h
index 42cdafde4d0..827d63bc670 100644
--- a/storage/xtradb/include/fil0crypt.h
+++ b/storage/xtradb/include/fil0crypt.h
@@ -530,6 +530,21 @@ fil_crypt_calculate_checksum(
 /*=========================*/
 	ulint	zip_size,	/*!< in: zip_size or 0 */
 	byte*	dst_frame);	/*!< in: page where to calculate */
+/**
+Add space to key rotation list for encryption.
+@param[in]	space_id	space to add to rotation */
+UNIV_INTERN
+void
+fil_crypt_add_space_to_keyrotation(
+	ulint	space_id);
+
+/**
+Adjust keyrotation
+@param	val		value to be set */
+UNIV_INTERN
+void
+fil_crypt_set_keyrotation(
+	my_bool	val);
 
 #ifndef UNIV_NONINL
 #include "fil0crypt.ic"
diff --git a/storage/xtradb/include/srv0srv.h b/storage/xtradb/include/srv0srv.h
index f65848fdf45..1cbcd4da3bd 100644
--- a/storage/xtradb/include/srv0srv.h
+++ b/storage/xtradb/include/srv0srv.h
@@ -192,6 +192,9 @@ struct srv_stats_t {
 
 	/** Number of encryption_get_latest_key_version calls */
 	ulint_ctr_64_t		n_key_requests;
+
+	/** Number of spaces in keyrotation list */
+	ulint_ctr_64_t		key_rotation_list_length;
 };
 
 extern const char*	srv_main_thread_op_info;
@@ -703,6 +706,9 @@ extern my_bool	srv_cmp_per_index_enabled;
 /* is encryption enabled */
 extern ulong	srv_encrypt_tables;
 
+/* is keyrotation enabled */
+extern my_bool	innodb_encryption_keyrotation;
+
 /** Status variables to be passed to MySQL */
 extern struct export_var_t export_vars;
 
@@ -1271,6 +1277,7 @@ struct export_var_t{
 	ulint innodb_encryption_rotation_pages_flushed;
 	ulint innodb_encryption_rotation_estimated_iops;
 	ib_int64_t innodb_encryption_key_requests;
+	ib_int64_t innodb_key_rotation_list_length;
 
 	ulint innodb_scrub_page_reorganizations;
 	ulint innodb_scrub_page_splits;
diff --git a/storage/xtradb/srv/srv0srv.cc b/storage/xtradb/srv/srv0srv.cc
index bc222b864ef..b63e9db108d 100644
--- a/storage/xtradb/srv/srv0srv.cc
+++ b/storage/xtradb/srv/srv0srv.cc
@@ -2076,6 +2076,8 @@ srv_export_innodb_status(void)
 		crypt_stat.estimated_iops;
 	export_vars.innodb_encryption_key_requests =
 		srv_stats.n_key_requests;
+	export_vars.innodb_key_rotation_list_length =
+		srv_stats.key_rotation_list_length;
 
 	export_vars.innodb_scrub_page_reorganizations =
 		scrub_stat.page_reorganizations;