Support explicit creation of Monasca Kafka topics

With this patch, Monasca no longer relies on automatic topic creation
in Kafka, and instead pre-creates all topics before bringing up the
containers. If the topic already exists then it will not be
changed, therefore existing users are not affected.

This patch allows per topic customisations, such as increasing the
number of partitions on particular topics and also works around
a race condition in automatic topic creation where multiple instances
of the same service could race to create a topic causing some of the
services to restart and throw an error before resuming normal
operation.

Change-Id: Ib15c95bb72cf79e9e55945d757b248e06f5f4065

ansible/group_vars/all.yml

@@ -1099,6 +1099,11 @@ infuxdb_internal_endpoint: "{{ internal_protocol }}://{{ kolla_internal_fqdn | p
 #################
 kafka_datadir_volume: "kafka"
 
+# The number of brokers in a Kafka cluster. This is used for automatically
+# setting quantities such as topic replicas and it is not recommended to
+# change it unless you know what you are doing.
+kafka_broker_count: "{{ groups['kafka'] | length }}"
+
 #########################
 # Internal Image options
 #########################

ansible/roles/kafka/defaults/main.yml

@@ -20,7 +20,6 @@ kafka_cluster_name: "kolla_kafka"
 kafka_log_dir: "/var/log/kolla/kafka"
 kafka_heap_opts: "-Xmx1G -Xms1G"
 kafka_zookeeper: "{% for host in groups['zookeeper'] %}{{ 'api' | kolla_address(host) | put_address_in_context('url') }}:{{ zookeeper_client_port }}{% if not loop.last %},{% endif %}{% endfor %}"
-kafka_broker_count: "{{ groups['kafka'] | length }}"
 
 ####################
 # Docker

ansible/roles/monasca/defaults/main.yml

@@ -131,7 +131,22 @@ monasca_storm_nimbus_servers: "{% for host in groups['storm-nimbus'] %}'{{ 'api'
 # NOTE(dszumski): Only one NTP server is currently supported by the Monasca Agent plugin
 monasca_ntp_server: "{{ external_ntp_servers | first }}"
 
-# Kafka topics used by Monasca services
+# The default number of Kafka topic partitions. This effectively limits
+# the maximum number of workers per topic, counted over all nodes in the
+# Monasca deployment. For example, if you have a 3 node Monasca
+# deployment, you will by default have 3 instances of Monasca Persister,
+# with each instance having 2 workers by default for the metrics topic.
+# In this case, each worker on the metrics topic will be assigned 5
+# partitions of the metrics topic. If you increase the worker or instance
+# count, you may need to increase the partition count to ensure that all
+# workers can get a share of the work.
+monasca_default_topic_partitions: 30
+
+# The default number of topic replicas. Generally you should not change
+# this.
+monasca_default_topic_replication_factor: "{{ kafka_broker_count if kafka_broker_count|int < 3 else 3 }}"
+
+# Kafka topic names used by Monasca services
 monasca_metrics_topic: "metrics"
 monasca_raw_logs_topic: "logs"
 monasca_transformed_logs_topic: "transformed-logs"
@@ -141,6 +156,47 @@ monasca_alarm_notifications_topic: "alarm-notifications"
 monasca_alarm_notifications_retry_topic: "retry-notifications"
 monasca_periodic_notifications_topic: "60-seconds-notifications"
 
+# Kafka topic configuration. Most users will not need to modify these
+# settings, however for deployments where resources are tightly
+# constrained, or very large deployments where there are many parallel
+# workers, it is worth considering changing them. Note that if you do
+# change these settings, then you will need to manually remove each
+# topic from the Kafka deployment for the change to take effect when
+# the Monasca service is reconfigured.
+monasca_all_topics:
+  - name: "{{ monasca_metrics_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_raw_logs_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_transformed_logs_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_events_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_alarm_state_transitions_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_alarm_notifications_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_alarm_notifications_retry_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+  - name: "{{ monasca_periodic_notifications_topic }}"
+    partitions: "{{ monasca_default_topic_partitions }}"
+    replication_factor: "{{ monasca_default_topic_replication_factor }}"
+    enabled: True
+
 # NOTE(dszumski): Due to the way monasca-notification is currently
 # implemented it is not recommended to change this period.
 monasca_periodic_notifications_period: 60

ansible/roles/monasca/tasks/bootstrap.yml

@@ -61,6 +61,47 @@
   delegate_to: "{{ groups['influxdb'][0] }}"
   when: monasca_influxdb_name not in monasca_influxdb_database.stdout_lines
 
-# NOTE(dszumski): The Monasca APIs write logs and messages to Kafka. Since
+# NOTE(dszumski): Although we can take advantage of automatic topic
-# Kafka has automatic topic generation enabled by default we don't need to
+# creation in Kafka, creating the topics manually allows unique settings
-# create topics here.
+# to be used per topic, rather than the defaults. It also avoids an edge
+# case where services on multiple nodes may race to create topics, and
+# paves the way for enabling things like compression on a per topic basis.
+- name: List monasca kafka topics
+  become: true
+  command: >
+    docker exec kafka /opt/kafka/bin/kafka-topics.sh
+    --zookeeper localhost
+    --list
+  register: kafka_topics
+  run_once: True
+  delegate_to: "{{ groups['kafka'][0] }}"
+
+- name: Create monasca kafka topics if they don't exist
+  become: true
+  command: >
+    docker exec kafka /opt/kafka/bin/kafka-topics.sh
+    --create
+    --topic {{ item.name }}
+    --partitions {{ item.partitions }}
+    --replication-factor {{ item.replication_factor }}
+    --zookeeper localhost
+  run_once: True
+  delegate_to: "{{ groups['kafka'][0] }}"
+  when:
+    - item.name not in kafka_topics.stdout_lines
+    - item.enabled | bool
+  with_items: "{{ monasca_all_topics }}"
+
+- name: Remove monasca kafka topics for disabled services
+  become: true
+  command: >
+    docker exec kafka /opt/kafka/bin/kafka-topics.sh
+    --delete
+    --topic "{{ item.name }}"
+    --zookeeper localhost
+  run_once: True
+  delegate_to: "{{ groups['kafka'][0] }}"
+  when:
+    - item.name in kafka_topics.stdout_lines
+    - not item.enabled | bool
+  with_items: "{{ monasca_all_topics }}"

releasenotes/notes/add-kafka-topic-creation-for-monasca-c8fccf0b6117eca5.yaml

@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    Adds support for explicitly creating individually customisable topics
+    in Kafka for Monasca.
+fixes:
+  - |
+    Fixes a trivial issue where some Monasca containers could momentarily
+    restart when initially racing to create topics in Kafka.