stack/scripts/multitenant_migrate.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
#
#
# To migrate multiple tenants within one cluster.
#
# STEP 1 - SETUP TENANT ONE TOMCAT RUNNING 2.1 NOT IN SERVICE AND INIT MIGRATION
#
#   Login to the Tomcat instance and run this command, specifying both superuser and tenant organization:
#
#       python multitenant_migrate.py --org <org1name> --super <user>:<pass> --init
#
#   This command will setup the database, setup the migration system and update index mappings:
#   - /system/database/setup
#   - /system/migrate/run/migration-system
#   - /system/migrate/run/index_mapping_migration
#
#   Then it will migrate appinfos, re-index the management app and then for each of the specified org's apps
#   it will de-dup connections and re-index the app.
#
#   Write down the 'Re-index start' timestamp when this is finished.
#
# STEP 2 - PUT TENANT ONE TOMCATS IN SERVICE AND DO DELTA MIGRATION
#
#   On the same Tomcat instance and run this command with the --date timestamp you noted in the previous step:
#
#       python multitenant_migrate.py --org <org1name> --super <user>:<pass> --date <timestamp>
#
#   Then it will migrate appinfos, re-index the management app and then for each of the specified org's apps
#   it will de-dup connections and re-index the app with a start-date specified so only data modified since
#   STEP 1 will be re-indexed.
#
# STEP 3 - SETUP TENANT TWO TOMCAT RUNNING 2.1 NOT IN SERVICE
#
#   Login to the Tomcat instance and run this command, specifying both superuser and tenant organization:
#
#       python multitenant_migrate.py --org <org2name> --super <user>:<pass>
#
#   This command will migrate appinfos, re-index the management app and then for each of the specified org's apps
#   it will de-dup connections and re-index the app.
#
#   Write down the 'Re-index start' timestamp when this is finished.

# STEP 4 - PUT TENANT TWO TOMCATS IN SERVICE AND DO DELTA MIGRATION
#
#   On the same Tomcat instance and run this command with the --date timestamp you noted in the previous step:
#
#       python multitenant_migrate.py --org <org2name> --super <user>:<pass> --date <timestamp>
#
#   Then it will migrate appinfos, re-index the management app and then for each of the specified org's apps
#   it will de-dup connections and re-index the app with a start-date specified so only data modified since
#   STEP 1 will be re-indexed.
#
# STEP 5 - FULL DATA MIGRATION (migrates entity data to new format)
#
#   Login to any Tomcat instance in the cluster and run this command:
#
#       python migrate_entity_data.py --super <user>:<pass> --full
#
#   This command will run the full data migration.
#

import sys
import logging
from logging.handlers import RotatingFileHandler
import argparse
import time
import requests
import json


# Version expected in status response post-migration for entity and app-info data
TARGET_APPINFO_VERSION=2
TARGET_ENTITY_DATA_VERSION=2
TARGET_CORE_DATA_VERSION=2
TARGET_MIGRATION_SYSTEM_VERSION = 1
TARGET_INDEX_MAPPING_VERSION = 2

# Set an interval (in seconds) for checking if re-index and/or migration has finished
STATUS_INTERVAL_SECONDS = 2

# Set plugin names
PLUGIN_MIGRATION_SYSTEM = 'migration-system'
PLUGIN_APPINFO = 'appinfo-migration'
PLUGIN_ENTITYDATA = 'collections-entity-data'
PLUGIN_INDEX_MAPPING = 'index_mapping_migration'
PLUGIN_CORE_DATA = 'core-data'

MANAGEMENT_APP_ID = 'b6768a08-b5d5-11e3-a495-11ddb1de66c8'



def parse_args():
    parser = argparse.ArgumentParser(description='Usergrid Migration Tool')

    parser.add_argument('--endpoint',
                        help='The endpoint to use for making API requests.',
                        type=str,
                        default='http://localhost:8080')

    parser.add_argument('--super',
                        help='Superuser username and creds <user:pass>',
                        type=str,
                        required=True)

    parser.add_argument('--init',
                        help='Init system and start first migration.',
                        action='store_true',
                        default=False)

    parser.add_argument('--org',
                        help='Name of organization on which to run migration.',
                        type=str,
                        required=False)

    parser.add_argument('--date',
                        help='A date from which to start the migration',
                        type=str)

    parser.add_argument('--full',
                        help='Run full data migration (last step in cluster migration).',
                        action='store_true',
                        default=False)

    my_args = parser.parse_args(sys.argv[1:])

    arg_vars = vars(my_args)

    creds = arg_vars['super'].split(':')
    if len(creds) != 2:
        print('Superuser credentials not properly specified.  Must be "-u <user:pass>". Exiting...')
        exit_on_error()
    else:
        arg_vars['superuser'] = creds[0]
        arg_vars['superpass'] = creds[1]

    return arg_vars



