modify-connectors.yml

#!/usr/bin/env ansible-playbook
#
# (c) 2016 DataNexus Inc.  All Rights Reserved
---
# Build our kafka and zookeeper host groups
- name: Create kafka and zookeeper host groups
  hosts: localhost
  gather_facts: no
  vars_files:
    - vars/kafka-operations.yml
  tasks:
    # load the 'configuration file' (if one was defined) to get any variables
    # we might need during this playbook run
    - name: Load configuration file
      include_vars:
        file: "{{config_file | default('config.yml')}}"
    # then, build our kafka and zookeeper host groups
    - include_role:
        name: build-app-host-groups
      vars:
        host_group_list:
          - name: kafka
          - name: zookeeper
      when: cloud is defined and (cloud == 'aws' or cloud == 'osp')
    - set_fact:
        num_kafka_nodes: "{{groups['kafka'] | default([]) | length}}"
    # if we didn't find any nodes in the kafka host group, then it's an error
    - name: Fail playbook run if a Kafka instance/cluster not found
      fail:
        msg: "An (external) Kafka cluster is required for telegraf deployments"
      when: num_kafka_nodes | int == 0

# Collect some Zookeeper related facts and determine the "private" IP addresses of
# the nodes in the Zookeeper ensemble (from their "public" IP addresses and the `data_iface`
# variable that was passed in as part of this playbook run) if a list of "public"  Zookeeper
# IP addresses was passed in.
- name: Gather facts from Zookeeper host group (if defined)
  hosts: zookeeper
  tasks: []

# Modify the connectors in the Kafka cluster (add new connectors, create new
# workers in the connector framework, create new instances of connectors, etc.)
- name: Modify Kafka connectors
  hosts: kafka
  gather_facts: no
  vars_files:
    - vars/kafka-operations.yml
  vars:
    - kafka_nodes: "{{groups['kafka'] | default([])}}"
    - zookeeper_nodes: "{{groups['zookeeper'] | default([])}}"
  pre_tasks:
    # first, initialize the play by loading any `config_file` that may have
    # been passed in, determining the `data_iface` and `api_iface` values from
    # the input `internal_subnet` and `external_subnet` (if values for these
    # two parameters were passed in), and determining the addresses of the
    # `data_iface` and `api_iface` interfaces (the `data_addr` and `api_addr`
    # values)
    - include_role:
        name: initialize-play
      vars:
        skip_network_restart: true
    # and now that we know we have our data_iface identified, we can construct
    # the list of zk_nodes (the data_iface IP addresses of our zookeeper_nodes)
    - set_fact:
        zk_nodes: "{{zookeeper_nodes | map('extract', hostvars, [('ansible_' + data_iface), 'ipv4', 'address']) | list}}"
    # Get a list of the actions where we're deploying a standalone worker and
    # the connection_list entry for that worker is empty or does not exist; if
    # any of the actions in the action_list have such entries, then throw an
    # error before we do anything
    - block:
        # first get a list of the actions where workers are started
        - name: Get a list of actions where workers are started or restarted
          set_fact:
            start_actions: "{{(action_list | default([])) | selectattr('action','equalto','start-workers') | list}}"
            restart_actions: "{{(action_list | default([])) | selectattr('action','equalto','restart-workers') | list}}"
        - name: Combine the two lists
          set_fact:
            all_start_actions: "{{start_actions + restart_actions}}"
        # then, if we found any, look for workers with a zero-length
        # (or missing) connection_list
        - block:
          - name: Get list of all workers from those actions
            set_fact:
              name_connector_list: '{{ (name_connector_list | default([])) + [{ "name": item.1.name, "connector_list_len": ((item.1.connector_list | default([])) or [] | length) }] }}'
            with_subelements:
              - "{{all_start_actions}}"
              - worker_list
          - name: Get a list of all of those workers with an empty (or no) connection_list
            set_fact:
              has_empty_connection_list: "{{name_connector_list | selectattr('connector_list_len', 'equalto', 0) | map(attribute='name') | list}}"
          - name: Fail if that list is not empty
            fail:
              msg="One or more workers ({{has_empty_connection_list | to_json}}) defined with an empty/missing connector_list; at least one connector must be provided when (re)starting connectors on a standalone Kafka node"
            when: has_empty_connection_list | length > 0
          when: all_start_actions | length > 0
      when: (kafka_nodes | length) == 1
    # Now that we have all of the facts we need we can proceed to modify the
    # connectors if that's what we're doing (add new connctors to the cluster,
    # managing connector instances, or managing worker instances)
    - name: Modify connectors
      include_role:
        name: modify-connectors
      vars:
        action_hash: "{{list_item}}"
      with_items: "{{action_list}}"
      loop_control:
        loop_var: list_item