def format(self, format_string, *args, **kwargs):
    variables = kwargs
    
    if Environment.has_instance():
      env = Environment.get_instance()
      params = env.config.params
  
      # don't use checked_unite for this as it would interfere with reload(module)
      # for things like params and status_params; instead, start out copying
      # the environment parameters and add in any locally declared variables to
      # override existing env parameters
      all_params = params.copy()
    else:
      all_params = {}
      
    all_params.update(variables)

    self.convert_field = self.convert_field_protected
    result_protected = self.vformat(format_string, args, all_params)
    
    self.convert_field = self.convert_field_unprotected
    result_unprotected = self.vformat(format_string, args, all_params)
    
    if result_protected != result_unprotected:
      Logger.sensitive_strings[result_unprotected] = result_protected
      
    return result_unprotected

def falcon(type, action = None):
  import params

  if action == 'config':
    env = Environment.get_instance()
    # These 2 parameters are used in ../templates/client.properties.j2
    env.config.params["falcon_host"] = params.falcon_host
    env.config.params["falcon_port"] = params.falcon_port
    File(os.path.join(params.falcon_conf_dir, 'falcon-env.sh'),
      content = InlineTemplate(params.falcon_env_sh_template))

    File(os.path.join(params.falcon_conf_dir, 'client.properties'),
      content = Template('client.properties.j2'))

    PropertiesFile(os.path.join(params.falcon_conf_dir, 'runtime.properties'),
      properties = params.falcon_runtime_properties)

    PropertiesFile(os.path.join(params.falcon_conf_dir, 'startup.properties'),
      properties = params.falcon_startup_properties)

  if type == 'server':
    ServiceConfig(params.falcon_win_service_name,
      action = "change_user",
      username = params.falcon_user,
      password = Script.get_password(params.falcon_user))

    if action == 'start':
      Service(params.falcon_win_service_name, action = "start")

    if action == 'stop':
      Service(params.falcon_win_service_name, action = "stop")

  def action_remove(self):
    with Environment.get_instance_copy() as env:
      repo_file_name = self.resource.repo_file_name
      repo_dir = get_repo_dir()

      File(format("{repo_dir}/{repo_file_name}.repo"),
           action="delete")

  def action_create(self):
    with Environment.get_instance_copy() as env:
      repo_file_name = self.resource.repo_file_name
      repo_dir = get_repo_dir()
      new_content = InlineTemplate(self.resource.repo_template, repo_id=self.resource.repo_id, repo_file_name=self.resource.repo_file_name,
                             base_url=self.resource.base_url, mirror_list=self.resource.mirror_list)
      repo_file_path = format("{repo_dir}/{repo_file_name}.repo")

      if os.path.isfile(repo_file_path):
        existing_content_str = sudo.read_file(repo_file_path)
        new_content_str = new_content.get_content()
        if existing_content_str != new_content_str and OSCheck.is_suse_family():
          # We need to reset package manager's cache when we replace base urls
          # at existing repo. That is a case at least under SLES
          Logger.info("Flushing package manager cache since repo file content is about to change")
          checked_call(self.update_cmd, sudo=True)
        if self.resource.append_to_file:
          content = existing_content_str + '\n' + new_content_str
        else:
          content = new_content_str
      else: # If repo file does not exist yet
        content = new_content

      File(repo_file_path,
           content=content
      )

    def action_create(self):
        with Environment.get_instance_copy() as env:
            with tempfile.NamedTemporaryFile() as tmpf:
                repo_file_name = format("{repo_file_name}.list", repo_file_name=self.resource.repo_file_name)
                repo_file_path = format("{repo_dir}/{repo_file_name}", repo_dir=self.repo_dir)

                new_content = Template(
                    self.resource.repo_template,
                    package_type=self.package_type,
                    base_url=self.resource.base_url,
                    components=" ".join(self.resource.components),
                ).get_content()
                old_content = ""
                if self.resource.append_to_file and os.path.isfile(repo_file_path):
                    with open(repo_file_path) as repo_file:
                        old_content = repo_file.read() + "\n"

                File(tmpf.name, content=old_content + new_content)

                if not os.path.isfile(repo_file_path) or not filecmp.cmp(tmpf.name, repo_file_path):
                    File(repo_file_path, content=StaticFile(tmpf.name))

                    update_cmd_formatted = [format(x) for x in self.update_cmd]
                    # this is time expensive
                    retcode, out = checked_call(update_cmd_formatted, sudo=True)

                    # add public keys for new repos
                    missing_pkeys = set(re.findall(self.missing_pkey_regex, out))
                    for pkey in missing_pkeys:
                        Execute(
                            format(self.add_pkey_cmd),
                            timeout=15,  # in case we are on the host w/o internet (using localrepo), we should ignore hanging
                            ignore_failures=True,
                        )

  def action_create(self):
    filename = self.resource.filename
    xml_config_provider_config_dir = self.resource.conf_dir

    # |e - for html-like escaping of <,>,',"
    config_content = InlineTemplate('''  <configuration>
    {% for key, value in configurations_dict|dictsort %}
    <property>
      <name>{{ key|e }}</name>
      <value>{{ resource_management.core.source.InlineTemplate(str(value)).get_content() |e }}</value>
      {%- if not configuration_attrs is none -%}
      {%- for attrib_name, attrib_occurances in  configuration_attrs.items() -%}
      {%- for property_name, attrib_value in  attrib_occurances.items() -%}
      {% if property_name == key and attrib_name %}
      <{{attrib_name|e}}>{{attrib_value|e}}</{{attrib_name|e}}>
      {%- endif -%}
      {%- endfor -%}
      {%- endfor -%}
      {%- endif %}
    </property>
    {% endfor %}
  </configuration>''', extra_imports=[time, resource_management, resource_management.core, resource_management.core.source], configurations_dict=self.resource.configurations,
                                    configuration_attrs=self.resource.configuration_attributes)

    xml_config_dest_file_path = os.path.join(xml_config_provider_config_dir, filename)
    Logger.info("Generating config: {0}".format(xml_config_dest_file_path))

    with Environment.get_instance_copy() as env:
      File (xml_config_dest_file_path,
        content = config_content,
        owner = self.resource.owner,
        group = self.resource.group,
        mode = self.resource.mode,
        encoding = self.resource.encoding
      )

