utils/usergrid-util-python/usergrid_tools/migration/usergrid_data_exporter.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.
# */

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
import time_uuid

import datetime
from cloghandler import ConcurrentRotatingFileHandler
import requests
import traceback
import time
from sys import platform as _platform

import signal

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()


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 = '%s/migrator.log' % config.get('log_dir')

    # 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 = '%s/migrator_errors.log' % config.get('log_dir')
    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}"
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}"

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

        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'].items():
                        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'].items():

                                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))

                except KeyboardInterrupt as e:
                    raise e

                except:
                    print(traceback.format_exc())

            except KeyboardInterrupt as 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))


class EntityExportWorker(Process):
    def __init__(self, work_queue, response_queue):
        super(EntityExportWorker, self).__init__()
        collection_worker_logger.debug('Creating worker!')
        self.work_queue = work_queue
        self.response_queue = response_queue

    def run(self):

        collection_worker_logger.info('starting run()...')
        keep_going = True

        empty_count = 0
        app = 'NOT SET'
        collection_name = 'NOT SET'
        status_map = {}
        entity_file = None

        try:
            while keep_going:

                try:
                    app, collection_name = self.work_queue.get(timeout=30)
                    empty_count = 0

                    status_map = self.process_collection(app, collection_name)

                    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 as 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:
            if entity_file is not None:
                entity_file.close()

            self.response_queue.put((app, collection_name, status_map))
            collection_worker_logger.info('FINISHED!')

    def process_collection(self, app, collection_name):

        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
            }
        }

        # 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'))
        counter = 0

        # 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'))

        directory = os.path.join(config['export_path'], ECID, config['org'], app)

        if not os.path.exists(directory):
            os.makedirs(directory)

        entity_filename = '_'.join([collection_name, 'entity-data'])
        entity_filename_base = os.path.join(directory, entity_filename)
        entity_file_number = 0
        entity_file_counter = 0
        entity_filename = '%s-%s.txt' % (entity_filename_base, entity_file_number)
        entity_file = open(entity_filename, 'w')

        edge_filename = '_'.join([collection_name, 'edge-data'])
        edge_filename_base = os.path.join(directory, edge_filename)
        edge_file_number = 0
        edge_file_counter = 0
        edge_filename = '%s-%s.txt' % (edge_filename_base, edge_file_number)
        edge_file = open(edge_filename, 'w')

        try:

            for entity in q:
                try:
                    entity_file_counter += 1
                    counter += 1

                    if entity_file_counter > config['entities_per_file']:
                        entity_file.close()
                        entity_file_number += 1
                        entity_file_counter = 0
                        entity_filename = '%s-%s.txt' % (entity_filename_base, entity_file_number)
                        entity_file = open(entity_filename, 'w')

                    entity_file.write('%s\n' % json.dumps(entity))

                    edge_names = get_edge_names(entity)

                    for edge_name in edge_names:
                        if not include_edge(collection_name, edge_name):
                            continue

                        connection_query_url = connection_query_url_template.format(
                                org=config.get('org'),
                                app=app,
                                verb=edge_name,
                                collection=collection_name,
                                uuid=entity.get('uuid'),
                                limit=config.get('limit'),
                                **config.get('source_endpoint'))

                        connection_query = UsergridQueryIterator(connection_query_url,
                                                                 sleep_time=config.get('error_retry_sleep'))

                        target_uuids = []

                        try:
                            for target_entity in connection_query:
                                target_uuids.append(target_entity.get('uuid'))
                        except:
                            logger.exception('Error processing edge [%s] of entity [ %s / %s / %s]' % (
                                edge_name, app, collection_name, entity.get('uuid')))

                        if len(target_uuids) > 0:
                            edge_file_counter += 1

                            edges = {
                                'entity': {
                                    'type': entity.get('type'),
                                    'uuid': entity.get('uuid')
                                },
                                'edge_name': edge_name,
                                'target_uuids': target_uuids
                            }

                            if entity_file_counter > config['entities_per_file']:
                                edge_file.close()
                                edge_file_number += 1
                                edge_file_counter = 0
                                edge_filename = '%s-%s.txt' % (edge_filename_base, edge_file_number)
                                edge_file = open(edge_filename, 'w')

                            edge_file.write('%s\n' % json.dumps(edges))

                    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 incremental 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))
                except KeyboardInterrupt:
                    raise

                except:
                    logger.exception(
                            'Error processing entity %s / %s / %s' % (app, collection_name, entity.get('uuid')))

        except KeyboardInterrupt:
            raise

        except:
            logger.exception('Error processing collection %s / %s ' % (app, collection_name))

        finally:
            if edge_file is not None:
                edge_file.close()

            if entity_file is not None:
                entity_file.close()

        return status_map


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 ['roles', 'followers', 'groups',
                                                               'feed', 'activities']) \
            or (collection_name in ['device', 'devices'] and edge_name in ['users']) \
            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 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):
    exclude = config.get('exclude_collection', [])

    if exclude is not None and collection_name in exclude:
        return False

    return True


