diff --git a/ansible/group_vars/all.yml b/ansible/group_vars/all.yml
index 3dca393799..8deda4476c 100644
--- a/ansible/group_vars/all.yml
+++ b/ansible/group_vars/all.yml
@@ -1102,6 +1102,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
#########################
diff --git a/ansible/roles/kafka/defaults/main.yml b/ansible/roles/kafka/defaults/main.yml
index c3b4de5301..5a6a42e70a 100644
--- a/ansible/roles/kafka/defaults/main.yml
+++ b/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
diff --git a/ansible/roles/monasca/defaults/main.yml b/ansible/roles/monasca/defaults/main.yml
index 24434e6670..5a4d506bb9 100644
--- a/ansible/roles/monasca/defaults/main.yml
+++ b/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
diff --git a/ansible/roles/monasca/tasks/bootstrap.yml b/ansible/roles/monasca/tasks/bootstrap.yml
index 2d0667b892..b116beb06d 100644
--- a/ansible/roles/monasca/tasks/bootstrap.yml
+++ b/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
-# Kafka has automatic topic generation enabled by default we don't need to
-# create topics here.
+# NOTE(dszumski): Although we can take advantage of automatic topic
+# creation in Kafka, creating the topics manually allows unique settings
+# 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 }}"
diff --git a/releasenotes/notes/add-kafka-topic-creation-for-monasca-c8fccf0b6117eca5.yaml b/releasenotes/notes/add-kafka-topic-creation-for-monasca-c8fccf0b6117eca5.yaml
new file mode 100644
index 0000000000..9b8bc81fe4
--- /dev/null
+++ b/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.