MDEV-11782: Redefine the innodb_encrypt_log formatbb-10.2-mdev-11782

Write only one encryption key to the checkpoint page.
Use 4 bytes of nonce. Encrypt more of each redo log block,
only skipping the 4-byte field LOG_BLOCK_HDR_NO which the
initialization vector is derived from.

Issue notes, not warning messages for rewriting the redo log files.

Issue a firm "Missing MLOG_CHECKPOINT" error message. Remove some
unreachable code.

LOG_HEADER_FORMAT_ENCRYPTED: A flag for identifying an encrypted redo
log format.

log_group_t::is_encrypted(), log_t::is_encrypted(): Determine
if the redo log is in encrypted format.

recv_find_max_checkpoint(): Interpret LOG_HEADER_FORMAT_ENCRYPTED.

srv_prepare_to_delete_redo_log_files(): Display NOTE messages about
adding or removing encryption. Do not issue warnings for redo log
resizing any more.

innobase_start_or_create_for_mysql(): Rebuild the redo logs also when
the encryption changes.

innodb_log_checksums_func_update(): Always use the CRC-32C checksum
if innodb_encrypt_log. Compute the checksum on the encrypted block
contents, so that transmission errors or incomplete blocks can be
detected without decrypting.

Rewrite most of the redo log encryption code. Only remember one
encryption key at a time (but remember up to 5 when upgrading from the
MariaDB 10.1 format.)

25 files changed, 399 insertions, 908 deletions

