# Apache Atlas Hive Plugin Configuration

    module.exports = (service) ->
      options = service.options

## Environment

      options.conf_dir = service.deps.hive_server2.options.conf_dir

## Libs
      
      options.oozie = service.deps.oozie
      options.oozie_user  = options.oozie.user if options.oozie
      options.atlas = service.deps.atlas[0]

## Configuration

Mechanism: The Apache Hive Hook (or Atlas hook) is running directly inside the
hiveserver2's JVM. As a consequence, Kafka Client configuration must be configured 
on the hiveserver2 side.
Ryba does the follwing steps:
- Configure Hiveserver2' hive-site.xml file with Atlas Props

Write Atlas Configuration file for hive's bridge. Atlas.Hive.Hook needs the atlas applcation
file in the hive-server2's configuration directory. 
Note: The file needs to be called atlas-application.properties as the name is
hard coded.

      service.deps.hive_server2.options.atlas ?= {}
      service.deps.hive_server2.options.atlas.client ?= {}
      options.client ?= {}
      options.client.properties ?= {}
      options.client.properties['atlas.http.authentication.enabled'] ?= options.atlas.options.application.properties['atlas.http.authentication.enabled']
      options.client.properties['atlas.http.authentication.type'] ?= options.atlas.options.application.properties['atlas.http.authentication.type']
      options.application ?= {}
      options.application.properties ?= {}
      options.application.properties['atlas.hook.hive.synchronous'] ?= 'false'
      options.application.properties['atlas.hook.hive.numRetries'] ?= '3'
      options.application.properties['atlas.hook.hive.minThreads'] ?= '5'
      options.application.properties['atlas.hook.hive.maxThreads'] ?= '5'
      options.application.properties['atlas.hook.hive.keepAliveTime'] ?= '10'
      options.application.properties['atlas.hook.hive.queueSize'] ?= '10000'
      service.deps.hive_server2.options.hive_site['atlas.cluster.name'] ?= "#{options.atlas.options.cluster_name}"
      # Step 1 - check if the rest adress already written
      # Step 2 (only if 1) Check if an url is already written
      service.deps.hive_server2.options.hive_site['atlas.rest.address'] = add_prop service.deps.hive_server2.options.hive_site['atlas.rest.address'], options.atlas.options.application.urls, ','
      service.deps.hive_server2.options.hive_site['hive.exec.post.hooks'] = add_prop service.deps.hive_server2.options.hive_site['hive.exec.post.hooks'], 'org.apache.atlas.hive.hook.HiveHook', ','
      # server2.aux_jars = add_prop server2.aux_jars, "/usr/hdp/current/atlas-client/hook/hive", ':'
      #Notifications
      chanels = []
      chanels.push 'SASL_SSL' if service.deps.hadoop_core.options.core_site['hadoop.security.authentication'] is 'kerberos' and service.deps.hive_server2.options.hive_site['hive.server2.use.SSL'] is 'true'
      chanels.push 'SASL_PLAINTEXT' if service.deps.hadoop_core.options.core_site['hadoop.security.authentication'] is 'kerberos'
      chanels.push 'SSL' if service.deps.hive_server2.options.hive_site['hive.server2.use.SSL'] is 'true'
      chanels.push 'PLAINTEXT'
      options.application.properties['atlas.kafka.security.protocol'] ?= chanels[0]
      options.application.properties['atlas.notification.topics'] ?= options.atlas.options.application.properties['atlas.notification.topics']
      options.application.properties['atlas.kafka.bootstrap.servers'] ?= options.atlas.options.application.properties['atlas.kafka.bootstrap.servers']
      # Configure Hive Server2 JAAS Properties for posting notifications to Kafka
      if options.application.properties['atlas.kafka.security.protocol'] in ['SASL_PLAINTEXT','SASL_SSL']
        options.application.properties['atlas.jaas.KafkaClient.loginModuleControlFlag'] ?= 'required'
        options.application.properties['atlas.jaas.KafkaClient.loginModuleName'] ?= 'com.sun.security.auth.module.Krb5LoginModule'
        options.application.properties['atlas.jaas.KafkaClient.option.keyTab'] ?= service.deps.hive_server2.options.hive_site['hive.server2.authentication.kerberos.keytab']
        options.application.properties['atlas.jaas.KafkaClient.option.principal'] ?= service.deps.hive_server2.options.hive_site['hive.server2.authentication.kerberos.principal']
        options.application.properties['atlas.jaas.KafkaClient.option.serviceName'] ?= 'kafka'
        options.application.properties['atlas.jaas.KafkaClient.option.storeKey'] ?= 'true'
        options.application.properties['atlas.jaas.KafkaClient.option.useKeyTab'] ?= 'true'
      if options.application.properties['atlas.kafka.security.protocol'] in ['SSL','SASL_SSL']
        #note: service.deps.hadoop_core.options should retrive the srv of the hive'server2 node
        options.application.properties['atlas.kafka.ssl.truststore.location'] ?= service.deps.hadoop_core.options.ssl_client['ssl.client.truststore.location']
        options.application.properties['atlas.kafka.ssl.truststore.password'] ?= service.deps.hadoop_core.options.ssl_client['ssl.client.truststore.password']
      #Administators can choose a different protocol for Atlas Kafka Notification
      protocol = options.application.properties['atlas.kafka.security.protocol']
      if protocol in service.deps.kafka_broker[0].options.protocols
        brokers = service.deps.kafka_broker.map( (srv) =>
          "#{srv.node.fqdn}:#{srv.options.ports[protocol]}"
        ).join ','
        # construcut the bootstrap listeners string base on channel
        # i.e.: SASL_SSL://master1.ryba:9096,master2.ryba:9096,master3.ryba:9096 for example
        options.application.properties['atlas.kafka.bootstrap.servers'] ?= "#{protocol}://#{brokers}"
      else
        throw Error "Atlas Hive Bridge Hook Selected Protocol #{options.application.kafka_chanel} is not allowed by Kafka Brokers configuration"
      #Kafka Ranger PLugin authorization
      if options.application.properties['atlas.kafka.security.protocol'] in  ['SASL_PLAINTEXT','SASL_SSL']
        options.atlas.options.kafka_policy.policyItems[0].users.push "#{service.deps.hive_server2.options.user.name}"
      if options.application.properties['atlas.kafka.security.protocol'] in  ['PLAINTEXT','SSL']
        options.atlas.options.kafka_policy.policyItems[0].users.push 'ANONYMOUS'
      if (options.application.properties['atlas.kafka.security.protocol'] in ['PLAINTEXT','SSL'])
        options.atlas.options.kafka_policy.policyItems[0].users.push 'ANONYMOUS'

## utility function

    add_prop = (value, add, separator) ->
      throw Error 'No separator provided' unless separator?
      value ?= ''
      return add if value.length is 0
      return if value.indexOf(add) is -1 then "#{value}#{separator}#{add}" else value