<commit_before>#!/usr/bin/env python
#
__author__ = 'Michalis'
__version__ = '0.11.2803'

import socket
import re
import urllib2
from optparse import OptionParser
import hashlib
import os
import sys
import random
from time import sleep

from cm_api.api_client import ApiResource, ApiException
from cm_api.endpoints.hosts import *
from cm_api.endpoints.services import ApiServiceSetupInfo, ApiService


def init_cluster():
    """
    Initialise Cluster
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=cmx.password)
    # Update Cloudera Manager configuration
    cm = api.get_cloudera_manager()
    cm.update_config({"REMOTE_PARCEL_REPO_URLS": "http://archive.cloudera.com/cdh5/parcels/{0}/,"
                                                 "http://archive.cloudera.com/impala/parcels/{0}/,"
                                                 "http://archive.cloudera.com/cdh4/parcels/{0}/,"
                                                 "http://archive.cloudera.com/search/parcels/{0}/,"
                                                 "http://archive.cloudera.com/spark/parcels/{0}/,"
                                                 "http://archive.cloudera.com/sqoop-connectors/parcels/{0}/,"
                                                 "http://archive.cloudera.com/accumulo/parcels/{0}/,"
                                                 "http://archive.cloudera.com/accumulo-c5/parcels/{0},"
                                                 "http://archive.cloudera.com/gplextras5/parcels/{0}".format(cdhver), 
                      "PHONE_HOME": False, "PARCEL_DISTRIBUTE_RATE_LIMIT_KBS_PER_SECOND": "1024000"})

    print "> Initialise Cluster"
    if cmx.cluster_name in [x.name for x in api.get_all_clusters()]:
        print "Cluster name: '%s' already exists" % cmx.cluster_name
    else:
        print "Creating cluster name '%s'" % cmx.cluster_name
        api.create_cluster(name=cmx.cluster_name, version=cmx.cluster_version)


def add_hosts_to_cluster():
    """
    Add hosts to cluster
    :return:
    """
    print "> Add hosts to Cluster: %s" % cmx.cluster_name
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    cm = api.get_cloudera_manager()

    # deploy agents into host_list
    host_list = list(set([socket.getfqdn(x) for x in cmx.host_names] + [socket.getfqdn(cmx.cm_server)]) -
                     set([x.hostname for x in api.get_all_hosts()]))
    if host_list:
        cmd = cm.host_install(user_name=cmx.ssh_root_user, host_names=host_list,
                              password=<PASSWORD>, private_key=cmx.ssh_private_key)
        print "Installing host(s) to cluster '%s' - [ http://%s:7180/cmf/command/%s/details ]" % \
              (socket.getfqdn(cmx.cm_server), cmx.cm_server, cmd.id)
        #check.status_for_command("Hosts: %s " % host_list, cmd)
        print "Installing hosts. This might take a while."
        while cmd.success == None:
            sleep(20)
            cmd = cmd.fetch()
            print "Installing hosts... Checking"

        if cmd.success != True:
            print "cm_host_install failed: " + cmd.resultMessage
            exit(0)

    print "Host install finish, agents installed"
    hosts = []
    for host in api.get_all_hosts():
        if host.hostId not in [x.hostId for x in cluster.list_hosts()]:
            print "Adding {'ip': '%s', 'hostname': '%s', 'hostId': '%s'}" % (host.ipAddress, host.hostname, host.hostId)
            hosts.append(host.hostId)

    print "adding new hosts to cluster"
    if hosts:
        print "Adding hostId(s) to '%s'" % cmx.cluster_name
        print "%s" % hosts
        cluster.add_hosts(hosts)


def host_rack():
    """
    Add host to rack
    :return:
    """
    # TODO: Add host to rack
    print "> Add host to rack"
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    hosts = []
    for h in api.get_all_hosts():
        # host = api.create_host(h.hostId, h.hostname,
        # socket.gethostbyname(h.hostname),
        # "/default_rack")
        h.set_rack_id("/default_rack")
        hosts.append(h)

    cluster.add_hosts(hosts)


def deploy_parcel(parcel_product, parcel_version):
    """
    Deploy parcels
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    parcel = cluster.get_parcel(parcel_product, parcel_version)
    if parcel.stage != 'ACTIVATED':
        print "> Deploying parcel: [ %s-%s ]" % (parcel_product, parcel_version)
        parcel.start_download()
        # unlike other commands, check progress by looking at parcel stage and status
        while True:
            parcel = cluster.get_parcel(parcel_product, parcel_version)
            if parcel.stage == 'DISTRIBUTED' or parcel.stage == 'DOWNLOADED' or parcel.stage == 'ACTIVATED':
                break
           # if parcel.state.errors:
            #    raise Exception(str(parcel.state.errors))
            msg = " [%s: %s / %s]" % (parcel.stage, parcel.state.progress, parcel.state.totalProgress)
            sys.stdout.write(msg + " " * (78 - len(msg)) + "\r")
            sys.stdout.flush()

        print ""
        print "1. Parcel Stage: %s" % parcel.stage
        parcel.start_distribution()

        while True:
            parcel = cluster.get_parcel(parcel_product, parcel_version)
            if parcel.stage == 'DISTRIBUTED' or parcel.stage == 'ACTIVATED':
                break
           # if parcel.state.errors:
               # raise Exception(str(parcel.state.errors))
            msg = " [%s: %s / %s]" % (parcel.stage, parcel.state.progress, parcel.state.totalProgress)
            sys.stdout.write(msg + " " * (78 - len(msg)) + "\r")
            sys.stdout.flush()

        print "2. Parcel Stage: %s" % parcel.stage
        if parcel.stage == 'DISTRIBUTED':
            parcel.activate()

        while True:
            parcel = cluster.get_parcel(parcel_product, parcel_version)
            if parcel.stage != 'ACTIVATED':
                msg = " [%s: %s / %s]" % (parcel.stage, parcel.state.progress, parcel.state.totalProgress)
                sys.stdout.write(msg + " " * (78 - len(msg)) + "\r")
                sys.stdout.flush()
           # elif parcel.state.errors:
             #   raise Exception(str(parcel.state.errors))
            else:
                print "3. Parcel Stage: %s" % parcel.stage
                break


