Cmak: Automating the cluster setup

[spicysomtam]

I am incorporating kafka-manager in a kafka/zookeeper openshift/kubernetes config, and wish to automate the cluster setup.

I am using a docker image similar to sheepkiller/kafka-manager-docker, whereby I am passing an env var for the zookeeper connection. Anyhow I am seeing kafka-manager connecting to zookeeper fine:

[info] o.a.z.ZooKeeper - Initiating client connection, connectString=zookeeper:2181 sessionTimeout=60000 watcher=org.apache.curator.ConnectionState@5c9fdc3a
[info] k.m.a.KafkaManagerActor - zk=zookeeper:2181
[info] k.m.a.KafkaManagerActor - baseZkPath=/kafka-manager
[info] o.a.z.ClientCnxn - Opening socket connection to server zookeeper.kafka.svc.cluster.local/172.30.151.6:2181. Will not attempt to authenticate using SASL (unknown error)
[info] o.a.z.ClientCnxn - Socket connection established to zookeeper.kafka.svc.cluster.local/172.30.151.6:2181, initiating session
[info] o.a.z.ClientCnxn - Session establishment complete on server zookeeper.kafka.svc.cluster.local/172.30.151.6:2181, sessionid = 0x100012e44a90005, negotiated timeout = 40000

But when I go in the gui, I see nothing. No clusters listed or anything. Is this expect behaviour, and if so, how can I pre-configure a cluster in kafka-manager to pass to the server?

[nu007a]

i think you want to auto "Add cluster" in config file。
you can ref https://github.com/yahoo/kafka-manager/issues/244

[montana-ua]

I do it by ansible in the following way:

This is my defaults (any variable can be overriden by extra-vars)

kafka_cluster_config:
  name: "{{ kafka_cluster_name }}"
  zkHosts: "{{ kafka_cluster_zkhosts }}"
  kafkaVersion: "{{ kafka_cluster_version }}"
  jmxEnabled: "true"
  jmxUser: ""
  jmxPass: ""
  jmxSsl: "false"
  logkafkaEnabled: "false"
  pollConsumers: "true"
  filterConsumers: "true"
  activeOffsetCacheEnabled: "true"
  displaySizeEnabled: "false"
  tuning.brokerViewUpdatePeriodSeconds: 30
  tuning.clusterManagerThreadPoolSize: 10
  tuning.clusterManagerThreadPoolQueueSize: 100
  tuning.kafkaCommandThreadPoolSize: 10
  tuning.kafkaCommandThreadPoolQueueSize: 100
  tuning.logkafkaCommandThreadPoolSize: 10
  tuning.logkafkaCommandThreadPoolQueueSize: 100
  tuning.logkafkaUpdatePeriodSeconds: 30
  tuning.partitionOffsetCacheTimeoutSecs: 5
  tuning.brokerViewThreadPoolSize: 10
  tuning.brokerViewThreadPoolQueueSize: 1000
  tuning.offsetCacheThreadPoolSize: 10
  tuning.offsetCacheThreadPoolQueueSize: 1000
  tuning.kafkaAdminClientThreadPoolSize: 10
  tuning.kafkaAdminClientThreadPoolQueueSize: 1000
  securityProtocol: PLAINTEXT

and this is my task

# Add kafka cluster into kafka manager via POST API
###################################################
---

- block:

  - name: prepare data for the POST request
    set_fact:
      kafka_cluster_post_data: "{{ kafka_cluster_post_data | default('') }}&{{ item }}={{ kafka_cluster_config[item] }}"
    with_items: "{{ kafka_cluster_config }}"

  - name: add kafka cluster {{ kafka_cluster_name }}
    uri:
      url: "http://{{ first_host }}:{{ kafka_manager_port }}/clusters"
      method: POST
      body: "{{ kafka_cluster_post_data }}"
      status_code: 200

  run_once: true
  delegate_to: localhost

[montana-ua]

Also you can do it by znode module from ansible

