#!/usr/bin/env 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 multiprocessing
import logging
import ambari_simplejson as json
import sys
import os
import socket
import time
import threading
import urllib2
import pprint
from random import randint
import re
import subprocess
import functools

import hostname
import security
import ssl
import AmbariConfig

from ambari_agent.Heartbeat import Heartbeat
from ambari_agent.Register import Register
from ambari_agent.ActionQueue import ActionQueue
from ambari_agent.FileCache import FileCache
from ambari_agent.NetUtil import NetUtil
from ambari_agent.LiveStatus import LiveStatus
from ambari_agent.AlertSchedulerHandler import AlertSchedulerHandler
from ambari_agent.ClusterConfiguration import  ClusterConfiguration
from ambari_agent.RecoveryManager import  RecoveryManager
from ambari_agent.HeartbeatHandlers import HeartbeatStopHandlers, bind_signal_handlers
from ambari_agent.ExitHelper import ExitHelper
from ambari_agent.StatusCommandsExecutor import MultiProcessStatusCommandsExecutor, SingleProcessStatusCommandsExecutor
from ambari_commons.network import reconfigure_urllib2_opener
from resource_management.libraries.functions.version import compare_versions
from ambari_commons.os_utils import get_used_ram

logger = logging.getLogger(__name__)

AGENT_AUTO_RESTART_EXIT_CODE = 77

AGENT_RAM_OVERUSE_MESSAGE = "Ambari-agent RAM usage {used_ram} MB went above {config_name}={max_ram} MB. Restarting ambari-agent to clean the RAM."