def setup_zookeeper():
    """
    Zookeeper
    > Waiting for ZooKeeper Service to initialize
    Starting ZooKeeper Service
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    service_type = "ZOOKEEPER"
    if cdh.get_service_type(service_type) is None:
        print "> %s" % service_type
        service_name = "zookeeper"
        print "Create %s service" % service_name
        cluster.create_service(service_name, service_type)
        service = cluster.get_service(service_name)
        hosts = management.get_hosts()
        cmhost= management.get_cmhost()
        service.update_config({"zookeeper_datadir_autocreate": True})

        # Role Config Group equivalent to Service Default Group
        for rcg in [x for x in service.get_all_role_config_groups()]:
            if rcg.roleType == "SERVER":
                rcg.update_config({"maxClientCnxns": "1024"})
                # Pick 3 hosts and deploy Zookeeper Server role
                # mingrui change install on primary, secondary, and CM
                print cmhost[0]
                print cmhost
                print [x for x in hosts if x.id == 0 ][0]
                print [x for x in hosts if x.id == 1 ][0]
                cdh.create_service_role(service, rcg.roleType, cmhost[0])
                cdh.create_service_role(service, rcg.roleType, [x for x in hosts if x.id == 0 ][0])
                cdh.create_service_role(service, rcg.roleType, [x for x in hosts if x.id == 1 ][0])

        # init_zookeeper not required as the API performs this when adding Zookeeper
        # check.status_for_command("Waiting for ZooKeeper Service to initialize", service.init_zookeeper())
        check.status_for_command("Starting ZooKeeper Service", service.start())


def setup_hdfs():
    """
    HDFS
    > Checking if the name directories of the NameNode are empty. Formatting HDFS only if empty.
    Starting HDFS Service
    > Creating HDFS /tmp directory
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    service_type = "HDFS"
    if cdh.get_service_type(service_type) is None:
        print "> %s" % service_type
        service_name = "hdfs"
        print "Create %s service" % service_name
        cluster.create_service(service_name, service_type)
        service = cluster.get_service(service_name)
        hosts = management.get_hosts()

        # Service-Wide
        service_config = cdh.dependencies_for(service)
        service_config.update({"dfs_replication": "3",
                               "dfs_block_local_path_access_user": "impala,hbase,mapred,spark"})<commit_msg>Update cmxDeployOnIbiza.py<commit_after>#!/usr/bin/env python
#
__author__ = 'Michalis'
__version__ = '0.11.2803'

import socket
import re
import urllib2
from optparse import OptionParser
import hashlib
import os
import sys
import random
from time import sleep