def get_check_command(oozie_url, host_name, configurations):
  if OOZIE_USER in configurations:
    oozie_user = configurations[OOZIE_USER]
  else:
    raise Exception("Oozie user is required")
    
  security_enabled = False
  if SECURITY_ENABLED in configurations:
    security_enabled = str(configurations[SECURITY_ENABLED]).upper() == 'TRUE'
  kerberos_env = None
  if security_enabled:
    if OOZIE_KEYTAB in configurations and OOZIE_PRINCIPAL in configurations:
      oozie_keytab = configurations[OOZIE_KEYTAB]
      oozie_principal = configurations[OOZIE_PRINCIPAL]

      # substitute _HOST in kerberos principal with actual fqdn
      oozie_principal = oozie_principal.replace('_HOST', host_name)
    else:
      raise KerberosPropertiesNotFound('The Oozie keytab and principal are required configurations when security is enabled.')

    # Create the kerberos credentials cache (ccache) file and set it in the environment to use
    # when executing curl
    env = Environment.get_instance()
    ccache_file = "{0}{1}oozie_alert_cc_{2}".format(env.tmp_dir, os.sep, os.getpid())
    kerberos_env = {'KRB5CCNAME': ccache_file}

    # Get the configured Kerberos executable search paths, if any
    if KERBEROS_EXECUTABLE_SEARCH_PATHS_KEY in configurations:
      kerberos_executable_search_paths = configurations[KERBEROS_EXECUTABLE_SEARCH_PATHS_KEY]
    else:
      kerberos_executable_search_paths = None

    klist_path_local = get_klist_path(kerberos_executable_search_paths)
    klist_command = format("{klist_path_local} -s {ccache_file}")

    # Determine if we need to kinit by testing to see if the relevant cache exists and has
    # non-expired tickets.  Tickets are marked to expire after 5 minutes to help reduce the number
    # it kinits we do but recover quickly when keytabs are regenerated
    return_code, _ = call(klist_command, user=oozie_user)
    if return_code != 0:
      kinit_path_local = get_kinit_path(kerberos_executable_search_paths)
      kinit_command = format("{kinit_path_local} -l 5m -kt {oozie_keytab} {oozie_principal}; ")

      # kinit
      Execute(kinit_command, 
              environment=kerberos_env,
              user=oozie_user,
      )

  # oozie configuration directory uses a symlink when > HDP 2.2
  oozie_config_directory = OOZIE_CONF_DIR_LEGACY
  if os.path.exists(OOZIE_CONF_DIR):
    oozie_config_directory = OOZIE_CONF_DIR

  command = "source {0}/oozie-env.sh ; oozie admin -oozie {1} -status".format(
    oozie_config_directory, oozie_url)

  return (command, kerberos_env, oozie_user)

    def action_remove(self):
        with Environment.get_instance_copy() as env:
            repo_file_name = format("{repo_file_name}.list", repo_file_name=self.resource.repo_file_name)
            repo_file_path = format("{repo_dir}/{repo_file_name}", repo_dir=self.repo_dir)

            if os.path.isfile(repo_file_path):
                File(repo_file_path, action="delete")

                # this is time expensive
                update_cmd_formatted = [format(x) for x in self.update_cmd]
                Execute(update_cmd_formatted)

    def format(self, format_string, *args, **kwargs):
        env = Environment.get_instance()
        variables = kwargs
        params = env.config.params
        all_params = checked_unite(variables, params)

        self.convert_field = self.convert_field_protected
        result_protected = self.vformat(format_string, args, all_params)

        self.convert_field = self.convert_field_unprotected
        result_unprotected = self.vformat(format_string, args, all_params)

        if result_protected != result_unprotected:
            Logger.sensitive_strings[result_unprotected] = result_protected

        return result_unprotected

  def action_create(self):
    template_tag = self.resource.template_tag
    qualified_file_name = self.resource.name
    file_name = os.path.basename(qualified_file_name)

    if not template_tag:
      template_name = format("{file_name}.j2")
    else:
      template_name = format("{file_name}-{template_tag}.j2")

    with Environment.get_instance_copy() as env:
      File( qualified_file_name,
       owner   = self.resource.owner,
       group   = self.resource.group,
       mode    = self.resource.mode,
       content = Template(template_name, extra_imports=self.resource.extra_imports)
      )

  def action_delayed(self, action_name, main_resource):
    resource = {}
    env = Environment.get_instance()
    if not 'hdfs_files' in env.config:
      env.config['hdfs_files'] = []

    # Put values in dictionary-resource
    for field_name, json_field_name in RESOURCE_TO_JSON_FIELDS.iteritems():
      if field_name == 'action':
        resource[json_field_name] = action_name
      elif field_name == 'mode' and main_resource.resource.mode:
        resource[json_field_name] = oct(main_resource.resource.mode)[1:]
      elif getattr(main_resource.resource, field_name):
        resource[json_field_name] = getattr(main_resource.resource, field_name)

    # Add resource to create
    env.config['hdfs_files'].append(resource)

  def __new__(cls, name, env=None, provider=None, **kwargs):
    if isinstance(name, list):
      while len(name) != 1:
        cls(name.pop(0), env, provider, **kwargs)
        
      name = name[0]
    
    env = env or Environment.get_instance()
    provider = provider or getattr(cls, 'provider', None)
    
    r_type = cls.__name__
    if r_type not in env.resources:
      env.resources[r_type] = {}

    obj = super(Resource, cls).__new__(cls)
    env.resources[r_type][name] = obj
    env.resource_list.append(obj)
    return obj

  def call_curl_request(self,user,keytab,principal, url, flag_http_response, request_method='GET',request_body='',header=''):
    """
    :param user: service user for which call is to be made
    :param keytab: keytab of service user
    :param principal: principal of service user
    :param url: url with which call is to be made
    :param flag_http_response: flag to get only response-code or response string
    :param request_method: http method (GET / POST / PUT / DELETE)
    :param request_body: data to be send along with the request
    :param header: http header required for the call
    :return: Returns the response error_msg , time_millis
    """
    response = None
    error_msg = None
    time_millis = 0
    response, error_msg, time_millis = curl_krb_request(Environment.get_instance().tmp_dir, keytab, principal, url, 'ranger_admin_calls',
                                                         None, flag_http_response, "Ranger-Admin API calls", user,kinit_timer_ms=0,method = request_method,body=request_body,header=header)

    return response, error_msg, time_millis

  def configure(self, env):
    import params
    env.set_params(params)

    if params.monitor_security_enabled and self.component == 'monitor':
      import os
      import random
      import string

      basedir = Environment.get_instance().config.basedir
      keystore_file = os.path.join(basedir, "files", "keystore.jks")
      truststore_file = os.path.join(basedir, "files", "cacerts.jks")
      cert_file = os.path.join(basedir, "files", "server.cer")

      if os.path.exists(keystore_file) or os.path.exists(truststore_file) or os.path.exists(cert_file):
        self.fail_with_error("trying to create monitor certs but they already existed")

      goodchars = string.lowercase + string.uppercase + string.digits + '#%+,-./:[email protected]^_'
      keypass = ''.join(random.choice(goodchars) for x in range(20))
      storepass = ''.join(random.choice(goodchars) for x in range(20))

      https_params = {}
      https_params[params.keystore_property] = params.keystore_path
      https_params[params.truststore_property] = params.truststore_path
      https_params[params.keystore_password_property] = keypass
      https_params[params.truststore_password_property] = storepass

      setup_conf_dir(name=self.component, extra_params=https_params)

      Execute( format("{java64_home}/bin/keytool -genkey -alias \"default\" -keyalg RSA -keypass {keypass} -storepass {storepass} -keystore {keystore_file} -dname \"CN=Unknown, OU=Unknown, O=Unknown, L=Unknown, ST=Unknown, C=Unknown\""),
               user=params.accumulo_user)
      Execute( format("{java64_home}/bin/keytool -export -alias \"default\" -storepass {storepass} -file {cert_file} -keystore {keystore_file}"),
               user=params.accumulo_user)
      Execute( format("echo \"yes\" | {java64_home}/bin/keytool -import -v -trustcacerts -alias \"default\" -file {cert_file} -keystore {truststore_file} -keypass {keypass} -storepass {storepass}"),
               user=params.accumulo_user)

      accumulo_StaticFile("keystore.jks")
      accumulo_StaticFile("cacerts.jks")

    else:
      setup_conf_dir(name=self.component)

  def action_create(self):
    filename = self.resource.filename
    comment_symbols = self.resource.comment_symbols
    delimiter = self.resource.key_value_delimiter
    properties = self.resource.properties
    unsaved_values = properties.keys()
    new_content_lines = []
    
    if sudo.path_isfile(filename):
      file_content = sudo.read_file(filename, encoding=self.resource.encoding)
      new_content_lines += file_content.split('\n')

      Logger.info(format("Modifying existing properties file: {filename}"))
      
      for line_num in range(len(new_content_lines)):
        line = new_content_lines[line_num]
        
        if line.lstrip() and not line.lstrip()[0] in comment_symbols and delimiter in line:
          in_var_name = line.split(delimiter)[0].strip()
          in_var_value = line.split(delimiter)[1].strip()
          
          if in_var_name in properties:
            value = InlineTemplate(unicode(properties[in_var_name])).get_content()
            new_content_lines[line_num] = u"{0}{1}{2}".format(unicode(in_var_name), delimiter, value)
            unsaved_values.remove(in_var_name)
    else:
      Logger.info(format("Creating new properties file as {filename} doesn't exist"))
       
    for property_name in unsaved_values:
      value = InlineTemplate(unicode(properties[property_name])).get_content()
      line = u"{0}{1}{2}".format(unicode(property_name), delimiter, value)
      new_content_lines.append(line)
          
    with Environment.get_instance_copy() as env:
      File (filename,
            content = u"\n".join(new_content_lines) + "\n",
            owner = self.resource.owner,
            group = self.resource.group,
            mode = self.resource.mode,
            encoding = self.resource.encoding,
      )

  def action_execute(self, main_resource):
    env = Environment.get_instance()

    # Check required parameters
    main_resource.assert_parameter_is_set('user')

    if not 'hdfs_files' in env.config or not env.config['hdfs_files']:
      Logger.info("No resources to create. 'create_on_execute' or 'delete_on_execute' wasn't triggered before this 'execute' action.")
      return

    hadoop_bin_dir = main_resource.resource.hadoop_bin_dir
    hadoop_conf_dir = main_resource.resource.hadoop_conf_dir
    user = main_resource.resource.user
    security_enabled = main_resource.resource.security_enabled
    keytab_file = main_resource.resource.keytab
    kinit_path = main_resource.resource.kinit_path_local
    logoutput = main_resource.resource.logoutput
    principal_name = main_resource.resource.principal_name
    jar_path=JAR_PATH
    timestamp = time.time()
    json_path=format(JSON_PATH)

    if security_enabled:
      main_resource.kinit()

    # Write json file to disk
    File(json_path,
         owner = user,
         content = json.dumps(env.config['hdfs_files'])
    )

    # Execute jar to create/delete resources in hadoop
    Execute(format("hadoop --config {hadoop_conf_dir} jar {jar_path} {json_path}"),
            user=user,
            path=[hadoop_bin_dir],
            logoutput=logoutput,
    )

    # Clean
    env.config['hdfs_files'] = []

  def action_create(self):
    filename = self.resource.filename
    dir = self.resource.dir
    if dir == None:
      filepath = filename
    else:
      filepath = os.path.join(dir, filename)

    config_content = InlineTemplate('''# Generated by Apache Ambari. {{time.asctime(time.localtime())}}
    {% for key, value in properties_dict|dictsort %}
{{key}}{{key_value_delimiter}}{{ resource_management.core.source.InlineTemplate(str(value)).get_content() }}{% endfor %}
    ''', extra_imports=[time, resource_management, resource_management.core, resource_management.core.source], properties_dict=self.resource.properties, key_value_delimiter=self.resource.key_value_delimiter)

    Logger.info(format("Generating properties file: {filepath}"))

    with Environment.get_instance_copy() as env:
      File (format("{filepath}"),
            content = config_content,
            owner = self.resource.owner,
            group = self.resource.group,
            mode = self.resource.mode
      )

  def __init__(self, name, env=None, provider=None, **kwargs):
    if isinstance(name, list):
      name = name.pop(0)
    
    if hasattr(self, 'name'):
      return

    self.env = env or Environment.get_instance()
    self.name = name
     
    self.provider = provider or getattr(self, 'provider', None)

    self.arguments = {}
    for key, value in kwargs.items():
      try:
        arg = self._arguments[key]
      except KeyError:
        raise Fail("%s received unsupported argument %s" % (self, key))
      else:
        try:
          self.arguments[key] = arg.validate(value)
        except InvalidArgument, exc:
          raise InvalidArgument("%s %s" % (self, exc))

    def action_run(self):
        kinit__path_local = self.resource.kinit_path_local
        keytab = self.resource.keytab
        conf_dir = self.resource.conf_dir
        command = self.resource.command
        principal = self.resource.principal

        if isinstance(command, (list, tuple)):
            command = " ".join(quote_bash_args(x) for x in command)

        with Environment.get_instance_copy() as env:
            if self.resource.security_enabled and not self.resource.kinit_override:
                Execute(format("{kinit__path_local} -kt {keytab} {principal}"), path=["/bin"], user=self.resource.user)

            Execute(
                format("hadoop --config {conf_dir} {command}"),
                user=self.resource.user,
                tries=self.resource.tries,
                try_sleep=self.resource.try_sleep,
                logoutput=self.resource.logoutput,
                path=self.resource.bin_dir,
                environment=self.resource.environment,
            )