diff --git a/mysql-test/mysql-test-run.pl b/mysql-test/mysql-test-run.pl
index b2c98692a22..718a8350d88 100755
--- a/mysql-test/mysql-test-run.pl
+++ b/mysql-test/mysql-test-run.pl
@@ -4342,10 +4342,7 @@ sub extract_warning_lines ($$) {
      qr/error .*connecting to master/,
      qr/InnoDB: Error: in ALTER TABLE `test`.`t[12]`/,
      qr/InnoDB: Error: table `test`.`t[12]` .*does not exist in the InnoDB internal/,
-     qr/InnoDB: Warning: Setting innodb_use_sys_malloc/,
      qr/InnoDB: Warning: a long semaphore wait:/,
-     qr/InnoDB: Disabling redo log encryption/,
-     qr/InnoDB: Redo log crypto: Can't initialize to key version -1u/,
      qr/InnoDB: Dumping buffer pool.*/,
      qr/InnoDB: Buffer pool.*/,
      qr/InnoDB: Warning: Writer thread is waiting this semaphore/,
@@ -4419,9 +4416,6 @@ sub extract_warning_lines ($$) {
      qr|InnoDB: TABLE to scan your table for corruption|,
      qr/InnoDB: See also */,
      qr/InnoDB: Cannot open .*ib_buffer_pool.* for reading: No such file or directory*/,
-     qr/InnoDB: Upgrading redo log:*/,
-     qr|InnoDB: Starting to delete and rewrite log files.|,
-     qr/InnoDB: New log files created, LSN=*/,
      qr|InnoDB: Creating foreign key constraint system tables.|,
      qr/InnoDB: Table .*mysql.*innodb_table_stats.* not found./,
      qr/InnoDB: User stopword table .* does not exist./
diff --git a/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result b/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result
index 16a5c91a953..65e03a028a0 100644
--- a/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result
+++ b/mysql-test/suite/encryption/r/innodb_encrypt_log_corruption.result
@@ -41,7 +41,7 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122/ in mysqld.1.err
-FOUND /InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
+FOUND /InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
 # --innodb-force-recovery=6 (skip the entire redo log)
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
@@ -54,7 +54,6 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND /InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
 # --innodb-force-recovery=6 (skip the entire redo log)
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
@@ -90,6 +89,7 @@ SELECT COUNT(*) `1` FROM INFORMATION_SCHEMA.ENGINES WHERE engine='innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 1
 1
+FOUND /InnoDB: Encrypting redo log/ in mysqld.1.err
 ib_buffer_pool
 ib_logfile0
 ib_logfile1
diff --git a/mysql-test/suite/encryption/t/innodb_encrypt_log_corruption.test b/mysql-test/suite/encryption/t/innodb_encrypt_log_corruption.test
index 85ce09e0901..f1642e83e32 100644
--- a/mysql-test/suite/encryption/t/innodb_encrypt_log_corruption.test
+++ b/mysql-test/suite/encryption/t/innodb_encrypt_log_corruption.test
@@ -3,9 +3,12 @@
 
 SELECT COUNT(*) `1` FROM INFORMATION_SCHEMA.ENGINES WHERE engine='innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
+--source include/shutdown_mysqld.inc
+--let SEARCH_PATTERN= InnoDB: Encrypting redo log
+--source include/search_pattern_in_file.inc
 
 --let $restart_parameters=
---source include/restart_mysqld.inc
+--source include/start_mysqld.inc
 
 --list_files $bugdir
 --remove_files_wildcard $bugdir
diff --git a/mysql-test/suite/innodb/r/innodb-32k.result b/mysql-test/suite/innodb/r/innodb-32k.result
index 29374689a3b..140b9b87c2f 100644
--- a/mysql-test/suite/innodb/r/innodb-32k.result
+++ b/mysql-test/suite/innodb/r/innodb-32k.result
@@ -1,7 +1,3 @@
-call mtr.add_suppression("InnoDB: Warning: innodb_page_size has been changed from default value ");
-call mtr.add_suppression("InnoDB: Resizing redo log from ");
-call mtr.add_suppression("InnoDB: Starting to delete and rewrite log files");
-call mtr.add_suppression("InnoDB: New log files created, LSN=");
 call mtr.add_suppression("Innodb: Cannot add field.*row size is");
 # Test 1) Show the page size from Information Schema
 SELECT variable_value FROM information_schema.global_status
diff --git a/mysql-test/suite/innodb/r/innodb-64k.result b/mysql-test/suite/innodb/r/innodb-64k.result
index 9271ad70fb4..f72ba8ef8b5 100644
--- a/mysql-test/suite/innodb/r/innodb-64k.result
+++ b/mysql-test/suite/innodb/r/innodb-64k.result
@@ -1,7 +1,3 @@
-call mtr.add_suppression("InnoDB: Warning: innodb_page_size has been changed from default value *");
-call mtr.add_suppression("InnoDB: Resizing redo log from *");
-call mtr.add_suppression("InnoDB: Starting to delete and rewrite log files.");
-call mtr.add_suppression("InnoDB: New log files created, LSN=*");
 # Test 1) Show the page size from Information Schema
 SELECT variable_value FROM information_schema.global_status
 WHERE LOWER(variable_name) = 'innodb_page_size';
diff --git a/mysql-test/suite/innodb/r/innodb-bigblob.result b/mysql-test/suite/innodb/r/innodb-bigblob.result
index 20fe0ce5c43..dcde9804cdc 100644
--- a/mysql-test/suite/innodb/r/innodb-bigblob.result
+++ b/mysql-test/suite/innodb/r/innodb-bigblob.result
@@ -1,7 +1,3 @@
-call mtr.add_suppression("Resizing redo log from *");
-call mtr.add_suppression("Starting to delete and rewrite log files.");
-call mtr.add_suppression("New log files created, LSN=*");
-call mtr.add_suppression("Writer thread is waiting this semaphore");
 create table foo (id varchar(37) not null, content longblob) engine=INNODB;
 insert into foo (id, content) values('xyz', '');
 update foo set content=repeat('a', 43941888) where id='xyz';
diff --git a/mysql-test/suite/innodb/r/log_corruption.result b/mysql-test/suite/innodb/r/log_corruption.result
index ad06398e4e2..ccf5f73a3a3 100644
--- a/mysql-test/suite/innodb/r/log_corruption.result
+++ b/mysql-test/suite/innodb/r/log_corruption.result
@@ -41,7 +41,7 @@ WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
 FOUND /InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122/ in mysqld.1.err
-FOUND /InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
+FOUND /InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
 # --innodb-force-recovery=6 (skip the entire redo log)
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
@@ -54,7 +54,6 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND /InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\./ in mysqld.1.err
 # --innodb-force-recovery=6 (skip the entire redo log)
 SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
@@ -90,7 +89,7 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND /InnoDB: Redo log crypto: getting mysqld crypto key from key version failed err = 4294967295/ in mysqld.1.err
+FOUND /InnoDB: Obtaining redo log encryption key version 1 failed/ in mysqld.1.err
 FOUND /InnoDB: Decrypting checkpoint failed/ in mysqld.1.err
 ib_buffer_pool
 ib_logfile0
diff --git a/mysql-test/suite/innodb/r/log_file.result b/mysql-test/suite/innodb/r/log_file.result
index b0351232ed9..eb96dc48199 100644
--- a/mysql-test/suite/innodb/r/log_file.result
+++ b/mysql-test/suite/innodb/r/log_file.result
@@ -378,7 +378,7 @@ WHERE engine='innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 1
 1
-FOUND /Resizing redo log from 2\*\d+ to 3\*\d+ pages, LSN=\d+/ in mysqld.1.err
+FOUND /Resizing redo log from 2\*\d+ to 3\*\d+ pages; LSN=\d+/ in mysqld.1.err
 # Cleanup
 bak_ib_logfile0
 bak_ib_logfile1
diff --git a/mysql-test/suite/innodb/t/innodb-32k.test b/mysql-test/suite/innodb/t/innodb-32k.test
index 53a2d3a7442..80a05c350d0 100644
--- a/mysql-test/suite/innodb/t/innodb-32k.test
+++ b/mysql-test/suite/innodb/t/innodb-32k.test
@@ -3,10 +3,6 @@
 --source include/have_innodb.inc
 --source include/have_innodb_32k.inc
 
-call mtr.add_suppression("InnoDB: Warning: innodb_page_size has been changed from default value ");
-call mtr.add_suppression("InnoDB: Resizing redo log from ");
-call mtr.add_suppression("InnoDB: Starting to delete and rewrite log files");
-call mtr.add_suppression("InnoDB: New log files created, LSN=");
 call mtr.add_suppression("Innodb: Cannot add field.*row size is");
 
 let $MYSQLD_DATADIR= `select @@datadir`;
diff --git a/mysql-test/suite/innodb/t/innodb-64k.test b/mysql-test/suite/innodb/t/innodb-64k.test
index c611b6cb2e2..13351450cfc 100644
--- a/mysql-test/suite/innodb/t/innodb-64k.test
+++ b/mysql-test/suite/innodb/t/innodb-64k.test
@@ -3,11 +3,6 @@
 --source include/have_innodb.inc
 --source include/have_innodb_64k.inc
 
-call mtr.add_suppression("InnoDB: Warning: innodb_page_size has been changed from default value *");
-call mtr.add_suppression("InnoDB: Resizing redo log from *");
-call mtr.add_suppression("InnoDB: Starting to delete and rewrite log files.");
-call mtr.add_suppression("InnoDB: New log files created, LSN=*");
-
 let $MYSQLD_DATADIR= `select @@datadir`;
 
 --echo # Test 1) Show the page size from Information Schema
diff --git a/mysql-test/suite/innodb/t/innodb-bigblob.test b/mysql-test/suite/innodb/t/innodb-bigblob.test
index d72e20487e4..799dfe42d48 100644
--- a/mysql-test/suite/innodb/t/innodb-bigblob.test
+++ b/mysql-test/suite/innodb/t/innodb-bigblob.test
@@ -6,11 +6,6 @@
 let $status_orig=`SELECT @@innodb_status_output`;
 --enable_query_log
 
-call mtr.add_suppression("Resizing redo log from *");
-call mtr.add_suppression("Starting to delete and rewrite log files.");
-call mtr.add_suppression("New log files created, LSN=*");
-call mtr.add_suppression("Writer thread is waiting this semaphore");
-
 create table foo (id varchar(37) not null, content longblob) engine=INNODB;
 insert into foo (id, content) values('xyz', '');
 update foo set content=repeat('a', 43941888) where id='xyz';
diff --git a/mysql-test/suite/innodb/t/log_corruption.test b/mysql-test/suite/innodb/t/log_corruption.test
index 0f59f64ebdf..abeadbd8f02 100644
--- a/mysql-test/suite/innodb/t/log_corruption.test
+++ b/mysql-test/suite/innodb/t/log_corruption.test
@@ -9,11 +9,11 @@ call mtr.add_suppression("Plugin 'InnoDB' registration as a STORAGE ENGINE faile
 call mtr.add_suppression("InnoDB: Unsupported redo log format");
 call mtr.add_suppression("InnoDB: No valid checkpoint found");
 call mtr.add_suppression("InnoDB: Invalid (log block|redo log header) checksum");
-call mtr.add_suppression("InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT");
+call mtr.add_suppression("InnoDB: Missing MLOG_CHECKPOINT");
 call mtr.add_suppression("InnoDB: MLOG_FILE_NAME incorrect");
 call mtr.add_suppression("InnoDB: ############### CORRUPT LOG RECORD FOUND");
 call mtr.add_suppression("InnoDB: Found corrupted log");
-call mtr.add_suppression("InnoDB: Redo log crypto: getting mysqld crypto key from key version failed");
+call mtr.add_suppression("InnoDB: Obtaining redo log encryption key version 1 failed");
 call mtr.add_suppression("InnoDB: Decrypting checkpoint failed");
 --enable_query_log
 
@@ -206,7 +206,7 @@ eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
 let SEARCH_PATTERN=InnoDB: Invalid log block checksum. block: 2372 checkpoint no: 1 expected: 3362026715 found: 144444122;
 --source include/search_pattern_in_file.inc
-let SEARCH_PATTERN=InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\.;
+let SEARCH_PATTERN=InnoDB: Missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\.;
 --source include/search_pattern_in_file.inc
 --echo # --innodb-force-recovery=6 (skip the entire redo log)
 --let $restart_parameters= $dirs --innodb-force-recovery=6
@@ -232,14 +232,12 @@ print OUT pack("H*x[5]", "C0DEBA5E0022000c0000000138");
 print OUT pack("H*x[475]H*", "12860cb7809781e80006626f677573", "089C0ADA");
 EOF
 --copy_file $bugdir/ib_logfile0 $bugdir/ib_logfile
-# Anything below innodb_force_recovery=6 must find a valid redo log.
+# Anything below innodb_force_recovery=6 must find an invalid redo log.
 # Missing tablespace files are tolerated already with innodb_force_recovery=1.
 --let $restart_parameters= $dirs --innodb-force-recovery=5
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
-let SEARCH_PATTERN=InnoDB: Ignoring the redo log due to missing MLOG_CHECKPOINT between the checkpoint 1213964 and the end 1213952\.;
---source include/search_pattern_in_file.inc
 --echo # --innodb-force-recovery=6 (skip the entire redo log)
 --let $restart_parameters= $dirs --innodb-force-recovery=6
 --source include/start_mysqld.inc
@@ -340,7 +338,7 @@ EOF
 if (!$no_cleanup) {
 eval $check_no_innodb;
 --source include/shutdown_mysqld.inc
---let SEARCH_PATTERN= InnoDB: Redo log crypto: getting mysqld crypto key from key version failed err = 4294967295
+--let SEARCH_PATTERN= InnoDB: Obtaining redo log encryption key version 1 failed
 --source include/search_pattern_in_file.inc
 --let SEARCH_PATTERN= InnoDB: Decrypting checkpoint failed
 --source include/search_pattern_in_file.inc
diff --git a/mysql-test/suite/innodb/t/log_file.test b/mysql-test/suite/innodb/t/log_file.test
index f7a8ef36cc0..bb131b506fc 100644
--- a/mysql-test/suite/innodb/t/log_file.test
+++ b/mysql-test/suite/innodb/t/log_file.test
@@ -231,7 +231,7 @@ let SEARCH_PATTERN=Only one log file found;
 --source include/start_mysqld.inc
 eval $check_yes_innodb;
 --source include/shutdown_mysqld.inc
-let SEARCH_PATTERN=Resizing redo log from 2\*\d+ to 3\*\d+ pages, LSN=\d+;
+--let SEARCH_PATTERN=Resizing redo log from 2\*\d+ to 3\*\d+ pages; LSN=\d+
 --source include/search_pattern_in_file.inc
 
 --let $restart_parameters=
diff --git a/mysql-test/suite/innodb/t/log_file_size.test b/mysql-test/suite/innodb/t/log_file_size.test
index 25988fc6fd8..1f109881eca 100644
--- a/mysql-test/suite/innodb/t/log_file_size.test
+++ b/mysql-test/suite/innodb/t/log_file_size.test
@@ -10,9 +10,6 @@ if (`SELECT @@innodb_log_file_size = 1048576`) {
 }
 
 --disable_query_log
-call mtr.add_suppression("InnoDB: Resizing redo log");
-call mtr.add_suppression("InnoDB: Starting to delete and rewrite log files");
-call mtr.add_suppression("InnoDB: New log files created");
 call mtr.add_suppression("InnoDB: The log sequence numbers [0-9]+ and [0-9]+ in ibdata files do not match the log sequence number [0-9]+ in the ib_logfiles");
 call mtr.add_suppression("syntax error in innodb_log_group_home_dir");
 call mtr.add_suppression("Plugin 'InnoDB' init function returned error");
diff --git a/mysql-test/suite/innodb/t/log_file_size_checkpoint.test b/mysql-test/suite/innodb/t/log_file_size_checkpoint.test
index 26e0bdf5e2c..16b71bfd1f4 100644
--- a/mysql-test/suite/innodb/t/log_file_size_checkpoint.test
+++ b/mysql-test/suite/innodb/t/log_file_size_checkpoint.test
@@ -8,11 +8,6 @@ let $n=250;
 let $t=veryLongTableNameToCreateMLOG_FILE_NAMErecords;
 
 --disable_query_log
-call mtr.add_suppression("InnoDB: Resizing redo log");
-call mtr.add_suppression("InnoDB: Starting to delete and rewrite log files");
-call mtr.add_suppression("InnoDB: New log files created");
-FLUSH TABLES;
-
 let $i=$n;
 while ($i)
 {
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 170a9d5b5a3..8ad0c308760 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -89,7 +89,7 @@ this program; if not, write to the Free Software Foundation, Inc.,
 #include "fts0types.h"
 #include "ibuf0ibuf.h"
 #include "lock0lock.h"
-#include "log0log.h"
+#include "log0crypt.h"
 #include "mem0mem.h"
 #include "mtr0mtr.h"
 #include "os0file.h"
@@ -3760,7 +3760,7 @@ static
 void
 innodb_log_checksums_func_update(bool	check)
 {
-	log_checksum_algorithm_ptr = check
+	log_checksum_algorithm_ptr = check || srv_encrypt_log
 		? log_block_calc_checksum_crc32
 		: log_block_calc_checksum_none;
 }
diff --git a/storage/innobase/include/fil0crypt.h b/storage/innobase/include/fil0crypt.h
index 72dd6c0d9c9..7185857c039 100644
--- a/storage/innobase/include/fil0crypt.h
+++ b/storage/innobase/include/fil0crypt.h
@@ -27,6 +27,7 @@ Created 04/01/2015 Jan Lindström
 #define fil0crypt_h
 
 #include "os0event.h"
+#include "my_crypt.h"
 
 /**
 * Magic pattern in start of crypt data on page 0
diff --git a/storage/innobase/include/log0crypt.h b/storage/innobase/include/log0crypt.h
index 6762b621155..d1282043665 100644
--- a/storage/innobase/include/log0crypt.h
+++ b/storage/innobase/include/log0crypt.h
@@ -22,27 +22,21 @@ Innodb log encrypt/decrypt
 
 Created 11/25/2013 Minli Zhu
 Modified           Jan Lindström jan.lindstrom@mariadb.com
+MDEV-11782: Rewritten for MariaDB 10.2 by Marko Mäkelä, MariaDB Corporation.
 *******************************************************/
 #ifndef log0crypt_h
 #define log0crypt_h
 
-#include "univ.i"
-#include "ut0byte.h"
-#include "my_crypt.h"
+#include "log0log.h"
 
-typedef int Crypt_result;
-
-/* If true, enable redo log encryption. */
+/** innodb_encrypt_log: whether to encrypt the redo log */
 extern my_bool srv_encrypt_log;
 
-/***********************************************************************
-Set next checkpoint's key version to latest one, and generate new key */
+/** Initialize the redo log encryption key.
+@return whether the operation succeeded */
 UNIV_INTERN
-void
-log_crypt_set_ver_and_key(
-/*======================*/
-	ib_uint64_t next_checkpoint_no);/*!< in: next checkpoint no */
-
+bool
+log_crypt_init();
 
 /*********************************************************************//**
 Writes the crypto (version, msg and iv) info, which has been used for
@@ -64,78 +58,23 @@ log_crypt_101_read_checkpoint(const byte* buf);
 /** Decrypt a MariaDB 10.1 redo log block.
 @param[in,out]	buf	log block
 @return	whether the decryption was successful */
+UNIV_INTERN
 bool
 log_crypt_101_read_block(byte* buf);
 
-/*********************************************************************//**
-Read the crypto (version, msg and iv) info, which has been used for
-log blocks with lsn <= this checkpoint's lsn, from a log header's
-checkpoint buf. */
+/** Read the checkpoint crypto (version, msg and iv) info.
+@param[in]	buf	checkpoint buffer
+@return	whether the operation was successful */
 UNIV_INTERN
 bool
-log_crypt_read_checkpoint_buf(
-/*===========================*/
-	const byte*	buf);		/*!< in: checkpoint buffer */
-
-/********************************************************
-Encrypt one or more log block before it is flushed to disk */
-UNIV_INTERN
-void
-log_encrypt_before_write(
-/*=====================*/
-	ib_uint64_t	next_checkpoint_no,	/*!< in: log group to be flushed */
-	byte* 		block,			/*!< in/out: pointer to a log block */
-	const ulint 	size);			/*!< in: size of log blocks */
-
-/********************************************************
-Decrypt a specified log segment after they are read from a log file to a buffer.
-*/
-UNIV_INTERN
-void
-log_decrypt_after_read(
-/*===================*/
-	byte*		frame,	/*!< in/out: log segment */
-	const ulint	size);	/*!< in: log segment size */
-
-/* Error codes for crypt info */
-typedef enum {
-	LOG_UNENCRYPTED = 0,
-	LOG_CRYPT_KEY_NOT_FOUND = 1,
-	LOG_DECRYPT_MAYBE_FAILED = 2
-} log_crypt_err_t;
+log_crypt_read_checkpoint_buf(const byte* buf);
 
-/********************************************************
-Check is the checkpoint information encrypted. This check
-is based on fact has log group crypt info and based
-on this crypt info was the key version different from
-unencrypted key version. There is no realible way to
-distinguish encrypted log block from corrupted log block,
-but if log block corruption is found this function is
-used to find out if log block is maybe encrypted but
-encryption key, key management plugin or encryption
-algorithm does not match.
-@return TRUE, if log block may be encrypted */
+/** Encrypt or decrypt log blocks.
+@param[in,out]	buf	log blocks to encrypt or decrypt
+@param[in]	size	size of the buffer, in bytes
+@param[in]	decrypt	whether to decrypt instead of encrypting */
 UNIV_INTERN
-ibool
-log_crypt_block_maybe_encrypted(
-/*============================*/
-	const byte* 		log_block, 	/*!< in: log block */
-	log_crypt_err_t* 	err_info); 	/*!< out: error info */
-
-/********************************************************
-Print crypt error message to error log */
-UNIV_INTERN
-void
-log_crypt_print_error(
-/*==================*/
-	log_crypt_err_t 	err_info); 	/*!< out: error info */
-
-/*********************************************************************//**
-Print checkpoint no from log block and all encryption keys from
-checkpoints if they are present. Used for problem analysis. */
 void
-log_crypt_print_checkpoint_keys(
-/*============================*/
-	const byte* log_block);
+log_crypt(byte* buf, ulint size, bool decrypt = false);
 
 #endif  // log0crypt.h
diff --git a/storage/innobase/include/log0log.h b/storage/innobase/include/log0log.h
index 9ba7ca48edc..ae484b36260 100644
--- a/storage/innobase/include/log0log.h
+++ b/storage/innobase/include/log0log.h
@@ -37,13 +37,9 @@ Created 12/9/1995 Heikki Tuuri
 #include "univ.i"
 #include "dyn0buf.h"
 #include "sync0rw.h"
-#include "log0crypt.h"
 #include "log0types.h"
 #include "os0event.h"
 
-/** Redo log buffer */
-struct log_t;
-
 /** Redo log group */
 struct log_group_t;
 
@@ -275,15 +271,6 @@ objects! */
 void
 log_check_margins(void);
 
-/******************************************************//**
-Reads a specified log segment to a buffer. */
-void
-log_group_read_log_seg(
-/*===================*/
-	byte*		buf,		/*!< in: buffer where to read */
-	log_group_t*	group,		/*!< in: log group */
-	lsn_t		start_lsn,	/*!< in: read area start */
-	lsn_t		end_lsn);	/*!< in: read area end */
 /********************************************************//**
 Sets the field values in group to correspond to a given lsn. For this function
 to work, the values must already be correctly initialized to correspond to
@@ -449,9 +436,6 @@ void
 log_mem_free(void);
 /*==============*/
 
-/** Redo log system */
-extern log_t*	log_sys;
-
 /** Whether to generate and require checksums on the redo log pages */
 extern my_bool	innodb_log_checksums;
 
@@ -508,6 +492,12 @@ extern my_bool	innodb_log_checksums;
 #define LOG_CHECKPOINT_LSN		8
 #define LOG_CHECKPOINT_OFFSET		16
 #define LOG_CHECKPOINT_LOG_BUF_SIZE	24
+/** MariaDB 10.2.5 encrypted redo log encryption key version (32 bits)*/
+#define LOG_CHECKPOINT_CRYPT_KEY	32
+/** MariaDB 10.2.5 encrypted redo log random nonce (32 bits) */
+#define LOG_CHECKPOINT_CRYPT_NONCE	36
+/** MariaDB 10.2.5 encrypted redo log random message (MY_AES_BLOCK_SIZE) */
+#define LOG_CHECKPOINT_CRYPT_MESSAGE	40
 
 /** Offsets of a log file header */
 /* @{ */
@@ -538,19 +528,11 @@ or the MySQL version that created the redo log file. */
 /** The redo log format identifier corresponding to the current format version.
 Stored in LOG_HEADER_FORMAT. */
 #define LOG_HEADER_FORMAT_CURRENT	1
+/** Encrypted MariaDB redo log */
+#define LOG_HEADER_FORMAT_ENCRYPTED	(1U<<31)
 
 /* @} */
 
-/** MariaDB Server 10.1 encrypted redo log offsets */
-/* @{ */
-#define LOG_CRYPT_VER			(20 + 32 * 9)
-#define LOG_CRYPT_MAX_ENTRIES           (5)
-#define LOG_CRYPT_ENTRY_SIZE            (4 + 4 + 2 * MY_AES_BLOCK_SIZE)
-#define LOG_CRYPT_SIZE                  (1 + 1 +			\
-					 (LOG_CRYPT_MAX_ENTRIES *	\
-					  LOG_CRYPT_ENTRY_SIZE))
-/* @} */
-
 #define LOG_CHECKPOINT_1	OS_FILE_LOG_BLOCK_SIZE
 					/* first checkpoint field in the log
 					header; we write alternately to the
@@ -609,6 +591,12 @@ struct log_group_t{
 	byte*				checkpoint_buf;
 	/** list of log groups */
 	UT_LIST_NODE_T(log_group_t)	log_groups;
+
+	/** @return whether the redo log is encrypted */
+	bool is_encrypted() const
+	{
+		return((format & LOG_HEADER_FORMAT_ENCRYPTED) != 0);
+	}
 };
 
 /** Redo log buffer */
@@ -750,8 +738,17 @@ struct log_t{
 	byte*		checkpoint_buf;	/*!< checkpoint header is read to this
 					buffer */
 	/* @} */
+
+	/** @return whether the redo log is encrypted */
+	bool is_encrypted() const
+	{
+		return(UT_LIST_GET_FIRST(log_groups)->is_encrypted());
+	}
 };
 
+/** Redo log system */
+extern log_t*	log_sys;
+
 /** Test if flush order mutex is owned. */
 #define log_flush_order_mutex_own()			\
 	mutex_own(&log_sys->log_flush_order_mutex)
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 6b4f817a9d0..24a83ec2ab1 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -297,13 +297,4 @@ use these free frames to read in pages when we start applying the
 log records to the database. */
 extern ulint	recv_n_pool_free_frames;
 
-/******************************************************//**
-Checks the 4-byte checksum to the trailer checksum field of a log
-block.  */
-bool
-log_block_checksum_is_ok(
-/*===================================*/
-	const byte*	block,	/*!< in: pointer to a log block */
-	bool            print_err); /*!< in print error ? */
-
 #endif
diff --git a/storage/innobase/log/log0crypt.cc b/storage/innobase/log/log0crypt.cc
index cde9768e78e..09c37459e1f 100644
--- a/storage/innobase/log/log0crypt.cc
+++ b/storage/innobase/log/log0crypt.cc
@@ -22,55 +22,44 @@ Innodb log encrypt/decrypt
 
 Created 11/25/2013 Minli Zhu Google
 Modified           Jan Lindström jan.lindstrom@mariadb.com
+MDEV-11782: Rewritten for MariaDB 10.2 by Marko Mäkelä, MariaDB Corporation.
 *******************************************************/
 #include "m_string.h"
 #include "log0crypt.h"
-#include <my_crypt.h>
-#include <my_crypt.h>
+#include "my_crypt.h"
 
-#include "log0log.h"
+#include "log0crypt.h"
 #include "srv0start.h" // for srv_start_lsn
 #include "log0recv.h"  // for recv_sys
 
-#include "ha_prototypes.h" // IB_LOG_
-
-#include "my_crypt.h"
-
-/* Used for debugging */
-// #define DEBUG_CRYPT 1
-#define UNENCRYPTED_KEY_VER	0
-
-/* If true, enable redo log encryption. */
-extern my_bool srv_encrypt_log;
-
-
-#include <algorithm>    // std::sort
-#include <deque>
-
-/* If true, enable redo log encryption. */
-UNIV_INTERN my_bool srv_encrypt_log = FALSE;
-/*
-  Sub system type for InnoDB redo log crypto.
-  Set and used to validate crypto msg.
-*/
-static const byte redo_log_purpose_byte = 0x02;
+/** innodb_encrypt_log: whether to encrypt the redo log */
+my_bool srv_encrypt_log;
 
+/** Redo log encryption key ID */
 #define LOG_DEFAULT_ENCRYPTION_KEY 1
 
-/*
-  Store this many keys into each checkpoint info
-*/
-static const size_t kMaxSavedKeys = LOG_CRYPT_MAX_ENTRIES;
-
 struct crypt_info_t {
-	ib_uint64_t     checkpoint_no; /*!< checkpoint no */
+	ulint		checkpoint_no; /*!< checkpoint no; 32 bits */
 	uint		key_version;   /*!< mysqld key version */
-	byte            crypt_msg[MY_AES_BLOCK_SIZE];
-	byte		crypt_key[MY_AES_BLOCK_SIZE];
-	byte            crypt_nonce[MY_AES_BLOCK_SIZE];
+	union {
+		uint32_t	u[MY_AES_BLOCK_SIZE / sizeof(uint32_t)];
+		byte		b[MY_AES_BLOCK_SIZE];
+	} crypt_msg;
+	union {
+		uint32_t	u[MY_AES_BLOCK_SIZE / sizeof(uint32_t)];
+		byte		b[MY_AES_BLOCK_SIZE];
+	} crypt_key;
+	union {
+		uint32_t	u;
+		byte		b[4];
+	} crypt_nonce;
 };
 
-static std::deque<crypt_info_t> crypt_info;
+/** The crypt info */
+static crypt_info_t info;
+
+/** Crypt info when upgrading from 10.1 */
+static crypt_info_t infos[5];
 
 /*********************************************************************//**
 Get a log block's start lsn.
@@ -93,145 +82,79 @@ Get crypt info from checkpoint.
 @return a crypt info or NULL if not present. */
 static
 const crypt_info_t*
-get_crypt_info(
-/*===========*/
-	ib_uint64_t checkpoint_no)
+get_crypt_info(ulint checkpoint_no)
 {
-	size_t items = crypt_info.size();
-
 	/* a log block only stores 4-bytes of checkpoint no */
 	checkpoint_no &= 0xFFFFFFFF;
-	for (size_t i = 0; i < items; i++) {
-		struct crypt_info_t* it = &crypt_info[i];
+	for (unsigned i = 0; i < 5; i++) {
+		const crypt_info_t* it = &infos[i];
 
-		if (it->checkpoint_no == checkpoint_no) {
+		if (it->key_version && it->checkpoint_no == checkpoint_no) {
 			return it;
 		}
 	}
 
 	/* If checkpoint contains more than one key and we did not
 	find the correct one use the first one. */
-	if (items) {
-		return (&crypt_info[0]);
-	}
-
-	return NULL;
+	return infos;
 }
 
-/*********************************************************************//**
-Get crypt info from log block
-@return a crypt info or NULL if not present. */
-static
-const crypt_info_t*
-get_crypt_info(
-/*===========*/
-	const byte* log_block)
-{
-	ib_uint64_t checkpoint_no = log_block_get_checkpoint_no(log_block);
-	return get_crypt_info(checkpoint_no);
-}
-
-/*********************************************************************//**
-Print checkpoint no from log block and all encryption keys from
-checkpoints if they are present. Used for problem analysis. */
+/** Encrypt or decrypt log blocks.
+@param[in,out]	buf	log blocks to encrypt or decrypt
+@param[in]	size	size of the buffer, in bytes
+@param[in]	decrypt	whether to decrypt instead of encrypting */
+UNIV_INTERN
 void
-log_crypt_print_checkpoint_keys(
-/*============================*/
-	const byte* log_block)
+log_crypt(byte* buf, ulint size, bool decrypt)
 {
-	ib_uint64_t checkpoint_no = log_block_get_checkpoint_no(log_block);
-
-	if (crypt_info.size()) {
-		ib::info() << "Redo log checkpoint encryption: " << checkpoint_no << " [ chk key ]: ";
-		for (size_t i = 0; i < crypt_info.size(); i++) {
-			struct crypt_info_t* it = &crypt_info[i];
-			ib::info() << "[" << it->checkpoint_no
-				   << "," << it->key_version
-				   << "]";
-		}
-	}
-}
+	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
+	ut_a(info.key_version);
 
-/*********************************************************************//**
-Call AES CTR to encrypt/decrypt log blocks. */
-static
-Crypt_result
-log_blocks_crypt(
-/*=============*/
-	const byte* block,	/*!< in: blocks before encrypt/decrypt*/
-	ulint size,		/*!< in: size of block */
-	byte* dst_block,	/*!< out: blocks after encrypt/decrypt */
-	int what,		/*!< in: encrypt or decrypt*/
-	const crypt_info_t* crypt_info) /*!< in: crypt info or NULL */
-{
-	byte *log_block = (byte*)block;
-	Crypt_result rc = MY_AES_OK;
 	uint dst_len;
-	byte aes_ctr_counter[MY_AES_BLOCK_SIZE];
-	byte is_encrypt= what == ENCRYPTION_FLAG_ENCRYPT;
-	lsn_t lsn = is_encrypt ? log_sys->lsn : srv_start_lsn;
+	uint32_t aes_ctr_iv[MY_AES_BLOCK_SIZE / sizeof(uint32_t)];
+	compile_time_assert(sizeof(uint32_t) == 4);
 
-	const uint src_len = OS_FILE_LOG_BLOCK_SIZE - LOG_BLOCK_HDR_SIZE;
-	for (ulint i = 0; i < size ; i += OS_FILE_LOG_BLOCK_SIZE) {
-		ulint log_block_no = log_block_get_hdr_no(log_block);
-		lsn_t log_block_start_lsn = log_block_get_start_lsn(
-			lsn, log_block_no);
-
-		const crypt_info_t* info = crypt_info == NULL ? get_crypt_info(log_block) :
-			crypt_info;
-#ifdef DEBUG_CRYPT
-		fprintf(stderr,
-			"%s %lu chkpt: %lu key: %u lsn: %lu\n",
-			is_encrypt ? "crypt" : "decrypt",
-			log_block_no,
-			log_block_get_checkpoint_no(log_block),
-			info ? info->key_version : 0,
-			log_block_start_lsn);
-#endif
-		/* If no key is found from checkpoint assume the log_block
-		to be unencrypted. If checkpoint contains the encryption key
-		compare log_block current checksum, if checksum matches,
-		block can't be encrypted. */
-		if (info == NULL ||
-		    info->key_version == UNENCRYPTED_KEY_VER ||
-			(log_block_checksum_is_ok(log_block, false) &&
-			 what == ENCRYPTION_FLAG_DECRYPT)) {
-			memcpy(dst_block, log_block, OS_FILE_LOG_BLOCK_SIZE);
-			goto next;
-		}
+#define LOG_CRYPT_HDR_SIZE 4
 
-		ut_ad(what == ENCRYPTION_FLAG_DECRYPT ? !log_block_checksum_is_ok(log_block, false) :
-			log_block_checksum_is_ok(log_block, false));
-
-		// Assume log block header is not encrypted
-		memcpy(dst_block, log_block, LOG_BLOCK_HDR_SIZE);
-
-		// aes_ctr_counter = nonce(3-byte) + start lsn to a log block
-		// (8-byte) + lbn (4-byte) + abn
-		// (1-byte, only 5 bits are used). "+" means concatenate.
-		bzero(aes_ctr_counter, MY_AES_BLOCK_SIZE);
-		memcpy(aes_ctr_counter, info->crypt_nonce, 3);
-		mach_write_to_8(aes_ctr_counter + 3, log_block_start_lsn);
-		mach_write_to_4(aes_ctr_counter + 11, log_block_no);
-		bzero(aes_ctr_counter + 15, 1);
-
-		int rc;
-		rc = encryption_crypt(log_block + LOG_BLOCK_HDR_SIZE, src_len,
-					dst_block + LOG_BLOCK_HDR_SIZE, &dst_len,
-					(unsigned char*)(info->crypt_key), 16,
-					aes_ctr_counter, MY_AES_BLOCK_SIZE,
-					what | ENCRYPTION_FLAG_NOPAD,
-					LOG_DEFAULT_ENCRYPTION_KEY,
-					info->key_version);
+	for (const byte* const end = buf + size; buf != end;
+	     buf += OS_FILE_LOG_BLOCK_SIZE) {
+		byte dst[OS_FILE_LOG_BLOCK_SIZE - LOG_CRYPT_HDR_SIZE];
+		const ulint log_block_no = log_block_get_hdr_no(buf);
+
+		/* The log block number is not encrypted. */
+		*aes_ctr_iv =
+#ifdef WORDS_BIGENDIAN
+			~LOG_BLOCK_FLUSH_BIT_MASK
+#else
+			~(LOG_BLOCK_FLUSH_BIT_MASK >> 24)
+#endif
+			& (*reinterpret_cast<uint32_t*>(dst)
+			   = *reinterpret_cast<const uint32_t*>(
+				   buf + LOG_BLOCK_HDR_NO));
+#if LOG_BLOCK_HDR_NO + 4 != LOG_CRYPT_HDR_SIZE
+# error "LOG_BLOCK_HDR_NO has been moved; redo log format affected!"
+#endif
+		aes_ctr_iv[1] = info.crypt_nonce.u;
+		mach_write_to_8(reinterpret_cast<byte*>(aes_ctr_iv + 2),
+				log_block_get_start_lsn(
+					decrypt ? srv_start_lsn : log_sys->lsn,
+					log_block_no));
+
+		int rc = encryption_crypt(
+			buf + LOG_CRYPT_HDR_SIZE, sizeof dst, dst, &dst_len,
+			const_cast<byte*>(info.crypt_key.b),
+			sizeof info.crypt_key,
+			reinterpret_cast<byte*>(aes_ctr_iv), sizeof aes_ctr_iv,
+			decrypt
+			? ENCRYPTION_FLAG_DECRYPT | ENCRYPTION_FLAG_NOPAD
+			: ENCRYPTION_FLAG_ENCRYPT | ENCRYPTION_FLAG_NOPAD,
+			LOG_DEFAULT_ENCRYPTION_KEY,
+			info.key_version);
 
 		ut_a(rc == MY_AES_OK);
-		ut_a(dst_len == src_len);
-next:
-		log_block += OS_FILE_LOG_BLOCK_SIZE;
-		dst_block += OS_FILE_LOG_BLOCK_SIZE;
+		ut_a(dst_len == sizeof dst);
+		memcpy(buf + LOG_CRYPT_HDR_SIZE, dst, sizeof dst);
 	}
-
-	return rc;
 }
 
 /*********************************************************************//**
@@ -243,216 +166,75 @@ init_crypt_key(
 /*===========*/
 	crypt_info_t* info)		/*< in/out: crypt info */
 {
-	if (info->key_version == UNENCRYPTED_KEY_VER) {
-		memset(info->crypt_key, 0, sizeof(info->crypt_key));
-		memset(info->crypt_msg, 0, sizeof(info->crypt_msg));
-		memset(info->crypt_nonce, 0, sizeof(info->crypt_nonce));
-		return true;
-	}
-
-	byte mysqld_key[MY_AES_MAX_KEY_LENGTH] = {0};
+	byte mysqld_key[MY_AES_MAX_KEY_LENGTH];
 	uint keylen= sizeof(mysqld_key);
 	uint rc;
 
+	compile_time_assert(16 == sizeof info->crypt_key);
+
 	rc = encryption_key_get(LOG_DEFAULT_ENCRYPTION_KEY, info->key_version, mysqld_key, &keylen);
 
 	if (rc) {
 		ib::error()
-			<< "Redo log crypto: getting mysqld crypto key "
-			<< "from key version failed err = " << rc
-			<< " Reason could be that requested key_version "
-			<< info->key_version
-			<< "is not found or required encryption "
-			<< " key management is not found.";
+			<< "Obtaining redo log encryption key version "
+			<< info->key_version << " failed (" << rc
+			<< "). Maybe the key or the required encryption "
+			<< " key management was not found.";
 		return false;
 	}
 
 	uint dst_len;
 	int err= my_aes_crypt(MY_AES_ECB, ENCRYPTION_FLAG_NOPAD|ENCRYPTION_FLAG_ENCRYPT,
-                             info->crypt_msg, sizeof(info->crypt_msg), //src, srclen
-                             info->crypt_key, &dst_len, //dst, &dstlen
-                             (unsigned char*)&mysqld_key, sizeof(mysqld_key),
-                             NULL, 0);
+			      info->crypt_msg.b, sizeof info->crypt_msg,
+			      info->crypt_key.b,
+			      &dst_len,
+			      mysqld_key, sizeof mysqld_key, NULL, 0);
 
 	if (err != MY_AES_OK || dst_len != MY_AES_BLOCK_SIZE) {
-		fprintf(stderr,
-			"\nInnodb redo log crypto: getting redo log crypto key "
-			"failed err = %d len = %u.\n", err, dst_len);
+		ib::error() << "Getting redo log crypto key failed: err = "
+			<< err << ", len = " << dst_len;
 		return false;
 	}
 
 	return true;
 }
 
-/*********************************************************************//**
-Compare function for checkpoint numbers
-@return true if first checkpoint is larger than second one */
-static
+/** Initialize the redo log encryption key.
+@return whether the operation succeeded */
+UNIV_INTERN
 bool
-mysort(const crypt_info_t& i,
-       const crypt_info_t& j)
+log_crypt_init()
 {
-	return i.checkpoint_no > j.checkpoint_no;
-}
+	ut_ad(log_mutex_own());
+	ut_ad(log_sys->is_encrypted());
 
-/*********************************************************************//**
-Add crypt info to set if it is not already present
-@return true if successfull, false if not- */
-static
-bool
-add_crypt_info(
-/*===========*/
-	crypt_info_t*	info,		/*!< in: crypt info */
-	bool		checkpoint_read)/*!< in: do we read checkpoint */
-{
-	const crypt_info_t* found=NULL;
-	/* so that no one is searching array while we modify it */
-	ut_ad(mutex_own(&(log_sys->mutex)));
-
-	found = get_crypt_info(info->checkpoint_no);
-
-	/* If one crypt info is found then we add a new one only if we
-	are reading checkpoint from the log. New checkpoints will always
-	use the first created crypt info. */
-	if (found != NULL &&
-		( found->checkpoint_no == info->checkpoint_no || !checkpoint_read)) {
-		// already present...
-		return true;
-	}
+	info.key_version = encryption_key_get_latest_version(
+		LOG_DEFAULT_ENCRYPTION_KEY);
 
-	if (!init_crypt_key(info)) {
+	if (my_random_bytes(info.crypt_msg.b, sizeof info.crypt_msg)
+	    != MY_AES_OK
+	    || my_random_bytes(info.crypt_nonce.b, sizeof info.crypt_nonce)
+	    != MY_AES_OK) {
+		ib::error() << "innodb_encrypt_log: my_random_bytes() failed";
 		return false;
 	}
 
-	crypt_info.push_back(*info);
-
-	/* a log block only stores 4-bytes of checkpoint no */
-	crypt_info.back().checkpoint_no &= 0xFFFFFFFF;
-
-	// keep keys sorted, assuming that last added key will be used most
-	std::sort(crypt_info.begin(), crypt_info.end(), mysort);
-
-	return true;
-}
-
-/*********************************************************************//**
-Encrypt log blocks. */
-UNIV_INTERN
-Crypt_result
-log_blocks_encrypt(
-/*===============*/
-	const byte* block,		/*!< in: blocks before encryption */
-	const ulint size,		/*!< in: size of blocks, must be multiple of a log block */
-	byte* dst_block)		/*!< out: blocks after encryption */
-{
-	return log_blocks_crypt(block, size, dst_block, ENCRYPTION_FLAG_ENCRYPT, NULL);
-}
-
-/*********************************************************************//**
-Set next checkpoint's key version to latest one, and generate current
-key. Key version 0 means no encryption. */
-UNIV_INTERN
-void
-log_crypt_set_ver_and_key(
-/*======================*/
-	ib_uint64_t next_checkpoint_no)
-{
-	crypt_info_t info;
-	info.checkpoint_no = next_checkpoint_no;
-
-	if (!srv_encrypt_log) {
-		info.key_version = UNENCRYPTED_KEY_VER;
-	} else {
-		info.key_version = encryption_key_get_latest_version(LOG_DEFAULT_ENCRYPTION_KEY);
-	}
-
-	if (info.key_version == UNENCRYPTED_KEY_VER) {
-		memset(info.crypt_msg, 0, sizeof(info.crypt_msg));
-		memset(info.crypt_nonce, 0, sizeof(info.crypt_nonce));
-	} else {
-		if (my_random_bytes(info.crypt_msg, MY_AES_BLOCK_SIZE) != MY_AES_OK) {
-			ib::error()
-				<< "Redo log crypto: generate "
-				<< MY_AES_BLOCK_SIZE
-				<< "-byte random number as crypto msg failed.";
-			ut_error;
-		}
-
-		if (my_random_bytes(info.crypt_nonce, MY_AES_BLOCK_SIZE) != MY_AES_OK) {
-			ib::error()
-				<< "Redo log crypto: generate "
-				<< MY_AES_BLOCK_SIZE
-				<< "-byte random number as AES_CTR nonce failed.";
-			ut_error;
-		}
-
-	}
-
-	add_crypt_info(&info, false);
+	return init_crypt_key(&info);
 }
 
-/********************************************************
-Encrypt one or more log block before it is flushed to disk */
+/** Encrypt a log segment.
+@param[in]	next_checkpoint_no	checkpoint number
+@param[in]	buf			log segment
+@param[in]	size			size of the buffer, in bytes
+					(integer multiple of log block size) */
 UNIV_INTERN
 void
-log_encrypt_before_write(
-/*=====================*/
-	ib_uint64_t next_checkpoint_no,	/*!< in: log group to be flushed */
-	byte* block,			/*!< in/out: pointer to a log block */
-	const ulint size)		/*!< in: size of log blocks */
+log_encrypt(
+	ib_uint64_t	next_checkpoint_no,
+	byte* 		buf,
+	const ulint 	size)
 {
-	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
-
-	const crypt_info_t* info = get_crypt_info(next_checkpoint_no);
-	if (info == NULL) {
-		return;
-	}
-
-	/* If the key is not encrypted or user has requested not to
-	encrypt, do not change log block. */
-	if (info->key_version == UNENCRYPTED_KEY_VER || !srv_encrypt_log) {
-		return;
-	}
-
-	byte* dst_frame = (byte*)malloc(size);
-
-	//encrypt log blocks content
-	Crypt_result result = log_blocks_crypt(block, size, dst_frame, ENCRYPTION_FLAG_ENCRYPT, NULL);
-
-	if (result == MY_AES_OK) {
-		ut_ad(block[0] == dst_frame[0]);
-		memcpy(block, dst_frame, size);
-	}
-	free(dst_frame);
-
-	if (unlikely(result != MY_AES_OK)) {
-		ut_error;
-	}
-}
-
-/********************************************************
-Decrypt a specified log segment after they are read from a log file to a buffer.
-*/
-void
-log_decrypt_after_read(
-/*===================*/
-	byte* frame,	        /*!< in/out: log segment */
-	const ulint size)	/*!< in: log segment size */
-{
-	ut_ad(size % OS_FILE_LOG_BLOCK_SIZE == 0);
-	byte* dst_frame = (byte*)malloc(size);
-
-	// decrypt log blocks content
-	Crypt_result result = log_blocks_crypt(frame, size, dst_frame, ENCRYPTION_FLAG_DECRYPT, NULL);
-
-	if (result == MY_AES_OK) {
-		memcpy(frame, dst_frame, size);
-	}
-	free(dst_frame);
-
-	if (unlikely(result != MY_AES_OK)) {
-		ut_error;
-	}
+	log_crypt(buf, size);
 }
 
 /** Read the MariaDB 10.1 checkpoint crypto (version, msg and iv) info.
@@ -467,13 +249,13 @@ log_crypt_101_read_checkpoint(const byte* buf)
 	const size_t n = *buf++ == 2 ? std::min(unsigned(*buf++), 5U) : 0;
 
 	for (size_t i = 0; i < n; i++) {
-		struct crypt_info_t info;
+		struct crypt_info_t& info = infos[i];
 		info.checkpoint_no = mach_read_from_4(buf);
 		info.key_version = mach_read_from_4(buf + 4);
-		memcpy(info.crypt_msg, buf + 8, MY_AES_BLOCK_SIZE);
-		memcpy(info.crypt_nonce, buf + 24, MY_AES_BLOCK_SIZE);
+		memcpy(info.crypt_msg.b, buf + 8, sizeof info.crypt_msg);
+		memcpy(info.crypt_nonce.b, buf + 24, sizeof info.crypt_nonce);
 
-		if (!add_crypt_info(&info, true)) {
+		if (!init_crypt_key(&info)) {
 			return false;
 		}
 		buf += 4 + 4 + 2 * MY_AES_BLOCK_SIZE;
@@ -491,7 +273,8 @@ log_crypt_101_read_block(byte* buf)
 {
 	ut_ad(log_block_calc_checksum_format_0(buf)
 	      != log_block_get_checksum(buf));
-	const crypt_info_t* info = get_crypt_info(buf);
+	const crypt_info_t* info = get_crypt_info(
+		log_block_get_checkpoint_no(buf));
 
 	if (!info || info->key_version == 0) {
 		return false;
@@ -499,27 +282,27 @@ log_crypt_101_read_block(byte* buf)
 
 	byte dst[OS_FILE_LOG_BLOCK_SIZE];
 	uint dst_len;
-	byte aes_ctr_counter[MY_AES_BLOCK_SIZE];
+	byte aes_ctr_iv[MY_AES_BLOCK_SIZE];
 
 	const uint src_len = OS_FILE_LOG_BLOCK_SIZE - LOG_BLOCK_HDR_SIZE;
 
 	ulint log_block_no = log_block_get_hdr_no(buf);
-	lsn_t log_block_start_lsn = log_block_get_start_lsn(
-		srv_start_lsn, log_block_no);
 
 	/* The log block header is not encrypted. */
 	memcpy(dst, buf, LOG_BLOCK_HDR_SIZE);
 
-	memcpy(aes_ctr_counter, info->crypt_nonce, 3);
-	mach_write_to_8(aes_ctr_counter + 3, log_block_start_lsn);
-	mach_write_to_4(aes_ctr_counter + 11, log_block_no);
-	aes_ctr_counter[15] = 0;
+	memcpy(aes_ctr_iv, info->crypt_nonce.b, 3);
+	mach_write_to_8(aes_ctr_iv + 3,
+			log_block_get_start_lsn(srv_start_lsn, log_block_no));
+	memcpy(aes_ctr_iv + 11, buf, 4);
+	aes_ctr_iv[11] &= ~(LOG_BLOCK_FLUSH_BIT_MASK >> 24);
+	aes_ctr_iv[15] = 0;
 
 	int rc = encryption_crypt(buf + LOG_BLOCK_HDR_SIZE, src_len,
 				  dst + LOG_BLOCK_HDR_SIZE, &dst_len,
-				  const_cast<byte*>(info->crypt_key),
+				  const_cast<byte*>(info->crypt_key.b),
 				  MY_AES_BLOCK_SIZE,
-				  aes_ctr_counter, MY_AES_BLOCK_SIZE,
+				  aes_ctr_iv, MY_AES_BLOCK_SIZE,
 				  ENCRYPTION_FLAG_DECRYPT
 				  | ENCRYPTION_FLAG_NOPAD,
 				  LOG_DEFAULT_ENCRYPTION_KEY,
@@ -535,184 +318,47 @@ log_crypt_101_read_block(byte* buf)
 	return true;
 }
 
-/*********************************************************************//**
-Writes the crypto (version, msg and iv) info, which has been used for
-log blocks with lsn <= this checkpoint's lsn, to a log header's
-checkpoint buf. */
+/** Add the encryption information to a redo log checkpoint buffer.
+@param[in,out]	buf	checkpoint buffer */
 UNIV_INTERN
 void
-log_crypt_write_checkpoint_buf(
-/*===========================*/
-	byte*	buf)			/*!< in/out: checkpoint buffer */
+log_crypt_write_checkpoint_buf(byte* buf)
 {
-	byte *save = buf;
-
-	// Only write kMaxSavedKeys (sort keys to remove oldest)
-	std::sort(crypt_info.begin(), crypt_info.end(), mysort);
-	while (crypt_info.size() > kMaxSavedKeys) {
-		crypt_info.pop_back();
-	}
-
-	bool encrypted = false;
-	for (size_t i = 0; i < crypt_info.size(); i++) {
-                const crypt_info_t & it = crypt_info[i];
-		if (it.key_version != UNENCRYPTED_KEY_VER) {
-			encrypted = true;
-			break;
-		}
-	}
-
-	if (encrypted == false) {
-		// if no encryption is inuse then zero out
-		// crypt data for upward/downward compability
-		memset(buf + LOG_CRYPT_VER, 0, LOG_CRYPT_SIZE);
-		return;
-	}
-
-	ib_uint64_t checkpoint_no = mach_read_from_8(buf + LOG_CHECKPOINT_NO);
-	buf += LOG_CRYPT_VER;
-
-	mach_write_to_1(buf + 0, redo_log_purpose_byte);
-	mach_write_to_1(buf + 1, crypt_info.size());
-	buf += 2;
-	for (size_t i = 0; i < crypt_info.size(); i++) {
-		struct crypt_info_t* it = &crypt_info[i];
-		mach_write_to_4(buf + 0, static_cast<ulint>(it->checkpoint_no));
-		mach_write_to_4(buf + 4, it->key_version);
-		memcpy(buf + 8, it->crypt_msg, MY_AES_BLOCK_SIZE);
-		memcpy(buf + 24, it->crypt_nonce, MY_AES_BLOCK_SIZE);
-		buf += LOG_CRYPT_ENTRY_SIZE;
-	}
-
-#ifdef DEBUG_CRYPT
-	fprintf(stderr, "write chk: %lu [ chk key ]: ", checkpoint_no);
-	for (size_t i = 0; i < crypt_info.size(); i++) {
-		struct crypt_info_t* it = &crypt_info[i];
-		fprintf(stderr, "[ %lu %u ] ",
-			it->checkpoint_no,
-			it->key_version);
-	}
-	fprintf(stderr, "\n");
-#else
-	(void)checkpoint_no; // unused variable
-#endif
-        ut_a((ulint)(buf - save) <= OS_FILE_LOG_BLOCK_SIZE);
+	ut_ad(info.key_version);
+	compile_time_assert(16 == sizeof info.crypt_msg);
+	compile_time_assert(LOG_CHECKPOINT_CRYPT_MESSAGE
+			    - LOG_CHECKPOINT_CRYPT_NONCE
+			    == sizeof info.crypt_nonce);
+
+	memcpy(buf + LOG_CHECKPOINT_CRYPT_MESSAGE, info.crypt_msg.b,
+	       sizeof info.crypt_msg);
+	memcpy(buf + LOG_CHECKPOINT_CRYPT_NONCE, info.crypt_nonce.b,
+	       sizeof info.crypt_nonce);
+	mach_write_to_4(buf + LOG_CHECKPOINT_CRYPT_KEY, info.key_version);
 }
 
-/*********************************************************************//**
-Read the crypto (version, msg and iv) info, which has been used for
-log blocks with lsn <= this checkpoint's lsn, from a log header's
-checkpoint buf. */
+/** Read the checkpoint crypto (version, msg and iv) info.
+@param[in]	buf	checkpoint buffer
+@return	whether the operation was successful */
 UNIV_INTERN
 bool
-log_crypt_read_checkpoint_buf(
-/*===========================*/
-	const byte*	buf) {			/*!< in: checkpoint buffer */
-
-	buf += LOG_CRYPT_VER;
-
-	byte scheme = buf[0];
-	if (scheme != redo_log_purpose_byte) {
-		return true;
-	}
-	buf++;
-	size_t n = buf[0];
-	buf++;
-
-	for (size_t i = 0; i < n; i++) {
-		struct crypt_info_t info;
-		info.checkpoint_no = mach_read_from_4(buf + 0);
-		info.key_version = mach_read_from_4(buf + 4);
-		memcpy(info.crypt_msg, buf + 8, MY_AES_BLOCK_SIZE);
-		memcpy(info.crypt_nonce, buf + 24, MY_AES_BLOCK_SIZE);
-
-		if (!add_crypt_info(&info, true)) {
-			return false;
-		}
-		buf += LOG_CRYPT_ENTRY_SIZE;
-	}
-
-#ifdef DEBUG_CRYPT
-	fprintf(stderr, "read [ chk key ]: ");
-	for (size_t i = 0; i < crypt_info.size(); i++) {
-		struct crypt_info_t* it = &crypt_info[i];
-		fprintf(stderr, "[ %lu %u ] ",
-			it->checkpoint_no,
-			it->key_version);
-	}
-	fprintf(stderr, "\n");
-#endif
-	return true;
-}
-
-/********************************************************
-Check is the checkpoint information encrypted. This check
-is based on fact has log group crypt info and based
-on this crypt info was the key version different from
-unencrypted key version. There is no realible way to
-distinguish encrypted log block from corrupted log block,
-but if log block corruption is found this function is
-used to find out if log block is maybe encrypted but
-encryption key, key management plugin or encryption
-algorithm does not match.
-@return TRUE, if log block may be encrypted */
-UNIV_INTERN
-ibool
-log_crypt_block_maybe_encrypted(
-/*============================*/
-	const byte* 		log_block,	/*!< in: log block */
-	log_crypt_err_t* 	err_info)	/*!< out: error info */
+log_crypt_read_checkpoint_buf(const byte* buf)
 {
-	ibool maybe_encrypted = FALSE;
-	const crypt_info_t* crypt_info;
+	info.checkpoint_no = mach_read_from_4(buf + (LOG_CHECKPOINT_NO + 4));
+	info.key_version = mach_read_from_4(buf + LOG_CHECKPOINT_CRYPT_KEY);
 
-	*err_info = LOG_UNENCRYPTED;
-	crypt_info = get_crypt_info(log_block);
-
-	if (crypt_info &&
-	    crypt_info->key_version != UNENCRYPTED_KEY_VER) {
-		byte mysqld_key[MY_AES_BLOCK_SIZE] = {0};
-		uint keylen= sizeof(mysqld_key);
-
-		/* Log block contains crypt info and based on key
-		version block could be encrypted. */
-		*err_info = LOG_DECRYPT_MAYBE_FAILED;
-		maybe_encrypted = TRUE;
-
-		if (encryption_key_get(LOG_DEFAULT_ENCRYPTION_KEY,
-				       crypt_info->key_version, mysqld_key, &keylen)) {
-			*err_info = LOG_CRYPT_KEY_NOT_FOUND;
-		}
-	}
+#if MY_AES_BLOCK_SIZE != 16
+# error "MY_AES_BLOCK_SIZE != 16; redo log checkpoint format affected"
+#endif
+	compile_time_assert(16 == sizeof info.crypt_msg);
+	compile_time_assert(LOG_CHECKPOINT_CRYPT_MESSAGE
+			    - LOG_CHECKPOINT_CRYPT_NONCE
+			    == sizeof info.crypt_nonce);
 
-	return (maybe_encrypted);
-}
+	memcpy(info.crypt_msg.b, buf + LOG_CHECKPOINT_CRYPT_MESSAGE,
+	       sizeof info.crypt_msg);
+	memcpy(info.crypt_nonce.b, buf + LOG_CHECKPOINT_CRYPT_NONCE,
+	       sizeof info.crypt_nonce);
 
-/********************************************************
-Print crypt error message to error log */
-UNIV_INTERN
-void
-log_crypt_print_error(
-/*==================*/
-	log_crypt_err_t 	err_info) 	/*!< out: error info */
-{
-	switch(err_info) {
-	case LOG_CRYPT_KEY_NOT_FOUND:
-		ib::error()
-			<< "Redo log crypto: getting mysqld crypto key "
-			<< "from key version failed. Reason could be that "
-			<< "requested key version is not found or required "
-			<< "encryption key management plugin is not found.";
-		break;
-	case LOG_DECRYPT_MAYBE_FAILED:
-		ib::error()
-			<< "Redo log crypto: failed to decrypt log block. "
-			<< "Reason could be that requested key version is "
-			<< "not found, required encryption key management "
-			<< "plugin is not found or configured encryption "
-			<< "algorithm and/or method does not match.";
-		break;
-	default:
-		ut_error; /* Real bug */
-	}
+	return init_crypt_key(&info);
 }
diff --git a/storage/innobase/log/log0log.cc b/storage/innobase/log/log0log.cc
index 1fe87f1250e..ca052138dd9 100644
--- a/storage/innobase/log/log0log.cc
+++ b/storage/innobase/log/log0log.cc
@@ -40,6 +40,7 @@ Created 12/9/1995 Heikki Tuuri
 #include "log0log.ic"
 #endif
 
+#include "log0crypt.h"
 #include "mem0mem.h"
 #include "buf0buf.h"
 #include "buf0flu.h"
@@ -56,9 +57,6 @@ Created 12/9/1995 Heikki Tuuri
 #include "srv0mon.h"
 #include "sync0sync.h"
 
-/* Used for debugging */
-// #define DEBUG_CRYPT 1
-
 /*
 General philosophy of InnoDB redo-logs:
 
@@ -898,7 +896,9 @@ log_group_init(
 
 	group->id = id;
 	group->n_files = n_files;
-	group->format = LOG_HEADER_FORMAT_CURRENT;
+	group->format = srv_encrypt_log
+		? LOG_HEADER_FORMAT_CURRENT | LOG_HEADER_FORMAT_ENCRYPTED
+		: LOG_HEADER_FORMAT_CURRENT;
 	group->file_size = file_size;
 	group->space_id = space_id;
 	group->state = LOG_GROUP_OK;
@@ -987,11 +987,13 @@ log_group_file_header_flush(
 	ut_ad(!recv_no_log_write);
 	ut_ad(group->id == 0);
 	ut_a(nth_file < group->n_files);
+	ut_ad((group->format & ~LOG_HEADER_FORMAT_ENCRYPTED)
+	      == LOG_HEADER_FORMAT_CURRENT);
 
 	buf = *(group->file_header_bufs + nth_file);
 
 	memset(buf, 0, OS_FILE_LOG_BLOCK_SIZE);
-	mach_write_to_4(buf + LOG_HEADER_FORMAT, LOG_HEADER_FORMAT_CURRENT);
+	mach_write_to_4(buf + LOG_HEADER_FORMAT, group->format);
 	mach_write_to_8(buf + LOG_HEADER_START_LSN, start_lsn);
 	strcpy(reinterpret_cast<char*>(buf) + LOG_HEADER_CREATOR,
 	       LOG_HEADER_CREATOR_CURRENT);
@@ -1115,6 +1117,10 @@ loop:
 	      || log_block_get_hdr_no(buf)
 		 == log_block_convert_lsn_to_no(start_lsn));
 
+	if (log_sys->is_encrypted()) {
+		log_crypt(buf, write_len);
+	}
+
 	/* Calculate the checksums for each log block and write them to
 	the trailer fields of the log blocks */
 
@@ -1135,8 +1141,6 @@ loop:
 
 	ut_a(next_offset / UNIV_PAGE_SIZE <= ULINT_MAX);
 
-	log_encrypt_before_write(log_sys->next_checkpoint_no,
-				buf, write_len);
 	const ulint	page_no
 		= (ulint) (next_offset / univ_page_size.physical());
 
@@ -1625,7 +1629,9 @@ log_group_checkpoint(
 	mach_write_to_8(buf + LOG_CHECKPOINT_NO, log_sys->next_checkpoint_no);
 	mach_write_to_8(buf + LOG_CHECKPOINT_LSN, log_sys->next_checkpoint_lsn);
 
-	log_crypt_write_checkpoint_buf(buf);
+	if (log_sys->is_encrypted()) {
+		log_crypt_write_checkpoint_buf(buf);
+	}
 
 	lsn_offset = log_group_calc_lsn_offset(log_sys->next_checkpoint_lsn,
 					       group);
@@ -1689,30 +1695,16 @@ log_group_header_read(
 /** Write checkpoint info to the log header and invoke log_mutex_exit().
 @param[in]	sync	whether to wait for the write to complete */
 void
-log_write_checkpoint_info(
-	bool	sync)
+log_write_checkpoint_info(bool sync)
 {
-	log_group_t*	group;
-
 	ut_ad(log_mutex_own());
+	ut_ad(!srv_read_only_mode);
 
-	if (!srv_read_only_mode) {
-		for (group = UT_LIST_GET_FIRST(log_sys->log_groups);
-		     group;
-		     group = UT_LIST_GET_NEXT(log_groups, group)) {
-
-			log_group_checkpoint(group);
-		}
-	}
+	for (log_group_t* group = UT_LIST_GET_FIRST(log_sys->log_groups);
+	     group;
+	     group = UT_LIST_GET_NEXT(log_groups, group)) {
 
-	/* generate key version and key used to encrypt future blocks,
-	*
-	* NOTE: the +1 is as the next_checkpoint_no will be updated once
-	* the checkpoint info has been written and THEN blocks will be encrypted
-	* with new key
-	*/
-	if (srv_encrypt_log) {
-		log_crypt_set_ver_and_key(log_sys->next_checkpoint_no + 1);
+		log_group_checkpoint(group);
 	}
 
 	log_mutex_exit();
@@ -1991,78 +1983,6 @@ loop:
 	}
 }
 
-/******************************************************//**
-Reads a specified log segment to a buffer. */
-void
-log_group_read_log_seg(
-/*===================*/
-	byte*		buf,		/*!< in: buffer where to read */
-	log_group_t*	group,		/*!< in: log group */
-	lsn_t		start_lsn,	/*!< in: read area start */
-	lsn_t		end_lsn)	/*!< in: read area end */
-{
-	ulint	len;
-	lsn_t	source_offset;
-
-	ut_ad(log_mutex_own());
-
-loop:
-	source_offset = log_group_calc_lsn_offset(start_lsn, group);
-
-	ut_a(end_lsn - start_lsn <= ULINT_MAX);
-	len = (ulint) (end_lsn - start_lsn);
-
-	ut_ad(len != 0);
-
-	if ((source_offset % group->file_size) + len > group->file_size) {
-
-		/* If the above condition is true then len (which is ulint)
-		is > the expression below, so the typecast is ok */
-		len = (ulint) (group->file_size -
-			(source_offset % group->file_size));
-	}
-
-	log_sys->n_log_ios++;
-
-	MONITOR_INC(MONITOR_LOG_IO);
-
-	ut_a(source_offset / UNIV_PAGE_SIZE <= ULINT_MAX);
-
-	const ulint	page_no
-		= (ulint) (source_offset / univ_page_size.physical());
-
-	fil_io(IORequestLogRead, true,
-	       page_id_t(group->space_id, page_no),
-	       univ_page_size,
-	       (ulint) (source_offset % univ_page_size.physical()),
-	       len, buf, NULL);
-
-#ifdef DEBUG_CRYPT
-	fprintf(stderr, "BEFORE DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx offset %lu\n",
-		log_block_get_hdr_no(buf),
-			log_block_get_checkpoint_no(buf),
-			log_block_calc_checksum(buf),
-		log_block_get_checksum(buf), source_offset);
-#endif
-
-	log_decrypt_after_read(buf, len);
-
-#ifdef DEBUG_CRYPT
-	fprintf(stderr, "AFTER DECRYPT: block: %lu checkpoint: %lu %.8lx %.8lx\n",
-			log_block_get_hdr_no(buf),
-			log_block_get_checkpoint_no(buf),
-			log_block_calc_checksum(buf),
-			log_block_get_checksum(buf));
-#endif
-	start_lsn += len;
-	buf += len;
-
-	if (start_lsn != end_lsn) {
-
-		goto loop;
-	}
-}
-
 /**
 Checks that there is enough free space in the log to start a new query step.
 Flushes the log buffer or makes a new checkpoint if necessary. NOTE: this
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 8c6080d4d7e..7c4a7f70945 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -688,6 +688,99 @@ recv_sys_debug_free(void)
 	mutex_exit(&(recv_sys->mutex));
 }
 
+/** Read a log segment to a buffer.
+@param[out]	buf		buffer
+@param[in]	group		redo log files
+@param[in]	start_lsn	read area start
+@param[in]	end_lsn		read area end
+@return	valid end_lsn */
+static
+lsn_t
+log_group_read_log_seg(
+	byte*			buf,
+	const log_group_t*	group,
+	lsn_t			start_lsn,
+	lsn_t			end_lsn)
+{
+	ulint	len;
+	lsn_t	source_offset;
+
+	ut_ad(log_mutex_own());
+
+loop:
+	source_offset = log_group_calc_lsn_offset(start_lsn, group);
+
+	ut_a(end_lsn - start_lsn <= ULINT_MAX);
+	len = (ulint) (end_lsn - start_lsn);
+
+	ut_ad(len != 0);
+
+	const bool at_eof = (source_offset % group->file_size) + len
+		> group->file_size;
+	if (at_eof) {
+		/* If the above condition is true then len (which is ulint)
+		is > the expression below, so the typecast is ok */
+		len = (ulint) (group->file_size -
+			(source_offset % group->file_size));
+	}
+
+	log_sys->n_log_ios++;
+
+	MONITOR_INC(MONITOR_LOG_IO);
+
+	ut_a(source_offset / UNIV_PAGE_SIZE <= ULINT_MAX);
+
+	const ulint	page_no
+		= (ulint) (source_offset / univ_page_size.physical());
+
+	fil_io(IORequestLogRead, true,
+	       page_id_t(group->space_id, page_no),
+	       univ_page_size,
+	       (ulint) (source_offset % univ_page_size.physical()),
+	       len, buf, NULL);
+
+	for (ulint l = 0; l < len; l += OS_FILE_LOG_BLOCK_SIZE,
+		     buf += OS_FILE_LOG_BLOCK_SIZE,
+		     start_lsn += OS_FILE_LOG_BLOCK_SIZE) {
+		const ulint block_number = log_block_get_hdr_no(buf);
+
+		if (block_number != log_block_convert_lsn_to_no(start_lsn)) {
+			/* Garbage or an incompletely written log block.
+			We will not report any error, because this can
+			happen when InnoDB was killed while it was
+			writing redo log. We simply treat this as an
+			abrupt end of the redo log. */
+			return(start_lsn);
+		}
+
+		if (innodb_log_checksums || group->is_encrypted()) {
+			ulint crc = log_block_calc_checksum_crc32(buf);
+			ulint cksum = log_block_get_checksum(buf);
+
+			if (crc != cksum) {
+				ib::error() << "Invalid log block checksum."
+					    << " block: " << block_number
+					    << " checkpoint no: "
+					    << log_block_get_checkpoint_no(buf)
+					    << " expected: " << crc
+					    << " found: " << cksum;
+				return(start_lsn);
+			}
+
+			if (group->is_encrypted()) {
+				log_crypt(buf, OS_FILE_LOG_BLOCK_SIZE, true);
+			}
+		}
+	}
+
+	if (start_lsn != end_lsn) {
+
+		goto loop;
+	}
+
+	return(start_lsn);
+}
+
 /********************************************************//**
 Copies a log segment from the most up-to-date log group to the other log
 groups, so that they all contain the latest log data. Also writes the info
@@ -695,8 +788,7 @@ about the latest checkpoint to the groups, and inits the fields in the group
 memory structs to up-to-date values. */
 static
 void
-recv_synchronize_groups(void)
-/*=========================*/
+recv_synchronize_groups()
 {
 	const lsn_t recovered_lsn = recv_sys->recovered_lsn;
 
@@ -730,8 +822,10 @@ recv_synchronize_groups(void)
 	over the max checkpoint info, thus making the preservation of max
 	checkpoint info on disk certain */
 
-	log_write_checkpoint_info(true);
-	log_mutex_enter();
+	if (!srv_read_only_mode) {
+		log_write_checkpoint_info(true);
+		log_mutex_enter();
+	}
 }
 
 /** Check the consistency of a log header block.
@@ -799,7 +893,7 @@ recv_find_max_checkpoint_0(
 			buf + LOG_CHECKPOINT_NO);
 
 		if (!log_crypt_101_read_checkpoint(buf)) {
-			ib::warn() << "Decrypting checkpoint failed";
+			ib::error() << "Decrypting checkpoint failed";
 			continue;
 		}
 
@@ -930,6 +1024,7 @@ recv_find_max_checkpoint(
 			return(recv_find_max_checkpoint_0(
 				       max_group, max_field));
 		case LOG_HEADER_FORMAT_CURRENT:
+		case LOG_HEADER_FORMAT_CURRENT | LOG_HEADER_FORMAT_ENCRYPTED:
 			break;
 		default:
 			/* Ensure that the string is NUL-terminated. */
@@ -963,6 +1058,13 @@ recv_find_max_checkpoint(
 				continue;
 			}
 
+			if (group->is_encrypted()
+			    && !log_crypt_read_checkpoint_buf(buf)) {
+				ib::error() << "Reading checkpoint"
+					" encryption info failed.";
+				continue;
+			}
+
 			group->state = LOG_GROUP_OK;
 
 			group->lsn = mach_read_from_8(
@@ -972,11 +1074,6 @@ recv_find_max_checkpoint(
 			checkpoint_no = mach_read_from_8(
 				buf + LOG_CHECKPOINT_NO);
 
-			if (!log_crypt_read_checkpoint_buf(buf)) {
-				ib::error() << "Reading checkpoint encryption info failed.";
-				return DB_ERROR;
-			}
-
 			DBUG_PRINT("ib_log",
 				   ("checkpoint " UINT64PF " at " LSN_PF
 				    " found in group " ULINTPF,
@@ -1008,28 +1105,6 @@ recv_find_max_checkpoint(
 	return(DB_SUCCESS);
 }
 
-/** Check the 4-byte checksum to the trailer checksum field of a log
-block.
-@param[in]	block		log block
-@param[in]	print_err	whether to report checksum mismatch
-@return whether the checksum matches */
-bool
-log_block_checksum_is_ok(const byte* block, bool print_err)
-{
-	bool	valid
-		= log_block_get_checksum(block) == log_block_calc_checksum(block);
-
-	if (!valid && print_err) {
-		ib::error() << "Invalid log block checksum."
-			    << " block: " << log_block_get_hdr_no(block)
-			    << " checkpoint no: " <<  log_block_get_checkpoint_no(block)
-			    << " expected: " << log_block_calc_checksum(block)
-			    << " found: " << log_block_get_checksum(block);
-	}
-
-	return(valid || !innodb_log_checksums);
-}
-
 /** Try to parse a single log record body and also applies it if
 specified.
 @param[in]	type		redo log entry type
@@ -2721,13 +2796,11 @@ recv_scan_log_recs(
 	dberr_t*	err)		/*!< out: error code or DB_SUCCESS */
 {
 	const byte*	log_block	= buf;
-	ulint		no;
 	lsn_t		scanned_lsn	= start_lsn;
 	bool		finished	= false;
 	ulint		data_len;
 	bool		more_data	= false;
 	bool		apply		= recv_sys->mlog_checkpoint_lsn != 0;
-	bool		maybe_encrypted = false;
 
 	ut_ad(start_lsn % OS_FILE_LOG_BLOCK_SIZE == 0);
 	ut_ad(len % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -2736,42 +2809,6 @@ recv_scan_log_recs(
 
 	do {
 		ut_ad(!finished);
-		no = log_block_get_hdr_no(log_block);
-		ulint expected_no = log_block_convert_lsn_to_no(scanned_lsn);
-		if (no != expected_no) {
-			/* Garbage or an incompletely written log block.
-
-			We will not report any error, because this can
-			happen when InnoDB was killed while it was
-			writing redo log. We simply treat this as an
-			abrupt end of the redo log. */
-			finished = true;
-			break;
-		}
-
-		if (!log_block_checksum_is_ok(log_block, true)) {
-			log_crypt_err_t log_crypt_err;
-
-			maybe_encrypted = log_crypt_block_maybe_encrypted(log_block,
-					&log_crypt_err);
-
-			/* Print checkpoint encryption keys if present */
-			log_crypt_print_checkpoint_keys(log_block);
-			if (maybe_encrypted) {
-				/* Log block maybe encrypted finish processing*/
-				log_crypt_print_error(log_crypt_err);
-				*err = DB_ERROR;
-				return (TRUE);
-			}
-
-			/* Garbage or an incompletely written log block.
-
-			This could be the result of killing the server
-			while it was writing this log block. We treat
-			this as an abrupt end of the redo log. */
-			finished = true;
-			break;
-		}
 
 		if (log_block_get_flush_bit(log_block)) {
 			/* This block was a start of a log flush operation:
@@ -2972,7 +3009,7 @@ recv_group_scan_log_recs(
 		   - (recv_n_pool_free_frames * srv_buf_pool_instances));
 
 	*err = DB_SUCCESS;
-	end_lsn = *contiguous_lsn = ut_uint64_align_down(
+	group->scanned_lsn = end_lsn = *contiguous_lsn = ut_uint64_align_down(
 		*contiguous_lsn, OS_FILE_LOG_BLOCK_SIZE);
 
 	do {
@@ -2990,18 +3027,18 @@ recv_group_scan_log_recs(
 		}
 
 		start_lsn = end_lsn;
-		end_lsn += RECV_SCAN_SIZE;
-
-		log_group_read_log_seg(
-			log_sys->buf, group, start_lsn, end_lsn);
-	} while (!recv_scan_log_recs(
+		end_lsn = log_group_read_log_seg(
+			log_sys->buf, group, start_lsn,
+			start_lsn + RECV_SCAN_SIZE);
+	} while (end_lsn != start_lsn
+		 && !recv_scan_log_recs(
 			 available_mem, &store_to_hash, log_sys->buf,
-			 RECV_SCAN_SIZE,
+			 end_lsn - start_lsn,
 			 checkpoint_lsn,
 			 start_lsn, contiguous_lsn, &group->scanned_lsn, err));
 
 	if (recv_sys->found_corrupt_log || recv_sys->found_corrupt_fs) {
-		ib::error() << "Found corrupted log when looking checkpoint lsn: "
+		ib::error() << "Found corrupted log when looking for checkpoint lsn: "
 			    << contiguous_lsn << " error = " << *err;
 		DBUG_RETURN(false);
 	}
@@ -3185,8 +3222,7 @@ recv_recovery_from_checkpoint_start(
 
 	if (srv_force_recovery >= SRV_FORCE_NO_LOG_REDO) {
 
-		ib::info() << "The user has set SRV_FORCE_NO_LOG_REDO on,"
-			" skipping log redo";
+		ib::info() << "innodb_force_recovery=6 skips redo log apply";
 
 		return(DB_SUCCESS);
 	}
@@ -3231,6 +3267,7 @@ recv_recovery_from_checkpoint_start(
 		log_mutex_exit();
 		return(recv_log_format_0_recover(checkpoint_lsn));
 	case LOG_HEADER_FORMAT_CURRENT:
+	case LOG_HEADER_FORMAT_CURRENT | LOG_HEADER_FORMAT_ENCRYPTED:
 		break;
 	default:
 		ut_ad(0);
@@ -3260,14 +3297,12 @@ recv_recovery_from_checkpoint_start(
 	if (recv_sys->mlog_checkpoint_lsn == 0) {
 		if (!srv_read_only_mode
 		    && group->scanned_lsn != checkpoint_lsn) {
-			ib::error() << "Ignoring the redo log due to missing"
+			ib::error() << "Missing"
 				" MLOG_CHECKPOINT between the checkpoint "
 				<< checkpoint_lsn << " and the end "
 				<< group->scanned_lsn << ".";
-			if (srv_force_recovery < SRV_FORCE_NO_LOG_REDO) {
-				log_mutex_exit();
-				return(DB_ERROR);
-			}
+			log_mutex_exit();
+			return(DB_ERROR);
 		}
 
 		group->scanned_lsn = checkpoint_lsn;
@@ -3380,10 +3415,6 @@ recv_recovery_from_checkpoint_start(
 	log_sys->next_checkpoint_lsn = checkpoint_lsn;
 	log_sys->next_checkpoint_no = checkpoint_no + 1;
 
-	/* here the checkpoint info is written without any redo logging ongoing
-	* and next_checkpoint_no is updated directly hence no +1 */
-	log_crypt_set_ver_and_key(log_sys->next_checkpoint_no);
-
 	recv_synchronize_groups();
 
 	if (!recv_needed_recovery) {
@@ -3407,8 +3438,7 @@ recv_recovery_from_checkpoint_start(
 	MONITOR_SET(MONITOR_LSN_CHECKPOINT_AGE,
 		    log_sys->lsn - log_sys->last_checkpoint_lsn);
 
-	log_sys->next_checkpoint_no = checkpoint_no + 1;
-	log_crypt_set_ver_and_key(log_sys->next_checkpoint_no);
+	log_sys->next_checkpoint_no = ++checkpoint_no;
 
 	mutex_enter(&recv_sys->mutex);
 
@@ -3546,18 +3576,14 @@ recv_reset_logs(
 					which we add
 					LOG_BLOCK_HDR_SIZE */
 {
-	log_group_t*	group;
-
 	ut_ad(log_mutex_own());
 
 	log_sys->lsn = ut_uint64_align_up(lsn, OS_FILE_LOG_BLOCK_SIZE);
 
-	group = UT_LIST_GET_FIRST(log_sys->log_groups);
-
-	while (group) {
+	for (log_group_t* group = UT_LIST_GET_FIRST(log_sys->log_groups);
+	     group; group = UT_LIST_GET_NEXT(log_groups, group)) {
 		group->lsn = log_sys->lsn;
 		group->lsn_offset = LOG_FILE_HDR_SIZE;
-		group = UT_LIST_GET_NEXT(log_groups, group);
 	}
 
 	log_sys->buf_next_to_write = 0;
diff --git a/storage/innobase/mysql-test/storage_engine/define_engine.inc b/storage/innobase/mysql-test/storage_engine/define_engine.inc
index 77e384d2351..7d7b0c7407a 100644
--- a/storage/innobase/mysql-test/storage_engine/define_engine.inc
+++ b/storage/innobase/mysql-test/storage_engine/define_engine.inc
@@ -41,9 +41,5 @@ let $default_char_type = CHAR(8);
 # e.g. creation of an additional schema or table, etc.
 # The cleanup part should be defined in cleanup_engine.inc
 
-CALL mtr.add_suppression("InnoDB: Resizing redo log from .* to .* pages, LSN=.*");
-CALL mtr.add_suppression("InnoDB: Starting to delete and rewrite log files.");
-CALL mtr.add_suppression("InnoDB: New log files created, LSN=.*");
-
 --enable_query_log
 --enable_result_log
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 69d094cdc86..849f45f59dc 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -63,7 +63,7 @@ Created 2/16/1996 Heikki Tuuri
 #include "fsp0fsp.h"
 #include "rem0rec.h"
 #include "mtr0mtr.h"
-#include "log0log.h"
+#include "log0crypt.h"
 #include "log0recv.h"
 #include "page0page.h"
 #include "page0cur.h"
@@ -481,6 +481,9 @@ create_log_files(
 
 	/* Create a log checkpoint. */
 	log_mutex_enter();
+	if (log_sys->is_encrypted() && !log_crypt_init()) {
+		return(DB_ERROR);
+	}
 	ut_d(recv_no_log_write = false);
 	recv_reset_logs(lsn);
 	log_mutex_exit();
@@ -536,7 +539,7 @@ create_log_files_rename(
 
 	fil_open_log_and_system_tablespace_files();
 
-	ib::warn() << "New log files created, LSN=" << lsn;
+	ib::info() << "New log files created, LSN=" << lsn;
 
 	return(err);
 }
@@ -1362,17 +1365,24 @@ srv_prepare_to_delete_redo_log_files(
 		flushed_lsn = log_sys->lsn;
 
 		{
-			ib::warn	warning;
+			ib::info	info;
 			if (srv_log_file_size == 0) {
-				warning << "Upgrading redo log: ";
+				info << "Upgrading redo log: ";
+			} else if (n_files != srv_n_log_files
+				   || srv_log_file_size
+				   != srv_log_file_size_requested) {
+				info << "Resizing redo log from "
+				     << n_files << "*" << srv_log_file_size
+				     << " to ";
+			} else if (srv_encrypt_log) {
+				info << "Encrypting redo log: ";
 			} else {
-				warning << "Resizing redo log from "
-					<< n_files << "*"
-					<< srv_log_file_size << " to ";
+				info << "Removing redo log encryption: ";
 			}
-			warning << srv_n_log_files << "*"
-				<< srv_log_file_size_requested
-				<< " pages, LSN=" << flushed_lsn;
+
+			info << srv_n_log_files << "*"
+			     << srv_log_file_size_requested
+			     << " pages; LSN=" << flushed_lsn;
 		}
 
 		/* Flush the old log files. */
@@ -2185,6 +2195,14 @@ files_checked:
 
 		recv_sys->dblwr.pages.clear();
 
+		if (err == DB_SUCCESS && !srv_read_only_mode) {
+			log_mutex_enter();
+			if (log_sys->is_encrypted() && !log_crypt_init()) {
+				err = DB_ERROR;
+			}
+			log_mutex_exit();
+		}
+
 		if (err == DB_SUCCESS) {
 			/* Initialize the change buffer. */
 			err = dict_boot();
@@ -2349,18 +2367,15 @@ files_checked:
 			return(srv_init_abort(err));
 		}
 
-		if (!srv_force_recovery
-		    && !recv_sys->found_corrupt_log
-		    && (srv_log_file_size_requested != srv_log_file_size
-			|| srv_n_log_files_found != srv_n_log_files)) {
-			/* Prepare to replace the redo log files. */
-
-			if (srv_read_only_mode) {
-				ib::error() << "Cannot resize log files"
-					" in read-only mode.";
-				return(srv_init_abort(DB_READ_ONLY));
-			}
-
+		if (srv_force_recovery == SRV_FORCE_NO_LOG_REDO) {
+			/* Completely ignore the redo log. */
+		} else if (srv_read_only_mode) {
+			/* Leave the redo log alone. */
+		} else if (srv_log_file_size_requested == srv_log_file_size
+			   && srv_n_log_files_found == srv_n_log_files
+			   && log_sys->is_encrypted() == srv_encrypt_log) {
+			/* No need to upgrade or resize the redo log. */
+		} else {
 			/* Prepare to delete the old redo log files */
 			flushed_lsn = srv_prepare_to_delete_redo_log_files(i);
 
@@ -2394,7 +2409,7 @@ files_checked:
 			/* Free the old log file space. */
 			log_group_close_all();
 
-			ib::warn() << "Starting to delete and rewrite log"
+			ib::info() << "Starting to delete and rewrite log"
 				" files.";
 
 			srv_log_file_size = srv_log_file_size_requested;