cruise control: severale enhancements

roles/amq_streams_cruise_control/defaults/main.yml

@@ -1,11 +1,16 @@
 ---
 amq_streams_cruise_control_home: /opt/cruise-control/
 amq_streams_cruise_control_service_name: amq_streams_cruise_control
-amq_streams_cruise_control_server_start: "{{ amq_streams_cruise_control_home  }}/kafka-cruise-control-start.sh"
+amq_streams_cruise_control_server_start: "{{ amq_streams_cruise_control_home }}/kafka-cruise-control-start.sh"
 amq_streams_cruise_control_server_stop: "{{ amq_streams_cruise_control_home }}/kafka-cruise-control-stop.sh"
 amq_streams_cruise_control_server_config: "/etc/amq_streams_cruise_control.properties"
 amq_streams_cruise_control_config_template: 'templates/service.conf.j2'
 amq_streams_cruise_control_config_properties_template: 'templates/cruisecontrol.properties.j2'
+amq_streams_cruise_control_path_to_log4j_properties: "{{ amq_streams_cruise_control_home }}/config/log4j.properties"
+amq_streams_cruise_control_log4j_template: 'templates/log4j.properties.j2'
+amq_streams_cruise_control_logfilename: '/var/log/cruise-control'
+amq_streams_cruise_control_capacity_template: 'templates/capacity.json.j2'
+amq_streams_cruise_control_path_to_capacity_file: "{{ amq_streams_cruise_control_home }}/config/capacity.json"
 amq_streams_cruise_control_user: root
 amq_streams_cruise_control_group: root
 amq_streams_cruise_control_zk_host: "{{ groups['zookeepers'][0] }}"

roles/amq_streams_cruise_control/handlers/main.yml

@@ -10,3 +10,6 @@
     name: "{{ amq_streams_cruise_control_service_name }}"
     state: stopped
   become: yes
+
+- name: "Refresh capacity.json"
+  ansible.builtin.include_tasks: capacity.yml

roles/amq_streams_cruise_control/tasks/capacity.yml

@@ -0,0 +1,16 @@
+---
+- name: "Ensure required parameters are provided"
+  ansible.builtin.assert:
+    that:
+      - amq_streams_cruise_control_capacity_template is defined and amq_streams_cruise_control_capacity_template | length > 0
+      - amq_streams_cruise_control_path_to_capacity_file is defined and amq_streams_cruise_control_path_to_capacity_file | length > 0
+    quiet: True
+
+- name: "Ensure capacity.json is properly configured"
+  ansible.builtin.template:
+    src: "{{ amq_streams_cruise_control_capacity_template }}"
+    dest: "{{ amq_streams_cruise_control_path_to_capacity_file }}"
+    owner: "{{ amq_streams_cruise_control_user | default(omit) }}"
+    group: "{{ amq_streams_cruise_control_group | default(omit) }}"
+    mode: 0644
+  notify: "Restart {{ amq_streams_cruise_control_service_name }}"

roles/amq_streams_cruise_control/tasks/main.yml

@@ -14,7 +14,29 @@
     name: amq_streams_common
     tasks_from: install.yml
 
-# TODO deal with FW, if enabled
+- name: "Deploy custom log4j config (if provided)"
+  when:
+    - amq_streams_cruise_control_logfilename is defined
+  block:
+    - name: "Ensure dirname for logfile exits {{ amq_streams_cruise_control_logfilename | dirname }}"
+      ansible.builtin.file:
+        path: "{{ amq_streams_cruise_control_logfilename | dirname }}"
+        state: directory
+
+    - name: "Ensure log4j logging is properly configured"
+      ansible.builtin.template:
+        src: "{{ amq_streams_cruise_control_log4j_template }}"
+        dest: "{{ amq_streams_cruise_control_path_to_log4j_properties }}"
+        owner: "{{ amq_streams_cruise_control_user | default(omit) }}"
+        group: "{{ amq_streams_cruise_control_group | default(omit) }}"
+        mode: 0644
+
+- name: "Ensure capacity.json is properly configured"
+  ansible.builtin.include_tasks: capacity.yml
+  when:
+    - amq_streams_broker_inventory_group is defined and amq_streams_broker_inventory_group | length > 0
+
+#TODO: Deal with FW if enabled
 
 - name: "Deploy Cruise Control as a systemd service."
   ansible.builtin.include_role:

roles/amq_streams_cruise_control/templates/capacity.json.j2