from cm_api.api_client import ApiResource, ApiException
from cm_api.endpoints.hosts import *
from cm_api.endpoints.services import ApiServiceSetupInfo, ApiService


def init_cluster():
    """
    Initialise Cluster
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    # Update Cloudera Manager configuration
    cm = api.get_cloudera_manager()
    cm.update_config({"REMOTE_PARCEL_REPO_URLS": "http://archive.cloudera.com/cdh5/parcels/{0}/,"
                                                 "http://archive.cloudera.com/impala/parcels/{0}/,"
                                                 "http://archive.cloudera.com/cdh4/parcels/{0}/,"
                                                 "http://archive.cloudera.com/search/parcels/{0}/,"
                                                 "http://archive.cloudera.com/spark/parcels/{0}/,"
                                                 "http://archive.cloudera.com/sqoop-connectors/parcels/{0}/,"
                                                 "http://archive.cloudera.com/accumulo/parcels/{0}/,"
                                                 "http://archive.cloudera.com/accumulo-c5/parcels/{0},"
                                                 "http://archive.cloudera.com/gplextras5/parcels/{0}".format(cdhver), 
                      "PHONE_HOME": False, "PARCEL_DISTRIBUTE_RATE_LIMIT_KBS_PER_SECOND": "1024000"})

    print "> Initialise Cluster"
    if cmx.cluster_name in [x.name for x in api.get_all_clusters()]:
        print "Cluster name: '%s' already exists" % cmx.cluster_name
    else:
        print "Creating cluster name '%s'" % cmx.cluster_name
        api.create_cluster(name=cmx.cluster_name, version=cmx.cluster_version)


def add_hosts_to_cluster():
    """
    Add hosts to cluster
    :return:
    """
    print "> Add hosts to Cluster: %s" % cmx.cluster_name
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    cm = api.get_cloudera_manager()

    # deploy agents into host_list
    host_list = list(set([socket.getfqdn(x) for x in cmx.host_names] + [socket.getfqdn(cmx.cm_server)]) -
                     set([x.hostname for x in api.get_all_hosts()]))
    if host_list:
        cmd = cm.host_install(user_name=cmx.ssh_root_user, host_names=host_list,
                              password=<PASSWORD>, private_key=cmx.ssh_private_key)
        print "Installing host(s) to cluster '%s' - [ http://%s:7180/cmf/command/%s/details ]" % \
              (socket.getfqdn(cmx.cm_server), cmx.cm_server, cmd.id)
        #check.status_for_command("Hosts: %s " % host_list, cmd)
        print "Installing hosts. This might take a while."
        while cmd.success == None:
            sleep(20)
            cmd = cmd.fetch()
            print "Installing hosts... Checking"

        if cmd.success != True:
            print "cm_host_install failed: " + cmd.resultMessage
            exit(0)

    print "Host install finish, agents installed"
    hosts = []
    for host in api.get_all_hosts():
        if host.hostId not in [x.hostId for x in cluster.list_hosts()]:
            print "Adding {'ip': '%s', 'hostname': '%s', 'hostId': '%s'}" % (host.ipAddress, host.hostname, host.hostId)
            hosts.append(host.hostId)

    print "adding new hosts to cluster"
    if hosts:
        print "Adding hostId(s) to '%s'" % cmx.cluster_name
        print "%s" % hosts
        cluster.add_hosts(hosts)


def host_rack():
    """
    Add host to rack
    :return:
    """
    # TODO: Add host to rack
    print "> Add host to rack"
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    hosts = []
    for h in api.get_all_hosts():
        # host = api.create_host(h.hostId, h.hostname,
        # socket.gethostbyname(h.hostname),
        # "/default_rack")
        h.set_rack_id("/default_rack")
        hosts.append(h)

    cluster.add_hosts(hosts)


def deploy_parcel(parcel_product, parcel_version):
    """
    Deploy parcels
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    parcel = cluster.get_parcel(parcel_product, parcel_version)
    if parcel.stage != 'ACTIVATED':
        print "> Deploying parcel: [ %s-%s ]" % (parcel_product, parcel_version)
        parcel.start_download()
        # unlike other commands, check progress by looking at parcel stage and status
        while True:
            parcel = cluster.get_parcel(parcel_product, parcel_version)
            if parcel.stage == 'DISTRIBUTED' or parcel.stage == 'DOWNLOADED' or parcel.stage == 'ACTIVATED':
                break
           # if parcel.state.errors:
            #    raise Exception(str(parcel.state.errors))
            msg = " [%s: %s / %s]" % (parcel.stage, parcel.state.progress, parcel.state.totalProgress)
            sys.stdout.write(msg + " " * (78 - len(msg)) + "\r")
            sys.stdout.flush()

        print ""
        print "1. Parcel Stage: %s" % parcel.stage
        parcel.start_distribution()

        while True:
            parcel = cluster.get_parcel(parcel_product, parcel_version)
            if parcel.stage == 'DISTRIBUTED' or parcel.stage == 'ACTIVATED':
                break
           # if parcel.state.errors:
               # raise Exception(str(parcel.state.errors))
            msg = " [%s: %s / %s]" % (parcel.stage, parcel.state.progress, parcel.state.totalProgress)
            sys.stdout.write(msg + " " * (78 - len(msg)) + "\r")
            sys.stdout.flush()

        print "2. Parcel Stage: %s" % parcel.stage
        if parcel.stage == 'DISTRIBUTED':
            parcel.activate()

        while True:
            parcel = cluster.get_parcel(parcel_product, parcel_version)
            if parcel.stage != 'ACTIVATED':
                msg = " [%s: %s / %s]" % (parcel.stage, parcel.state.progress, parcel.state.totalProgress)
                sys.stdout.write(msg + " " * (78 - len(msg)) + "\r")
                sys.stdout.flush()
           # elif parcel.state.errors:
             #   raise Exception(str(parcel.state.errors))
            else:
                print "3. Parcel Stage: %s" % parcel.stage
                break


