# */ # * 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. # */ from __future__ import print_function from __future__ import print_function from __future__ import print_function import os import uuid from Queue import Empty import argparse import json import logging import sys from multiprocessing import Queue, Process from sets import Set import time_uuid import datetime from cloghandler import ConcurrentRotatingFileHandler import requests import traceback import redis import time from sys import platform as _platform import signal from requests.auth import HTTPBasicAuth from usergrid import UsergridQueryIterator import urllib3 __author__ = 'Jeff.West@yahoo.com' ECID = str(uuid.uuid1()) key_version = 'v4' logger = logging.getLogger('GraphMigrator') worker_logger = logging.getLogger('Worker') collection_worker_logger = logging.getLogger('CollectionWorker') error_logger = logging.getLogger('ErrorLogger') audit_logger = logging.getLogger('AuditLogger') status_logger = logging.getLogger('StatusLogger') urllib3.disable_warnings() DEFAULT_CREATE_APPS = False DEFAULT_RETRY_SLEEP = 10 DEFAULT_PROCESSING_SLEEP = 1 queue = Queue() QSIZE_OK = False try: queue.qsize() QSIZE_OK = True except: pass session_source = requests.Session() session_target = requests.Session() cache = None def total_seconds(td): return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10 ** 6) / 10 ** 6 def init_logging(stdout_enabled=True): root_logger = logging.getLogger() root_logger.setLevel(logging.getLevelName(config.get('log_level', 'INFO'))) # root_logger.setLevel(logging.WARN) logging.getLogger('requests.packages.urllib3.connectionpool').setLevel(logging.ERROR) logging.getLogger('boto').setLevel(logging.ERROR) logging.getLogger('urllib3.connectionpool').setLevel(logging.WARN) log_formatter = logging.Formatter( fmt='%(asctime)s | ' + ECID + ' | %(name)s | %(processName)s | %(levelname)s | %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p') stdout_logger = logging.StreamHandler(sys.stdout) stdout_logger.setFormatter(log_formatter) root_logger.addHandler(stdout_logger) if stdout_enabled: stdout_logger.setLevel(logging.getLevelName(config.get('log_level', 'INFO'))) # base log file log_file_name = os.path.join(config.get('log_dir'), '%s-%s-%s-migrator.log' % (config.get('org'), config.get('migrate'), ECID)) # ConcurrentRotatingFileHandler rotating_file = ConcurrentRotatingFileHandler(filename=log_file_name, mode='a', maxBytes=404857600, backupCount=0) rotating_file.setFormatter(log_formatter) rotating_file.setLevel(logging.INFO) root_logger.addHandler(rotating_file) error_log_file_name = os.path.join(config.get('log_dir'), '%s-%s-%s-migrator-errors.log' % ( config.get('org'), config.get('migrate'), ECID)) error_rotating_file = ConcurrentRotatingFileHandler(filename=error_log_file_name, mode='a', maxBytes=404857600, backupCount=0) error_rotating_file.setFormatter(log_formatter) error_rotating_file.setLevel(logging.ERROR) root_logger.addHandler(error_rotating_file) entity_name_map = { 'users': 'username' } config = {} # URL Templates for Usergrid org_management_app_url_template = "{api_url}/management/organizations/{org}/applications?client_id={client_id}&client_secret={client_secret}" org_management_url_template = "{api_url}/management/organizations/{org}/applications?client_id={client_id}&client_secret={client_secret}" org_url_template = "{api_url}/{org}?client_id={client_id}&client_secret={client_secret}" app_url_template = "{api_url}/{org}/{app}?client_id={client_id}&client_secret={client_secret}" collection_url_template = "{api_url}/{org}/{app}/{collection}?client_id={client_id}&client_secret={client_secret}" collection_query_url_template = "{api_url}/{org}/{app}/{collection}?ql={ql}&client_id={client_id}&client_secret={client_secret}&limit={limit}" collection_graph_url_template = "{api_url}/{org}/{app}/{collection}?client_id={client_id}&client_secret={client_secret}&limit={limit}" connection_query_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}/{verb}?client_id={client_id}&client_secret={client_secret}" connecting_query_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}/connecting/{verb}?client_id={client_id}&client_secret={client_secret}" connection_create_by_uuid_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}/{verb}/{target_uuid}?client_id={client_id}&client_secret={client_secret}" connection_create_by_name_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}/{verb}/{target_type}/{target_name}?client_id={client_id}&client_secret={client_secret}" connection_create_by_pairs_url_template = "{api_url}/{org}/{app}/{source_type_id}/{verb}/{target_type_id}?client_id={client_id}&client_secret={client_secret}" get_entity_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}?client_id={client_id}&client_secret={client_secret}&connections=none" get_entity_url_with_connections_template = "{api_url}/{org}/{app}/{collection}/{uuid}?client_id={client_id}&client_secret={client_secret}" put_entity_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}?client_id={client_id}&client_secret={client_secret}" permissions_url_template = "{api_url}/{org}/{app}/{collection}/{uuid}/permissions?client_id={client_id}&client_secret={client_secret}" user_credentials_url_template = "{api_url}/{org}/{app}/users/{uuid}/credentials" ignore_collections = ['activities', 'queues', 'events', 'notifications'] class StatusListener(Process): def __init__(self, status_queue, worker_queue): super(StatusListener, self).__init__() self.status_queue = status_queue self.worker_queue = worker_queue def run(self): keep_going = True org_results = { 'name': config.get('org'), 'apps': {}, } empty_count = 0 status_file_name = os.path.join(config.get('log_dir'), '%s-%s-%s-status.json' % (config.get('org'), config.get('migrate'), ECID)) while keep_going: try: app, collection, status_map = self.status_queue.get(timeout=60) status_logger.info('Received status update for app/collection: [%s / %s]' % (app, collection)) empty_count = 0 org_results['summary'] = { 'max_created': -1, 'max_modified': -1, 'min_created': 1584946416000, 'min_modified': 1584946416000, 'count': 0, 'bytes': 0 } if app not in org_results['apps']: org_results['apps'][app] = { 'collections': {} } org_results['apps'][app]['collections'].update(status_map) try: for app, app_data in org_results['apps'].iteritems(): app_data['summary'] = { 'max_created': -1, 'max_modified': -1, 'min_created': 1584946416000, 'min_modified': 1584946416000, 'count': 0, 'bytes': 0 } if 'collections' in app_data: for collection, collection_data in app_data['collections'].iteritems(): app_data['summary']['count'] += collection_data['count'] app_data['summary']['bytes'] += collection_data['bytes'] org_results['summary']['count'] += collection_data['count'] org_results['summary']['bytes'] += collection_data['bytes'] # APP if collection_data.get('max_modified') > app_data['summary']['max_modified']: app_data['summary']['max_modified'] = collection_data.get('max_modified') if collection_data.get('min_modified') < app_data['summary']['min_modified']: app_data['summary']['min_modified'] = collection_data.get('min_modified') if collection_data.get('max_created') > app_data['summary']['max_created']: app_data['summary']['max_created'] = collection_data.get('max_created') if collection_data.get('min_created') < app_data['summary']['min_created']: app_data['summary']['min_created'] = collection_data.get('min_created') # ORG if collection_data.get('max_modified') > org_results['summary']['max_modified']: org_results['summary']['max_modified'] = collection_data.get('max_modified') if collection_data.get('min_modified') < org_results['summary']['min_modified']: org_results['summary']['min_modified'] = collection_data.get('min_modified') if collection_data.get('max_created') > org_results['summary']['max_created']: org_results['summary']['max_created'] = collection_data.get('max_created') if collection_data.get('min_created') < org_results['summary']['min_created']: org_results['summary']['min_created'] = collection_data.get('min_created') if QSIZE_OK: status_logger.warn('CURRENT Queue Depth: %s' % self.worker_queue.qsize()) status_logger.warn('UPDATED status of org processed: %s' % json.dumps(org_results)) try: logger.info('Writing status to file: %s' % status_file_name) with open(status_file_name, 'w') as f: json.dump(org_results, f, indent=2) except: print traceback.format_exc() except KeyboardInterrupt, e: raise e except: print traceback.format_exc() except KeyboardInterrupt, e: status_logger.warn('FINAL status of org processed: %s' % json.dumps(org_results)) raise e except Empty: if QSIZE_OK: status_logger.warn('CURRENT Queue Depth: %s' % self.worker_queue.qsize()) status_logger.warn('CURRENT status of org processed: %s' % json.dumps(org_results)) status_logger.warning('EMPTY! Count=%s' % empty_count) empty_count += 1 if empty_count >= 120: keep_going = False except: print traceback.format_exc() logger.warn('FINAL status of org processed: %s' % json.dumps(org_results)) try: logger.info('Writing final status to file: %s' % status_file_name) with open(status_file_name, 'w') as f: json.dump(org_results, f, indent=2) except: print traceback.format_exc() class EntityWorker(Process): def __init__(self, queue, handler_function): super(EntityWorker, self).__init__() worker_logger.debug('Creating worker!') self.queue = queue self.handler_function = handler_function def run(self): worker_logger.info('starting run()...') keep_going = True count_processed = 0 empty_count = 0 start_time = int(time.time()) while keep_going: try: # get an entity with the app and collection name app, collection_name, entity = self.queue.get(timeout=120) empty_count = 0 # if entity.get('type') == 'user': # entity = confirm_user_entity(app, entity) # the handler operation is the specified operation such as migrate_graph if self.handler_function is not None: try: message_start_time = int(time.time()) processed = self.handler_function(app, collection_name, entity) message_end_time = int(time.time()) if processed: count_processed += 1 total_time = message_end_time - start_time avg_time_per_message = total_time / count_processed message_time = message_end_time - message_start_time worker_logger.debug('Processed [%sth] entity = %s / %s / %s' % ( count_processed, app, collection_name, entity.get('uuid'))) if count_processed % 1000 == 1: worker_logger.info( 'Processed [%sth] entity = [%s / %s / %s] in [%s]s - avg time/message [%s]' % ( count_processed, app, collection_name, entity.get('uuid'), message_time, avg_time_per_message)) except KeyboardInterrupt, e: raise e except Exception, e: logger.exception('Error in EntityWorker processing message') print traceback.format_exc() except KeyboardInterrupt, e: raise e except Empty: worker_logger.warning('EMPTY! Count=%s' % empty_count) empty_count += 1 if empty_count >= 2: keep_going = False except Exception, e: logger.exception('Error in EntityWorker run()') print traceback.format_exc() class CollectionWorker(Process): def __init__(self, work_queue, entity_queue, response_queue): super(CollectionWorker, self).__init__() collection_worker_logger.debug('Creating worker!') self.work_queue = work_queue self.response_queue = response_queue self.entity_queue = entity_queue def run(self): collection_worker_logger.info('starting run()...') keep_going = True counter = 0 # max_created = 0 empty_count = 0 app = 'ERROR' collection_name = 'NOT SET' status_map = {} sleep_time = 10 try: while keep_going: try: app, collection_name = self.work_queue.get(timeout=30) status_map = { collection_name: { 'iteration_started': str(datetime.datetime.now()), 'max_created': -1, 'max_modified': -1, 'min_created': 1584946416000, 'min_modified': 1584946416000, 'count': 0, 'bytes': 0 } } empty_count = 0 # added a flag for using graph vs query/index if config.get('graph', False): source_collection_url = collection_graph_url_template.format(org=config.get('org'), app=app, collection=collection_name, limit=config.get('limit'), **config.get('source_endpoint')) else: source_collection_url = collection_query_url_template.format(org=config.get('org'), app=app, collection=collection_name, limit=config.get('limit'), ql="select * %s" % config.get( 'ql'), **config.get('source_endpoint')) logger.info('Iterating URL: %s' % source_collection_url) # use the UsergridQuery from the Python SDK to iterate the collection q = UsergridQueryIterator(source_collection_url, page_delay=config.get('page_sleep_time'), sleep_time=config.get('error_retry_sleep')) for entity in q: # begin entity loop self.entity_queue.put((app, collection_name, entity)) counter += 1 if 'created' in entity: try: entity_created = long(entity.get('created')) if entity_created > status_map[collection_name]['max_created']: status_map[collection_name]['max_created'] = entity_created status_map[collection_name]['max_created_str'] = str( datetime.datetime.fromtimestamp(entity_created / 1000)) if entity_created < status_map[collection_name]['min_created']: status_map[collection_name]['min_created'] = entity_created status_map[collection_name]['min_created_str'] = str( datetime.datetime.fromtimestamp(entity_created / 1000)) except ValueError: pass if 'modified' in entity: try: entity_modified = long(entity.get('modified')) if entity_modified > status_map[collection_name]['max_modified']: status_map[collection_name]['max_modified'] = entity_modified status_map[collection_name]['max_modified_str'] = str( datetime.datetime.fromtimestamp(entity_modified / 1000)) if entity_modified < status_map[collection_name]['min_modified']: status_map[collection_name]['min_modified'] = entity_modified status_map[collection_name]['min_modified_str'] = str( datetime.datetime.fromtimestamp(entity_modified / 1000)) except ValueError: pass status_map[collection_name]['bytes'] += count_bytes(entity) status_map[collection_name]['count'] += 1 if counter % 1000 == 1: try: collection_worker_logger.warning( 'Sending stats for app/collection [%s / %s]: %s' % ( app, collection_name, status_map)) self.response_queue.put((app, collection_name, status_map)) if QSIZE_OK: collection_worker_logger.info( 'Counter=%s, collection queue depth=%s' % ( counter, self.work_queue.qsize())) except: pass collection_worker_logger.warn( 'Current status of collections processed: %s' % json.dumps(status_map)) if config.get('entity_sleep_time') > 0: collection_worker_logger.debug( 'sleeping for [%s]s per entity...' % (config.get('entity_sleep_time'))) time.sleep(config.get('entity_sleep_time')) collection_worker_logger.debug( 'STOPPED sleeping for [%s]s per entity...' % (config.get('entity_sleep_time'))) # end entity loop status_map[collection_name]['iteration_finished'] = str(datetime.datetime.now()) collection_worker_logger.warning( 'Collection [%s / %s / %s] loop complete! Max Created entity %s' % ( config.get('org'), app, collection_name, status_map[collection_name]['max_created'])) collection_worker_logger.warning( 'Sending FINAL stats for app/collection [%s / %s]: %s' % (app, collection_name, status_map)) self.response_queue.put((app, collection_name, status_map)) collection_worker_logger.info('Done! Finished app/collection: %s / %s' % (app, collection_name)) except KeyboardInterrupt, e: raise e except Empty: collection_worker_logger.warning('EMPTY! Count=%s' % empty_count) empty_count += 1 if empty_count >= 2: keep_going = False except Exception as e: logger.exception('Error in CollectionWorker processing collection [%s]' % collection_name) print traceback.format_exc() finally: self.response_queue.put((app, collection_name, status_map)) collection_worker_logger.info('FINISHED!') def use_name_for_collection(collection_name): return collection_name in config.get('use_name_for_collection', []) def include_edge(collection_name, edge_name): include_edges = config.get('include_edge', []) if include_edges is None: include_edges = [] exclude_edges = config.get('exclude_edge', []) if exclude_edges is None: exclude_edges = [] if len(include_edges) > 0 and edge_name not in include_edges: logger.debug( 'Skipping edge [%s] since it is not in INCLUDED list: %s' % (edge_name, include_edges)) return False if edge_name in exclude_edges: logger.debug( 'Skipping edge [%s] since it is in EXCLUDED list: %s' % (edge_name, exclude_edges)) return False if (collection_name in ['users', 'user'] and edge_name in ['followers', 'feed', 'activities']) \ or (collection_name in ['receipts', 'receipt'] and edge_name in ['device', 'devices']): # feed and activities are not retrievable... # roles and groups will be more efficiently handled from the role/group -> user # followers will be handled by 'following' # do only this from user -> device return False return True def exclude_edge(collection_name, edge_name): exclude_edges = config.get('exclude_edge', []) if exclude_edges is None: exclude_edges = [] if edge_name in exclude_edges: logger.debug('Skipping edge [%s] since it is in EXCLUDED list: %s' % (edge_name, exclude_edges)) return True if (collection_name in ['users', 'user'] and edge_name in ['followers', 'feed', 'activities']) \ or (collection_name in ['receipts', 'receipt'] and edge_name in ['device', 'devices']): # feed and activities are not retrievable... # roles and groups will be more efficiently handled from the role/group -> user # followers will be handled by 'following' # do only this from user -> device return True return False def confirm_user_entity(app, source_entity, attempts=0): attempts += 1 source_entity_url = get_entity_url_template.format(org=config.get('org'), app=app, collection='users', uuid=source_entity.get('username'), **config.get('source_endpoint')) if attempts >= 5: logger.warning('Punting after [%s] attempts to confirm user at URL [%s], will use the source entity...' % ( attempts, source_entity_url)) return source_entity r = requests.get(url=source_entity_url) if r.status_code == 200: retrieved_entity = r.json().get('entities')[0] if retrieved_entity.get('uuid') != source_entity.get('uuid'): logger.info( 'UUID of Source Entity [%s] differs from uuid [%s] of retrieved entity at URL=[%s] and will be substituted' % ( source_entity.get('uuid'), retrieved_entity.get('uuid'), source_entity_url)) return retrieved_entity elif 'service_resource_not_found' in r.text: logger.warn('Unable to retrieve user at URL [%s], and will use source entity. status=[%s] response: %s...' % ( source_entity_url, r.status_code, r.text)) return source_entity else: logger.error('After [%s] attempts to confirm user at URL [%s], received status [%s] message: %s...' % ( attempts, source_entity_url, r.status_code, r.text)) time.sleep(DEFAULT_RETRY_SLEEP) return confirm_user_entity(app, source_entity, attempts) def create_connection(app, collection_name, source_entity, edge_name, target_entity): target_app, target_collection, target_org = get_target_mapping(app, collection_name) source_identifier = get_source_identifier(source_entity) target_identifier = get_source_identifier(target_entity) source_type_id = '%s/%s' % (source_entity.get('type'), source_identifier) target_type_id = '%s/%s' % (target_entity.get('type'), target_identifier) if source_entity.get('type') == 'user': source_type_id = '%s/%s' % ('users', source_entity.get('username')) if target_entity.get('type') == 'user': if edge_name == 'users': target_type_id = target_entity.get('uuid') else: target_type_id = '%s/%s' % ('users', target_entity.get('uuid')) if target_entity.get('type') == 'device': if edge_name == 'devices': target_type_id = target_entity.get('uuid') else: target_type_id = '%s/%s' % ('devices', target_entity.get('uuid')) if target_entity.get('type') == 'receipt': if edge_name == 'receipts': target_type_id = target_entity.get('uuid') else: target_type_id = '%s/%s' % ('receipts', target_entity.get('uuid')) create_connection_url = connection_create_by_pairs_url_template.format( org=target_org, app=target_app, source_type_id=source_type_id, verb=edge_name, target_type_id=target_type_id, **config.get('target_endpoint')) if not config.get('skip_cache_read', False): processed = cache.get(create_connection_url) if processed not in [None, 'None']: logger.debug('Skipping visited Edge: [%s / %s / %s] --[%s]--> [%s / %s / %s]: %s ' % ( app, collection_name, source_identifier, edge_name, target_app, target_entity.get('type'), target_entity.get('name'), create_connection_url)) return True logger.info('Connecting entity [%s / %s / %s] --[%s]--> [%s / %s / %s]: %s ' % ( app, collection_name, source_identifier, edge_name, target_app, target_entity.get('type'), target_entity.get('name', target_entity.get('uuid')), create_connection_url)) attempts = 0 while attempts < 5: attempts += 1 r_create = session_target.post(create_connection_url) if r_create.status_code == 200: if not config.get('skip_cache_write', False): cache.set(create_connection_url, 1) return True else: if r_create.status_code >= 500: if attempts < 5: logger.warning('FAILED [%s] (will retry) to create connection at URL=[%s]: %s' % ( r_create.status_code, create_connection_url, r_create.text)) time.sleep(DEFAULT_RETRY_SLEEP) else: logger.critical( 'FAILED [%s] (WILL NOT RETRY - max attempts) to create connection at URL=[%s]: %s' % ( r_create.status_code, create_connection_url, r_create.text)) return False elif r_create.status_code in [401, 404]: if config.get('repair_data', False): logger.warning('FAILED [%s] (WILL attempt repair) to create connection at URL=[%s]: %s' % ( r_create.status_code, create_connection_url, r_create.text)) migrate_data(app, source_entity.get('type'), source_entity, force=True) migrate_data(app, target_entity.get('type'), target_entity, force=True) else: logger.critical('FAILED [%s] (WILL NOT attempt repair) to create connection at URL=[%s]: %s' % ( r_create.status_code, create_connection_url, r_create.text)) else: logger.warning('FAILED [%s] (will retry) to create connection at URL=[%s]: %s' % ( r_create.status_code, create_connection_url, r_create.text)) return False def process_edges(app, collection_name, source_entity, edge_name, connection_stack): source_identifier = get_source_identifier(source_entity) while len(connection_stack) > 0: target_entity = connection_stack.pop() if exclude_collection(collection_name) or exclude_collection(target_entity.get('type')): logger.debug('EXCLUDING Edge (collection): [%s / %s / %s] --[%s]--> ?' % ( app, collection_name, source_identifier, edge_name )) continue create_connection(app, collection_name, source_entity, edge_name, target_entity) def migrate_out_graph_edge_type(app, collection_name, source_entity, edge_name, depth=0): if not include_edge(collection_name, edge_name): return True source_uuid = source_entity.get('uuid') key = '%s:edge:out:%s:%s' % (key_version, source_uuid, edge_name) if not config.get('skip_cache_read', False): date_visited = cache.get(key) if date_visited not in [None, 'None']: logger.info('Skipping EDGE [%s / %s --%s-->] - visited at %s' % ( collection_name, source_uuid, edge_name, date_visited)) return True else: cache.delete(key) if not config.get('skip_cache_write', False): cache.set(name=key, value=str(int(time.time())), ex=config.get('visit_cache_ttl', 3600 * 2)) logger.debug('Visiting EDGE [%s / %s (%s) --%s-->] at %s' % ( collection_name, source_uuid, get_uuid_time(source_uuid), edge_name, str(datetime.datetime.utcnow()))) response = True source_identifier = get_source_identifier(source_entity) count_edges = 0 logger.debug( 'Processing edge type=[%s] of entity [%s / %s / %s]' % (edge_name, app, collection_name, source_identifier)) target_app, target_collection, target_org = get_target_mapping(app, collection_name) connection_query_url = connection_query_url_template.format( org=config.get('org'), app=app, verb=edge_name, collection=collection_name, uuid=source_identifier, limit=config.get('limit'), **config.get('source_endpoint')) connection_query = UsergridQueryIterator(connection_query_url, sleep_time=config.get('error_retry_sleep')) connection_stack = [] for target_entity in connection_query: target_connection_collection = config.get('collection_mapping', {}).get(target_entity.get('type'), target_entity.get('type')) target_ok = migrate_graph(app, target_entity.get('type'), source_entity=target_entity, depth=depth) if not target_ok: logger.critical( 'Error migrating TARGET entity data for connection [%s / %s / %s] --[%s]--> [%s / %s / %s]' % ( app, collection_name, source_identifier, edge_name, app, target_connection_collection, target_entity.get('name', target_entity.get('uuid')))) count_edges += 1 connection_stack.append(target_entity) process_edges(app, collection_name, source_entity, edge_name, connection_stack) return response def get_source_identifier(source_entity): entity_type = source_entity.get('type') source_identifier = source_entity.get('uuid') if use_name_for_collection(entity_type): if entity_type in ['user']: source_identifier = source_entity.get('username') else: source_identifier = source_entity.get('name') if source_identifier is None: source_identifier = source_entity.get('uuid') logger.warn('Using UUID for entity [%s / %s]' % (entity_type, source_identifier)) return source_identifier def include_collection(collection_name): if collection_name in ['events']: return False include = config.get('collection', []) if include is not None and len(include) > 0 and collection_name not in include: return False exclude = config.get('exclude_collection', []) if exclude is not None and collection_name in exclude: return False return True def exclude_collection(collection_name): exclude = config.get('exclude_collection', []) if exclude is not None and collection_name in exclude: return True return False def migrate_in_graph_edge_type(app, collection_name, source_entity, edge_name, depth=0): source_uuid = source_entity.get('uuid') key = '%s:edges:in:%s:%s' % (key_version, source_uuid, edge_name) if not config.get('skip_cache_read', False): date_visited = cache.get(key) if date_visited not in [None, 'None']: logger.info('Skipping EDGE [--%s--> %s / %s] - visited at %s' % ( collection_name, source_uuid, edge_name, date_visited)) return True else: cache.delete(key) if not config.get('skip_cache_write', False): cache.set(name=key, value=str(int(time.time())), ex=config.get('visit_cache_ttl', 3600 * 2)) logger.debug('Visiting EDGE [--%s--> %s / %s (%s)] at %s' % ( edge_name, collection_name, source_uuid, get_uuid_time(source_uuid), str(datetime.datetime.utcnow()))) source_identifier = get_source_identifier(source_entity) if exclude_collection(collection_name): logger.debug('Excluding (Collection) entity [%s / %s / %s]' % (app, collection_name, source_uuid)) return True if not include_edge(collection_name, edge_name): return True logger.debug( 'Processing edge type=[%s] of entity [%s / %s / %s]' % (edge_name, app, collection_name, source_identifier)) logger.debug('Processing IN edges type=[%s] of entity [ %s / %s / %s]' % ( edge_name, app, collection_name, source_uuid)) connecting_query_url = connecting_query_url_template.format( org=config.get('org'), app=app, collection=collection_name, uuid=source_uuid, verb=edge_name, limit=config.get('limit'), **config.get('source_endpoint')) connection_query = UsergridQueryIterator(connecting_query_url, sleep_time=config.get('error_retry_sleep')) response = True for e_connection in connection_query: logger.debug('Triggering IN->OUT edge migration on entity [%s / %s / %s] ' % ( app, e_connection.get('type'), e_connection.get('uuid'))) response = migrate_graph(app, e_connection.get('type'), e_connection, depth) and response return response def migrate_graph(app, collection_name, source_entity, depth=0): depth += 1 source_uuid = source_entity.get('uuid') # short circuit if the graph depth exceeds what was specified if depth > config.get('graph_depth', 1): logger.debug( 'Reached Max Graph Depth, stopping after [%s] on [%s / %s]' % (depth, collection_name, source_uuid)) return True else: logger.debug('Processing @ Graph Depth [%s]' % depth) if exclude_collection(collection_name): logger.warn('Ignoring entity in filtered collection [%s]' % collection_name) return True key = '%s:graph:%s' % (key_version, source_uuid) entity_tag = '[%s / %s / %s (%s)]' % (app, collection_name, source_uuid, get_uuid_time(source_uuid)) if not config.get('skip_cache_read', False): date_visited = cache.get(key) if date_visited not in [None, 'None']: logger.debug('Skipping GRAPH %s at %s' % (entity_tag, date_visited)) return True else: cache.delete(key) logger.info('Visiting GRAPH %s at %s' % (entity_tag, str(datetime.datetime.utcnow()))) if not config.get('skip_cache_write', False): cache.set(name=key, value=str(int(time.time())), ex=config.get('visit_cache_ttl', 3600 * 2)) # first, migrate data for current node response = migrate_data(app, collection_name, source_entity) # gather the outbound edge names out_edge_names = [edge_name for edge_name in source_entity.get('metadata', {}).get('collections', [])] out_edge_names += [edge_name for edge_name in source_entity.get('metadata', {}).get('connections', [])] logger.debug('Entity %s has [%s] OUT edges' % (entity_tag, len(out_edge_names))) # migrate each outbound edge type for edge_name in out_edge_names: if not exclude_edge(collection_name, edge_name): response = migrate_out_graph_edge_type(app, collection_name, source_entity, edge_name, depth) and response if config.get('prune', False): prune_edge_by_name(edge_name, app, collection_name, source_entity) # gather the inbound edge names in_edge_names = [edge_name for edge_name in source_entity.get('metadata', {}).get('connecting', [])] logger.debug('Entity %s has [%s] IN edges' % (entity_tag, len(in_edge_names))) # migrate each inbound edge type for edge_name in in_edge_names: if not exclude_edge(collection_name, edge_name): response = migrate_in_graph_edge_type(app, collection_name, source_entity, edge_name, depth) and response return response def collect_entities(q): response = {} for e in q: response[e.get('uuid')] = e return response def prune_edge_by_name(edge_name, app, collection_name, source_entity): if not include_edge(collection_name, edge_name): return True source_identifier = get_source_identifier(source_entity) source_uuid = source_entity.get('uuid') entity_tag = '[%s / %s / %s (%s)]' % (app, collection_name, source_uuid, get_uuid_time(source_uuid)) target_app, target_collection, target_org = get_target_mapping(app, collection_name) target_connection_query_url = connection_query_url_template.format( org=target_org, app=target_app, verb=edge_name, collection=target_collection, uuid=source_identifier, limit=config.get('limit'), **config.get('target_endpoint')) source_connection_query_url = connection_query_url_template.format( org=config.get('org'), app=app, verb=edge_name, collection=collection_name, uuid=source_identifier, limit=config.get('limit'), **config.get('source_endpoint')) source_connections = collect_entities( UsergridQueryIterator(source_connection_query_url, sleep_time=config.get('error_retry_sleep'))) target_connections = collect_entities( UsergridQueryIterator(target_connection_query_url, sleep_time=config.get('error_retry_sleep'))) delete_uuids = Set(target_connections.keys()) - Set(source_connections.keys()) if len(delete_uuids) > 0: logger.info('Found [%s] edges to delete for entity %s' % (len(delete_uuids), entity_tag)) for delete_uuid in delete_uuids: delete_connection_url = connection_create_by_uuid_url_template.format( org=target_org, app=target_app, verb=edge_name, collection=target_collection, uuid=source_identifier, target_uuid=delete_uuid, **config.get('target_endpoint')) attempts = 0 while attempts < 5: attempts += 1 r = session_target.delete(delete_connection_url) if not config.get('skip_cache_write'): cache.delete(delete_connection_url) if r.status_code == 200: logger.info('Pruned edge on attempt [%s] URL=[%s]' % (attempts, delete_connection_url)) break else: logger.error('Error [%s] on attempt [%s] deleting connection at URL=[%s]: %s' % ( r.status_code, attempts, delete_connection_url, r.text)) time.sleep(DEFAULT_RETRY_SLEEP) return True def prune_graph(app, collection_name, source_entity): source_uuid = source_entity.get('uuid') key = '%s:prune_graph:%s' % (key_version, source_uuid) entity_tag = '[%s / %s / %s (%s)]' % (app, collection_name, source_uuid, get_uuid_time(source_uuid)) if not config.get('skip_cache_read', False): date_visited = cache.get(key) if date_visited not in [None, 'None']: logger.debug('Skipping PRUNE %s at %s' % (entity_tag, date_visited)) return True else: cache.delete(key) logger.debug('pruning GRAPH %s at %s' % (entity_tag, str(datetime.datetime.utcnow()))) if not config.get('skip_cache_write', False): cache.set(name=key, value=str(int(time.time())), ex=config.get('visit_cache_ttl', 3600 * 2)) if collection_name in config.get('exclude_collection', []): logger.debug('Excluding (Collection) entity %s' % entity_tag) return True out_edge_names = [edge_name for edge_name in source_entity.get('metadata', {}).get('collections', [])] out_edge_names += [edge_name for edge_name in source_entity.get('metadata', {}).get('connections', [])] for edge_name in out_edge_names: prune_edge_by_name(edge_name, app, collection_name, source_entity) def reput(app, collection_name, source_entity, attempts=0): source_identifier = source_entity.get('uuid') target_app, target_collection, target_org = get_target_mapping(app, collection_name) try: target_entity_url_by_name = put_entity_url_template.format(org=target_org, app=target_app, collection=target_collection, uuid=source_identifier, **config.get('target_endpoint')) r = session_source.put(target_entity_url_by_name, data=json.dumps({})) if r.status_code != 200: logger.info('HTTP [%s]: %s' % (target_entity_url_by_name, r.status_code)) else: logger.debug('HTTP [%s]: %s' % (target_entity_url_by_name, r.status_code)) except: pass def get_uuid_time(the_uuid_string): return time_uuid.TimeUUID(the_uuid_string).get_datetime() def migrate_permissions(app, collection_name, source_entity, attempts=0): if collection_name not in ['roles', 'role', 'group', 'groups']: return True target_app, target_collection, target_org = get_target_mapping(app, collection_name) source_identifier = get_source_identifier(source_entity) source_permissions_url = permissions_url_template.format(org=config.get('org'), app=app, collection=collection_name, uuid=source_identifier, **config.get('source_endpoint')) r = session_source.get(source_permissions_url) if r.status_code != 200: logger.error('Unable to get permissions at URL [%s]: %s' % (source_permissions_url, r.text)) return False perm_response = r.json() perms = perm_response.get('data', []) logger.info('Migrating [%s / %s] with permissions %s' % (collection_name, source_identifier, perms)) if len(perms) > 0: target_permissions_url = permissions_url_template.format(org=target_org, app=target_app, collection=target_collection, uuid=source_identifier, **config.get('target_endpoint')) for permission in perms: data = {'permission': permission} logger.info('Posting permission %s to %s' % (json.dumps(data), target_permissions_url)) r = session_target.post(target_permissions_url, json.dumps(data)) if r.status_code != 200: logger.error( 'ERROR posting permission %s to URL=[%s]: %s' % ( json.dumps(data), target_permissions_url, r.text)) return True def migrate_data(app, collection_name, source_entity, attempts=0, force=False): if config.get('skip_data') and not force: return True # check the cache to see if this entity has changed if not config.get('skip_cache_read', False) and not force: try: str_modified = cache.get(source_entity.get('uuid')) if str_modified not in [None, 'None']: modified = long(str_modified) logger.debug('FOUND CACHE: %s = %s ' % (source_entity.get('uuid'), modified)) if modified <= source_entity.get('modified'): modified_date = datetime.datetime.utcfromtimestamp(modified / 1000) e_uuid = source_entity.get('uuid') uuid_datetime = time_uuid.TimeUUID(e_uuid).get_datetime() logger.debug('Skipping ENTITY: %s / %s / %s / %s (%s) / %s (%s)' % ( config.get('org'), app, collection_name, e_uuid, uuid_datetime, modified, modified_date)) return True else: logger.debug('DELETING CACHE: %s ' % (source_entity.get('uuid'))) cache.delete(source_entity.get('uuid')) except: logger.error('Error on checking cache for uuid=[%s]' % source_entity.get('uuid')) logger.error(traceback.format_exc()) if exclude_collection(collection_name): logger.warn('Excluding entity in filtered collection [%s]' % collection_name) return True # handle duplicate user case if collection_name in ['users', 'user']: source_entity = confirm_user_entity(app, source_entity) source_identifier = get_source_identifier(source_entity) logger.info('Visiting ENTITY data [%s / %s (%s) ] at %s' % ( collection_name, source_identifier, get_uuid_time(source_entity.get('uuid')), str(datetime.datetime.utcnow()))) entity_copy = source_entity.copy() if 'metadata' in entity_copy: entity_copy.pop('metadata') target_app, target_collection, target_org = get_target_mapping(app, collection_name) try: target_entity_url_by_name = put_entity_url_template.format(org=target_org, app=target_app, collection=target_collection, uuid=source_identifier, **config.get('target_endpoint')) r = session_target.put(url=target_entity_url_by_name, data=json.dumps(entity_copy)) if attempts > 1: logger.warn('Attempt [%s] to migrate entity [%s / %s] at URL [%s]' % ( attempts, collection_name, source_identifier, target_entity_url_by_name)) else: logger.debug('Attempt [%s] to migrate entity [%s / %s] at URL [%s]' % ( attempts, collection_name, source_identifier, target_entity_url_by_name)) if r.status_code == 200: # Worked => WE ARE DONE logger.info( 'migrate_data | success=[%s] | attempts=[%s] | entity=[%s / %s / %s] | created=[%s] | modified=[%s]' % ( True, attempts, config.get('org'), app, source_identifier, source_entity.get('created'), source_entity.get('modified'),)) if not config.get('skip_cache_write', False): logger.debug('SETTING CACHE | uuid=[%s] | modified=[%s]' % ( source_entity.get('uuid'), str(source_entity.get('modified')))) cache.set(source_entity.get('uuid'), str(source_entity.get('modified'))) if collection_name in ['role', 'group', 'roles', 'groups']: migrate_permissions(app, collection_name, source_entity, attempts=0) if collection_name in ['users', 'user']: migrate_user_credentials(app, collection_name, source_entity, attempts=0) return True else: logger.error('Failure [%s] on attempt [%s] to PUT url=[%s], entity=[%s] response=[%s]' % ( r.status_code, attempts, target_entity_url_by_name, json.dumps(source_entity), r.text)) if attempts >= 5: logger.critical( 'ABORT migrate_data | success=[%s] | attempts=[%s] | created=[%s] | modified=[%s] %s / %s / %s' % ( True, attempts, source_entity.get('created'), source_entity.get('modified'), app, collection_name, source_identifier)) return False if r.status_code == 400: if target_collection in ['roles', 'role']: return repair_user_role(app, collection_name, source_entity) elif target_collection in ['users', 'user']: return handle_user_migration_conflict(app, collection_name, source_entity) elif 'duplicate_unique_property_exists' in r.text: logger.error( 'WILL NOT RETRY (duplicate) [%s] attempts to PUT url=[%s], entity=[%s] response=[%s]' % ( attempts, target_entity_url_by_name, json.dumps(source_entity), r.text)) return False elif r.status_code == 403: logger.critical( 'ABORT migrate_data | success=[%s] | attempts=[%s] | created=[%s] | modified=[%s] %s / %s / %s' % ( False, attempts, source_entity.get('created'), source_entity.get('modified'), app, collection_name, source_identifier)) return False except: logger.error(traceback.format_exc()) logger.error('error in migrate_data on entity: %s' % json.dumps(source_entity)) logger.warn( 'UNSUCCESSFUL migrate_data | success=[%s] | attempts=[%s] | entity=[%s / %s / %s] | created=[%s] | modified=[%s]' % ( True, attempts, config.get('org'), app, source_identifier, source_entity.get('created'), source_entity.get('modified'),)) return migrate_data(app, collection_name, source_entity, attempts=attempts + 1) def handle_user_migration_conflict(app, collection_name, source_entity, attempts=0, depth=0): if collection_name in ['users', 'user']: return False username = source_entity.get('username') target_app, target_collection, target_org = get_target_mapping(app, collection_name) target_entity_url = get_entity_url_template.format(org=target_org, app=target_app, collection=target_collection, uuid=username, **config.get('target_endpoint')) # There is retry build in, here is the short circuit if attempts >= 5: logger.critical( 'Aborting after [%s] attempts to audit user [%s] at URL [%s]' % (attempts, username, target_entity_url)) return False r = session_target.get(url=target_entity_url) if r.status_code == 200: target_entity = r.json().get('entities')[0] if source_entity.get('created') < target_entity.get('created'): return repair_user_role(app, collection_name, source_entity) elif r.status_code / 100 == 5: audit_logger.warning( 'CONFLICT: handle_user_migration_conflict failed attempt [%s] GET [%s] on TARGET URL=[%s] - : %s' % ( attempts, r.status_code, target_entity_url, r.text)) time.sleep(DEFAULT_RETRY_SLEEP) return handle_user_migration_conflict(app, collection_name, source_entity, attempts) else: audit_logger.error( 'CONFLICT: Failed handle_user_migration_conflict attempt [%s] GET [%s] on TARGET URL=[%s] - : %s' % ( attempts, r.status_code, target_entity_url, r.text)) return False def get_best_source_entity(app, collection_name, source_entity, depth=0): target_app, target_collection, target_org = get_target_mapping(app, collection_name) target_pk = 'uuid' if target_collection in ['users', 'user']: target_pk = 'username' elif target_collection in ['roles', 'role']: target_pk = 'name' target_name = source_entity.get(target_pk) # there should be no target entity now, we just need to decide which one from the source to use source_entity_url_by_name = get_entity_url_template.format(org=config.get('org'), app=app, collection=collection_name, uuid=target_name, **config.get('source_endpoint')) r_get_source_entity = session_source.get(source_entity_url_by_name) # if we are able to get at the source by PK... if r_get_source_entity.status_code == 200: # extract the entity from the response entity_from_get = r_get_source_entity.json().get('entities')[0] return entity_from_get elif r_get_source_entity.status_code / 100 == 4: # wasn't found, get by QL and sort source_entity_query_url = collection_query_url_template.format(org=config.get('org'), app=app, collection=collection_name, ql='select * where %s=\'%s\' order by created asc' % ( target_pk, target_name), limit=config.get('limit'), **config.get('source_endpoint')) logger.info('Attempting to determine best entity from query on URL %s' % source_entity_query_url) q = UsergridQueryIterator(source_entity_query_url, sleep_time=config.get('error_retry_sleep')) desired_entity = None entity_counter = 0 for e in q: entity_counter += 1 if desired_entity is None: desired_entity = e elif e.get('created') < desired_entity.get('created'): desired_entity = e if desired_entity is None: logger.warn('Unable to determine best of [%s] entities from query on URL %s' % ( entity_counter, source_entity_query_url)) return source_entity else: return desired_entity else: return source_entity def repair_user_role(app, collection_name, source_entity, attempts=0, depth=0): target_app, target_collection, target_org = get_target_mapping(app, collection_name) # For the users collection, there seemed to be cases where a USERNAME was created/existing with the a # different UUID which caused a 'collision' - so the point is to delete the entity with the differing # UUID by UUID and then do a recursive call to migrate the data - now that the collision has been cleared target_pk = 'uuid' if target_collection in ['users', 'user']: target_pk = 'username' elif target_collection in ['roles', 'role']: target_pk = 'name' target_name = source_entity.get(target_pk) target_entity_url_by_name = get_entity_url_template.format(org=target_org, app=target_app, collection=target_collection, uuid=target_name, **config.get('target_endpoint')) logger.warning('Repairing: Deleting name=[%s] entity at URL=[%s]' % (target_name, target_entity_url_by_name)) r = session_target.delete(target_entity_url_by_name) if r.status_code == 200 or (r.status_code in [404, 401] and 'service_resource_not_found' in r.text): logger.info('Deletion of entity at URL=[%s] was [%s]' % (target_entity_url_by_name, r.status_code)) best_source_entity = get_best_source_entity(app, collection_name, source_entity) target_entity_url_by_uuid = get_entity_url_template.format(org=target_org, app=target_app, collection=target_collection, uuid=best_source_entity.get('uuid'), **config.get('target_endpoint')) r = session_target.put(target_entity_url_by_uuid, data=json.dumps(best_source_entity)) if r.status_code == 200: logger.info('Successfully repaired user at URL=[%s]' % target_entity_url_by_uuid) return True else: logger.critical('Failed to PUT [%s] the desired entity at URL=[%s]: %s' % ( r.status_code, target_entity_url_by_name, r.text)) return False else: # log an error and keep going if we cannot delete the entity at the specified URL. Unlikely, but if so # then this entity is borked logger.critical( 'Deletion of entity at URL=[%s] FAILED [%s]: %s' % (target_entity_url_by_name, r.status_code, r.text)) return False def get_target_mapping(app, collection_name): target_org = config.get('org_mapping', {}).get(config.get('org'), config.get('org')) target_app = config.get('app_mapping', {}).get(app, app) target_collection = config.get('collection_mapping', {}).get(collection_name, collection_name) return target_app, target_collection, target_org def parse_args(): parser = argparse.ArgumentParser(description='Usergrid Org/App Migrator') parser.add_argument('--log_dir', help='path to the place where logs will be written', default='./', type=str, required=False) parser.add_argument('--log_level', help='log level - DEBUG, INFO, WARN, ERROR, CRITICAL', default='INFO', type=str, required=False) parser.add_argument('-o', '--org', help='Name of the org to migrate', type=str, required=True) parser.add_argument('-a', '--app', help='Name of one or more apps to include, specify none to include all apps', required=False, action='append') parser.add_argument('-e', '--include_edge', help='Name of one or more edges/connection types to INCLUDE, specify none to include all edges', required=False, action='append') parser.add_argument('--exclude_edge', help='Name of one or more edges/connection types to EXCLUDE, specify none to include all edges', required=False, action='append') parser.add_argument('--exclude_collection', help='Name of one or more collections to EXCLUDE, specify none to include all collections', required=False, action='append') parser.add_argument('-c', '--collection', help='Name of one or more collections to include, specify none to include all collections', default=[], action='append') parser.add_argument('--force_app', help='Necessary for using 2.0 as a source at times due to API issues. Forces the specified app(s) to be processed, even if they are not returned from the list of apps in the API call', default=[], action='append') parser.add_argument('--use_name_for_collection', help='Name of one or more collections to use [name] instead of [uuid] for creating entities and edges', default=[], action='append') parser.add_argument('-m', '--migrate', help='Specifies what to migrate: data, connections, credentials, audit or none (just iterate ' 'the apps/collections)', type=str, choices=[ 'data', 'prune', 'none', 'reput', 'credentials', 'graph', 'permissions' ], default='data') parser.add_argument('-s', '--source_config', help='The path to the source endpoint/org configuration file', type=str, default='source.json') parser.add_argument('-d', '--target_config', help='The path to the target endpoint/org configuration file', type=str, default='destination.json') parser.add_argument('--redis_socket', help='The path to the socket for redis to use', type=str) parser.add_argument('--limit', help='The number of entities to return per query request', type=int, default=100) parser.add_argument('-w', '--entity_workers', help='The number of worker processes to do the migration', type=int, default=16) parser.add_argument('--visit_cache_ttl', help='The TTL of the cache of visiting nodes in the graph for connections', type=int, default=3600 * 2) parser.add_argument('--error_retry_sleep', help='The number of seconds to wait between retrieving after an error', type=float, default=30) parser.add_argument('--page_sleep_time', help='The number of seconds to wait between retrieving pages from the UsergridQueryIterator', type=float, default=0) parser.add_argument('--entity_sleep_time', help='The number of seconds to wait between retrieving pages from the UsergridQueryIterator', type=float, default=0) parser.add_argument('--collection_workers', help='The number of worker processes to do the migration', type=int, default=2) parser.add_argument('--queue_size_max', help='The max size of entities to allow in the queue', type=int, default=100000) parser.add_argument('--graph_depth', help='The graph depth to traverse to copy', type=int, default=3) parser.add_argument('--queue_watermark_high', help='The point at which publishing to the queue will PAUSE until it is at or below low watermark', type=int, default=25000) parser.add_argument('--min_modified', help='Break when encountering a modified date before this, per collection', type=int, default=0) parser.add_argument('--max_modified', help='Break when encountering a modified date after this, per collection', type=long, default=3793805526000) parser.add_argument('--queue_watermark_low', help='The point at which publishing to the queue will RESUME after it has reached the high watermark', type=int, default=5000) parser.add_argument('--ql', help='The QL to use in the filter for reading data from collections', type=str, default='select * order by created asc') # default='select * order by created asc') parser.add_argument('--repair_data', help='Repair data when iterating/migrating graph but skipping data', action='store_true') parser.add_argument('--prune', help='Prune the graph while processing (instead of the prune operation)', action='store_true') parser.add_argument('--skip_data', help='Skip migrating data (useful for connections only)', action='store_true') parser.add_argument('--skip_credentials', help='Skip migrating credentials', action='store_true') parser.add_argument('--skip_cache_read', help='Skip reading the cache (modified timestamps and graph edges)', dest='skip_cache_read', action='store_true') parser.add_argument('--skip_cache_write', help='Skip updating the cache with modified timestamps of entities and graph edges', dest='skip_cache_write', action='store_true') parser.add_argument('--create_apps', help='Create apps at the target if they do not exist', dest='create_apps', action='store_true') parser.add_argument('--nohup', help='specifies not to use stdout for logging', action='store_true') parser.add_argument('--graph', help='Use GRAPH instead of Query', dest='graph', action='store_true') parser.add_argument('--su_username', help='Superuser username', required=False, type=str) parser.add_argument('--su_password', help='Superuser Password', required=False, type=str) parser.add_argument('--inbound_connections', help='Name of the org to migrate', action='store_true') parser.add_argument('--map_app', help="Multiple allowed: A colon-separated string such as 'apples:oranges' which indicates to" " put data from the app named 'apples' from the source endpoint into app named 'oranges' " "in the target endpoint", default=[], action='append') parser.add_argument('--map_collection', help="One or more colon-separated string such as 'cats:dogs' which indicates to put data from " "collections named 'cats' from the source endpoint into a collection named 'dogs' in the " "target endpoint, applicable globally to all apps", default=[], action='append') parser.add_argument('--map_org', help="One or more colon-separated strings such as 'red:blue' which indicates to put data from " "org named 'red' from the source endpoint into a collection named 'blue' in the target " "endpoint", default=[], action='append') my_args = parser.parse_args(sys.argv[1:]) return vars(my_args) def init(): global config if config.get('migrate') == 'credentials': if config.get('su_password') is None or config.get('su_username') is None: message = 'ABORT: In order to migrate credentials, Superuser parameters (su_password, su_username) are required' print message logger.critical(message) exit() config['collection_mapping'] = {} config['app_mapping'] = {} config['org_mapping'] = {} for mapping in config.get('map_collection', []): parts = mapping.split(':') if len(parts) == 2: config['collection_mapping'][parts[0]] = parts[1] else: logger.warning('Skipping Collection mapping: [%s]' % mapping) for mapping in config.get('map_app', []): parts = mapping.split(':') if len(parts) == 2: config['app_mapping'][parts[0]] = parts[1] else: logger.warning('Skipping App mapping: [%s]' % mapping) for mapping in config.get('map_org', []): parts = mapping.split(':') if len(parts) == 2: config['org_mapping'][parts[0]] = parts[1] logger.info('Mapping Org [%s] to [%s] from mapping [%s]' % (parts[0], parts[1], mapping)) else: logger.warning('Skipping Org mapping: [%s]' % mapping) with open(config.get('source_config'), 'r') as f: config['source_config'] = json.load(f) with open(config.get('target_config'), 'r') as f: config['target_config'] = json.load(f) if config['exclude_collection'] is None: config['exclude_collection'] = [] config['source_endpoint'] = config['source_config'].get('endpoint').copy() config['source_endpoint'].update(config['source_config']['credentials'][config['org']]) target_org = config.get('org_mapping', {}).get(config.get('org'), config.get('org')) config['target_endpoint'] = config['target_config'].get('endpoint').copy() config['target_endpoint'].update(config['target_config']['credentials'][target_org]) def wait_for(threads, label, sleep_time=60): wait = True logger.info('Starting to wait for [%s] threads with sleep time=[%s]' % (len(threads), sleep_time)) while wait: wait = False alive_count = 0 for t in threads: if t.is_alive(): alive_count += 1 logger.info('Thread [%s] is still alive' % t.name) if alive_count > 0: wait = True logger.info('Continuing to wait for [%s] threads with sleep time=[%s]' % (alive_count, sleep_time)) time.sleep(sleep_time) logger.warn('All workers [%s] done!' % label) def count_bytes(entity): entity_copy = entity.copy() if 'metadata' in entity_copy: del entity_copy['metadata'] entity_str = json.dumps(entity_copy) return len(entity_str) def migrate_user_credentials(app, collection_name, source_entity, attempts=0): # this only applies to users if collection_name not in ['users', 'user'] \ or config.get('skip_credentials', False): return False source_identifier = get_source_identifier(source_entity) target_app, target_collection, target_org = get_target_mapping(app, collection_name) # get the URLs for the source and target users source_url = user_credentials_url_template.format(org=config.get('org'), app=app, uuid=source_identifier, **config.get('source_endpoint')) target_url = user_credentials_url_template.format(org=target_org, app=target_app, uuid=source_identifier, **config.get('target_endpoint')) # this endpoint for some reason uses basic auth... r = requests.get(source_url, auth=HTTPBasicAuth(config.get('su_username'), config.get('su_password'))) if r.status_code != 200: logger.error('Unable to migrate credentials due to HTTP [%s] on GET URL [%s]: %s' % ( r.status_code, source_url, r.text)) return False source_credentials = r.json() logger.info('Putting credentials to [%s]...' % target_url) r = requests.put(target_url, data=json.dumps(source_credentials), auth=HTTPBasicAuth(config.get('su_username'), config.get('su_password'))) if r.status_code != 200: logger.error( 'Unable to migrate credentials due to HTTP [%s] on PUT URL [%s]: %s' % ( r.status_code, target_url, r.text)) return False logger.info('migrate_user_credentials | success=[%s] | app/collection/name = %s/%s/%s' % ( True, app, collection_name, source_entity.get('uuid'))) return True def check_response_status(r, url, exit_on_error=True): if r.status_code != 200: logger.critical('HTTP [%s] on URL=[%s]' % (r.status_code, url)) logger.critical('Response: %s' % r.text) if exit_on_error: exit() def do_operation(apps_and_collections, operation): status_map = {} logger.info('Creating queues...') # Mac, for example, does not support the max_size for a queue in Python if _platform == "linux" or _platform == "linux2": entity_queue = Queue(maxsize=config.get('queue_size_max')) collection_queue = Queue(maxsize=config.get('queue_size_max')) collection_response_queue = Queue(maxsize=config.get('queue_size_max')) else: entity_queue = Queue() collection_queue = Queue() collection_response_queue = Queue() logger.info('Starting entity_workers...') collection_count = 0 # create the entity workers, but only start them (later) if there is work to do entity_workers = [EntityWorker(entity_queue, operation) for x in xrange(config.get('entity_workers'))] # create the collection workers, but only start them (later) if there is work to do collection_workers = [CollectionWorker(collection_queue, entity_queue, collection_response_queue) for x in xrange(config.get('collection_workers'))] status_listener = StatusListener(collection_response_queue, entity_queue) try: # for each app, publish the (app_name, collection_name) to the queue. # this is received by a collection worker who iterates the collection and publishes # entities into a queue. These are received by an individual entity worker which # executes the specified operation on the entity for app, app_data in apps_and_collections.get('apps', {}).iteritems(): logger.info('Processing app=[%s]' % app) status_map[app] = { 'iteration_started': str(datetime.datetime.now()), 'max_created': -1, 'max_modified': -1, 'min_created': 1584946416000, 'min_modified': 1584946416000, 'count': 0, 'bytes': 0, 'collections': {} } # iterate the collections which are returned. for collection_name in app_data.get('collections'): logger.info('Publishing app / collection: %s / %s' % (app, collection_name)) collection_count += 1 collection_queue.put((app, collection_name)) logger.info('Finished publishing [%s] collections for app [%s] !' % (collection_count, app)) # only start the threads if there is work to do if collection_count > 0: status_listener.start() # start the worker processes which will iterate the collections [w.start() for w in collection_workers] # start the worker processes which will do the work of migrating [w.start() for w in entity_workers] # allow collection workers to finish wait_for(collection_workers, label='collection_workers', sleep_time=60) # allow entity workers to finish wait_for(entity_workers, label='entity_workers', sleep_time=60) status_listener.terminate() except KeyboardInterrupt: logger.warning('Keyboard Interrupt, aborting...') entity_queue.close() collection_queue.close() collection_response_queue.close() [os.kill(super(EntityWorker, p).pid, signal.SIGINT) for p in entity_workers] [os.kill(super(CollectionWorker, p).pid, signal.SIGINT) for p in collection_workers] os.kill(super(StatusListener, status_listener).pid, signal.SIGINT) [w.terminate() for w in entity_workers] [w.terminate() for w in collection_workers] status_listener.terminate() logger.info('entity_workers DONE!') def filter_apps_and_collections(org_apps): app_collecitons = { 'apps': { } } try: selected_apps = config.get('app') # iterate the apps retrieved from the org for org_app in sorted(org_apps.keys()): logger.info('Found SOURCE App: %s' % org_app) time.sleep(3) for org_app in sorted(org_apps.keys()): parts = org_app.split('/') app = parts[1] # if apps are specified and the current app is not in the list, skip it if selected_apps and len(selected_apps) > 0 and app not in selected_apps: logger.warning('Skipping app [%s] not included in process list [%s]' % (app, selected_apps)) continue app_collecitons['apps'][app] = { 'collections': [] } # get the list of collections from the source org/app source_app_url = app_url_template.format(org=config.get('org'), app=app, **config.get('source_endpoint')) logger.info('GET %s' % source_app_url) r_collections = session_source.get(source_app_url) collection_attempts = 0 # sometimes this call was not working so I put it in a loop to force it... while r_collections.status_code != 200 and collection_attempts < 5: collection_attempts += 1 logger.warning('FAILED: GET (%s) [%s] URL: %s' % (r_collections.elapsed, r_collections.status_code, source_app_url)) time.sleep(DEFAULT_RETRY_SLEEP) r_collections = session_source.get(source_app_url) if collection_attempts >= 5: logger.critical('Unable to get collections at URL %s, skipping app' % source_app_url) continue app_response = r_collections.json() logger.info('App Response: ' + json.dumps(app_response)) app_entities = app_response.get('entities', []) if len(app_entities) > 0: app_entity = app_entities[0] collections = app_entity.get('metadata', {}).get('collections', {}) logger.info('App=[%s] starting Collections=[%s]' % (app, collections)) app_collecitons['apps'][app]['collections'] = [c for c in collections if include_collection(c)] logger.info('App=[%s] filtered Collections=[%s]' % (app, collections)) except: print(traceback.format_exc()) return app_collecitons def confirm_target_org_apps(apps_and_collections): for app in apps_and_collections.get('apps'): # it is possible to map source orgs and apps to differently named targets. This gets the # target names for each target_org = config.get('org_mapping', {}).get(config.get('org'), config.get('org')) target_app = config.get('app_mapping', {}).get(app, app) # Check that the target Org/App exists. If not, move on to the next target_app_url = app_url_template.format(org=target_org, app=target_app, **config.get('target_endpoint')) logger.info('GET %s' % target_app_url) r_target_apps = session_target.get(target_app_url) if r_target_apps.status_code != 200: if config.get('create_apps', DEFAULT_CREATE_APPS): create_app_url = org_management_app_url_template.format(org=target_org, app=target_app, **config.get('target_endpoint')) app_request = {'name': target_app} r = session_target.post(create_app_url, data=json.dumps(app_request)) if r.status_code != 200: logger.critical('--create_apps specified and unable to create app [%s] at URL=[%s]: %s' % ( target_app, create_app_url, r.text)) logger.critical('Process will now exit') exit() else: logger.warning('Created app=[%s] at URL=[%s]: %s' % (target_app, create_app_url, r.text)) else: logger.critical('Target application DOES NOT EXIST at [%s] URL=%s' % ( r_target_apps.status_code, target_app_url)) continue def main(): global config, cache config = parse_args() init() init_logging() logger.warn('Script starting') try: if config.get('redis_socket') is not None: cache = redis.Redis(unix_socket_path=config.get('redis_socket')) else: # this does not try to connect to redis cache = redis.StrictRedis(host='localhost', port=6379, db=0) # this is necessary to test the connection to redis cache.get('usergrid') except: logger.error( 'Error connecting to Redis cache, consider using Redis to be able to optimize the migration process...') logger.error( 'Error connecting to Redis cache, consider using Redis to be able to optimize the migration process...') time.sleep(3) config['use_cache'] = False config['skip_cache_read'] = True config['skip_cache_write'] = True org_apps = { } force_apps = config.get('force_app', []) if force_apps is not None and len(force_apps) > 0: logger.warn('Forcing only the following apps to be processed: %s' % force_apps) for app in force_apps: key = '%s/%s' % (app, app) org_apps[key] = app if len(org_apps) == 0: source_org_mgmt_url = org_management_url_template.format(org=config.get('org'), limit=config.get('limit'), **config.get('source_endpoint')) print('Retrieving apps from [%s]' % source_org_mgmt_url) logger.info('Retrieving apps from [%s]' % source_org_mgmt_url) try: # list the apps for the SOURCE org logger.info('GET %s' % source_org_mgmt_url) r = session_source.get(source_org_mgmt_url) if r.status_code != 200: logger.critical( 'Abort processing: Unable to retrieve apps from [%s]: %s' % (source_org_mgmt_url, r.text)) exit() logger.info(json.dumps(r.text)) org_apps = r.json().get('data') except Exception: logger.exception('ERROR Retrieving apps from [%s]' % source_org_mgmt_url) print(traceback.format_exc()) logger.critical('Unable to retrieve apps from [%s] and will exit' % source_org_mgmt_url) exit() # Check the specified configuration for what to migrate/audit if config.get('migrate') == 'graph': operation = migrate_graph elif config.get('migrate') == 'data': operation = migrate_data elif config.get('migrate') == 'prune': operation = prune_graph elif config.get('migrate') == 'permissions': operation = migrate_permissions config['collection'] = ['roles', 'groups'] logger.warn( 'Since permissions migration was specified, overwriting included collections to be %s...' % config[ 'collection']) elif config.get('migrate') == 'credentials': operation = migrate_user_credentials config['collection'] = ['users'] logger.warn('Since credential migration was specified, overwriting included collections to be %s' % config[ 'collection']) elif config.get('migrate') == 'reput': operation = reput else: operation = None # filter out the apps and collections based on the -c and --exclude_collection directives apps_and_collections = filter_apps_and_collections(org_apps) logger.warn('The following apps/collections will be processed: %s' % json.dumps(apps_and_collections)) # confirm the apps exist at the target/destination org confirm_target_org_apps(apps_and_collections) # execute the operation over apps and collections do_operation(apps_and_collections, operation) logger.warn('Script finished') if __name__ == "__main__": main()