def get_edge_names(entity):
    out_edge_names = [edge_name for edge_name in entity.get('metadata', {}).get('collections', [])]
    out_edge_names += [edge_name for edge_name in entity.get('metadata', {}).get('connections', [])]

    return out_edge_names


def get_uuid_time(the_uuid_string):
    return time_uuid.TimeUUID(the_uuid_string).get_datetime()


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('-s', '--source_config',
                        help='The path to the source endpoint/org configuration file',
                        type=str,
                        default='source.json')

    parser.add_argument('--export_path',
                        help='The path to save the export files',
                        type=str,
                        default='.')

    parser.add_argument('--limit',
                        help='The number of entities to return per query request',
                        type=int,
                        default=100)

    parser.add_argument('--entities_per_file',
                        help='The number of entities to put in one JSON file',
                        type=int,
                        default=10000)

    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=.5)

    parser.add_argument('--entity_sleep_time',
                        help='The number of seconds to wait between retrieving pages from the UsergridQueryIterator',
                        type=float,
                        default=.1)

    parser.add_argument('--workers',
                        dest='collection_workers',
                        help='The number of worker processes to do the migration',
                        type=int,
                        default=4)

    parser.add_argument('--queue_size_max',
                        help='The max size of entities to allow in the queue',
                        type=int,
                        default=100000)

    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('--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')

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

    return vars(my_args)


def init():
    global config

    config['collection_mapping'] = {}
    config['app_mapping'] = {}
    config['org_mapping'] = {}

    with open(config.get('source_config'), 'r') as f:
        config['source_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']])


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 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 main():
    global config

    config = parse_args()
    init()
    init_logging()

    status_map = {}

    org_apps = {
    }

    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 as e:
            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()

    if _platform == "linux" or _platform == "linux2":
        collection_queue = Queue(maxsize=config.get('queue_size_max'))
        collection_response_queue = Queue(maxsize=config.get('queue_size_max'))
    else:
        collection_queue = Queue()
        collection_response_queue = Queue()

    logger.info('Starting entity_workers...')

    status_listener = StatusListener(collection_response_queue, collection_queue)
    status_listener.start()

    # start the worker processes which will iterate the collections
    collection_workers = [EntityExportWorker(collection_queue, collection_response_queue) for x in
                          xrange(config.get('collection_workers'))]
    [w.start() for w in collection_workers]

    try:
        apps_to_process = config.get('app')
        collections_to_process = config.get('collection')

        # 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 apps_to_process and len(apps_to_process) > 0 and app not in apps_to_process:
                logger.warning('Skipping app [%s] not included in process list [%s]' % (app, apps_to_process))
                continue

            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': {}
            }

            # 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('Collection List: %s' % collections)

                # iterate the collections which are returned.
                for collection_name, collection_data in collections.iteritems():
                    exclude_collections = config.get('exclude_collection', [])

                    if exclude_collections is None:
                        exclude_collections = []

                    # filter out collections as configured...
                    if collection_name in ignore_collections \
                            or (len(collections_to_process) > 0 and collection_name not in collections_to_process) \
                            or (len(exclude_collections) > 0 and collection_name in exclude_collections) \
                            or (config.get('migrate') == 'credentials' and collection_name != 'users'):
                        logger.warning('Skipping collection=[%s]' % collection_name)

                        continue

                    logger.info('Publishing app / collection: %s / %s' % (app, collection_name))

                    collection_queue.put((app, collection_name))

            status_map[app]['iteration_finished'] = str(datetime.datetime.now())

            logger.info('Finished publishing collections for app [%s] !' % app)

        # allow collection workers to finish
        wait_for(collection_workers, label='collection_workers', sleep_time=30)

        status_listener.terminate()

    except KeyboardInterrupt:
        logger.warning('Keyboard Interrupt, aborting...')
        collection_queue.close()
        collection_response_queue.close()

        [os.kill(super(EntityExportWorker, p).pid, signal.SIGINT) for p in collection_workers]
        os.kill(super(StatusListener, status_listener).pid, signal.SIGINT)

        [w.terminate() for w in collection_workers]
        status_listener.terminate()

    logger.info('entity_workers DONE!')


if __name__ == "__main__":
    main()