# Kafka Check

    module.exports = header: 'Kafka Client Check', handler: ({options}) ->

## Register

      @registry.register 'ranger_policy', '@rybajs/metal/ranger/actions/ranger_policy'

## Wait

      @call once: true, 'masson/core/krb5_client/wait', options.wait_krb5_client
      @call once: true, '@rybajs/metal/zookeeper/server/wait', options.wait_zookeeper_server
      @call once: true, '@rybajs/metal/kafka/broker/wait', options.wait_kafka_broker

## Add Ranger Policy

      @call
        header: 'Ranger'
        if: !!options.ranger_admin
      , ->
        @call '@rybajs/metal/ranger/admin/wait', once: true, options.wait_ranger_admin
        topics = options.protocols.map (prot) =>
          "check-#{options.hostname}-client-#{prot.toLowerCase().split('_').join('-')}-topic"
        users = ["#{options.test.user.name}", options.superusers...]
        users.push 'ANONYMOUS' if ('PLAINTEXT' in options.protocols) or ('SSL' in options.protocols)
        @wait.execute
          header: 'Wait Service'
          cmd: """
          curl --fail -H "Content-Type: application/json" -k -X GET \
            -u #{options.ranger_admin.username}:#{options.ranger_admin.password} \
            "#{options.ranger_install['POLICY_MGR_URL']}/service/public/v2/api/service/name/#{options.ranger_install['REPOSITORY_NAME']}"
          """
          code_skipped: [1,7,22] #22 is for 404 not found,7 is for not connected to host
        @ranger_policy
          header: 'Create Policy'
          username: options.ranger_admin.username
          password: options.ranger_admin.password
          url: options.ranger_install['POLICY_MGR_URL']
          policy:
            service: "#{options.ranger_install['REPOSITORY_NAME']}"
            name: "test-ryba-client-#{options.hostname}"
            description: "Policy for ryba kafka Client test"
            isAuditEnabled: true
            resources:
              topic:
                values: topics
                isExcludes: false
                isRecursive: false
            policyItems: [
                "accesses": [
                  'type': 'publish'
                  'isAllowed': true
                ,
                  'type': 'consume'
                  'isAllowed': true
                ,
                  'type': 'configure'
                  'isAllowed': true
                ,
                  'type': 'describe'
                  'isAllowed': true
                ,
                  'type': 'create'
                  'isAllowed': true
                ,
                  'type': 'delete'
                  'isAllowed': true
                ,
                  'type': 'kafka_admin'
                  'isAllowed': true
                ],
                'users': users
                'groups': []
                'conditions': []
                'delegateAdmin': false
              ]
        @wait
          time: 10000
          if: -> @status -1

## PLAINTEXT Protocol

