File: base.py

package info (click to toggle)
openstack-trove 1%3A24.0.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 6,976 kB
  • sloc: python: 50,665; sh: 2,866; makefile: 71
file content (256 lines) | stat: -rw-r--r-- 8,903 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
# Copyright 2020 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
import re
import signal
import subprocess

from oslo_config import cfg
from oslo_log import log as logging

CONF = cfg.CONF
LOG = logging.getLogger(__name__)


class BaseRunner(object):
    """Base class for Backup Strategy implementations."""

    # Subclass should provide the commands.
    cmd = ''
    restore_cmd = ''
    prepare_cmd = ''
    backup_log = ''

    encrypt_key = CONF.backup_encryption_key

    def __init__(self, *args, **kwargs):
        self.process = None
        self.pid = None
        self.base_filename = kwargs.get('filename')
        self.storage = kwargs.pop('storage', None)
        self.location = kwargs.pop('location', '')
        self.checksum = kwargs.pop('checksum', '')
        self._gzip = False

        if 'restore_location' not in kwargs:
            kwargs['restore_location'] = self.datadir
        self.restore_location = kwargs['restore_location']
        self.restore_content_length = 0

        self.command = self.cmd % kwargs

        if self.location.endswith('.enc') and not self.encrypt_key:
            raise Exception("Encryption key not provided with an encrypted "
                            "backup.")

        self.restore_command = ''
        # Only decrypt if the object name ends with .enc
        if self.location.endswith('.enc'):
            self.restore_command = self.decrypt_cmd
        self.restore_command = self.restore_cmd % kwargs
        self.prepare_command = self.prepare_cmd % kwargs

    @property
    def filename(self):
        """Subclasses may overwrite this to declare a format (.tar)."""
        return self.base_filename

    @property
    def manifest(self):
        """Target file name."""
        return "%s%s%s" % (self.filename,
                           self.zip_manifest,
                           self.encrypt_manifest)

    @property
    def zip_cmd(self):
        return self._gzip

    @property
    def unzip_cmd(self):
        return self._gzip

    @property
    def zip_manifest(self):
        return '.gz'

    @property
    def encrypt_cmd(self):
        """Encryption command.

        Since Victoria, trove no longer encrypts the backup data for the end
        user. This could be improved by giving users the capability to specify
        password when creating the backups.
        """
        return ""

    @property
    def decrypt_cmd(self):
        """Decryption command.

        Since Victoria, trove no longer encrypts the backup data for the end
        user. This command is only for backward compatibility.
        """
        if self.encrypt_key:
            return ('openssl enc -d -aes-256-cbc -md sha512 -pbkdf2 -iter '
                    '10000 -salt -pass pass:%s | '
                    % self.encrypt_key)
        else:
            return ''

    @property
    def encrypt_manifest(self):
        return '.enc' if self.encrypt_key else ''

    def _run(self):
        """Running backup cmd"""
        LOG.info("Running backup cmd: %s", self.command)
        with open(self.backup_log, "w+") as fp:
            if not self._gzip:
                self.process = subprocess.Popen(self.command.split(),
                                                shell=False,
                                                stdout=subprocess.PIPE,
                                                stderr=fp,
                                                preexec_fn=os.setsid)
            else:
                bkup_process = subprocess.Popen(self.command.split(),
                                                shell=False,
                                                stdout=subprocess.PIPE,
                                                stderr=fp)
                self.process = subprocess.Popen(["gzip"], shell=False,
                                                stdin=bkup_process.stdout,
                                                stdout=subprocess.PIPE,
                                                stderr=fp)
                bkup_process.stdout.close()
            self.pid = self.process.pid

    def __enter__(self):
        """Start up the process."""
        self.pre_backup()
        self._run()
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        """Clean up everything."""
        if getattr(self, 'process', None):
            try:
                # Send a sigterm to the session leader, so that all
                # child processes are killed and cleaned up on terminate
                os.killpg(self.process.pid, signal.SIGTERM)
                self.process.terminate()
            except OSError:
                pass

            if exc_type is not None:
                return False

            if not self.check_process():
                with open(self.backup_log, "r") as fp:
                    err = fp.read()
                    if err:
                        raise Exception(err)
                raise Exception()

        self.post_backup()

        return True

    def read(self, chunk_size):
        return self.process.stdout.read(chunk_size)

    def get_metadata(self):
        """Hook for subclasses to get metadata from the backup."""
        return {}

    def check_process(self):
        """Hook for subclasses to check process for errors."""
        return True

    def check_restore_process(self):
        """Hook for subclasses to check the restore process for errors."""
        return True

    def pre_backup(self):
        """Hook for subclasses to run commands before backup."""
        pass

    def post_backup(self):
        """Hook for subclasses to run commands after backup."""
        pass

    def pre_restore(self):
        """Hook that is called before the restore command."""
        pass

    def post_restore(self):
        """Hook that is called after the restore command."""
        pass

    def unpack(self, location, checksum, command):
        stream = self.storage.load(location, checksum)

        LOG.info('Running restore from stream, command: %s', command)
        content_length = 0
        if not re.match(r'.*.gz', location) or not self._gzip:
            LOG.info('gz processor without gz file or with gzip disabled')
            self.process = subprocess.Popen(command.split(), shell=False,
                                            stdin=subprocess.PIPE,
                                            stdout=subprocess.PIPE,
                                            stderr=subprocess.PIPE)
            for chunk in stream:
                self.process.stdin.write(chunk)  # write data to mbstream
                content_length += len(chunk)
            stdout, stderr = self.process.communicate()
        else:
            LOG.info('gz processor with gz file')
            gunzip = subprocess.Popen(["gzip", "-d", "-c"], shell=False,
                                      stdin=subprocess.PIPE,
                                      stdout=subprocess.PIPE,
                                      stderr=subprocess.PIPE)
            self.process = subprocess.Popen(command.split(), shell=False,
                                            stdin=gunzip.stdout,
                                            stdout=subprocess.PIPE,
                                            stderr=subprocess.PIPE)
            for chunk in stream:
                gunzip.stdin.write(chunk)  # write data to mbstream
                content_length += len(chunk)
            gunzip.stdin.close()
            gunzip.stdout.close()
            stdout, stderr = self.process.communicate()
        stdout_str = stdout.decode()
        stderr_str = stderr.decode()
        LOG.info("command: %s, stdout: %s, stderr: %s",
                 command, stdout_str, stderr_str)

        if not self.check_restore_process():
            LOG.info('self.check_restore_process() False')
            if stderr_str:
                raise Exception(stderr_str)
            raise Exception()

        return content_length

    def run_restore(self):
        return self.unpack(self.location, self.checksum, self.restore_command)

    def restore(self):
        """Restore backup to data directory.

        :returns Restored data size.
        """
        self.pre_restore()
        content_length = self.run_restore()
        self.post_restore()
        return content_length