Add periodic task to remove postgres archived wal files

* Added a periodic task for postgresql datastore to clean up the
  archived WAL files.
* Added a check when creating incremental backups for postgresql.
* A new container image ``openstacktrove/db-backup-postgresql:1.1.2``
  is uploaded to docker hub.

Story: 2009066
Task: 42871
Change-Id: I235e2abf8c0405e143ded6fb48017d596b8b41a1

11 files changed, 268 insertions, 34 deletions

diff --git a/trove/common/cfg.py b/trove/common/cfg.py
index 49ea8b82..eadb8514 100644
--- a/trove/common/cfg.py
+++ b/trove/common/cfg.py
@@ -1091,13 +1091,18 @@ postgresql_group = cfg.OptGroup(
     'postgresql', title='PostgreSQL options',
     help="Oslo option group for the PostgreSQL datastore.")
 postgresql_opts = [
+    cfg.BoolOpt(
+        'enable_clean_wal_archives',
+        default=True,
+        help='Enable the periodic job to clean up WAL archive folder.'
+    ),
     cfg.StrOpt(
         'docker_image', default='postgres',
         help='Database docker image.'
     ),
     cfg.StrOpt(
         'backup_docker_image',
-        default='openstacktrove/db-backup-postgresql:1.1.1',
+        default='openstacktrove/db-backup-postgresql:1.1.2',
         help='The docker image used for backup and restore.'
     ),
     cfg.BoolOpt('icmp', default=False,
@@ -1131,7 +1136,11 @@ postgresql_opts = [
     cfg.StrOpt('wal_archive_location', default='/mnt/wal_archive',
                help="Filesystem path storing WAL archive files when "
                     "WAL-shipping based backups or replication "
-                    "is enabled."),
+                    "is enabled.",
+               deprecated_for_removal=True,
+               deprecated_reason='Option is not used any more, will be '
+                                 'removed in future release.'
+               ),
     cfg.BoolOpt('root_on_create', default=False,
                 help='Enable the automatic creation of the root user for the '
                 'service during instance-create. The generated password for '
@@ -1154,7 +1163,7 @@ postgresql_opts = [
                     "statement logging.",
                deprecated_for_removal=True,
                deprecated_reason='Will be replaced by configuration group '
-               'option: log_min_duration_statement'),
+                                 'option: log_min_duration_statement'),
     cfg.IntOpt('default_password_length', default=36,
                help='Character length of generated passwords.',
                deprecated_name='default_password_length',
diff --git a/trove/guestagent/common/operating_system.py b/trove/guestagent/common/operating_system.py
index 433ed603..6ca9c36a 100644
--- a/trove/guestagent/common/operating_system.py
+++ b/trove/guestagent/common/operating_system.py
@@ -17,6 +17,7 @@ from functools import reduce
 import inspect
 import operator
 import os
+from pathlib import Path
 import pwd
 import re
 import stat
@@ -904,3 +905,19 @@ def remove_dir_contents(folder):
     """
     path = os.path.join(folder, '*')
     execute_shell_cmd(f'rm -rf {path}', [], shell=True, as_root=True)
+
+
+def get_dir_size(path):
+    """Get the directory size in bytes."""
+    root_directory = Path(path)
+    return sum(f.stat().st_size for f in root_directory.glob('**/*')
+               if f.is_file())
+
+
+def get_filesystem_size(path):
+    """Get size(bytes) of a mounted filesystem the given path locates.
+
+    path is the pathname of any file within the mounted filesystem.
+    """
+    ret = os.statvfs(path)
+    return ret.f_blocks * ret.f_frsize
diff --git a/trove/guestagent/datastore/manager.py b/trove/guestagent/datastore/manager.py
index 2bdd3c48..097ccc99 100644
--- a/trove/guestagent/datastore/manager.py
+++ b/trove/guestagent/datastore/manager.py
@@ -64,7 +64,15 @@ class Manager(periodic_task.PeriodicTasks):
 
     MODULE_APPLY_TO_ALL = module_manager.ModuleManager.MODULE_APPLY_TO_ALL
 
-    docker_client = docker.from_env()
+    _docker_client = None
+
+    @property
+    def docker_client(self):
+        if self._docker_client:
+            return self._docker_client
+
+        self._docker_client = docker.from_env()
+        return self._docker_client
 
     def __init__(self, manager_name):
         super(Manager, self).__init__(CONF)
diff --git a/trove/guestagent/datastore/mysql_common/service.py b/trove/guestagent/datastore/mysql_common/service.py
index a110d222..483e9a1c 100644
--- a/trove/guestagent/datastore/mysql_common/service.py
+++ b/trove/guestagent/datastore/mysql_common/service.py
@@ -423,11 +423,19 @@ class BaseMySqlAdmin(object, metaclass=abc.ABCMeta):
 
 
 class BaseMySqlApp(service.BaseDbApp):
-    configuration_manager = ConfigurationManager(
-        MYSQL_CONFIG, CONF.database_service_uid, CONF.database_service_uid,
-        service.BaseDbApp.CFG_CODEC, requires_root=True,
-        override_strategy=ImportOverrideStrategy(CNF_INCLUDE_DIR, CNF_EXT)
-    )
+    _configuration_manager = None
+
+    @property
+    def configuration_manager(self):
+        if self._configuration_manager:
+            return self._configuration_manager
+
+        self._configuration_manager = ConfigurationManager(
+            MYSQL_CONFIG, CONF.database_service_uid, CONF.database_service_uid,
+            service.BaseDbApp.CFG_CODEC, requires_root=True,
+            override_strategy=ImportOverrideStrategy(CNF_INCLUDE_DIR, CNF_EXT)
+        )
+        return self._configuration_manager
 
     def get_engine(self):
         """Create the default engine with the updated admin user.
@@ -460,14 +468,12 @@ class BaseMySqlApp(service.BaseDbApp):
         with mysql_util.SqlClient(self.get_engine()) as client:
             return client.execute(sql_statement)
 
-    @classmethod
-    def get_data_dir(cls):
-        return cls.configuration_manager.get_value(
+    def get_data_dir(self):
+        return self.configuration_manager.get_value(
             'datadir', section=MySQLConfParser.SERVER_CONF_SECTION)
 
-    @classmethod
-    def set_data_dir(cls, value):
-        cls.configuration_manager.apply_system_override(
+    def set_data_dir(self, value):
+        self.configuration_manager.apply_system_override(
             {MySQLConfParser.SERVER_CONF_SECTION: {'datadir': value}})
 
     def _create_admin_user(self, client, password):
diff --git a/trove/guestagent/datastore/postgres/manager.py b/trove/guestagent/datastore/postgres/manager.py
index b233d5b3..d6a043b0 100644
--- a/trove/guestagent/datastore/postgres/manager.py
+++ b/trove/guestagent/datastore/postgres/manager.py
@@ -12,8 +12,10 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 import os
+import re
 
 from oslo_log import log as logging
+from oslo_service import periodic_task
 
 from trove.common import cfg
 from trove.common import exception
@@ -40,6 +42,82 @@ class PostgresManager(manager.Manager):
     def configuration_manager(self):
         return self.app.configuration_manager
 
+    def _check_wal_archive_size(self, archive_path, data_path):
+        """Check wal archive folder size.
+
+        Return True if the size is greater than half of the data volume size.
+        """
+        archive_size = operating_system.get_dir_size(archive_path)
+        data_volume_size = operating_system.get_filesystem_size(data_path)
+
+        if archive_size > (data_volume_size / 2):
+            LOG.info(f"The size({archive_size}) of wal archive folder is "
+                     f"greater than half of the data volume "
+                     f"size({data_volume_size})")
+            return True
+
+        return False
+
+    def _remove_older_files(self, archive_path, files, cur_file):
+        """Remove files older than cur_file.
+
+        :param archive_path: The archive folder
+        :param files: List of the ordered file names.
+        :param cur_file: The compared file name.
+        """
+        cur_seq = os.path.basename(cur_file).split('.')[0]
+        wal_re = re.compile(r"^([0-9A-F]{24}).*")
+
+        for wal_file in files:
+            m = wal_re.search(wal_file)
+            if m and m.group(1) < cur_seq:
+                file_path = os.path.join(archive_path, wal_file)
+                LOG.info(f"Removing wal file {file_path}")
+                operating_system.remove(
+                    path=file_path, force=True, recursive=False, as_root=True)
+
+    def _remove_wals(self, archive_path, force=False):
+        """Remove wal files.
+
+        If force=True, do not consider backup.
+        """
+        files = os.listdir(archive_path)
+        files = sorted(files, reverse=True)
+        wal_files = []
+
+        if not force:
+            # Get latest backup file
+            backup_re = re.compile("[0-9A-F]{24}.*.backup")
+            wal_files = [wal_file for wal_file in files
+                         if backup_re.search(wal_file)]
+
+        # If there is no backup file or force=True, remove all except the
+        # latest one, otherwise, remove all the files older than the backup
+        # file
+        wal_files = wal_files or files
+        self._remove_older_files(archive_path, files, wal_files[0])
+
+    def _clean_wals(self, archive_path, data_path, force=False):
+        if self._check_wal_archive_size(archive_path, data_path):
+            self._remove_wals(archive_path, force)
+
+            # check again with force=True
+            self._clean_wals(archive_path, data_path, force=True)
+
+    @periodic_task.periodic_task(
+        enabled=CONF.postgresql.enable_clean_wal_archives,
+        spacing=180)
+    def clean_wal_archives(self, context):
+        """Clean up the wal archives to free up disk space."""
+        archive_path = service.WAL_ARCHIVE_DIR
+        data_path = cfg.get_configuration_property('mount_point')
+
+        if not operating_system.exists(archive_path, is_directory=True,
+                                       as_root=True):
+            return
+
+        self._clean_wals(archive_path, data_path)
+
     def do_prepare(self, context, packages, databases, memory_mb, users,
                    device_path, mount_point, backup_info,
                    config_contents, root_password, overrides,
diff --git a/trove/guestagent/datastore/postgres/service.py b/trove/guestagent/datastore/postgres/service.py
index a2cc9865..863c179a 100644
--- a/trove/guestagent/datastore/postgres/service.py
+++ b/trove/guestagent/datastore/postgres/service.py
@@ -74,18 +74,26 @@ class PgSqlAppStatus(service.BaseDbStatus):
 
 
 class PgSqlApp(service.BaseDbApp):
-    configuration_manager = configuration.ConfigurationManager(
-        CONFIG_FILE,
-        CONF.database_service_uid,
-        CONF.database_service_uid,
-        stream_codecs.KeyValueCodec(
-            value_quoting=True,
-            bool_case=stream_codecs.KeyValueCodec.BOOL_LOWER,
-            big_ints=True),
-        requires_root=True,
-        override_strategy=configuration.ImportOverrideStrategy(
-            CNF_INCLUDE_DIR, CNF_EXT)
-    )
+    _configuration_manager = None
+
+    @property
+    def configuration_manager(self):
+        if self._configuration_manager:
+            return self._configuration_manager
+
+        self._configuration_manager = configuration.ConfigurationManager(
+            CONFIG_FILE,
+            CONF.database_service_uid,
+            CONF.database_service_uid,
+            stream_codecs.KeyValueCodec(
+                value_quoting=True,
+                bool_case=stream_codecs.KeyValueCodec.BOOL_LOWER,
+                big_ints=True),
+            requires_root=True,
+            override_strategy=configuration.ImportOverrideStrategy(
+                CNF_INCLUDE_DIR, CNF_EXT)
+        )
+        return self._configuration_manager
 
     def __init__(self, status, docker_client):
         super(PgSqlApp, self).__init__(status, docker_client)
@@ -96,13 +104,11 @@ class PgSqlApp(service.BaseDbApp):
         self.datadir = f"{mount_point}/data/pgdata"
         self.adm = PgSqlAdmin(SUPER_USER_NAME)
 
-    @classmethod
-    def get_data_dir(cls):
-        return cls.configuration_manager.get_value('data_directory')
+    def get_data_dir(self):
+        return self.configuration_manager.get_value('data_directory')
 
-    @classmethod
-    def set_data_dir(cls, value):
-        cls.configuration_manager.apply_system_override(
+    def set_data_dir(self, value):
+        self.configuration_manager.apply_system_override(
             {'data_directory': value})
 
     def reload(self):
diff --git a/trove/guestagent/datastore/service.py b/trove/guestagent/datastore/service.py
index a10f821d..3de1afad 100644
--- a/trove/guestagent/datastore/service.py
+++ b/trove/guestagent/datastore/service.py
@@ -509,7 +509,16 @@ class BaseDbApp(object):
                     'success': False,
                     'state': BackupState.FAILED,
                 })
-                raise Exception(msg)
+
+                # The exception message is visible to the user
+                user_msg = msg
+                ex_regex = re.compile(r'.+Exception: (.+)')
+                for line in output[-5:-1]:
+                    m = ex_regex.search(line)
+                    if m:
+                        user_msg = m.group(1)
+                        break
+                raise Exception(user_msg)
         except Exception as err:
             LOG.error("Failed to create backup %s", backup_id)
             backup_state.update({
diff --git a/trove/tests/unittests/guestagent/__init__.py b/trove/tests/unittests/guestagent/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/trove/tests/unittests/guestagent/__init__.py
diff --git a/trove/tests/unittests/guestagent/datastore/__init__.py b/trove/tests/unittests/guestagent/datastore/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/trove/tests/unittests/guestagent/datastore/__init__.py
diff --git a/trove/tests/unittests/guestagent/datastore/postgres/__init__.py b/trove/tests/unittests/guestagent/datastore/postgres/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/trove/tests/unittests/guestagent/datastore/postgres/__init__.py
diff --git a/trove/tests/unittests/guestagent/datastore/postgres/test_manager.py b/trove/tests/unittests/guestagent/datastore/postgres/test_manager.py
new file mode 100644
index 00000000..343eaee9
--- /dev/null
+++ b/trove/tests/unittests/guestagent/datastore/postgres/test_manager.py
@@ -0,0 +1,101 @@
+# Copyright 2021 Catalyst Cloud
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+import os
+from unittest import mock
+
+from trove.guestagent.datastore.postgres import manager
+from trove.guestagent.datastore.postgres import service
+from trove.tests.unittests import trove_testtools
+
+
+class TestPostgresManager(trove_testtools.TestCase):
+    def setUp(self):
+        super(TestPostgresManager, self).setUp()
+        manager.PostgresManager._docker_client = mock.MagicMock()
+        self.patch_datastore_manager('postgresql')
+
+    @mock.patch('trove.guestagent.common.operating_system.remove')
+    @mock.patch('os.listdir')
+    @mock.patch('trove.guestagent.common.operating_system.get_filesystem_size')
+    @mock.patch('trove.guestagent.common.operating_system.get_dir_size')
+    @mock.patch('trove.guestagent.common.operating_system.exists')
+    def test_clean_wal_archives(self, mock_exists, mock_get_dir_size,
+                                mock_get_filesystem_size, mock_listdir,
+                                mock_remove):
+        mock_exists.return_value = True
+        mock_get_dir_size.side_effect = [6, 1]
+        mock_get_filesystem_size.return_value = 10
+        mock_listdir.return_value = [
+            '0000000100000002000000D4',
+            '00000001000000000000008D',
+            '0000000100000000000000A7.00000028.backup',
+            '0000000100000000000000A7',
+            '0000000100000002000000E7'
+        ]
+
+        psql_manager = manager.PostgresManager()
+        psql_manager.clean_wal_archives(mock.ANY)
+
+        self.assertEqual(1, mock_remove.call_count)
+
+        archive_path = service.WAL_ARCHIVE_DIR
+        expected_calls = [
+            mock.call(
+                path=os.path.join(archive_path, '00000001000000000000008D'),
+                force=True, recursive=False,
+                as_root=True),
+        ]
+        self.assertEqual(expected_calls, mock_remove.call_args_list)
+
+    @mock.patch('trove.guestagent.common.operating_system.remove')
+    @mock.patch('os.listdir')
+    @mock.patch('trove.guestagent.common.operating_system.get_filesystem_size')
+    @mock.patch('trove.guestagent.common.operating_system.get_dir_size')
+    @mock.patch('trove.guestagent.common.operating_system.exists')
+    def test_clean_wal_archives_no_backups(self, mock_exists,
+                                           mock_get_dir_size,
+                                           mock_get_filesystem_size,
+                                           mock_listdir,
+                                           mock_remove):
+        mock_exists.return_value = True
+        mock_get_dir_size.side_effect = [6, 1]
+        mock_get_filesystem_size.return_value = 10
+        mock_listdir.return_value = [
+            '0000000100000002000000D4',
+            '00000001000000000000008D',
+            '0000000100000000000000A7',
+            '0000000100000002000000E7'
+        ]
+
+        psql_manager = manager.PostgresManager()
+        psql_manager.clean_wal_archives(mock.ANY)
+
+        self.assertEqual(3, mock_remove.call_count)
+
+        archive_path = service.WAL_ARCHIVE_DIR
+        expected_calls = [
+            mock.call(
+                path=os.path.join(archive_path, '0000000100000002000000D4'),
+                force=True, recursive=False,
+                as_root=True),
+            mock.call(
+                path=os.path.join(archive_path, '0000000100000000000000A7'),
+                force=True, recursive=False,
+                as_root=True),
+            mock.call(
+                path=os.path.join(archive_path, '00000001000000000000008D'),
+                force=True, recursive=False,
+                as_root=True),
+        ]
+        self.assertEqual(expected_calls, mock_remove.call_args_list)