Check Message by writing to a test topic on the PLAINTEXT channel.
Since new API (0.8-0.9) and security features, kafka broker are able to deal with
multiple channel with different protocols. For its internal functionment, it associates
an not authenticated user to ANONYMOUS name when client communicates on PLAINTEXT-SSL
protocols.

      @call
        header: 'PLAINTEXT'
        if: 'PLAINTEXT' in options.protocols
        retry: 3
      , ->
        test_topic = "check-#{options.hostname}-client-plaintext-topic"
        @system.execute
          if: options.env['KAFKA_KERBEROS_PARAMS']?
          cmd: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
            --partitions #{Math.min(options.brokers['PLAINTEXT'].length-1,1)} \
            --replication-factor #{Math.min(options.brokers['PLAINTEXT'].length-1,0)} \
            --topic #{test_topic}
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --list \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
          | grep #{test_topic}
          """
        @system.execute
          unless: options.env['KAFKA_KERBEROS_PARAMS']? or !!options.ranger_admin
          cmd: """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
            --partitions #{Math.min(options.brokers['PLAINTEXT'].length-1,1)} \
            --replication-factor #{Math.min(options.brokers['PLAINTEXT'].length-1,0)} \
            --topic #{test_topic}
          """
          unless_exec: """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --list \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
          | grep #{test_topic}
          """
        @system.execute
          if:  options.env['KAFKA_KERBEROS_PARAMS']?
          cmd: mkcmd.kafka options.admin, """
          (
            sleep 1
            /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
              --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
              --add \
              --allow-principal User:ANONYMOUS \
              --operation Read \
              --operation Write \
              --topic #{test_topic}
          )&
          (
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --add \
            --allow-principal User:ANONYMOUS \
            --consumer \
            --producer \
            --group #{options.consumer.config['group.id']} \
            --topic #{test_topic}
          )
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh --list \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --topic #{test_topic} \
          | grep 'User:ANONYMOUS has Allow permission for operations: Write from hosts: *'
          """
        @system.execute
          unless:  options.env['KAFKA_KERBEROS_PARAMS']? or !!options.ranger_admin
          cmd: """
          (
            sleep 1
            /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
              --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
              --add --allow-principal User:ANONYMOUS \
              --operation Read \
              --operation Write \
              --topic #{test_topic}
          )&
          (
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --add \
            --allow-principal User:ANONYMOUS \
            --consumer \
            --producer \
            --group #{options.consumer.config['group.id']} \
            --topic #{test_topic}
          )
          """
          unless_exec: """
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh --list \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --topic #{test_topic} \
          | grep 'User:ANONYMOUS has Allow permission for operations: Write from hosts: *'
          """
        @system.execute
          cmd: """
          (
            sleep 5
            echo 'hello #{options.hostname}' | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh \
              --producer-property security.protocol=PLAINTEXT \
              --broker-list #{options.brokers['PLAINTEXT'].join ','} \
              --security-protocol PLAINTEXT \
              --producer.config #{options.conf_dir}/producer.properties \
              --topic #{test_topic}
          )&
          /usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh \
            --delete-consumer-offsets \
            --bootstrap-server #{options.brokers['PLAINTEXT'].join ','} \
            --topic #{test_topic} \
            --security-protocol PLAINTEXT \
            --property security.protocol=PLAINTEXT \
            --consumer.config #{options.conf_dir}/consumer.properties \
            --from-beginning \
            --max-messages 1 \
          | grep 'hello #{options.hostname}'
          """

## SSL Protocol

Check Message by writing to a test topic on the SSL channel. Truststore location 
and password given to line command because if executed before producer install
'/etc/kafka/conf/producer.properties' might be empty.

      @call
        header: 'SSL'
        retry: 3
        if: 'SSL' in options.protocols
      , ->
        test_topic = "check-#{options.hostname}-client-ssl-topic"
        @system.execute
          cmd: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
            --partitions #{Math.min(options.brokers['SSL'].length-1,1)} \
            --replication-factor #{Math.min(options.brokers['SSL'].length-1,0)} \
            --topic #{test_topic}
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --list \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
          | grep #{test_topic}
          """
        @system.execute
          unless: !!options.ranger_admin
          cmd: mkcmd.kafka options.admin, """
          (
            sleep 1
            /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
              --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
              --add --allow-principal User:ANONYMOUS \
              --operation Read \
              --operation Write \
              --topic #{test_topic}
          )&
          (
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --add \
            --allow-principal User:ANONYMOUS \
            --consumer \
            --producer \
            --group #{options.consumer.config['group.id']} \
            --topic #{test_topic}
          )
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh --list \
            --authorizer-properties \
            zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --topic #{test_topic} \
          | grep 'User:ANONYMOUS has Allow permission for operations: Write from hosts: *'
          """
        @system.execute
          cmd: """
          (
            sleep 5
            echo 'hello #{options.hostname}' | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh \
              --producer-property security.protocol=SSL \
              --broker-list #{options.brokers['SSL'].join ','} \
              --security-protocol SSL \
              --producer-property ssl.truststore.location=#{options.config['ssl.truststore.location']} \
              --producer-property ssl.truststore.password=#{options.config['ssl.truststore.password']} \
              --producer.config #{options.conf_dir}/producer.properties \
              --topic #{test_topic}
          )&
          /usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh \
            --delete-consumer-offsets \
            --bootstrap-server #{options.brokers['SSL'].join ','} \
            --topic #{test_topic} \
            --security-protocol SSL \
            --property security.protocol=SSL \
            --property ssl.truststore.location=#{options.config['ssl.truststore.location']} \
            --property ssl.truststore.password=#{options.config['ssl.truststore.password']} \
            --consumer.config #{options.conf_dir}/consumer.properties \
            --from-beginning \
            --max-messages 1 \
          | grep 'hello #{options.hostname}'
          """

## SASL_PLAINTEXT Protocol

