blob: 2832691ee999ffe990ebdc5683c9bb2c9c760c85 [file] [log] [blame]
#!/usr/bin/python
'''
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.
'''
import copy
import glob
import logging
import os
import sys
import urllib2, ssl
import json
import base64
import optparse
import socket
import time
import traceback
import ConfigParser
from random import randrange
from subprocess import Popen, PIPE
HTTP_PROTOCOL = 'http'
HTTPS_PROTOCOL = 'https'
AMBARI_SUDO = "/var/lib/ambari-agent/ambari-sudo.sh"
SOLR_SERVICE_NAME = 'AMBARI_INFRA_SOLR'
SOLR_COMPONENT_NAME ='INFRA_SOLR'
LOGSEARCH_SERVICE_NAME = 'LOGSEARCH'
LOGSEARCH_SERVER_COMPONENT_NAME ='LOGSEARCH_SERVER'
LOGSEARCH_LOGFEEDER_COMPONENT_NAME ='LOGSEARCH_LOGFEEDER'
CLUSTERS_URL = '/api/v1/clusters/{0}'
GET_HOSTS_COMPONENTS_URL = '/services/{0}/components/{1}?fields=host_components'
REQUESTS_API_URL = '/requests'
BATCH_REQUEST_API_URL = "/api/v1/clusters/{0}/request_schedules"
LIST_SOLR_COLLECTION_URL = '{0}/admin/collections?action=LIST&wt=json'
CREATE_SOLR_COLLECTION_URL = '{0}/admin/collections?action=CREATE&name={1}&collection.configName={2}&numShards={3}&replicationFactor={4}&maxShardsPerNode={5}&wt=json'
DELETE_SOLR_COLLECTION_URL = '{0}/admin/collections?action=DELETE&name={1}&wt=json'
RELOAD_SOLR_COLLECTION_URL = '{0}/admin/collections?action=RELOAD&name={1}&wt=json'
INFRA_SOLR_CLIENT_BASE_PATH = '/usr/lib/ambari-infra-solr-client/'
RANGER_NEW_SCHEMA = 'migrate/managed-schema'
SOLR_CLOUD_CLI_SCRIPT = 'solrCloudCli.sh'
COLLECTIONS_DATA_JSON_LOCATION = INFRA_SOLR_CLIENT_BASE_PATH + "migrate/data/{0}"
logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = logging.Formatter("%(asctime)s - %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
class colors:
OKGREEN = '\033[92m'
WARNING = '\033[38;5;214m'
FAIL = '\033[91m'
ENDC = '\033[0m'
def api_accessor(host, username, password, protocol, port):
def do_request(api_url, request_type, request_body=''):
try:
url = '{0}://{1}:{2}{3}'.format(protocol, host, port, api_url)
logger.debug('Execute {0} {1}'.format(request_type, url))
if request_body:
logger.debug('Request body: {0}'.format(request_body))
admin_auth = base64.encodestring('%s:%s' % (username, password)).replace('\n', '')
request = urllib2.Request(url)
request.add_header('Authorization', 'Basic %s' % admin_auth)
request.add_header('X-Requested-By', 'ambari')
request.add_data(request_body)
request.get_method = lambda: request_type
response = None
if protocol == 'https':
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
response = urllib2.urlopen(request, context=ctx)
else:
response = urllib2.urlopen(request)
response_body = response.read()
except Exception as exc:
raise Exception('Problem with accessing api. Reason: {0}'.format(exc))
return response_body
return do_request
def set_log_level(verbose):
if verbose:
logger.setLevel(logging.DEBUG)
else:
logger.setLevel(logging.INFO)
def retry(func, *args, **kwargs):
retry_count = kwargs.pop("count", 10)
delay = kwargs.pop("delay", 5)
context = kwargs.pop("context", "")
for r in range(retry_count):
try:
result = func(*args, **kwargs)
if result is not None: return result
except Exception as e:
logger.error("Error occurred during {0} operation: {1}".format(context, str(traceback.format_exc())))
logger.info("\n{0}: waiting for {1} seconds before retyring again (retry count: {2})".format(context, delay, r+1))
time.sleep(delay)
print '{0} operation {1}FAILED{2}'.format(context, colors.FAIL, colors.ENDC)
exit(1)
def create_solr_api_request_command(request_url, config, output=None):
user='infra-solr'
kerberos_enabled='false'
keytab=None
principal=None
if config.has_section('cluster') and config.has_option('cluster', 'kerberos_enabled'):
kerberos_enabled=config.get('cluster', 'kerberos_enabled')
if config.has_section('infra_solr'):
if config.has_option('infra_solr', 'user'):
user=config.get('infra_solr', 'user')
if kerberos_enabled == 'true':
if config.has_option('infra_solr', 'keytab'):
keytab=config.get('infra_solr', 'keytab')
if config.has_option('infra_solr', 'principal'):
principal=config.get('infra_solr', 'principal')
use_infra_solr_user="sudo -u {0}".format(user)
curl_prefix = "curl -k"
if output is not None:
curl_prefix+=" -o {0}".format(output)
api_cmd = '{0} kinit -kt {1} {2} && {3} {4} --negotiate -u : "{5}"'.format(use_infra_solr_user, keytab, principal, use_infra_solr_user, curl_prefix, request_url) \
if kerberos_enabled == 'true' else '{0} {1} "{2}"'.format(use_infra_solr_user, curl_prefix, request_url)
logger.debug("Solr API command: {0}".format(api_cmd))
return api_cmd
def create_infra_solr_client_command(options, config, command, appendZnode=False):
user='infra-solr'
kerberos_enabled='false'
infra_solr_cli_opts = ''
java_home=None
jaasOption=None
zkConnectString=None
if config.has_section('cluster') and config.has_option('cluster', 'kerberos_enabled'):
kerberos_enabled=config.get('cluster', 'kerberos_enabled')
if config.has_section('infra_solr'):
if config.has_option('infra_solr', 'user'):
user=config.get('infra_solr', 'user')
if config.has_option('infra_solr', 'external_zk_connect_string'):
zkConnectString=config.get('infra_solr', 'external_zk_connect_string')
elif config.has_option('infra_solr', 'zk_connect_string'):
zkConnectString=config.get('infra_solr', 'zk_connect_string')
if kerberos_enabled == 'true':
zk_principal_user = config.get('infra_solr', 'zk_principal_user') if config.has_option('infra_solr', 'zk_principal_user') else 'zookeeper'
infra_solr_cli_opts= '-Dzookeeper.sasl.client=true -Dzookeeper.sasl.client.username={0} -Dzookeeper.sasl.clientconfig=Client'.format(zk_principal_user)
jaasOption=" --jaas-file /etc/ambari-infra-solr/conf/infra_solr_jaas.conf"
command+=jaasOption
if config.has_section('local') and config.has_option('local', 'java_home'):
java_home=config.get('local', 'java_home')
if not java_home:
raise Exception("'local' section or 'java_home' is missing (or empty) from the configuration")
if not zkConnectString:
raise Exception("'zk_connect_string' section or 'external_zk_connect_string' is missing (or empty) from the configuration")
if appendZnode:
if config.has_option('infra_solr', 'znode'):
znode_to_append=config.get('infra_solr', 'znode')
zkConnectString+="{0}".format(znode_to_append)
else:
raise Exception("'znode' option is required for infra_solr section")
set_java_home_= 'JAVA_HOME={0}'.format(java_home)
set_infra_solr_cli_opts = ' INFRA_SOLR_CLI_OPTS="{0}"'.format(infra_solr_cli_opts) if infra_solr_cli_opts != '' else ''
solr_cli_cmd = '{0} {1}{2} /usr/lib/ambari-infra-solr-client/solrCloudCli.sh --zookeeper-connect-string {3} {4}'\
.format(AMBARI_SUDO, set_java_home_, set_infra_solr_cli_opts, zkConnectString, command)
return solr_cli_cmd
def get_random_solr_url(solr_urls, options):
random_index = randrange(0, len(solr_urls))
result = solr_urls[random_index]
logger.debug("Use {0} solr address for next request.".format(result))
return result
def format_json(dictionary, tab_level=0):
output = ''
tab = ' ' * 2 * tab_level
for key, value in dictionary.iteritems():
output += ',\n{0}"{1}": '.format(tab, key)
if isinstance(value, dict):
output += '{\n' + format_json(value, tab_level + 1) + tab + '}'
else:
output += '"{0}"'.format(value)
output += '\n'
return output[2:]
def read_json(json_file):
with open(json_file) as data_file:
data = json.load(data_file)
return data
def get_json(accessor, url):
response = accessor(url, 'GET')
logger.debug('GET ' + url + ' response: ')
logger.debug('----------------------------')
logger.debug(str(response))
json_resp = json.loads(response)
return json_resp
def post_json(accessor, url, request_body):
response = accessor(url, 'POST', json.dumps(request_body))
logger.debug('POST ' + url + ' response: ')
logger.debug( '----------------------------')
logger.debug(str(response))
json_resp = json.loads(response)
return json_resp
def get_component_hosts(host_components_json):
hosts = []
if "host_components" in host_components_json and len(host_components_json['host_components']) > 0:
for host_component in host_components_json['host_components']:
if 'HostRoles' in host_component:
hosts.append(host_component['HostRoles']['host_name'])
return hosts
def create_batch_command(command, hosts, cluster, service_name, component_name, interval_seconds, fault_tolerance, context):
request_schedules = []
request_schedule = {}
batch = []
requests = []
order_id = 1
all = len(hosts)
for host in hosts:
request = {}
request['order_id'] = order_id
request['type'] = 'POST'
request['uri'] = "/clusters/{0}/requests".format(cluster)
request_body_info = {}
request_info = {}
request_info["context"] = context + " ({0} of {1})".format(order_id, all)
request_info["command"] = command
order_id = order_id + 1
resource_filter = {}
resource_filter["service_name"] = service_name
resource_filter["component_name"] = component_name
resource_filter["hosts"] = host
resource_filters = []
resource_filters.append(resource_filter)
request_body_info["Requests/resource_filters"] = resource_filters
request_body_info['RequestInfo'] = request_info
request['RequestBodyInfo'] = request_body_info
requests.append(request)
batch_requests_item = {}
batch_requests_item['requests'] = requests
batch.append(batch_requests_item)
batch_settings_item = {}
batch_settings = {}
batch_settings['batch_separation_in_seconds'] = interval_seconds
batch_settings['task_failure_tolerance'] = fault_tolerance
batch_settings_item['batch_settings'] = batch_settings
batch.append(batch_settings_item)
request_schedule['batch'] = batch
request_schedule_item = {}
request_schedule_item['RequestSchedule'] = request_schedule
request_schedules.append(request_schedule_item)
return request_schedules
def create_command_request(command, parameters, hosts, cluster, context, service=SOLR_SERVICE_NAME, component=SOLR_COMPONENT_NAME):
request = {}
request_info = {}
request_info["context"] = context
request_info["command"] = command
request_info["parameters"] = parameters
operation_level = {}
operation_level["level"] = "HOST_COMPONENT"
operation_level["cluster_name"] = cluster
request_info["operation_level"] = operation_level
request["RequestInfo"] = request_info
resource_filter = {}
resource_filter["service_name"] = service
resource_filter["component_name"] = component
resource_filter["hosts"] = ','.join(hosts)
resource_filters = []
resource_filters.append(resource_filter)
request["Requests/resource_filters"] = resource_filters
return request
def fill_params_for_backup(params, collection):
collections_data = get_collections_data(COLLECTIONS_DATA_JSON_LOCATION.format("backup_collections.json"))
if collection in collections_data and 'leaderHostCoreMap' in collections_data[collection]:
params["solr_backup_host_cores_map"] = json.dumps(collections_data[collection]['leaderHostCoreMap'])
if collection in collections_data and 'leaderCoreHostMap' in collections_data[collection]:
params["solr_backup_core_host_map"] = json.dumps(collections_data[collection]['leaderCoreHostMap'])
return params
def fill_params_for_restore(params, original_collection, collection, config_set):
backup_collections_data = get_collections_data(COLLECTIONS_DATA_JSON_LOCATION.format("backup_collections.json"))
if original_collection in backup_collections_data and 'leaderHostCoreMap' in backup_collections_data[original_collection]:
params["solr_backup_host_cores_map"] = json.dumps(backup_collections_data[original_collection]['leaderHostCoreMap'])
if original_collection in backup_collections_data and 'leaderCoreHostMap' in backup_collections_data[original_collection]:
params["solr_backup_core_host_map"] = json.dumps(backup_collections_data[original_collection]['leaderCoreHostMap'])
collections_data = get_collections_data(COLLECTIONS_DATA_JSON_LOCATION.format("restore_collections.json"))
if collection in collections_data and 'leaderHostCoreMap' in collections_data[collection]:
params["solr_restore_host_cores_map"] = json.dumps(collections_data[collection]['leaderHostCoreMap'])
if collection in collections_data and 'leaderCoreHostMap' in collections_data[collection]:
params["solr_restore_core_host_map"] = json.dumps(collections_data[collection]['leaderCoreHostMap'])
if collection in collections_data and 'leaderSolrCoreDataMap' in collections_data[collection]:
params["solr_restore_core_data"] = json.dumps(collections_data[collection]['leaderSolrCoreDataMap'])
if config_set:
params["solr_restore_config_set"] = config_set
return params
def fill_parameters(options, config, collection, index_location, hdfs_path=None, shards=None):
params = {}
if collection:
params['solr_collection'] = collection
params['solr_backup_name'] = collection
if index_location:
params['solr_index_location'] = index_location
if options.index_version:
params['solr_index_version'] = options.index_version
if options.force:
params['solr_index_upgrade_force'] = options.force
if options.async:
params['solr_request_async'] = options.request_async
if options.request_tries:
params['solr_request_tries'] = options.request_tries
if options.request_time_interval:
params['solr_request_time_interval'] = options.request_time_interval
if options.disable_solr_host_check:
params['solr_check_hosts'] = False
if options.core_filter:
params['solr_core_filter'] = options.core_filter
if options.core_filter:
params['solr_skip_cores'] = options.skip_cores
if shards:
params['solr_shards'] = shards
if options.shared_drive:
params['solr_shared_fs'] = True
elif config.has_section('local') and config.has_option('local', 'shared_drive') and config.get('local', 'shared_drive') == 'true':
params['solr_shared_fs'] = True
if hdfs_path:
params['solr_hdfs_path'] = hdfs_path
if options.keep_backup:
params['solr_keep_backup'] = True
return params
def validte_common_options(options, parser, config):
if not options.index_location:
parser.print_help()
print 'index-location option is required'
sys.exit(1)
if not options.collection:
parser.print_help()
print 'collection option is required'
sys.exit(1)
def get_service_components(options, accessor, cluster, service, component):
host_components_json = get_json(accessor, CLUSTERS_URL.format(cluster) + GET_HOSTS_COMPONENTS_URL.format(service, component))
component_hosts = get_component_hosts(host_components_json)
return component_hosts
def get_solr_hosts(options, accessor, cluster):
component_hosts = get_service_components(options, accessor, cluster, SOLR_SERVICE_NAME, SOLR_COMPONENT_NAME)
if options.include_solr_hosts:
new_component_hosts = []
include_solr_hosts_list = options.include_solr_hosts.split(',')
for include_host in include_solr_hosts_list:
if include_host in component_hosts:
new_component_hosts.append(include_host)
component_hosts = new_component_hosts
if options.exclude_solr_hosts:
exclude_solr_hosts_list = options.exclude_solr_hosts.split(',')
for exclude_host in exclude_solr_hosts_list:
if exclude_host in component_hosts:
component_hosts.remove(exclude_host)
return component_hosts
def restore(options, accessor, parser, config, original_collection, collection, config_set, index_location, hdfs_path, shards):
"""
Send restore solr collection custom command request to ambari-server
"""
cluster = config.get('ambari_server', 'cluster')
component_hosts = get_solr_hosts(options, accessor, cluster)
parameters = fill_parameters(options, config, collection, index_location, hdfs_path, shards)
parameters = fill_params_for_restore(parameters, original_collection, collection, config_set)
cmd_request = create_command_request("RESTORE", parameters, component_hosts, cluster, 'Restore Solr Collection: ' + collection)
return post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
def migrate(options, accessor, parser, config, collection, index_location):
"""
Send migrate lucene index custom command request to ambari-server
"""
cluster = config.get('ambari_server', 'cluster')
component_hosts = get_solr_hosts(options, accessor, cluster)
parameters = fill_parameters(options, config, collection, index_location)
cmd_request = create_command_request("MIGRATE", parameters, component_hosts, cluster, 'Migrating Solr Collection: ' + collection)
return post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
def backup(options, accessor, parser, config, collection, index_location):
"""
Send backup solr collection custom command request to ambari-server
"""
cluster = config.get('ambari_server', 'cluster')
component_hosts = get_solr_hosts(options, accessor, cluster)
parameters = fill_parameters(options, config, collection, index_location)
parameters = fill_params_for_backup(parameters, collection)
cmd_request = create_command_request("BACKUP", parameters, component_hosts, cluster, 'Backup Solr Collection: ' + collection)
return post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
def upgrade_solr_instances(options, accessor, parser, config):
"""
Upgrade (remove & re-install) infra solr instances
"""
cluster = config.get('ambari_server', 'cluster')
solr_instance_hosts = get_service_components(options, accessor, cluster, "AMBARI_INFRA_SOLR", "INFRA_SOLR")
context = "Upgrade Solr Instances"
sys.stdout.write("Sending upgrade request: [{0}] ".format(context))
sys.stdout.flush()
cmd_request = create_command_request("UPGRADE_SOLR_INSTANCE", {}, solr_instance_hosts, cluster, context)
response = post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Upgrade command request id: {0}'.format(request_id)
if options.async:
print "Upgrade request sent to Ambari server. Check Ambari UI about the results."
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, context)
print "{0}... {1} DONE{2}".format(context, colors.OKGREEN, colors.ENDC)
def upgrade_solr_clients(options, accessor, parser, config):
"""
Upgrade (remove & re-install) infra solr clients
"""
cluster = config.get('ambari_server', 'cluster')
solr_client_hosts = get_service_components(options, accessor, cluster, "AMBARI_INFRA_SOLR", "INFRA_SOLR_CLIENT")
fqdn = socket.getfqdn()
if fqdn in solr_client_hosts:
solr_client_hosts.remove(fqdn)
host = socket.gethostname()
if host in solr_client_hosts:
solr_client_hosts.remove(host)
context = "Upgrade Solr Clients"
sys.stdout.write("Sending upgrade request: [{0}] ".format(context))
sys.stdout.flush()
cmd_request = create_command_request("UPGRADE_SOLR_CLIENT", {}, solr_client_hosts, cluster, context, component="INFRA_SOLR_CLIENT")
response = post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Upgrade command request id: {0}'.format(request_id)
if options.async:
print "Upgrade request sent to Ambari server. Check Ambari UI about the results."
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, context)
print "{0}... {1}DONE{2}".format(context, colors.OKGREEN, colors.ENDC)
def upgrade_logfeeders(options, accessor, parser, config):
"""
Upgrade (remove & re-install) logfeeders
"""
cluster = config.get('ambari_server', 'cluster')
logfeeder_hosts = get_service_components(options, accessor, cluster, "LOGSEARCH", "LOGSEARCH_LOGFEEDER")
context = "Upgrade Log Feeders"
sys.stdout.write("Sending upgrade request: [{0}] ".format(context))
sys.stdout.flush()
cmd_request = create_command_request("UPGRADE_LOGFEEDER", {}, logfeeder_hosts, cluster, context, service="LOGSEARCH", component="LOGSEARCH_LOGFEEDER")
response = post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Upgrade command request id: {0}'.format(request_id)
if options.async:
print "Upgrade request sent to Ambari server. Check Ambari UI about the results."
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, context)
print "{0}... {1} DONE{2}".format(context, colors.OKGREEN, colors.ENDC)
def upgrade_logsearch_portal(options, accessor, parser, config):
"""
Upgrade (remove & re-install) logsearch server instances
"""
cluster = config.get('ambari_server', 'cluster')
logsearch_portal_hosts = get_service_components(options, accessor, cluster, "LOGSEARCH", "LOGSEARCH_SERVER")
context = "Upgrade Log Search Portal"
sys.stdout.write("Sending upgrade request: [{0}] ".format(context))
sys.stdout.flush()
cmd_request = create_command_request("UPGRADE_LOGSEARCH_PORTAL", {}, logsearch_portal_hosts, cluster, context, service="LOGSEARCH", component="LOGSEARCH_SERVER")
response = post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Upgrade command request id: {0}'.format(request_id)
if options.async:
print "Upgrade request sent to Ambari server. Check Ambari UI about the results."
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, context)
print "{0}... {1} DONE{2}".format(context, colors.OKGREEN, colors.ENDC)
def service_components_command(options, accessor, parser, config, service, component, command, command_str):
"""
Run command on service components
"""
cluster = config.get('ambari_server', 'cluster')
service_components = get_service_components(options, accessor, cluster, service, component)
context = "{0} {1}".format(command_str, component)
sys.stdout.write("Sending '{0}' request: [{1}] ".format(command, context))
sys.stdout.flush()
cmd_request = create_command_request(command, {}, service_components, cluster, context, service=service, component=component)
response = post_json(accessor, CLUSTERS_URL.format(cluster) + REQUESTS_API_URL, cmd_request)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print '{0} command request id: {1}'.format(command_str, request_id)
if options.async:
print "{0} request sent to Ambari server. Check Ambari UI about the results.".format(command_str)
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, context)
print "{0}... {1} DONE{2}".format(context, colors.OKGREEN, colors.ENDC)
def monitor_request(options, accessor, cluster, request_id, context):
while True:
request_response=get_json(accessor, "/api/v1/clusters/{0}{1}/{2}".format(cluster, REQUESTS_API_URL, request_id))
if 'Requests' in request_response and 'request_status' in request_response['Requests']:
request_status = request_response['Requests']['request_status']
logger.debug("\nMonitoring '{0}' request (id: '{1}') status is {2}".format(context, request_id, request_status))
if request_status in ['FAILED', 'TIMEDOUT', 'ABORTED', 'COMPLETED', 'SKIPPED_FAILED']:
if request_status == 'COMPLETED':
print "\nRequest (id: {0}) {1}COMPLETED{2}".format(request_id, colors.OKGREEN, colors.ENDC)
time.sleep(4)
else:
print "\nRequest (id: {0}) {1}FAILED{2} (checkout Ambari UI about the failed tasks)\n".format(request_id, colors.FAIL, colors.ENDC)
sys.exit(1)
break
else:
if not options.verbose:
sys.stdout.write(".")
sys.stdout.flush()
logger.debug("Sleep 5 seconds ...")
time.sleep(5)
else:
print "'Requests' or 'request_status' cannot be found in JSON response: {0}".format(request_response)
sys.exit(1)
def get_request_id(json_response):
if "Requests" in json_response:
if "id" in json_response['Requests']:
return json_response['Requests']['id']
raise Exception("Cannot access request id from Ambari response: {0}".format(json_response))
def filter_collections(options, collections):
if options.collection is not None:
filtered_collections = []
if options.collection in collections:
filtered_collections.append(options.collection)
return filtered_collections
else:
return collections
def get_solr_urls(options, config, collection, collections_json):
solr_urls = []
solr_hosts = None
solr_port = "8886"
solr_protocol = "http"
if config.has_section("infra_solr") and config.has_option("infra_solr", "port"):
solr_port = config.get('infra_solr', 'port')
if config.has_section("infra_solr") and config.has_option("infra_solr", "protocol"):
solr_protocol = config.get('infra_solr', 'protocol')
if config.has_section('infra_solr') and config.has_option('infra_solr', 'hosts'):
solr_hosts = config.get('infra_solr', 'hosts')
splitted_solr_hosts = solr_hosts.split(',')
if options.include_solr_hosts:
# keep only included ones, do not override any
include_solr_hosts_list = options.include_solr_hosts.split(',')
new_splitted_hosts = []
for host in splitted_solr_hosts:
if any(inc_solr_host in host for inc_solr_host in include_solr_hosts_list):
new_splitted_hosts.append(host)
splitted_solr_hosts = new_splitted_hosts
if options.exclude_solr_hosts:
exclude_solr_hosts_list = options.exclude_solr_hosts.split(',')
hosts_to_exclude = []
for host in splitted_solr_hosts:
if any(exc_solr_host in host for exc_solr_host in exclude_solr_hosts_list):
hosts_to_exclude.append(host)
for excluded_url in hosts_to_exclude:
splitted_solr_hosts.remove(excluded_url)
for solr_host in splitted_solr_hosts:
solr_addr = "{0}://{1}:{2}/solr".format(solr_protocol, solr_host, solr_port)
solr_urls.append(solr_addr)
return solr_urls
def is_atlas_available(config, service_filter):
return 'ATLAS' in service_filter and config.has_section('atlas_collections') \
and config.has_option('atlas_collections', 'enabled') and config.get('atlas_collections', 'enabled') == 'true'
def is_ranger_available(config, service_filter):
return 'RANGER' in service_filter and config.has_section('ranger_collection') \
and config.has_option('ranger_collection', 'enabled') and config.get('ranger_collection', 'enabled') == 'true'
def is_logsearch_available(config, service_filter):
return 'LOGSEARCH' in service_filter and config.has_section('logsearch_collections') \
and config.has_option('logsearch_collections', 'enabled') and config.get('logsearch_collections', 'enabled') == 'true'
def delete_collection(options, config, collection, solr_urls):
request = DELETE_SOLR_COLLECTION_URL.format(get_random_solr_url(solr_urls, options), collection)
logger.debug("Solr request: {0}".format(request))
delete_collection_json_cmd=create_solr_api_request_command(request, config)
process = Popen(delete_collection_json_cmd, stdout=PIPE, stderr=PIPE, shell=True)
out, err = process.communicate()
if process.returncode != 0:
raise Exception("{0} command failed: {1}".format(delete_collection_json_cmd, str(err)))
response=json.loads(str(out))
if 'success' in response:
print 'Deleting collection {0} was {1}SUCCESSFUL{2}'.format(collection, colors.OKGREEN, colors.ENDC)
return collection
else:
raise Exception("DELETE collection ('{0}') failed. Response: {1}".format(collection, str(out)))
def create_collection(options, config, solr_urls, collection, config_set, shards, replica, max_shards_per_node):
request = CREATE_SOLR_COLLECTION_URL.format(get_random_solr_url(solr_urls, options), collection, config_set, shards, replica, max_shards_per_node)
logger.debug("Solr request: {0}".format(request))
create_collection_json_cmd=create_solr_api_request_command(request, config)
process = Popen(create_collection_json_cmd, stdout=PIPE, stderr=PIPE, shell=True)
out, err = process.communicate()
if process.returncode != 0:
raise Exception("{0} command failed: {1}".format(create_collection_json_cmd, str(err)))
response=json.loads(str(out))
if 'success' in response:
print 'Creating collection {0} was {1}SUCCESSFUL{2}'.format(collection, colors.OKGREEN, colors.ENDC)
return collection
else:
raise Exception("CREATE collection ('{0}') failed. ({1}) Response: {1}".format(collection, str(out)))
def reload_collection(options, config, solr_urls, collection):
request = RELOAD_SOLR_COLLECTION_URL.format(get_random_solr_url(solr_urls, options), collection)
logger.debug("Solr request: {0}".format(request))
reload_collection_json_cmd=create_solr_api_request_command(request, config)
process = Popen(reload_collection_json_cmd, stdout=PIPE, stderr=PIPE, shell=True)
out, err = process.communicate()
if process.returncode != 0:
raise Exception("{0} command failed: {1}".format(reload_collection_json_cmd, str(err)))
response=json.loads(str(out))
if 'success' in response:
print 'Reloading collection {0} was {1}SUCCESSFUL{2}'.format(collection, colors.OKGREEN, colors.ENDC)
return collection
else:
raise Exception("RELOAD collection ('{0}') failed. ({1}) Response: {1}".format(collection, str(out)))
def delete_znode(options, config, znode):
solr_cli_command=create_infra_solr_client_command(options, config, '--delete-znode --znode {0}'.format(znode))
logger.debug("Solr cli command: {0}".format(solr_cli_command))
sys.stdout.write('Deleting znode {0} ... '.format(znode))
sys.stdout.flush()
process = Popen(solr_cli_command, stdout=PIPE, stderr=PIPE, shell=True)
out, err = process.communicate()
if process.returncode != 0:
sys.stdout.write(colors.FAIL + 'FAILED\n' + colors.ENDC)
sys.stdout.flush()
raise Exception("{0} command failed: {1}".format(solr_cli_command, str(err)))
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
logger.debug(str(out))
def copy_znode(options, config, copy_src, copy_dest, copy_from_local=False, copy_to_local=False):
solr_cli_command=create_infra_solr_client_command(options, config, '--transfer-znode --copy-src {0} --copy-dest {1}'.format(copy_src, copy_dest))
if copy_from_local:
solr_cli_command+=" --transfer-mode copyFromLocal"
elif copy_to_local:
solr_cli_command+=" --transfer-mode copyToLocal"
logger.debug("Solr cli command: {0}".format(solr_cli_command))
sys.stdout.write('Transferring data from {0} to {1} ... '.format(copy_src, copy_dest))
sys.stdout.flush()
process = Popen(solr_cli_command, stdout=PIPE, stderr=PIPE, shell=True)
out, err = process.communicate()
if process.returncode != 0:
sys.stdout.write(colors.FAIL + 'FAILED\n' + colors.ENDC)
sys.stdout.flush()
raise Exception("{0} command failed: {1}".format(solr_cli_command, str(err)))
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
logger.debug(str(out))
def list_collections(options, config, output_file):
dump_json_files_list=[]
skip_dump=False
if options.skip_json_dump_files:
dump_json_files_list=options.skip_json_dump_files.split(',')
if dump_json_files_list:
for dump_json_file in dump_json_files_list:
if output_file.endswith(dump_json_file):
skip_dump=True
if skip_dump:
print 'Skipping collection dump file generation: {0}'.format(output_file)
if not os.path.exists(output_file):
print "{0}FAIL{1}: Collection dump file '{2}' does not exist.".format(colors.FAIL, colors.ENDC, output_file)
sys.exit(1)
else:
solr_cli_command=create_infra_solr_client_command(options, config, '--dump-collections --output {0}'.format(output_file), appendZnode=True)
logger.debug("Solr cli command: {0}".format(solr_cli_command))
sys.stdout.write('Dumping collections data to {0} ... '.format(output_file))
sys.stdout.flush()
process = Popen(solr_cli_command, stdout=PIPE, stderr=PIPE, shell=True)
out, err = process.communicate()
if process.returncode != 0:
sys.stdout.write(colors.FAIL + 'FAILED\n' + colors.ENDC)
sys.stdout.flush()
raise Exception("{0} command failed: {1}".format(solr_cli_command, str(err)))
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
logger.debug(str(out))
collections_data = get_collections_data(output_file)
return collections_data.keys() if collections_data is not None else []
def get_collections_data(output_file):
return read_json(output_file)
def get_collection_data(collections_data, collection):
return collections_data[collection] if collection in collections_data else None
def delete_logsearch_collections(options, config, collections_json_location, collections):
service_logs_collection = config.get('logsearch_collections', 'hadoop_logs_collection_name')
audit_logs_collection = config.get('logsearch_collections', 'audit_logs_collection_name')
history_collection = config.get('logsearch_collections', 'history_collection_name')
if service_logs_collection in collections:
solr_urls = get_solr_urls(options, config, service_logs_collection, collections_json_location)
retry(delete_collection, options, config, service_logs_collection, solr_urls, context='[Delete {0} collection]'.format(service_logs_collection))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation.'.format(service_logs_collection)
if audit_logs_collection in collections:
solr_urls = get_solr_urls(options, config, audit_logs_collection, collections_json_location)
retry(delete_collection, options, config, audit_logs_collection, solr_urls, context='[Delete {0} collection]'.format(audit_logs_collection))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation.'.format(audit_logs_collection)
if history_collection in collections:
solr_urls = get_solr_urls(options, config, history_collection, collections_json_location)
retry(delete_collection, options, config, history_collection, solr_urls, context='[Delete {0} collection]'.format(history_collection))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation.'.format(history_collection)
def delete_atlas_collections(options, config, collections_json_location, collections):
fulltext_collection = config.get('atlas_collections', 'fulltext_index_name')
edge_index_collection = config.get('atlas_collections', 'edge_index_name')
vertex_index_collection = config.get('atlas_collections', 'vertex_index_name')
if fulltext_collection in collections:
solr_urls = get_solr_urls(options, config, fulltext_collection, collections_json_location)
retry(delete_collection, options, config, fulltext_collection, solr_urls, context='[Delete {0} collection]'.format(fulltext_collection))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation.'.format(fulltext_collection)
if edge_index_collection in collections:
solr_urls = get_solr_urls(options, config, edge_index_collection, collections_json_location)
retry(delete_collection, options, config, edge_index_collection, solr_urls, context='[Delete {0} collection]'.format(edge_index_collection))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation.'.format(edge_index_collection)
if vertex_index_collection in collections:
solr_urls = get_solr_urls(options, config, vertex_index_collection, collections_json_location)
retry(delete_collection, options, config, vertex_index_collection, solr_urls, context='[Delete {0} collection]'.format(vertex_index_collection))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation.'.format(vertex_index_collection)
def delete_ranger_collection(options, config, collections_json_location, collections):
ranger_collection_name = config.get('ranger_collection', 'ranger_collection_name')
if ranger_collection_name in collections:
solr_urls = get_solr_urls(options, config, ranger_collection_name, collections_json_location)
retry(delete_collection, options, config, ranger_collection_name, solr_urls, context='[Delete {0} collection]'.format(ranger_collection_name))
else:
print 'Collection {0} does not exist or filtered out. Skipping delete operation'.format(ranger_collection_name)
def delete_collections(options, config, service_filter):
collections_json_location = COLLECTIONS_DATA_JSON_LOCATION.format("delete_collections.json")
collections=list_collections(options, config, collections_json_location)
collections=filter_collections(options, collections)
if is_ranger_available(config, service_filter):
delete_ranger_collection(options, config, collections_json_location, collections)
if is_atlas_available(config, service_filter):
delete_atlas_collections(options, config, collections_json_location, collections)
if is_logsearch_available(config, service_filter):
delete_logsearch_collections(options, config, collections_json_location, collections)
def upgrade_ranger_schema(options, config, service_filter):
solr_znode='/infra-solr'
if is_ranger_available(config, service_filter):
if config.has_section('infra_solr') and config.has_option('infra_solr', 'znode'):
solr_znode=config.get('infra_solr', 'znode')
ranger_config_set_name = config.get('ranger_collection', 'ranger_config_set_name')
copy_znode(options, config, "{0}{1}".format(INFRA_SOLR_CLIENT_BASE_PATH, RANGER_NEW_SCHEMA),
"{0}/configs/{1}/managed-schema".format(solr_znode, ranger_config_set_name), copy_from_local=True)
def backup_ranger_configs(options, config, service_filter):
solr_znode='/infra-solr'
if is_ranger_available(config, service_filter):
if config.has_section('infra_solr') and config.has_option('infra_solr', 'znode'):
solr_znode=config.get('infra_solr', 'znode')
ranger_config_set_name = config.get('ranger_collection', 'ranger_config_set_name')
backup_ranger_config_set_name = config.get('ranger_collection', 'backup_ranger_config_set_name')
copy_znode(options, config, "{0}/configs/{1}".format(solr_znode, ranger_config_set_name),
"{0}/configs/{1}".format(solr_znode, backup_ranger_config_set_name))
def upgrade_ranger_solrconfig_xml(options, config, service_filter):
solr_znode='/infra-solr'
if is_ranger_available(config, service_filter):
if config.has_section('infra_solr') and config.has_option('infra_solr', 'znode'):
solr_znode=config.get('infra_solr', 'znode')
ranger_config_set_name = config.get('ranger_collection', 'ranger_config_set_name')
backup_ranger_config_set_name = config.get('ranger_collection', 'backup_ranger_config_set_name')
copy_znode(options, config, "{0}/configs/{1}/solrconfig.xml".format(solr_znode, ranger_config_set_name),
"{0}/configs/{1}/solrconfig.xml".format(solr_znode, backup_ranger_config_set_name))
def check_shard_for_collection(collection):
active_shards = []
all_shards = []
collections_data = get_collections_data(COLLECTIONS_DATA_JSON_LOCATION.format("check_collections.json"))
print "Checking available shards for '{0}' collection...".format(collection)
if collection in collections_data:
collection_details = collections_data[collection]
if 'shards' in collection_details:
for shard in collection_details['shards']:
all_shards.append(shard)
if 'replicas' in collection_details['shards'][shard]:
for replica in collection_details['shards'][shard]['replicas']:
if 'state' in collection_details['shards'][shard]['replicas'][replica] \
and collection_details['shards'][shard]['replicas'][replica]['state'].lower() == 'active':
logger.debug("Found active shard for {0} (collection: {1})".format(shard, collection))
active_shards.append(shard)
for shard in all_shards:
if shard in active_shards:
print "{0}OK{1}: Found active replica for {2}" \
.format(colors.OKGREEN, colors.ENDC, shard)
else:
print "{0}WARNING{1}: Not found any active replicas for {2}, migration will probably fail, fix or delete the shard if it is possible." \
.format(colors.WARNING, colors.ENDC, shard)
def generate_core_pairs(original_collection, collection, config, options):
core_pairs_data={}
original_cores={}
original_collections_data = get_collections_data(COLLECTIONS_DATA_JSON_LOCATION.format("backup_collections.json"))
if original_collection in original_collections_data and 'leaderCoreHostMap' in original_collections_data[original_collection]:
original_cores = original_collections_data[original_collection]['leaderCoreHostMap']
sorted_original_cores=[]
for key in sorted(original_cores):
sorted_original_cores.append((key, original_cores[key]))
new_cores={}
collections_data = get_collections_data(COLLECTIONS_DATA_JSON_LOCATION.format("restore_collections.json"))
if collection in collections_data and 'leaderCoreHostMap' in collections_data[collection]:
new_cores = collections_data[collection]['leaderCoreHostMap']
sorted_new_cores=[]
for key in sorted(new_cores):
sorted_new_cores.append((key, new_cores[key]))
if len(new_cores) < len(original_cores):
raise Exception("Old collection core size is: " + str(len(new_cores)) +
". You will need at least: " + str(len(original_cores)))
else:
for index, original_core_data in enumerate(sorted_original_cores):
core_pairs_data[sorted_new_cores[index][0]]=original_core_data[0]
with open(COLLECTIONS_DATA_JSON_LOCATION.format(collection + "/restore_core_pairs.json"), 'w') as outfile:
json.dump(core_pairs_data, outfile)
return core_pairs_data
def update_state_json(original_collection, collection, config, options):
solr_znode='/infra-solr'
if config.has_section('infra_solr') and config.has_option('infra_solr', 'znode'):
solr_znode=config.get('infra_solr', 'znode')
coll_data_dir = "{0}migrate/data/{1}".format(INFRA_SOLR_CLIENT_BASE_PATH, collection)
if not os.path.exists(coll_data_dir):
os.makedirs(coll_data_dir)
copy_znode(options, config, "{0}/collections/{1}/state.json".format(solr_znode, collection), "{0}/state.json".format(coll_data_dir), copy_to_local=True)
copy_znode(options, config, "{0}/restore_metadata/{1}".format(solr_znode, collection), "{0}".format(coll_data_dir), copy_to_local=True)
json_file_list=glob.glob("{0}/*.json".format(coll_data_dir))
logger.debug("Downloaded json files list: {0}".format(str(json_file_list)))
cores_data_json_list = [k for k in json_file_list if 'state.json' not in k and 'new_state.json' not in k and 'restore_core_pairs.json' not in k]
state_json_list = [k for k in json_file_list if '/state.json' in k]
if not cores_data_json_list:
raise Exception('Cannot find any downloaded restore core metadata for {0}'.format(collection))
if not state_json_list:
raise Exception('Cannot find any downloaded restore collection state metadata for {0}'.format(collection))
core_pairs = generate_core_pairs(original_collection, collection, config, options)
cores_to_skip = []
logger.debug("Generated core pairs: {0}".format(str(core_pairs)))
if options.skip_cores:
cores_to_skip = options.skip_cores.split(',')
logger.debug("Cores to skip: {0}".format(str(cores_to_skip)))
state_json_file=state_json_list[0]
state_data = read_json(state_json_file)
core_json_data=[]
for core_data_json_file in cores_data_json_list:
core_json_data.append(read_json(core_data_json_file))
logger.debug("collection data content: {0}".format(str(state_data)))
core_details={}
for core in core_json_data:
core_details[core['core_node']]=core
logger.debug("core data contents: {0}".format(str(core_details)))
collection_data = state_data[collection]
shards = collection_data['shards']
new_state_json_data=copy.deepcopy(state_data)
for shard in shards:
replicas = shards[shard]['replicas']
for replica in replicas:
core_data = replicas[replica]
core = core_data['core']
base_url = core_data['base_url']
node_name = core_data['node_name']
data_dir = core_data['dataDir'] if 'dataDir' in core_data else None
ulog_dir = core_data['ulogDir'] if 'ulogDir' in core_data else None
if cores_to_skip and (core in cores_to_skip or (core in core_pairs and core_pairs[core] in cores_to_skip)):
print "Skipping core '{0}' as it is in skip-cores list (or its original pair: '{1}')".format(core, core_pairs[core])
elif replica in core_details:
old_core_node=core_details[replica]['core_node']
new_core_node=core_details[replica]['new_core_node']
new_state_core = copy.deepcopy(state_data[collection]['shards'][shard]['replicas'][replica])
new_state_json_data[collection]['shards'][shard]['replicas'][new_core_node]=new_state_core
if old_core_node != new_core_node:
if old_core_node in new_state_json_data[collection]['shards'][shard]['replicas']:
del new_state_json_data[collection]['shards'][shard]['replicas'][old_core_node]
if data_dir:
new_state_json_data[collection]['shards'][shard]['replicas'][new_core_node]['dataDir']=data_dir.replace(old_core_node, new_core_node)
if ulog_dir:
new_state_json_data[collection]['shards'][shard]['replicas'][new_core_node]['ulogDir']=ulog_dir.replace(old_core_node, new_core_node)
old_host=core_details[replica]['old_host']
new_host=core_details[replica]['new_host']
if old_host != new_host and old_core_node != new_core_node:
new_state_json_data[collection]['shards'][shard]['replicas'][new_core_node]['base_url']=base_url.replace(old_host, new_host)
new_state_json_data[collection]['shards'][shard]['replicas'][new_core_node]['node_name']=node_name.replace(old_host, new_host)
elif old_host != new_host:
new_state_json_data[collection]['shards'][shard]['replicas'][replica]['base_url']=base_url.replace(old_host, new_host)
new_state_json_data[collection]['shards'][shard]['replicas'][replica]['node_name']=node_name.replace(old_host, new_host)
with open("{0}/new_state.json".format(coll_data_dir), 'w') as outfile:
json.dump(new_state_json_data, outfile)
copy_znode(options, config, "{0}/new_state.json".format(coll_data_dir), "{0}/collections/{1}/state.json".format(solr_znode, collection), copy_from_local=True)
def delete_znodes(options, config, service_filter):
solr_znode='/infra-solr'
if is_logsearch_available(config, service_filter):
if config.has_section('infra_solr') and config.has_option('infra_solr', 'znode'):
solr_znode=config.get('infra_solr', 'znode')
delete_znode(options, config, "{0}/configs/hadoop_logs".format(solr_znode))
delete_znode(options, config, "{0}/configs/audit_logs".format(solr_znode))
delete_znode(options, config, "{0}/configs/history".format(solr_znode))
def do_backup_request(options, accessor, parser, config, collection, index_location):
sys.stdout.write("Sending backup collection request ('{0}') to Ambari to process (backup destination: '{1}')..."
.format(collection, index_location))
sys.stdout.flush()
response = backup(options, accessor, parser, config, collection, index_location)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Backup command request id: {0}'.format(request_id)
if options.async:
print "Backup {0} collection request sent to Ambari server. Check Ambari UI about the results.".format(collection)
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, 'Backup Solr collection: ' + collection)
print "Backup collection '{0}'... {1}DONE{2}".format(collection, colors.OKGREEN, colors.ENDC)
def do_migrate_request(options, accessor, parser, config, collection, index_location):
sys.stdout.write("Sending migrate collection request ('{0}') to Ambari to process (migrate folder: '{1}')..."
.format(collection, index_location))
sys.stdout.flush()
response = migrate(options, accessor, parser, config, collection, index_location)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Migrate command request id: {0}'.format(request_id)
if options.async:
print "Migrate {0} collection index request sent to Ambari server. Check Ambari UI about the results.".format(collection)
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, 'Migrate Solr collection index: ' + collection)
print "Migrate index '{0}'... {1}DONE{2}".format(collection, colors.OKGREEN, colors.ENDC)
def do_restore_request(options, accessor, parser, config, original_collection, collection, config_set, index_location, shards, hdfs_path):
sys.stdout.write("Sending restore collection request ('{0}') to Ambari to process (backup location: '{1}')..."
.format(collection, index_location))
sys.stdout.flush()
response = restore(options, accessor, parser, config, original_collection, collection, config_set, index_location, hdfs_path, shards)
request_id = get_request_id(response)
sys.stdout.write(colors.OKGREEN + 'DONE\n' + colors.ENDC)
sys.stdout.flush()
print 'Restore command request id: {0}'.format(request_id)
if options.async:
print "Restore {0} collection request sent to Ambari server. Check Ambari UI about the results.".format(collection)
sys.exit(0)
else:
sys.stdout.write("Start monitoring Ambari request with id {0} ...".format(request_id))
sys.stdout.flush()
cluster = config.get('ambari_server', 'cluster')
monitor_request(options, accessor, cluster, request_id, 'Restore Solr collection: ' + collection)
print "Restoring collection '{0}'... {1}DONE{2}".format(collection, colors.OKGREEN, colors.ENDC)
def get_ranger_index_location(collection, config, options):
ranger_index_location = None
if options.index_location:
ranger_index_location = os.path.join(options.index_location, "ranger")
elif options.ranger_index_location:
ranger_index_location = options.ranger_index_location
elif config.has_option('ranger_collection', 'backup_path'):
ranger_index_location = config.get('ranger_collection', 'backup_path')
else:
print "'backup_path'is missing from config file and --index-location or --ranger-index-location options are missing as well. Backup collection {0} {1}FAILED{2}." \
.format(collection, colors.FAIL, colors.ENDC)
sys.exit(1)
return ranger_index_location
def get_atlas_index_location(collection, config, options):
atlas_index_location = None
if options.index_location:
atlas_index_location = os.path.join(options.index_location, "atlas", collection)
elif options.ranger_index_location:
atlas_index_location = os.path.join(options.atlas_index_location, collection)
elif config.has_option('atlas_collections', 'backup_path'):
atlas_index_location = os.path.join(config.get('atlas_collections', 'backup_path'), collection)
else:
print "'backup_path'is missing from config file and --index-location or --atlas-index-location options are missing as well. Backup collection {0} {1}FAILED{2}." \
.format(collection, colors.FAIL, colors.ENDC)
sys.exit(1)
return atlas_index_location
def backup_collections(options, accessor, parser, config, service_filter):
collections=list_collections(options, config, COLLECTIONS_DATA_JSON_LOCATION.format("backup_collections.json"))
collections=filter_collections(options, collections)
if is_ranger_available(config, service_filter):
collection_name = config.get('ranger_collection', 'ranger_collection_name')
if collection_name in collections:
ranger_index_location=get_ranger_index_location(collection_name, config, options)
do_backup_request(options, accessor, parser, config, collection_name, ranger_index_location)
else:
print 'Collection {0} does not exist or filtered out. Skipping backup operation.'.format(collection_name)
if is_atlas_available(config, service_filter):
fulltext_index_collection = config.get('atlas_collections', 'fulltext_index_name')
if fulltext_index_collection in collections:
fulltext_index_location = get_atlas_index_location(fulltext_index_collection, config, options)
do_backup_request(options, accessor, parser, config, fulltext_index_collection, fulltext_index_location)
else:
print 'Collection {0} does not exist or filtered out. Skipping backup operation.'.format(fulltext_index_collection)
vertex_index_collection = config.get('atlas_collections', 'vertex_index_name')
if vertex_index_collection in collections:
vertex_index_location = get_atlas_index_location(vertex_index_collection, config, options)
do_backup_request(options, accessor, parser, config, vertex_index_collection, vertex_index_location)
else:
print 'Collection {0} does not exist or filtered out. Skipping backup operation.'.format(vertex_index_collection)
edge_index_collection = config.get('atlas_collections', 'edge_index_name')
if edge_index_collection in collections:
edge_index_location = get_atlas_index_location(edge_index_collection, config, options)
do_backup_request(options, accessor, parser, config, edge_index_collection, edge_index_location)
else:
print 'Collection {0} does not exist or filtered out. Skipping backup operation.'.format(edge_index_collection)
def migrate_snapshots(options, accessor, parser, config, service_filter):
if is_ranger_available(config, service_filter):
collection_name = config.get('ranger_collection', 'ranger_collection_name')
if options.collection is None or options.collection == collection_name:
ranger_index_location=get_ranger_index_location(collection_name, config, options)
do_migrate_request(options, accessor, parser, config, collection_name, ranger_index_location)
else:
print "Collection ('{0}') backup index has filtered out. Skipping migrate operation.".format(collection_name)
if is_atlas_available(config, service_filter):
fulltext_index_collection = config.get('atlas_collections', 'fulltext_index_name')
if options.collection is None or options.collection == fulltext_index_collection:
fulltext_index_location=get_atlas_index_location(fulltext_index_collection, config, options)
do_migrate_request(options, accessor, parser, config, fulltext_index_collection, fulltext_index_location)
else:
print "Collection ('{0}') backup index has filtered out. Skipping migrate operation.".format(fulltext_index_collection)
vertex_index_collection = config.get('atlas_collections', 'vertex_index_name')
if options.collection is None or options.collection == vertex_index_collection:
vertex_index_location=get_atlas_index_location(vertex_index_collection, config, options)
do_migrate_request(options, accessor, parser, config, fulltext_index_collection, vertex_index_location)
else:
print "Collection ('{0}') backup index has filtered out. Skipping migrate operation.".format(vertex_index_collection)
edge_index_collection = config.get('atlas_collections', 'edge_index_name')
if options.collection is None or options.collection == edge_index_collection:
edge_index_location=get_atlas_index_location(edge_index_collection, config, options)
do_migrate_request(options, accessor, parser, config, edge_index_collection, edge_index_location)
else:
print "Collection ('{0}') backup index has filtered out. Skipping migrate operation.".format(edge_index_collection)
def create_backup_collections(options, accessor, parser, config, service_filter):
collections_json_location = COLLECTIONS_DATA_JSON_LOCATION.format("before_restore_collections.json")
collections=list_collections(options, config, collections_json_location)
replica_number = "1" # hard coded
if is_ranger_available(config, service_filter):
backup_ranger_collection = config.get('ranger_collection', 'backup_ranger_collection_name')
if backup_ranger_collection not in collections:
if options.collection is not None and options.collection != backup_ranger_collection:
print "Collection {0} has filtered out. Skipping create operation.".format(backup_ranger_collection)
else:
solr_urls = get_solr_urls(options, config, backup_ranger_collection, collections_json_location)
backup_ranger_config_set = config.get('ranger_collection', 'backup_ranger_config_set_name')
backup_ranger_shards = config.get('ranger_collection', 'ranger_collection_shards')
backup_ranger_max_shards = config.get('ranger_collection', 'ranger_collection_max_shards_per_node')
retry(create_collection, options, config, solr_urls, backup_ranger_collection, backup_ranger_config_set,
backup_ranger_shards, replica_number, backup_ranger_max_shards, context="[Create Solr Collections]")
else:
print "Collection {0} has already exist. Skipping create operation.".format(backup_ranger_collection)
if is_atlas_available(config, service_filter):
backup_atlas_config_set = config.get('atlas_collections', 'config_set')
backup_fulltext_index_name = config.get('atlas_collections', 'backup_fulltext_index_name')
if backup_fulltext_index_name not in collections:
if options.collection is not None and options.collection != backup_fulltext_index_name:
print "Collection {0} has filtered out. Skipping create operation.".format(backup_fulltext_index_name)
else:
solr_urls = get_solr_urls(options, config, backup_fulltext_index_name, collections_json_location)
backup_fulltext_index_shards = config.get('atlas_collections', 'fulltext_index_shards')
backup_fulltext_index_max_shards = config.get('atlas_collections', 'fulltext_index_max_shards_per_node')
retry(create_collection, options, config, solr_urls, backup_fulltext_index_name, backup_atlas_config_set,
backup_fulltext_index_shards, replica_number, backup_fulltext_index_max_shards, context="[Create Solr Collections]")
else:
print "Collection {0} has already exist. Skipping create operation.".format(backup_fulltext_index_name)
backup_edge_index_name = config.get('atlas_collections', 'backup_edge_index_name')
if backup_edge_index_name not in collections:
if options.collection is not None and options.collection != backup_edge_index_name:
print "Collection {0} has filtered out. Skipping create operation.".format(backup_edge_index_name)
else:
solr_urls = get_solr_urls(options, config, backup_edge_index_name, collections_json_location)
backup_edge_index_shards = config.get('atlas_collections', 'edge_index_shards')
backup_edge_index_max_shards = config.get('atlas_collections', 'edge_index_max_shards_per_node')
retry(create_collection, options, config, solr_urls, backup_edge_index_name, backup_atlas_config_set,
backup_edge_index_shards, replica_number, backup_edge_index_max_shards, context="[Create Solr Collections]")
else:
print "Collection {0} has already exist. Skipping create operation.".format(backup_edge_index_name)
backup_vertex_index_name = config.get('atlas_collections', 'backup_vertex_index_name')
if backup_vertex_index_name not in collections:
if options.collection is not None and options.collection != backup_vertex_index_name:
print "Collection {0} has filtered out. Skipping create operation.".format(backup_vertex_index_name)
else:
solr_urls = get_solr_urls(options, config, backup_vertex_index_name, collections_json_location)
backup_vertex_index_shards = config.get('atlas_collections', 'vertex_index_shards')
backup_vertex_index_max_shards = config.get('atlas_collections', 'vertex_index_max_shards_per_node')
retry(create_collection, options, config, solr_urls, backup_vertex_index_name, backup_atlas_config_set,
backup_vertex_index_shards, replica_number, backup_vertex_index_max_shards, context="[Create Solr Collections]")
else:
print "Collection {0} has already exist. Skipping create operation.".format(backup_fulltext_index_name)
def restore_collections(options, accessor, parser, config, service_filter):
collections=list_collections(options, config, COLLECTIONS_DATA_JSON_LOCATION.format("restore_collections.json"))
collections=filter_collections(options, collections)
if 'RANGER' in service_filter and config.has_section('ranger_collection') and config.has_option('ranger_collection', 'enabled') \
and config.get('ranger_collection', 'enabled') == 'true':
collection_name = config.get('ranger_collection', 'ranger_collection_name')
backup_ranger_collection = config.get('ranger_collection', 'backup_ranger_collection_name')
backup_ranger_config_set_name = config.get('ranger_collection', 'backup_ranger_config_set_name')
hdfs_base_path = None
if options.ranger_hdfs_base_path:
hdfs_base_path = options.ranger_hdfs_base_path
elif options.hdfs_base_path:
hdfs_base_path = options.hdfs_base_path
elif config.has_option('ranger_collection', 'hdfs_base_path'):
hdfs_base_path = config.get('ranger_collection', 'hdfs_base_path')
if backup_ranger_collection in collections:
backup_ranger_shards = config.get('ranger_collection', 'ranger_collection_shards')
ranger_index_location=get_ranger_index_location(collection_name, config, options)
do_restore_request(options, accessor, parser, config, collection_name, backup_ranger_collection, backup_ranger_config_set_name, ranger_index_location, backup_ranger_shards, hdfs_base_path)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping restore operation.".format(backup_ranger_collection)
if is_atlas_available(config, service_filter):
hdfs_base_path = None
if options.ranger_hdfs_base_path:
hdfs_base_path = options.atlas_hdfs_base_path
elif options.hdfs_base_path:
hdfs_base_path = options.hdfs_base_path
elif config.has_option('atlas_collections', 'hdfs_base_path'):
hdfs_base_path = config.get('atlas_collections', 'hdfs_base_path')
atlas_config_set = config.get('atlas_collections', 'config_set')
fulltext_index_collection = config.get('atlas_collections', 'fulltext_index_name')
backup_fulltext_index_name = config.get('atlas_collections', 'backup_fulltext_index_name')
if backup_fulltext_index_name in collections:
backup_fulltext_index_shards = config.get('atlas_collections', 'fulltext_index_shards')
fulltext_index_location=get_atlas_index_location(fulltext_index_collection, config, options)
do_restore_request(options, accessor, parser, config, fulltext_index_collection, backup_fulltext_index_name, atlas_config_set, fulltext_index_location, backup_fulltext_index_shards, hdfs_base_path)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping restore operation.".format(fulltext_index_collection)
edge_index_collection = config.get('atlas_collections', 'edge_index_name')
backup_edge_index_name = config.get('atlas_collections', 'backup_edge_index_name')
if backup_edge_index_name in collections:
backup_edge_index_shards = config.get('atlas_collections', 'edge_index_shards')
edge_index_location=get_atlas_index_location(edge_index_collection, config, options)
do_restore_request(options, accessor, parser, config, edge_index_collection, backup_edge_index_name, atlas_config_set, edge_index_location, backup_edge_index_shards, hdfs_base_path)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping restore operation.".format(edge_index_collection)
vertex_index_collection = config.get('atlas_collections', 'vertex_index_name')
backup_vertex_index_name = config.get('atlas_collections', 'backup_vertex_index_name')
if backup_vertex_index_name in collections:
backup_vertex_index_shards = config.get('atlas_collections', 'vertex_index_shards')
vertex_index_location=get_atlas_index_location(vertex_index_collection, config, options)
do_restore_request(options, accessor, parser, config, vertex_index_collection, backup_vertex_index_name, atlas_config_set, vertex_index_location, backup_vertex_index_shards, hdfs_base_path)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping restore operation.".format(vertex_index_collection)
def reload_collections(options, accessor, parser, config, service_filter):
collections_json_location = config, COLLECTIONS_DATA_JSON_LOCATION.format("reload_collections.json")
collections=list_collections(options, config, collections_json_location)
collections=filter_collections(options, collections)
if is_ranger_available(config, service_filter):
backup_ranger_collection = config.get('ranger_collection', 'backup_ranger_collection_name')
if backup_ranger_collection in collections:
solr_urls = get_solr_urls(options, config, backup_ranger_collection, collections_json_location)
retry(reload_collection, options, config, solr_urls, backup_ranger_collection, context="[Reload Solr Collections]")
else:
print "Collection ('{0}') does not exist or filtered out. Skipping reload operation.".format(backup_ranger_collection)
if is_atlas_available(config, service_filter):
backup_fulltext_index_name = config.get('atlas_collections', 'backup_fulltext_index_name')
if backup_fulltext_index_name in collections:
solr_urls = get_solr_urls(options, config, backup_fulltext_index_name, collections_json_location)
retry(reload_collection, options, config, solr_urls, backup_fulltext_index_name, context="[Reload Solr Collections]")
else:
print "Collection ('{0}') does not exist or filtered out. Skipping reload operation.".format(backup_fulltext_index_name)
backup_edge_index_name = config.get('atlas_collections', 'backup_edge_index_name')
if backup_edge_index_name in collections:
solr_urls = get_solr_urls(options, config, backup_edge_index_name, collections_json_location)
retry(reload_collection, options, config, solr_urls, backup_edge_index_name, context="[Reload Solr Collections]")
else:
print "Collection ('{0}') does not exist or filtered out. Skipping reload operation.".format(backup_edge_index_name)
backup_vertex_index_name = config.get('atlas_collections', 'backup_vertex_index_name')
if backup_vertex_index_name in collections:
solr_urls = get_solr_urls(options, config, backup_vertex_index_name, collections_json_location)
retry(reload_collection, options, config, solr_urls, backup_vertex_index_name, context="[Reload Solr Collections]")
else:
print "Collection ('{0}') does not exist or filtered out. Skipping reload operation.".format(backup_fulltext_index_name)
def validate_ini_file(options, parser):
if options.ini_file is None:
parser.print_help()
print 'ini-file option is missing'
sys.exit(1)
elif not os.path.isfile(options.ini_file):
parser.print_help()
print 'ini file ({0}) does not exist'.format(options.ini_file)
sys.exit(1)
def rolling_restart_solr(options, accessor, parser, config):
cluster = config.get('ambari_server', 'cluster')
component_hosts = get_solr_hosts(options, accessor, cluster)
interval_secs = options.batch_interval
fault_tolerance = options.batch_fault_tolerance
request_body = create_batch_command("RESTART", component_hosts, cluster, "AMBARI_INFRA_SOLR", "INFRA_SOLR", interval_secs, fault_tolerance, "Rolling restart Infra Solr Instances")
post_json(accessor, BATCH_REQUEST_API_URL.format(cluster), request_body)
print "Rolling Restart Infra Solr Instances request sent. (check Ambari UI about the requests)"
def update_state_jsons(options, accessor, parser, config, service_filter):
collections=list_collections(options, config, COLLECTIONS_DATA_JSON_LOCATION.format("collections.json"))
collections=filter_collections(options, collections)
if is_ranger_available(config, service_filter):
original_ranger_collection = config.get('ranger_collection', 'ranger_collection_name')
backup_ranger_collection = config.get('ranger_collection', 'backup_ranger_collection_name')
if backup_ranger_collection in collections:
update_state_json(original_ranger_collection, backup_ranger_collection, config, options)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(backup_ranger_collection)
if is_atlas_available(config, service_filter):
original_fulltext_index_name = config.get('atlas_collections', 'fulltext_index_name')
backup_fulltext_index_name = config.get('atlas_collections', 'backup_fulltext_index_name')
if backup_fulltext_index_name in collections:
update_state_json(original_fulltext_index_name, backup_fulltext_index_name, config, options)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(backup_fulltext_index_name)
original_edge_index_name = config.get('atlas_collections', 'edge_index_name')
backup_edge_index_name = config.get('atlas_collections', 'backup_edge_index_name')
if backup_edge_index_name in collections:
update_state_json(original_edge_index_name, backup_edge_index_name, config, options)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(backup_edge_index_name)
original_vertex_index_name = config.get('atlas_collections', 'vertex_index_name')
backup_vertex_index_name = config.get('atlas_collections', 'backup_vertex_index_name')
if backup_vertex_index_name in collections:
update_state_json(original_vertex_index_name, backup_vertex_index_name, config, options)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(backup_fulltext_index_name)
def disable_solr_authorization(options, accessor, parser, config):
solr_znode='/infra-solr'
if config.has_section('infra_solr') and config.has_option('infra_solr', 'znode'):
solr_znode=config.get('infra_solr', 'znode')
kerberos_enabled='false'
if config.has_section('cluster') and config.has_option('cluster', 'kerberos_enabled'):
kerberos_enabled=config.get('cluster', 'kerberos_enabled')
if kerberos_enabled == 'true':
print "Disable Solr authorization by uploading a new security.json ..."
copy_znode(options, config, COLLECTIONS_DATA_JSON_LOCATION.format("security-without-authr.json"),
"{0}/security.json".format(solr_znode), copy_from_local=True)
else:
print "Security is not enabled. Skipping disable Solr authorization operation."
def check_shards(options, accessor, parser, config):
collections=list_collections(options, config, COLLECTIONS_DATA_JSON_LOCATION.format("check_collections.json"))
collections=filter_collections(options, collections)
if is_ranger_available(config, service_filter):
ranger_collection = config.get('ranger_collection', 'ranger_collection_name')
if ranger_collection in collections:
check_shard_for_collection(ranger_collection)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(ranger_collection)
if is_atlas_available(config, service_filter):
fulltext_index_name = config.get('atlas_collections', 'fulltext_index_name')
if fulltext_index_name in collections:
check_shard_for_collection(fulltext_index_name)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(fulltext_index_name)
edge_index_name = config.get('atlas_collections', 'edge_index_name')
if edge_index_name in collections:
check_shard_for_collection(edge_index_name)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(edge_index_name)
vertex_index_name = config.get('atlas_collections', 'vertex_index_name')
if vertex_index_name in collections:
check_shard_for_collection(vertex_index_name)
else:
print "Collection ('{0}') does not exist or filtered out. Skipping update collection state operation.".format(fulltext_index_name)
if __name__=="__main__":
parser = optparse.OptionParser("usage: %prog [options]")
parser.add_option("-a", "--action", dest="action", type="string", help="delete-collections | backup | cleanup-znodes | backup-and-cleanup | migrate | restore |' \
' rolling-restart-solr | check-shards | disable-solr-authorization | upgrade-solr-clients | upgrade-solr-instances | upgrade-logsearch-portal | upgrade-logfeeders | stop-logsearch | restart-logsearch")
parser.add_option("-i", "--ini-file", dest="ini_file", type="string", help="Config ini file to parse (required)")
parser.add_option("-f", "--force", dest="force", default=False, action="store_true", help="force index upgrade even if it's the right version")
parser.add_option("-v", "--verbose", dest="verbose", action="store_true", help="use for verbose logging")
parser.add_option("-s", "--service-filter", dest="service_filter", default=None, type="string", help="run commands only selected services (comma separated: LOGSEARCH,ATLAS,RANGER)")
parser.add_option("-c", "--collection", dest="collection", default=None, type="string", help="selected collection to run an operation")
parser.add_option("--async", dest="async", action="store_true", default=False, help="async Ambari operations (backup | restore | migrate)")
parser.add_option("--index-location", dest="index_location", type="string", help="location of the index backups. add ranger/atlas prefix after the path. required only if no backup path in the ini file")
parser.add_option("--atlas-index-location", dest="atlas_index_location", type="string", help="location of the index backups (for atlas). required only if no backup path in the ini file")
parser.add_option("--ranger-index-location", dest="ranger_index_location", type="string", help="location of the index backups (for ranger). required only if no backup path in the ini file")
parser.add_option("--version", dest="index_version", type="string", default="6.6.2", help="lucene index version for migration (6.6.2 or 7.3.1)")
parser.add_option("--request-tries", dest="request_tries", type="int", help="number of tries for BACKUP/RESTORE status api calls in the request")
parser.add_option("--request-time-interval", dest="request_time_interval", type="int", help="time interval between BACKUP/RESTORE status api calls in the request")
parser.add_option("--request-async", dest="request_async", action="store_true", default=False, help="skip BACKUP/RESTORE status api calls from the command")
parser.add_option("--include-solr-hosts", dest="include_solr_hosts", type="string", help="comma separated list of included solr hosts")
parser.add_option("--exclude-solr-hosts", dest="exclude_solr_hosts", type="string", help="comma separated list of excluded solr hosts")
parser.add_option("--disable-solr-host-check", dest="disable_solr_host_check", action="store_true", default=False, help="Disable to check solr hosts are good for the collection backups")
parser.add_option("--core-filter", dest="core_filter", default=None, type="string", help="core filter for replica folders")
parser.add_option("--skip-cores", dest="skip_cores", default=None, type="string", help="specific cores to skip (comma separated)")
parser.add_option("--hdfs-base-path", dest="hdfs_base_path", default=None, type="string", help="hdfs base path where the collections are located (e.g.: /user/infrasolr). Use if both atlas and ranger collections are on hdfs.")
parser.add_option("--ranger-hdfs-base-path", dest="ranger_hdfs_base_path", default=None, type="string", help="hdfs base path where the ranger collection is located (e.g.: /user/infra-solr). Use if only ranger collection is on hdfs.")
parser.add_option("--atlas-hdfs-base-path", dest="atlas_hdfs_base_path", default=None, type="string", help="hdfs base path where the atlas collections are located (e.g.: /user/infra-solr). Use if only atlas collections are on hdfs.")
parser.add_option("--keep-backup", dest="keep_backup", default=False, action="store_true", help="If it is turned on, Snapshot Solr data will not be deleted from the filesystem during restore.")
parser.add_option("--batch-interval", dest="batch_interval", type="int", default=60 ,help="batch time interval (seconds) between requests (for restarting INFRA SOLR, default: 60)")
parser.add_option("--batch-fault-tolerance", dest="batch_fault_tolerance", type="int", default=0 ,help="fault tolerance of tasks for batch request (for restarting INFRA SOLR, default: 0)")
parser.add_option("--shared-drive", dest="shared_drive", default=False, action="store_true", help="Use if the backup location is shared between hosts. (override config from config ini file)")
parser.add_option("--skip-json-dump-files", dest="skip_json_dump_files", type="string", help="comma separated list of files that won't be download during collection dump (could be useful if it is required to change something in manually in the already downloaded file)")
(options, args) = parser.parse_args()
set_log_level(options.verbose)
validate_ini_file(options, parser)
config = ConfigParser.RawConfigParser()
config.read(options.ini_file)
service_filter=options.service_filter.upper().split(',') if options.service_filter is not None else ['LOGSEARCH', 'ATLAS', 'RANGER']
if options.action is None:
parser.print_help()
print 'action option is missing'
sys.exit(1)
else:
if config.has_section('ambari_server'):
host = config.get('ambari_server', 'host')
port = config.get('ambari_server', 'port')
protocol = config.get('ambari_server', 'protocol')
username = config.get('ambari_server', 'username')
password = config.get('ambari_server', 'password')
accessor = api_accessor(host, username, password, protocol, port)
if options.action.lower() == 'backup':
backup_ranger_configs(options, config, service_filter)
backup_collections(options, accessor, parser, config, service_filter)
elif options.action.lower() == 'delete-collections':
delete_collections(options, config, service_filter)
delete_znodes(options, config, service_filter)
upgrade_ranger_schema(options, config, service_filter)
elif options.action.lower() == 'cleanup-znodes':
delete_znodes(options, config, service_filter)
upgrade_ranger_schema(options, config, service_filter)
elif options.action.lower() == 'backup-and-cleanup':
backup_ranger_configs(options, config, service_filter)
backup_collections(options, accessor, parser, config, service_filter)
delete_collections(options, config, service_filter)
delete_znodes(options, config, service_filter)
upgrade_ranger_schema(options, config, service_filter)
elif options.action.lower() == 'restore':
upgrade_ranger_solrconfig_xml(options, config, service_filter)
create_backup_collections(options, accessor, parser, config, service_filter)
restore_collections(options, accessor, parser, config, service_filter)
update_state_jsons(options, accessor, parser, config, service_filter)
elif options.action.lower() == 'update-collection-state':
update_state_jsons(options, accessor, parser, config, service_filter)
elif options.action.lower() == 'reload':
reload_collections(options, accessor, parser, config, service_filter)
elif options.action.lower() == 'migrate':
migrate_snapshots(options, accessor, parser, config, service_filter)
elif options.action.lower() == 'upgrade-solr-clients':
upgrade_solr_clients(options, accessor, parser, config)
elif options.action.lower() == 'upgrade-solr-instances':
upgrade_solr_instances(options, accessor, parser, config)
elif options.action.lower() == 'upgrade-logsearch-portal':
upgrade_logsearch_portal(options, accessor, parser, config)
elif options.action.lower() == 'upgrade-logfeeders':
upgrade_logfeeders(options, accessor, parser, config)
elif options.action.lower() == 'stop-logsearch':
service_components_command(options, accessor, parser, config, LOGSEARCH_SERVICE_NAME, LOGSEARCH_SERVER_COMPONENT_NAME, "STOP", "Stop")
service_components_command(options, accessor, parser, config, LOGSEARCH_SERVICE_NAME, LOGSEARCH_LOGFEEDER_COMPONENT_NAME, "STOP", "Stop")
elif options.action.lower() == 'restart-logsearch':
service_components_command(options, accessor, parser, config, LOGSEARCH_SERVICE_NAME, LOGSEARCH_SERVER_COMPONENT_NAME, "RESTART", "Restart")
service_components_command(options, accessor, parser, config, LOGSEARCH_SERVICE_NAME, LOGSEARCH_LOGFEEDER_COMPONENT_NAME, "RESTART", "Restart")
elif options.action.lower() == 'rolling-restart-solr':
rolling_restart_solr(options, accessor, parser, config)
elif options.action.lower() == 'disable-solr-authorization':
disable_solr_authorization(options, accessor, parser, config)
elif options.action.lower() == 'check-shards':
check_shards(options, accessor, parser, config)
else:
parser.print_help()
print 'action option is invalid (available actions: delete-collections | backup | cleanup-znodes | backup-and-cleanup | migrate | restore |' \
' rolling-restart-solr | check-shards | disable-solr-authorization | upgrade-solr-clients | upgrade-solr-instances | upgrade-logsearch-portal | upgrade-logfeeders | stop-logsearch | restart-logsearch)'
sys.exit(1)
print "Migration helper command {0}FINISHED{1}".format(colors.OKGREEN, colors.ENDC)