def execute(configurations={}, parameters={}, host_name=None):
  """
  Returns a tuple containing the result code and a pre-formatted result label

  Keyword arguments:
  configurations (dictionary): a mapping of configuration key to value
  parameters (dictionary): a mapping of script parameter key to value
  host_name (string): the name of this host where the alert is running
  """

  result_code = RESULT_CODE_UNKNOWN

  if configurations is None:
    return (result_code, ['There were no configurations supplied to the script.'])

  webhcat_port = WEBHCAT_PORT_DEFAULT
  if TEMPLETON_PORT_KEY in configurations:
    webhcat_port = int(configurations[TEMPLETON_PORT_KEY])

  security_enabled = False
  if SECURITY_ENABLED_KEY in configurations:
    security_enabled = configurations[SECURITY_ENABLED_KEY].lower() == 'true'

  # parse script arguments
  connection_timeout = CONNECTION_TIMEOUT_DEFAULT
  curl_connection_timeout = CURL_CONNECTION_TIMEOUT_DEFAULT
  if CONNECTION_TIMEOUT_KEY in parameters:
    connection_timeout = float(parameters[CONNECTION_TIMEOUT_KEY])
    curl_connection_timeout = str(int(connection_timeout))


  # the alert will always run on the webhcat host
  if host_name is None:
    host_name = socket.getfqdn()

  smokeuser = SMOKEUSER_DEFAULT

  if SMOKEUSER_KEY in configurations:
    smokeuser = configurations[SMOKEUSER_KEY]

  # webhcat always uses http, never SSL
  query_url = "http://{0}:{1}/templeton/v1/status?user.name={2}".format(host_name, webhcat_port, smokeuser)

  # initialize
  total_time = 0
  json_response = {}

  if security_enabled:
    try:
      # defaults
      smokeuser_keytab = SMOKEUSER_KEYTAB_DEFAULT
      smokeuser_principal = SMOKEUSER_PRINCIPAL_DEFAULT

      # check script params
      if SMOKEUSER_PRINCIPAL_SCRIPT_PARAM_KEY in parameters:
        smokeuser_principal = parameters[SMOKEUSER_PRINCIPAL_SCRIPT_PARAM_KEY]
      if SMOKEUSER_KEYTAB_SCRIPT_PARAM_KEY in parameters:
        smokeuser_keytab = parameters[SMOKEUSER_KEYTAB_SCRIPT_PARAM_KEY]

      # check configurations last as they should always take precedence
      if SMOKEUSER_PRINCIPAL_KEY in configurations:
        smokeuser_principal = configurations[SMOKEUSER_PRINCIPAL_KEY]
      if SMOKEUSER_KEYTAB_KEY in configurations:
        smokeuser_keytab = configurations[SMOKEUSER_KEYTAB_KEY]

      # Get the configured Kerberos executable search paths, if any
      kerberos_executable_search_paths = None
      if KERBEROS_EXECUTABLE_SEARCH_PATHS_KEY in configurations:
        kerberos_executable_search_paths = configurations[KERBEROS_EXECUTABLE_SEARCH_PATHS_KEY]

      env = Environment.get_instance()
      stdout, stderr, time_millis = curl_krb_request(env.tmp_dir, smokeuser_keytab, smokeuser_principal,
                                                      query_url, "webhcat_alert_cc_", kerberos_executable_search_paths, True,
                                                      "WebHCat Server Status", smokeuser,
                                                      connection_timeout=curl_connection_timeout)

      # check the response code
      response_code = int(stdout)

      # 0 indicates no connection
      if response_code == 0:
        label = CRITICAL_CONNECTION_MESSAGE.format(query_url)
        return (RESULT_CODE_CRITICAL, [label])

      # any other response aside from 200 is a problem
      if response_code != 200:
        label = CRITICAL_HTTP_MESSAGE.format(response_code, query_url)
        return (RESULT_CODE_CRITICAL, [label])

      # now that we have the http status and it was 200, get the content
      stdout, stderr, total_time = curl_krb_request(env.tmp_dir, smokeuser_keytab, smokeuser_principal,
                                                      query_url, "webhcat_alert_cc_", kerberos_executable_search_paths,
                                                      False, "WebHCat Server Status", smokeuser,
                                                      connection_timeout=curl_connection_timeout)
      json_response = json.loads(stdout)
    except Exception, exception:
      return (RESULT_CODE_CRITICAL, [str(exception)])