@@ -0,0 +1,25 @@
+{
+  "brokerCapacities":[
+    {%if amq_streams_broker_inventory_group is defined and amq_streams_broker_inventory_group | length > 0 %}{% for broker in amq_streams_broker_inventory_group %}{
+      "brokerId": "{{ amq_streams_broker_inventory_group.index(broker) }}",
+      "capacity": {
+        "DISK": "500000",
+        "CPU": "100",
+        "NW_IN": "50000",
+        "NW_OUT": "50000"
+      },
+      "doc": "This overrides the capacity for broker {{ amq_streams_broker_inventory_group.index(broker) }}."
+    },{% endfor %}{% endif %}
+    {
+      "brokerId": "-1",
+      "capacity": {
+        "DISK": "100000",
+        "CPU": "100",
+        "NW_IN": "10000",
+        "NW_OUT": "10000"
+      },
+      "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+    }
+  ]
+}
+

roles/amq_streams_cruise_control/templates/cruisecontrol.properties.j2

@@ -87,7 +87,7 @@ min.samples.per.broker.metrics.window=1
 
 # The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
 #capacity.config.file={{ amq_stream_cruise_control_home }}/config/capacity.json
-capacity.config.file={{ amq_stream_cruise_control_home }}/config/capacityJBOD.json
+capacity.config.file={{ amq_stream_cruise_control_home }}/config/capacity.json
 
 # Configurations for the analyzer
 # =======================================

roles/amq_streams_cruise_control/templates/log4j.properties.j2

@@ -0,0 +1,65 @@
+#
+# Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+#
+
+rootLogger.level=INFO
+appenders=console, kafkaCruiseControlAppender, operationAppender, requestAppender
+
+{%if amq_streams_cruise_control_logfilename is defined and amq_streams_cruise_control_logfilename | length > 0 %}property.filename={{ amq_streams_cruise_control_logfilename }}"
+{% endif %}
+
+appender.console.type=Console
+appender.console.name=STDOUT
+appender.console.layout.type=PatternLayout
+appender.console.layout.pattern=[%d] %p %m (%c)%n
+
+appender.kafkaCruiseControlAppender.type=RollingFile
+appender.kafkaCruiseControlAppender.name=kafkaCruiseControlFile
+appender.kafkaCruiseControlAppender.fileName=${filename}/kafkacruisecontrol.log
+appender.kafkaCruiseControlAppender.filePattern=${filename}/kafkacruisecontrol.log.%d{yyyy-MM-dd-HH}
+appender.kafkaCruiseControlAppender.layout.type=PatternLayout
+appender.kafkaCruiseControlAppender.layout.pattern=[%d] %p %m (%c)%n
+appender.kafkaCruiseControlAppender.policies.type=Policies
+appender.kafkaCruiseControlAppender.policies.time.type=TimeBasedTriggeringPolicy
+appender.kafkaCruiseControlAppender.policies.time.interval=1
+
+appender.operationAppender.type=RollingFile
+appender.operationAppender.name=operationFile
+appender.operationAppender.fileName=${filename}/kafkacruisecontrol-operation.log
+appender.operationAppender.filePattern=${filename}/kafkacruisecontrol-operation.log.%d{yyyy-MM-dd}
+appender.operationAppender.layout.type=PatternLayout
+appender.operationAppender.layout.pattern=[%d] %p [%c] %m %n
+appender.operationAppender.policies.type=Policies
+appender.operationAppender.policies.time.type=TimeBasedTriggeringPolicy
+appender.operationAppender.policies.time.interval=1
+
+appender.requestAppender.type=RollingFile
+appender.requestAppender.name=requestFile
+appender.requestAppender.fileName=${filename}/kafkacruisecontrol-request.log
+appender.requestAppender.filePattern=${filename}/kafkacruisecontrol-request.log.%d{yyyy-MM-dd-HH}
+appender.requestAppender.layout.type=PatternLayout
+appender.requestAppender.layout.pattern=[%d] %p %m (%c)%n
+appender.requestAppender.policies.type=Policies
+appender.requestAppender.policies.time.type=TimeBasedTriggeringPolicy
+appender.requestAppender.policies.time.interval=1
+
+# Loggers
+logger.cruisecontrol.name=com.linkedin.kafka.cruisecontrol
+logger.cruisecontrol.level=debug
+logger.cruisecontrol.appenderRef.kafkaCruiseControlAppender.ref=kafkaCruiseControlFile
+
+logger.detector.name=com.linkedin.kafka.cruisecontrol.detector
+logger.detector.level=info
+logger.detector.appenderRef.kafkaCruiseControlAppender.ref=kafkaCruiseControlFile
+
+logger.operationLogger.name=operationLogger
+logger.operationLogger.level=info
+logger.operationLogger.appenderRef.operationAppender.ref=operationFile
+
+logger.CruiseControlPublicAccessLogger.name=CruiseControlPublicAccessLogger
+logger.CruiseControlPublicAccessLogger.level=info
+logger.CruiseControlPublicAccessLogger.appenderRef.requestAppender.ref=requestFile
+
+rootLogger.appenderRefs=console, kafkaCruiseControlAppender
+rootLogger.appenderRef.console.ref=STDOUT
+rootLogger.appenderRef.kafkaCruiseControlAppender.ref=kafkaCruiseControlFile