def setup_zookeeper():
    """
    Zookeeper
    > Waiting for ZooKeeper Service to initialize
    Starting ZooKeeper Service
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    service_type = "ZOOKEEPER"
    if cdh.get_service_type(service_type) is None:
        print "> %s" % service_type
        service_name = "zookeeper"
        print "Create %s service" % service_name
        cluster.create_service(service_name, service_type)
        service = cluster.get_service(service_name)
        hosts = management.get_hosts()
        cmhost= management.get_cmhost()
        service.update_config({"zookeeper_datadir_autocreate": True})

        # Role Config Group equivalent to Service Default Group
        for rcg in [x for x in service.get_all_role_config_groups()]:
            if rcg.roleType == "SERVER":
                rcg.update_config({"maxClientCnxns": "1024"})
                # Pick 3 hosts and deploy Zookeeper Server role
                # mingrui change install on primary, secondary, and CM
                
                print cmhost
                print [x for x in hosts if x.id == 0 ][0]
                print [x for x in hosts if x.id == 1 ][0]
                cdh.create_service_role(service, rcg.roleType, cmhost)
                cdh.create_service_role(service, rcg.roleType, [x for x in hosts if x.id == 0 ][0])
                cdh.create_service_role(service, rcg.roleType, [x for x in hosts if x.id == 1 ][0])

        # init_zookeeper not required as the API performs this when adding Zookeeper
        # check.status_for_command("Waiting for ZooKeeper Service to initialize", service.init_zookeeper())
        check.status_for_command("Starting ZooKeeper Service", service.start())


def setup_hdfs():
    """
    HDFS
    > Checking if the name directories of the NameNode are empty. Formatting HDFS only if empty.
    Starting HDFS Service
    > Creating HDFS /tmp directory
    :return:
    """
    api = ApiResource(server_host=cmx.cm_server, username=cmx.username, password=<PASSWORD>)
    cluster = api.get_cluster(cmx.cluster_name)
    service_type = "HDFS"
    if cdh.get_service_type(service_type) is None:
        print "> %s" % service_type
        service_name = "hdfs"
        print "Create %s service" % service_name
        cluster.create_service(service_name, service_type)
        service = cluster.get_service(service_name)
        hosts = management.get_hosts()

        # Service-Wide
        service_config = cdh.dependencies_for(service)
        service_config.update({"dfs_replication": "3",
                               "dfs_block_local_path_access_user": "impala,hbase,mapred,spark"})
{'fraction_non_alphanumeric': 0.1157419697371914, 'fraction_numerical': 0.006264932306875498, 'mean_word_length': 2.2684365781710913, 'pattern_counts': {'":': 16, '<': 16, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7458111', 'n_tokens_mistral': 6015, 'n_tokens_neox': 5721, 'n_words': 1254}