Check Message by writing to a test topic on the SASL_PLAINTEXT channel.

      @call
        header: 'SASL_PLAINTEXT'
        retry: 3
        if: 'SASL_PLAINTEXT' in options.protocols
      , ->
        test_topic = "check-#{options.hostname}-client-sasl-plaintext-topic"
        @system.execute
          cmd: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
            --partitions #{Math.min(options.brokers['SASL_PLAINTEXT'].length-1,1)} \
            --replication-factor #{Math.min(options.brokers['SASL_PLAINTEXT'].length-1,0)} \
            --topic #{test_topic}
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --list \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
          | grep #{test_topic}
          """
        @system.execute
          unless: !!options.ranger_admin
          cmd: mkcmd.kafka options.admin, """
          (
            sleep 1
            /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
              --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
              --add \
              --allow-principal User:#{options.test.user.name} \
              --operation Read \
              --operation Write \
              --topic #{test_topic}
          )&
          (
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --add \
            --allow-principal User:#{options.test.user.name} \
            --consumer --group #{options.config['group.id']} \
            --topic #{test_topic}
          )
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh  --list \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --topic #{test_topic} \
          | grep 'User:#{options.test.user.name} has Allow permission for operations: Write from hosts: *'
          """
        @system.execute
          cmd:  mkcmd.test options.test_krb5_user, """
          (
            sleep 5
            echo 'hello #{options.hostname}' | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh \
              --producer-property security.protocol=SASL_PLAINTEXT \
              --broker-list #{options.brokers['SASL_PLAINTEXT'].join ','} \
              --security-protocol SASL_PLAINTEXT \
              --producer.config #{options.conf_dir}/producer.properties \
              --topic #{test_topic}
          )&
          /usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh \
            --delete-consumer-offsets \
            --bootstrap-server #{options.brokers['SASL_PLAINTEXT'].join ','} \
            --topic #{test_topic} \
            --security-protocol SASL_PLAINTEXT \
            --consumer.config #{options.conf_dir}/consumer.properties \
            --from-beginning \
            --max-messages 1 \
          | grep 'hello #{options.hostname}'
          """

## SASL_SSL Protocol

Check Message by writing to a test topic on the SASL_SSL channel.
Trustore location and password given to line command because if executed before producer install
'/etc/kafka/conf/producer.properties' might be empty.

      @call
        header: 'SASL_SSL'
        retry: 3
        if: 'SASL_SSL' in options.protocols
      , ->
        test_topic = "check-#{options.hostname}-client-sasl-ssl-topic"
        @system.execute
          cmd: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
            --partitions #{Math.min(options.brokers['SASL_SSL'].length-1,1)} \
            --replication-factor #{Math.min(options.brokers['SASL_SSL'].length-1,0)} \
            --topic #{test_topic}
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-topics.sh \
            --list \
            --zookeeper #{options.consumer.config['zookeeper.connect']} \
          | grep #{test_topic}
          """
        @system.execute
          unless: !!options.ranger_admin
          cmd: mkcmd.kafka options.admin, """
          (
            sleep 1
            /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
              --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
              --add \
              --allow-principal User:#{options.test.user.name} \
              --operation Read \
              --operation Write \
              --topic #{test_topic}
          )&
          (
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --add \
            --allow-principal User:#{options.test.user.name} \
            --consumer \
            --producer \
            --group #{options.consumer.config['group.id']} \
            --topic #{test_topic}
          )
          """
          unless_exec: mkcmd.kafka options.admin, """
          /usr/hdp/current/kafka-broker/bin/kafka-acls.sh  --list \
            --authorizer-properties zookeeper.connect=#{options.consumer.config['zookeeper.connect']} \
            --topic #{test_topic} \
          | grep 'User:#{options.test.user.name} has Allow permission for operations: Write from hosts: *'
          """
        @system.execute
          cmd:  mkcmd.test options.test_krb5_user, """
          (
            sleep 5
            echo 'hello #{options.hostname}' | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh \
              --producer-property security.protocol=SASL_SSL \
              --broker-list #{options.brokers['SASL_SSL'].join ','} \
              --security-protocol SASL_SSL \
              --producer-property ssl.truststore.location=#{options.config['ssl.truststore.location']} \
              --producer-property ssl.truststore.password=#{options.config['ssl.truststore.password']} \
              --producer.config #{options.conf_dir}/producer.properties \
              --topic #{test_topic}
          )&
          /usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh \
            --delete-consumer-offsets \
            --bootstrap-server #{options.brokers['SASL_SSL'].join ','} \
            --topic #{test_topic} \
            --security-protocol SASL_SSL \
            --property security.protocol=SASL_SSL \
            --property ssl.truststore.location=#{options.config['ssl.truststore.location']} \
            --property ssl.truststore.password=#{options.config['ssl.truststore.password']} \
            --consumer.config #{options.conf_dir}/consumer.properties \
            --from-beginning \
            --max-messages 1 \
          | grep 'hello #{options.hostname}'
          """

## Dependencies

    mkcmd = require '../../lib/mkcmd'