confluent/confluent_server/bin/confluentdbutil

#!/usr/bin/python3
# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2017,2024 Lenovo
#
# 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 getpass
import optparse
import sys
import os
path = os.path.dirname(os.path.realpath(__file__))
path = os.path.realpath(os.path.join(path, '..', 'lib', 'python'))
if path.startswith('/opt'):
    # if installed into system path, do not muck with things
    sys.path.append(path)
import confluent.config.configmanager as cfm
import confluent.config.conf as conf
import confluent.main as main

argparser = optparse.OptionParser(
    usage="Usage: %prog [options] [dump|restore|merge] [path]")
argparser.add_option('-p', '--password',
                     help='Password to use to protect/unlock a protected dump')
argparser.add_option('-i', '--interactivepassword', help='Prompt for password',
                     action='store_true')
argparser.add_option('-r', '--redact', action='store_true',
                     help='Redact potentially sensitive data rather than store')
argparser.add_option('-u', '--unprotected', action='store_true',
                      help='Specify that no password should be used to protect'
                           ' the key information.  Fields will be encrypted, '
                           'but keys.json will contain unencrypted decryption'
                           ' keys that may be used to read the dump')
argparser.add_option('-s', '--skipkeys', action='store_true',
                      help='This specifies to dump the encrypted data without '
                           'dumping the keys needed to decrypt it.  This is '
                           'suitable for an automated incremental backup, '
                           'where an earlier password protected dump has a '
                           'protected keys.json file, and only the protected '
                           'data is needed.  keys do not change and as such '
                           'they do not require incremental backup')
(options, args) = argparser.parse_args()
if len(args) != 2 or args[0] not in ('dump', 'restore', 'merge'):
    argparser.print_help()
    sys.exit(1)
dumpdir = args[1]


if args[0] in ('restore', 'merge'):
    pid = main.is_running()
    if pid is not None:
        print("Confluent is running, must shut down to restore db")
        sys.exit(1)
    stinf = os.stat('/etc/confluent')
    owner = stinf.st_uid
    group = stinf.st_gid
    password = options.password
    if options.interactivepassword:
        password = getpass.getpass('Enter password to restore backup: ')
    try:
        stateless = args[0] == 'restore'
        cfm.init(stateless)
        cfm.statelessmode = stateless
        skipped = {'nodes': [], 'nodegroups': []}
        cfm.restore_db_from_directory(
            dumpdir, password,
            merge="skip" if args[0] == 'merge' else False, skipped=skipped)
        if skipped['nodes']:
            skippedn = ','.join(skipped['nodes'])
            print('The following nodes were skipped during merge: '
                  '{}'.format(skippedn))
        if skipped['nodegroups']:
            skippedn = ','.join(skipped['nodegroups'])
            print('The following node groups were skipped during merge: '
                  '{}'.format(skippedn))

        cfm.statelessmode = False
        cfm.ConfigManager.wait_for_sync(True)
        if owner != 0:
            for targdir in os.walk('/etc/confluent'):
                os.chown(targdir[0], owner, group)
                for f in targdir[2]:
                    os.chown(os.path.join(targdir[0], f), owner, group)
    except Exception as e:
        print(str(e))
        sys.exit(1)
elif args[0] == 'dump':
    password = options.password
    if not password and options.interactivepassword:
        passcfm = None
        while passcfm is None or password != passcfm:
            password = getpass.getpass(
                'Enter password to protect the backup: ')
            passcfm = getpass.getpass('Confirm password to protect the backup: ')
    if password is None and not (options.unprotected or options.redact
                                         or options.skipkeys):
        print("Must indicate a password to protect or -u to opt opt of "
              "secure value protection or -r to redact sensitive information, "
              "or -s to do encrypted backup that requires keys.json from "
              "another backup to restore.")
        sys.exit(1)
    os.umask(0o77)
    main._initsecurity(conf.get_config())
    if not os.path.exists(dumpdir):
        os.makedirs(dumpdir)
    cfm.dump_db_to_directory(dumpdir, password, options.redact,
                             options.skipkeys)