class Controller(threading.Thread):

  def __init__(self, config, server_hostname, heartbeat_stop_callback = None):
    threading.Thread.__init__(self)
    logger.debug('Initializing Controller RPC thread.')
    if heartbeat_stop_callback is None:
      heartbeat_stop_callback = HeartbeatStopHandlers()

    self.version = self.read_agent_version(config)
    self.lock = threading.Lock()
    self.safeMode = True
    self.credential = None
    self.config = config
    self.hostname = hostname.hostname(config)
    self.serverHostname = server_hostname
    server_secured_url = 'https://' + self.serverHostname + \
                         ':' + config.get('server', 'secured_url_port')
    self.registerUrl = server_secured_url + '/agent/v1/register/' + self.hostname
    self.heartbeatUrl = server_secured_url + '/agent/v1/heartbeat/' + self.hostname
    self.componentsUrl = server_secured_url + '/agent/v1/components/'
    self.netutil = NetUtil(self.config, heartbeat_stop_callback)
    self.responseId = -1
    self.repeatRegistration = False
    self.isRegistered = False
    self.cachedconnect = None
    self.max_reconnect_retry_delay = int(config.get('server','max_reconnect_retry_delay', default=30))
    self.hasMappedComponents = True
    self.statusCommandsExecutor = None

    # Event is used for synchronizing heartbeat iterations (to make possible
    # manual wait() interruption between heartbeats )
    self.heartbeat_stop_callback = heartbeat_stop_callback
    # List of callbacks that are called at agent registration
    self.registration_listeners = []

    # pull config directory out of config
    cache_dir = config.get('agent', 'cache_dir')
    if cache_dir is None:
      cache_dir = '/var/lib/ambari-agent/cache'

    self.max_ram_soft = int(config.get('agent','memory_threshold_soft_mb', default=0))
    self.max_ram_hard = int(config.get('agent','memory_threshold_hard_mb', default=0))

    stacks_cache_dir = os.path.join(cache_dir, FileCache.STACKS_CACHE_DIRECTORY)
    common_services_cache_dir = os.path.join(cache_dir, FileCache.COMMON_SERVICES_DIRECTORY)
    extensions_cache_dir = os.path.join(cache_dir, FileCache.EXTENSIONS_CACHE_DIRECTORY)
    host_scripts_cache_dir = os.path.join(cache_dir, FileCache.HOST_SCRIPTS_CACHE_DIRECTORY)
    alerts_cache_dir = os.path.join(cache_dir, FileCache.ALERTS_CACHE_DIRECTORY)
    cluster_config_cache_dir = os.path.join(cache_dir, FileCache.CLUSTER_CONFIGURATION_CACHE_DIRECTORY)
    recovery_cache_dir = os.path.join(cache_dir, FileCache.RECOVERY_CACHE_DIRECTORY)

    self.heartbeat_idle_interval_min = int(self.config.get('heartbeat', 'idle_interval_min')) if self.config.get('heartbeat', 'idle_interval_min') else self.netutil.HEARTBEAT_IDLE_INTERVAL_DEFAULT_MIN_SEC
    if self.heartbeat_idle_interval_min < 1:
      self.heartbeat_idle_interval_min = 1

    self.heartbeat_idle_interval_max = int(self.config.get('heartbeat', 'idle_interval_max')) if self.config.get('heartbeat', 'idle_interval_max') else self.netutil.HEARTBEAT_IDLE_INTERVAL_DEFAULT_MAX_SEC

    if self.heartbeat_idle_interval_min > self.heartbeat_idle_interval_max:
      raise Exception("Heartbeat minimum interval={0} seconds can not be greater than the maximum interval={1} seconds !".format(self.heartbeat_idle_interval_min, self.heartbeat_idle_interval_max))

    self.get_heartbeat_interval = functools.partial(self.netutil.get_agent_heartbeat_idle_interval_sec, self.heartbeat_idle_interval_min, self.heartbeat_idle_interval_max)

    self.recovery_manager = RecoveryManager(recovery_cache_dir)

    self.cluster_configuration = ClusterConfiguration(cluster_config_cache_dir)

    self.move_data_dir_mount_file()

    if not config.use_system_proxy_setting():
      reconfigure_urllib2_opener(ignore_system_proxy=True)

    self.alert_scheduler_handler = AlertSchedulerHandler(alerts_cache_dir,
      stacks_cache_dir, common_services_cache_dir, extensions_cache_dir,
      host_scripts_cache_dir, self.cluster_configuration, config,
      self.recovery_manager)

    self.alert_scheduler_handler.start()


  def read_agent_version(self, config):
    data_dir = config.get('agent', 'prefix')
    ver_file = os.path.join(data_dir, 'version')
    f = open(ver_file, "r")
    version = f.read().strip()
    f.close()
    return version


  def __del__(self):
    logger.info("Server connection disconnected.")

  def registerWithServer(self):
    """
    :return: returning from current method without setting self.isRegistered
    to True will lead to agent termination.
    """
    LiveStatus.SERVICES = []
    LiveStatus.CLIENT_COMPONENTS = []
    LiveStatus.COMPONENTS = []
    ret = {}

    while not self.isRegistered:
      try:
        data = json.dumps(self.register.build(self.version))
        prettyData = pprint.pformat(data)

        try:
          server_ip = socket.gethostbyname(self.hostname)
          logger.info("Registering with %s (%s) (agent=%s)", self.hostname, server_ip, prettyData)
        except socket.error:
          logger.warn("Unable to determine the IP address of '%s', agent registration may fail (agent=%s)",
                      self.hostname, prettyData)

        ret = self.sendRequest(self.registerUrl, data)
        prettyData = pprint.pformat(ret)
        logger.debug("Registration response is %s", prettyData)

        # exitstatus is a code of error which was raised on server side.
        # exitstatus = 0 (OK - Default)
        # exitstatus = 1 (Registration failed because different version of agent and server)
        exitstatus = 0
        if 'exitstatus' in ret.keys():
          exitstatus = int(ret['exitstatus'])

        if exitstatus == 1:
          # log - message, which will be printed to agents log
          if 'log' in ret.keys():
            log = ret['log']
            logger.error(log)
          self.isRegistered = False
          self.repeatRegistration = False
          return ret

        self.responseId = int(ret['responseId'])
        logger.info("Registration Successful (response id = %s)", self.responseId)

        self.isRegistered = True

        # always update cached cluster configurations on registration
        # must be prior to any other operation
        self.cluster_configuration.update_configurations_from_heartbeat(ret)
        self.recovery_manager.update_configuration_from_registration(ret)
        self.config.update_configuration_from_registration(ret)
        logger.debug("Updated config:" + str(self.config))

        # Start StatusCommandExecutor child process or restart it if already running
        # in order to receive up to date agent config.
        self.statusCommandsExecutor.relaunch("REGISTER_WITH_SERVER")

        if 'statusCommands' in ret.keys():
          logger.debug("Got status commands on registration.")
          self.addToStatusQueue(ret['statusCommands'])
        else:
          self.hasMappedComponents = False

        # always update alert definitions on registration
        self.alert_scheduler_handler.update_definitions(ret)
      except ssl.SSLError:
        self.repeatRegistration = False
        self.isRegistered = False
        return
      except Exception, ex:
        # try a reconnect only after a certain amount of random time
        delay = randint(0, self.max_reconnect_retry_delay)
        logger.error("Unable to connect to: " + self.registerUrl, exc_info=True)
        logger.error("Error:" + str(ex))
        logger.warn(""" Sleeping for {0} seconds and then trying again """.format(delay,))
        time.sleep(delay)

    return ret

  def cancelCommandInQueue(self, commands):
    """ Remove from the queue commands, kill the process if it's in progress """
    if commands:
      try:
        self.actionQueue.cancel(commands)
      except Exception, err:
        logger.error("Exception occurred on commands cancel: %s", err.message)

  def addToQueue(self, commands):
    """Add to the queue for running the commands """
    """ Put the required actions into the Queue """
    """ Verify if the action is to reboot or not """
    if not commands:
      logger.debug("No commands received from %s", self.serverHostname)
    else:
      """Only add to the queue if not empty list """
      logger.info("Adding %s commands. Heartbeat id = %s", len(commands), self.responseId)
      if 'clusterName' in commands[0].keys():
        self.updateComponents(commands[0]['clusterName'])
      self.actionQueue.put(commands)

  def addToStatusQueue(self, commands):
    if not commands:
      logger.debug("No status commands received from %s", self.serverHostname)
    else:
      logger.info("Adding %s status commands. Heartbeat id = %s", len(commands), self.responseId)
      if 'clusterName' in commands[0].keys():
        self.updateComponents(commands[0]['clusterName'])
      self.recovery_manager.process_status_commands(commands)
      self.actionQueue.put_status(commands)
    pass

  # For testing purposes
  DEBUG_HEARTBEAT_RETRIES = 0
  DEBUG_SUCCESSFULL_HEARTBEATS = 0
  DEBUG_STOP_HEARTBEATING = False

  def trigger_heartbeat(self):
    self.heartbeat_stop_callback.set_heartbeat()

  def heartbeatWithServer(self):
    self.DEBUG_HEARTBEAT_RETRIES = 0
    self.DEBUG_SUCCESSFULL_HEARTBEATS = 0
    retry = False
    certVerifFailed = False
    state_interval = int(self.config.get('heartbeat', 'state_interval_seconds', '60'))

    # last time when state was successfully sent to server
    last_state_timestamp = 0.0

    # in order to be able to check from logs that heartbeats processing
    # still running we log a message. However to avoid generating too
    # much log when the heartbeat runs at a higher rate (e.g. 1 second intervals)
    # we log the message at the same interval as 'state interval'
    heartbeat_running_msg_timestamp = 0.0

    # Prevent excessive logging by logging only at specific intervals
    getrecoverycommands_timestamp = 0.0
    getrecoverycommands_interval = self.netutil.HEARTBEAT_IDLE_INTERVAL_DEFAULT_MAX_SEC

    heartbeat_interval = self.netutil.HEARTBEAT_IDLE_INTERVAL_DEFAULT_MAX_SEC

    while not self.DEBUG_STOP_HEARTBEATING:
      current_time = time.time()
      logging_level = logging.DEBUG
      if current_time - heartbeat_running_msg_timestamp > state_interval:
        # log more steps every minute or so
        logging_level = logging.INFO
        heartbeat_running_msg_timestamp = current_time

      try:
        logger.log(logging_level, "Heartbeat (response id = %s) with server is running...", self.responseId)

        send_state = False
        if not retry:
          if current_time - last_state_timestamp > state_interval:
            send_state = True

          logger.log(logging_level, "Building heartbeat message")

          data = json.dumps(self.heartbeat.build(self.responseId, send_state, self.hasMappedComponents))
        else:
          self.DEBUG_HEARTBEAT_RETRIES += 1

        if logger.isEnabledFor(logging.DEBUG):
          logger.log(logging_level, "Sending Heartbeat (id = %s): %s", self.responseId, data)
        else:
          logger.log(logging_level, "Sending Heartbeat (id = %s)", self.responseId)

        response = self.sendRequest(self.heartbeatUrl, data)
        exitStatus = 0
        if 'exitstatus' in response.keys():
          exitStatus = int(response['exitstatus'])

        if exitStatus != 0:
          raise Exception(response)

        serverId = int(response['responseId'])

        logger.log(logging_level, 'Heartbeat response received (id = %s)', serverId)

        cluster_size = int(response['clusterSize']) if 'clusterSize' in response.keys() else -1

        # TODO: this needs to be revised if hosts can be shared across multiple clusters
        heartbeat_interval = self.get_heartbeat_interval(cluster_size) \
          if cluster_size > 0 \
          else self.netutil.HEARTBEAT_IDLE_INTERVAL_DEFAULT_MAX_SEC

        logger.log(logging_level, "Heartbeat interval is %s seconds", heartbeat_interval)

        if 'hasMappedComponents' in response.keys():
          self.hasMappedComponents = response['hasMappedComponents'] is not False

        if 'hasPendingTasks' in response.keys():
          has_pending_tasks = bool(response['hasPendingTasks'])
          self.recovery_manager.set_paused(has_pending_tasks)

        if 'registrationCommand' in response.keys():
          # check if the registration command is None. If none skip
          if response['registrationCommand'] is not None:
            logger.info("RegistrationCommand received - repeat agent registration")
            self.isRegistered = False
            self.repeatRegistration = True
            return

        used_ram = get_used_ram()/1000
        # dealing with a possible memory leaks
        if self.max_ram_soft and used_ram >= self.max_ram_soft and not self.actionQueue.tasks_in_progress_or_pending():
          logger.error(AGENT_RAM_OVERUSE_MESSAGE.format(used_ram=used_ram, config_name="memory_threshold_soft_mb", max_ram=self.max_ram_soft))
          self.restartAgent()
        if self.max_ram_hard and used_ram >= self.max_ram_hard:
          logger.error(AGENT_RAM_OVERUSE_MESSAGE.format(used_ram=used_ram, config_name="memory_threshold_hard_mb", max_ram=self.max_ram_hard))
          self.restartAgent()

        if serverId != self.responseId + 1:
          logger.error("Error in responseId sequence - restarting")
          self.restartAgent()
        else:
          self.responseId = serverId
          if send_state:
            last_state_timestamp = current_time

        # if the response contains configurations, update the in-memory and
        # disk-based configuration cache (execution and alert commands have this)
        logger.log(logging_level, "Updating configurations from heartbeat")
        self.cluster_configuration.update_configurations_from_heartbeat(response)

        response_keys = response.keys()

        # there's case when canceled task can be processed in Action Queue.execute before adding rescheduled task to queue
        # this can cause command failure instead result suppression
        # so canceling and putting rescheduled commands should be executed atomically
        if 'cancelCommands' in response_keys or 'executionCommands' in response_keys:
          logger.log(logging_level, "Adding cancel/execution commands")
        with self.actionQueue.lock:
          if 'cancelCommands' in response_keys:
            self.cancelCommandInQueue(response['cancelCommands'])

          if 'executionCommands' in response_keys:
            execution_commands = response['executionCommands']
            self.recovery_manager.process_execution_commands(execution_commands)
            self.addToQueue(execution_commands)

        if 'statusCommands' in response_keys:
          # try storing execution command details and desired state
          self.addToStatusQueue(response['statusCommands'])

        if current_time - getrecoverycommands_timestamp > getrecoverycommands_interval:
          getrecoverycommands_timestamp = current_time
          if not self.actionQueue.tasks_in_progress_or_pending():
            logger.log(logging_level, "Adding recovery commands")
            recovery_commands = self.recovery_manager.get_recovery_commands()
            for recovery_command in recovery_commands:
              logger.info("Adding recovery command %s for component %s",
                          recovery_command['roleCommand'], recovery_command['role'])
              self.addToQueue([recovery_command])

        if 'alertDefinitionCommands' in response_keys:
          logger.log(logging_level, "Updating alert definitions")
          self.alert_scheduler_handler.update_definitions(response)

        if 'alertExecutionCommands' in response_keys:
          logger.log(logging_level, "Executing alert commands")
          self.alert_scheduler_handler.execute_alert(response['alertExecutionCommands'])

        if "true" == response['restartAgent']:
          logger.error("Received the restartAgent command")
          self.restartAgent()
        else:
          logger.debug("No commands sent from %s", self.serverHostname)

        if retry:
          logger.info("Reconnected to %s", self.heartbeatUrl)

        if "recoveryConfig" in response:
          # update the list of components enabled for recovery
          logger.log(logging_level, "Updating recovery config")
          self.recovery_manager.update_configuration_from_registration(response)

        retry = False
        certVerifFailed = False
        self.DEBUG_SUCCESSFULL_HEARTBEATS += 1
        self.DEBUG_HEARTBEAT_RETRIES = 0
        self.heartbeat_stop_callback.reset_heartbeat()
      except ssl.SSLError:
        self.repeatRegistration=False
        self.isRegistered = False
        logger.exception("SSLError while trying to heartbeat.")
        return
      except Exception, err:
        if "code" in err:
          logger.error(err.code)
        else:
          logException = False
          if logger.isEnabledFor(logging.DEBUG):
            logException = True

          exceptionMessage = str(err)
          errorMessage = "Unable to reconnect to {0} (attempts={1}, details={2})".format(self.heartbeatUrl, self.DEBUG_HEARTBEAT_RETRIES, exceptionMessage)

          if not retry:
            errorMessage = "Connection to {0} was lost (details={1})".format(self.serverHostname, exceptionMessage)

          logger.error(errorMessage, exc_info=logException)

          if 'certificate verify failed' in str(err) and not certVerifFailed:
            logger.warn("Server certificate verify failed. Did you regenerate server certificate?")
            certVerifFailed = True

        self.cachedconnect = None  # Previous connection is broken now
        retry = True

        #randomize the heartbeat
        delay = randint(0, self.max_reconnect_retry_delay)
        time.sleep(delay)

      # Sleep for some time
      timeout = heartbeat_interval - self.netutil.MINIMUM_INTERVAL_BETWEEN_HEARTBEATS
      logger.log(logging_level, "Waiting %s for next heartbeat", timeout)

      if 0 == self.heartbeat_stop_callback.wait(timeout, self.netutil.MINIMUM_INTERVAL_BETWEEN_HEARTBEATS):
        # Stop loop when stop event received
        logger.info("Stop event received")
        self.DEBUG_STOP_HEARTBEATING=True

      logger.log(logging_level, "Wait for next heartbeat over")

  def run(self):
    try:
      self.actionQueue = ActionQueue(self.config, controller=self)
      if self.config.get_multiprocess_status_commands_executor_enabled():
        self.statusCommandsExecutor = MultiProcessStatusCommandsExecutor(self.config, self.actionQueue)
      else:
        self.statusCommandsExecutor = SingleProcessStatusCommandsExecutor(self.config, self.actionQueue)
      ExitHelper().register(self.statusCommandsExecutor.kill, "CLEANUP_KILLING", can_relaunch=False)
      self.actionQueue.start()
      self.register = Register(self.config)
      self.heartbeat = Heartbeat(self.actionQueue, self.config, self.alert_scheduler_handler.collector())

      opener = urllib2.build_opener()
      urllib2.install_opener(opener)

      while True:
        self.repeatRegistration = False
        self.registerAndHeartbeat()
        if not self.repeatRegistration:
          logger.info("Finished heartbeating and registering cycle")
          break
    except:
      logger.exception("Controller thread failed with exception:")
      raise

    logger.info("Controller thread has successfully finished")

  def registerAndHeartbeat(self):
    registerResponse = self.registerWithServer()

    if "response" in registerResponse:
      message = registerResponse["response"]
      logger.info("Registration response from %s was %s", self.serverHostname, message)

      if self.isRegistered:
        # Clearing command queue to stop executing "stale" commands
        # after registration
        logger.info('Resetting ActionQueue...')
        self.actionQueue.reset()

        # Process callbacks
        for callback in self.registration_listeners:
          callback()

        time.sleep(self.netutil.HEARTBEAT_IDLE_INTERVAL_DEFAULT_MAX_SEC)
        self.heartbeatWithServer()
      else:
        logger.info("Registration response from %s didn't contain 'response' as a key".format(self.serverHostname))

  def restartAgent(self):
    ExitHelper().exit(AGENT_AUTO_RESTART_EXIT_CODE)


  def sendRequest(self, url, data):
    response = None

    try:
      if self.cachedconnect is None: # Lazy initialization
        self.cachedconnect = security.CachedHTTPSConnection(self.config, self.serverHostname)
      req = urllib2.Request(url, data, {'Content-Type': 'application/json',
                                        'Accept-encoding': 'gzip'})
      response = self.cachedconnect.request(req)
      return json.loads(response)
    except Exception, exception:
      if response is None:
        raise IOError('Request to {0} failed due to {1}'.format(url, str(exception)))
      else:
        raise IOError('Response parsing failed! Request data: ' + str(data)
                      + '; Response: ' + str(response))


  def updateComponents(self, cluster_name):
    if LiveStatus.SERVICES:
      return

    logger.debug("Updating components map of cluster " + cluster_name)

    # May throw IOError on server connection error
    response = self.sendRequest(self.componentsUrl + cluster_name, None)
    logger.debug("Response from %s was %s", self.serverHostname, str(response))

    services, client_components, server_components = [], [], []
    for service, components in response['components'].items():
      services.append(service)
      for component, category in components.items():
        service_component = {"serviceName": service, "componentName": component}
        if category == 'CLIENT':
          client_components.append(service_component)
        else:
          server_components.append(service_component)

    LiveStatus.SERVICES = services
    LiveStatus.CLIENT_COMPONENTS = client_components
    LiveStatus.COMPONENTS = server_components

    logger.debug("Components map updated")
    logger.debug("LiveStatus.SERVICES" + str(LiveStatus.SERVICES))
    logger.debug("LiveStatus.CLIENT_COMPONENTS" + str(LiveStatus.CLIENT_COMPONENTS))
    logger.debug("LiveStatus.COMPONENTS" + str(LiveStatus.COMPONENTS))

  def get_status_commands_executor(self):
    return self.statusCommandsExecutor

  def move_data_dir_mount_file(self):
    """
    In Ambari 2.1.2, we moved the dfs_data_dir_mount.hist to a static location
    because /etc/hadoop/conf points to a symlink'ed location that would change during
    Stack Upgrade.
    """
    try:
      version = self.get_version()
      logger.debug("Ambari Agent version {0}".format(version))
      if compare_versions(version, "2.1.2") >= 0:
        source_file = "/etc/hadoop/conf/dfs_data_dir_mount.hist"
        destination_file = "/var/lib/ambari-agent/data/datanode/dfs_data_dir_mount.hist"
        if os.path.exists(source_file) and not os.path.exists(destination_file):
          command = "mkdir -p %s" % os.path.dirname(destination_file)
          logger.info("Moving Data Dir Mount History file. Executing command: %s" % command)
          return_code = subprocess.call(command, shell=True)
          logger.info("Return code: %d" % return_code)

          command = "mv %s %s" % (source_file, destination_file)
          logger.info("Moving Data Dir Mount History file. Executing command: %s" % command)
          return_code = subprocess.call(command, shell=True)
          logger.info("Return code: %d" % return_code)
    except Exception, e:
      logger.error("Exception in move_data_dir_mount_file(). Error: {0}".format(str(e)))

  def get_version(self):
    version = self.version
    matches = re.findall(r"[\d+.]+",version)
    if not matches:
      logger.warning("No version match result, use original version {0}".format(version))
      return version
    else:
      return matches[0]

def main(argv=None):
  # Allow Ctrl-C

  formatter = logging.Formatter("%(asctime)s %(filename)s:%(lineno)d - \
    %(message)s")
  stream_handler = logging.StreamHandler()
  stream_handler.setFormatter(formatter)
  logger.addHandler(stream_handler)

  logger.info('Starting Server RPC Thread: %s' % ' '.join(sys.argv))

  config = AmbariConfig.config
  heartbeat_stop_callback = bind_signal_handlers()
  collector = Controller(config, heartbeat_stop_callback)
  collector.start()
  collector.run()

if __name__ == '__main__':
  main()