class Migrate:
    def __init__(self):
        self.args = parse_args()
        self.start_date = self.args['date']
        self.endpoint = self.args['endpoint']
        self.metrics = {'reindex_start': '',
                        'reindex_end': '',
                        'appinfo_migration_start': '',
                        'appinfo_migration_end': '',
                        'full_data_migration_start': '',
                        'full_data_migration_end': ''}
        self.logger = init_logging(self.__class__.__name__)
        self.super_user = self.args['superuser']
        self.super_pass = self.args['superpass']
        self.org = self.args['org']
        self.init = self.args['init']
        self.full = self.args['full']

    def run(self):
        self.logger.info('Initializing...')

        if not self.is_endpoint_available():
            exit_on_error('Endpoint is not available, aborting')

        if self.start_date is not None:
            self.logger.info("Date Provided.  Re-index will run from date=[%s]", self.start_date)

        try:

            if self.full:

                # Do full data migration and exit

                self.start_fulldata_migration()

                self.metrics['full_data_migration_start'] = get_current_time()
                self.logger.info("Full Data Migration Started")
                is_migrated = False
                while not is_migrated:
                    time.sleep(STATUS_INTERVAL_SECONDS)
                    is_migrated = self.is_data_migrated()
                    if is_migrated:
                        break

                self.metrics['full_data_migration_end'] = get_current_time()
                self.logger.info("Full Data Migration completed")

                self.log_metrics()
                self.logger.info("Finished...")

                return

            if self.init:

                # Init the migration system as this is the first migration done on the cluster
                self.run_database_setup()

                migration_system_updated = self.is_migration_system_updated()

                if not migration_system_updated:
                    self.logger.info('Migration system needs to be updated.  Updating migration system..')
                    self.start_migration_system_update()
                    while not migration_system_updated:
                        time.sleep(STATUS_INTERVAL_SECONDS)
                        migration_system_updated = self.is_migration_system_updated()
                        if migration_system_updated:
                            break

                index_mapping_updated = self.is_index_mapping_updated()

                if not index_mapping_updated:
                    self.logger.info('Index Mapping needs to be updated.  Updating index mapping..')
                    self.start_index_mapping_migration()
                    while not index_mapping_updated:
                        time.sleep(STATUS_INTERVAL_SECONDS)
                        index_mapping_updated = self.is_index_mapping_updated()
                        if index_mapping_updated:
                            break

            # Migrate app info
            if self.is_appinfo_migrated():
                self.logger.info('AppInfo already migrated. Resetting version for re-migration.')
                self.reset_appinfo_migration()
                time.sleep(STATUS_INTERVAL_SECONDS)

            self.start_appinfo_migration()
            self.logger.info('AppInfo Migration Started.')
            self.metrics['appinfo_migration_start'] = get_current_time()

            is_appinfo_migrated = False
            while not is_appinfo_migrated:
                is_appinfo_migrated = self.is_appinfo_migrated()
                time.sleep(STATUS_INTERVAL_SECONDS)
                if is_appinfo_migrated:
                    self.metrics['appinfo_migration_end'] = get_current_time()
                    break
            self.logger.info('AppInfo Migration Ended.')

            # De-dup management app
            job = self.start_dedup(MANAGEMENT_APP_ID)
            self.logger.info('Started management dedup.  App=[%s], Job=[%s]', MANAGEMENT_APP_ID, job)
            is_running = True
            while is_running:
                time.sleep(STATUS_INTERVAL_SECONDS)
                is_running = self.is_dedup_running(job)
                if not is_running:
                    break

            self.logger.info("Finished dedup. App=[%s], Job=[%s]", MANAGEMENT_APP_ID, job)
            self.metrics['dedup_end_' + MANAGEMENT_APP_ID] = get_current_time()

            # Reindex management app

            job = self.start_app_reindex(MANAGEMENT_APP_ID)
            self.metrics['reindex_start'] = get_current_time()
            self.logger.info('Started management Re-index.  Job=[%s]', job)
            is_running = True
            while is_running:
                time.sleep(STATUS_INTERVAL_SECONDS)
                is_running = self.is_reindex_running(job)
                if not is_running:
                    break

            self.logger.info("Finished management Re-index. Job=[%s]", job)
            self.metrics['reindex_end'] = get_current_time()

            # Dedup and re-index all of organization's apps

            app_ids = self.get_app_ids()
            for app_id in app_ids:

                # De-dup app
                job = self.start_dedup(app_id)
                self.logger.info('Started dedup.  App=[%s], Job=[%s]', app_id, job)
                is_running = True
                while is_running:
                    time.sleep(STATUS_INTERVAL_SECONDS)
                    is_running = self.is_dedup_running(job)
                    if not is_running:
                        break

                self.logger.info("Finished dedup. App=[%s], Job=[%s]", app_id, job)
                self.metrics['dedup_end_' + app_id] = get_current_time()

                # Re-index app
                job = self.start_app_reindex(app_id)
                self.metrics['reindex_start_' + app_id] = get_current_time()
                self.logger.info('Started Re-index.  App=[%s], Job=[%s]', app_id, job)
                is_running = True
                while is_running:
                    time.sleep(STATUS_INTERVAL_SECONDS)
                    is_running = self.is_reindex_running(job)
                    if not is_running:
                        break

                self.logger.info("Finished Re-index. App=[%s], Job=[%s]", app_id, job)
                self.metrics['reindex_end_' + app_id] = get_current_time()

            self.log_metrics()
            self.logger.info("Finished...")

        except KeyboardInterrupt:
            self.log_metrics()
            self.logger.error('Keyboard interrupted migration. Please run again to ensure the migration finished.')

    def get_database_setup_url(self):
        url = self.endpoint + '/system/database/setup'
        return url

    def get_migration_url(self):
        url = self.endpoint + '/system/migrate/run'
        return url

    def get_reset_migration_url(self):
        url = self.endpoint + '/system/migrate/set'
        return url

    def get_migration_status_url(self):
        url = self.endpoint + '/system/migrate/status'
        return url

    def get_dedup_url(self):
        url = self.endpoint + '/system/connection/dedup'
        return url

    def get_reindex_url(self):
        url = self.endpoint + '/system/index/rebuild'
        return url

    def get_management_reindex_url(self):
          url = self.get_reindex_url() + "/management"
          return url

    def start_core_data_migration(self):
           try:
               r = requests.put(url=self.get_migration_url(), auth=(self.super_user, self.super_pass))
               response = r.json()
               return response
           except requests.exceptions.RequestException as e:
               self.logger.error('Failed to start migration, %s', e)
               exit_on_error(str(e))

    def start_fulldata_migration(self):
        try:
            r = requests.put(url=self.get_migration_url(), auth=(self.super_user, self.super_pass))
            response = r.json()
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to start migration, %s', e)
            exit_on_error(str(e))

    def start_migration_system_update(self):
        try:
            # TODO fix this URL
            migrateUrl = self.get_migration_url() + '/' + PLUGIN_MIGRATION_SYSTEM
            r = requests.put(url=migrateUrl, auth=(self.super_user, self.super_pass))
            response = r.json()
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to start migration, %s', e)
            exit_on_error(str(e))

    def run_database_setup(self):
        try:
            setupUrl = self.get_database_setup_url()
            r = requests.put(url=setupUrl, auth=(self.super_user, self.super_pass))
            if r.status_code != 200:
                exit_on_error('Database Setup Failed')

        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to run database setup, %s', e)
            exit_on_error(str(e))

    def start_index_mapping_migration(self):
        try:
            migrateUrl = self.get_migration_url() + '/' + PLUGIN_INDEX_MAPPING
            r = requests.put(url=migrateUrl, auth=(self.super_user, self.super_pass))
            response = r.json()
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to start migration, %s', e)
            exit_on_error(str(e))

    def start_appinfo_migration(self):
        try:
            migrateUrl = self.get_migration_url() + '/' + PLUGIN_APPINFO
            r = requests.put(url=migrateUrl, auth=(self.super_user, self.super_pass))
            response = r.json()
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to start migration, %s', e)
            exit_on_error(str(e))

    def reset_data_migration(self):
        version = TARGET_ENTITY_DATA_VERSION - 1
        body = json.dumps({PLUGIN_ENTITYDATA: version, PLUGIN_APPINFO: version})
        try:
            r = requests.put(url=self.get_reset_migration_url(), data=body, auth=(self.super_user, self.super_pass))
            response = r.json()
            self.logger.info('Resetting data migration versions to %s=[%s] '
                             'and %s=[%s]', PLUGIN_ENTITYDATA, version, PLUGIN_APPINFO, version)
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to reset full data migration versions, %s', e)
            exit_on_error(str(e))

    def reset_appinfo_migration(self):
        version = TARGET_APPINFO_VERSION - 1
        body = json.dumps({PLUGIN_APPINFO: version})
        try:
            r = requests.put(url=self.get_reset_migration_url(), data=body, auth=(self.super_user, self.super_pass))
            response = r.json()
            self.logger.info('Resetting appinfo migration versions to %s=[%s]', PLUGIN_APPINFO, version)
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to reset appinfo migration version, %s', e)
            exit_on_error(str(e))

    def is_data_migrated(self):
        status = self.check_data_migration_status()
        if status is not None:
            entity_version = status['data'][PLUGIN_ENTITYDATA]
            appinfo_version = status['data'][PLUGIN_APPINFO]
            core_data_version = status['data'][PLUGIN_CORE_DATA]

            if entity_version == TARGET_ENTITY_DATA_VERSION and appinfo_version == TARGET_APPINFO_VERSION and core_data_version == TARGET_CORE_DATA_VERSION:
                self.logger.info('Full Data Migration status=[COMPLETE], %s=[%s], '
                                 '%s=[%s], %s=%s',
                                 PLUGIN_ENTITYDATA,
                                 entity_version,
                                 PLUGIN_APPINFO,
                                 appinfo_version,
                                 PLUGIN_CORE_DATA,
                                 core_data_version)
                return True
            else:
                self.logger.info('Full Data Migration status=[NOTSTARTED/INPROGRESS]')
        return False

    def is_appinfo_migrated(self):
        status = self.check_data_migration_status()
        if status is not None:
            appinfo_version = status['data'][PLUGIN_APPINFO]

            if appinfo_version == TARGET_APPINFO_VERSION:
                self.logger.info('AppInfo Migration status=[COMPLETE],'
                                 '%s=[%s]',
                                 PLUGIN_APPINFO,
                                 appinfo_version)
                return True
            else:
                self.logger.info('AppInfo Migration status=[NOTSTARTED/INPROGRESS]')
        return False

    def is_migration_system_updated(self):
        status = self.check_data_migration_status()
        if status is not None:
            migration_system_version = status['data'][PLUGIN_MIGRATION_SYSTEM]

            if migration_system_version == TARGET_MIGRATION_SYSTEM_VERSION:
                self.logger.info('Migration System CURRENT, %s=[%s]',
                                 PLUGIN_MIGRATION_SYSTEM,
                                 migration_system_version)
                return True
            else:
                self.logger.info('Migration System OLD, %s=[%s]',
                                 PLUGIN_MIGRATION_SYSTEM,
                                 migration_system_version)
        return False

    def is_index_mapping_updated(self):
        status = self.check_data_migration_status()
        if status is not None:
            index_mapping_version = status['data'][PLUGIN_INDEX_MAPPING]

            if index_mapping_version == TARGET_INDEX_MAPPING_VERSION:
                self.logger.info('Index Mapping CURRENT, %s=[%s]',
                                 PLUGIN_INDEX_MAPPING,
                                 index_mapping_version)
                return True
            else:
                self.logger.info('Index Mapping OLD, %s=[%s]',
                                 PLUGIN_INDEX_MAPPING,
                                 index_mapping_version)
        return False

    def check_data_migration_status(self):

        try:
            r = requests.get(url=self.get_migration_status_url(), auth=(self.super_user, self.super_pass))
            if r.status_code == 200:
                response = r.json()
                return response
            else:
                self.logger.error('Failed to check migration status, %s', r)
                return
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to check migration status, %s', e)
            # exit_on_error()

    def get_reindex_status(self, job):
        status_url = self.get_reindex_url()+'/' + job

        try:
            r = requests.get(url=status_url, auth=(self.super_user, self.super_pass))
            response = r.json()
            return response['status']
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to get reindex status, %s', e)
            # exit_on_error()

    def start_app_reindex(self, appId):
        body = ""
        if self.start_date is not None:
            body = json.dumps({'updated': self.start_date})

        try:
            r = requests.post(url=self.get_reindex_url() + "/" + appId, data=body, auth=(self.super_user, self.super_pass))

            if r.status_code == 200:
                response = r.json()
                return response['jobId']
            else:
                self.logger.error('Failed to start reindex, %s', r)
                exit_on_error(str(r))
        except requests.exceptions.RequestException as e:
            self.logger.error('Unable to make API request for reindex, %s', e)
            exit_on_error(str(e))

    def is_reindex_running(self, job):
        status = self.get_reindex_status(job)
        self.logger.info('Re-index status=[%s]', status)
        if status != "COMPLETE":
            return True
        else:
            return False

    def get_dedup_status(self, job):
        status_url = self.get_dedup_url()+'/' + job
        try:
            r = requests.get(url=status_url, auth=(self.super_user, self.super_pass))
            response = r.json()
            return response['status']['status']
        except requests.exceptions.RequestException as e:
            self.logger.error('Failed to get dedup status, %s', e)
            # exit_on_error()

    def start_dedup(self, app_id):
        body = ""
        try:
            r = requests.post(url=self.get_dedup_url() + "/" + app_id, data=body, auth=(self.super_user, self.super_pass))
            if r.status_code == 200:
                response = r.json()
                return response['status']['jobStatusId']
            else:
                self.logger.error('Failed to start dedup, %s', r)
                exit_on_error(str(r))

        except requests.exceptions.RequestException as e:
            self.logger.error('Unable to make API request for dedup, %s', e)
            exit_on_error(str(e))

    def is_dedup_running(self, job):
        status = self.get_dedup_status(job)
        self.logger.info('Dedup status=[%s]', status)
        if status != "COMPLETE":
            return True
        else:
            return False

    def is_endpoint_available(self):

        try:
            r = requests.get(url=self.endpoint+'/status')
            if r.status_code == 200:
                return True
        except requests.exceptions.RequestException as e:
            self.logger.error('Endpoint is unavailable, %s', str(e))
            return False

    def log_metrics(self):
        self.logger.info(
            'Re-index start=[%s], ' +
            'Re-index end =[%s], ' +
            'Full Data Migration start=[%s], ' +
            'Full Data Migration end=[%s] ' +
            'AppInfo Migration start=[%s], ' +
            'AppInfo Migration end=[%s] ',
            self.metrics['reindex_start'],
            self.metrics['reindex_end'],
            self.metrics['full_data_migration_start'],
            self.metrics['full_data_migration_end'],
            self.metrics['appinfo_migration_start'],
            self.metrics['appinfo_migration_end']

        )

    def get_app_ids(self):

        try:

            url = self.endpoint + "/management/organizations"
            r = requests.get(url=url, auth=(self.super_user, self.super_pass))

            if r.status_code != 200:
                exit_on_error('Cannot get app ids: ' + r.text)

            response_json = r.json()

            app_ids = []
            orgs = response_json["organizations"]
            if orgs is not None:
                for org in orgs:
                    if org["name"] == self.org:
                        for app in org["applications"]:
                            app_ids.append(org["applications"][app])
            else:
                e = 'No Orgs in this system'
                self.logger.error(e)
                exit_on_error(e)

            return app_ids

        except requests.exceptions.RequestException as e:
            self.logger.error('Unable to get list of application ids, %s', e)
            exit_on_error(str(e))

def get_current_time():
    return str(int(time.time()*1000))


def exit_on_error(e=""):
    print ('Exiting migration script due to error: ' + str(e))
    sys.exit(1)


def init_logging(name):

    logger = logging.getLogger(name)
    log_file_name = './migration.log'
    log_formatter = logging.Formatter(fmt='%(asctime)s [%(name)s] %(levelname)s %(message)s',
                                      datefmt='%Y-%m-%d %H:%M:%S')

    rotating_file = logging.handlers.RotatingFileHandler(filename=log_file_name,
                                                         mode='a',
                                                         maxBytes=104857600,
                                                         backupCount=10)
    rotating_file.setFormatter(log_formatter)
    rotating_file.setLevel(logging.INFO)
    logger.addHandler(rotating_file)
    logger.setLevel(logging.INFO)

    stdout_logger = logging.StreamHandler(sys.stdout)
    stdout_logger.setFormatter(log_formatter)
    stdout_logger.setLevel(logging.INFO)
    logger.addHandler(stdout_logger)

    return logger

if __name__ == '__main__':

    migration = Migrate()
    migration.run()