- name: add kafka cluster {{ kafka_cluster_name }}
   znode:
     hosts: "{{ kafka_manager_zkhosts }}"
     name: "{{ kafka_manager_baseZkPath }}/configs/{{ kafka_cluster_name }}"
     value: "{{ kafka_cluster_data | to_json | string }}"
     state: present
   run_once: true
   delegate_to: 127.0.0.1

[lconnell]

@montana-ua Thanks for sharing. I was able to get this to work with the uri module. I could get the znode module to work, however Kafka Manager was not picking up the cluster config. Have you got it working?

[montana-ua]

@lconnell yes, both approaches works for me but I prefer to use API (POST) to create a cluster into kafka manager.

[lconnell]

@montana-ua Could you share your kafka_cluster_data variable so I can see where it differs from what I tried using?

[montana-ua]

@lconnell

kafka_cluster_data:
  name: "{{ kafka_cluster_name }}"
  curatorConfig:
    zkConnect: "{{ kafka_cluster_zkhosts }}"
    zkMaxRetry: 100
    baseSleepTimeMs: 100
    maxSleepTimeMs: 1000
  enabled: true
  kafkaVersion: "{{ kafka_cluster_version }}"
  jmxEnabled: true
  jmxUser: null
  jmxPass: null
  jmxSsl: false
  pollConsumers: true
  filterConsumers: true
  logkafkaEnabled: false
  activeOffsetCacheEnabled: true
  displaySizeEnabled: false
  tuning:
    brokerViewUpdatePeriodSeconds: 30
    clusterManagerThreadPoolSize: 10
    clusterManagerThreadPoolQueueSize: 100
    kafkaCommandThreadPoolSize: 10
    kafkaCommandThreadPoolQueueSize: 100
    logkafkaCommandThreadPoolSize: 10
    logkafkaCommandThreadPoolQueueSize: 100
    logkafkaUpdatePeriodSeconds: 30
    partitionOffsetCacheTimeoutSecs: 5
    brokerViewThreadPoolSize: 10
    brokerViewThreadPoolQueueSize: 1000
    offsetCacheThreadPoolSize: 10
    offsetCacheThreadPoolQueueSize: 1000
    kafkaAdminClientThreadPoolSize: 10
    kafkaAdminClientThreadPoolQueueSize: 1000
  securityProtocol: "PLAINTEXT"

[montana-ua]

@lconnell I recommend to use API instead of create znode directly (it's not good way).

[lconnell]

@montana-ua Yes, I am going to stick with the API. I was more curious as to "why" it wasn't working with znode. I see one difference from your dict. I was wrapping the booleans in quotes and the jmxuser/jmxpass was an empty string.

Thanks again for your quick replies!

[LozanoMatheus]

@montana-ua Thanks for sharing !

I only change the var with zkHosts and url in uri module.

zkHosts: "{{ groups['zookeeper'] | join(':2181,') }}:2181"
_output: "zkHosts": "192.168.0.10:2181,192.168.0.11:2181,192.168.0.12:2181"_

url: "http://{{ groups['zookeeper'] | first }}/clusters"
_output: 192.168.0.10_

The first zk server will be a zm nodes too.

[dynnamitt]

Where is the doc for this API?

[svergun]

@dynnamitt I guess there is no official documentation for the API but if you open routes file in config folder, you find all necessary to understand how it works.

[dynnamitt]

I'm decoding what you do and it's "posting the payload as json" to /clusters.
But #244 suggest that this currently isn't working and has a "x-www-form-urlencoded" sample script
that contains a huge amount of code..
I beg the developers here to commit to documenting this API and keep it backwards compatible .
Reading routes(code) is ok but it makes me feel like whatever I create "on top" now will be (eventually) wasted .

[eshepelyuk]

Not sure this is a good place to post - but could you take a look at CMAK operator, I've created. It allows to setup CMAK in K8S via Helm chart and configure clusters connection settings simply via Helm values.

https://github.com/eshepelyuk/cmak-operator