Added role for postgres DB and init scripts. Changes to conductor setup accordingly to vars

ansible/roles/common/defaults/main.yaml

@@ -1,3 +1,4 @@
 ---
-target_path: /tmp/conductor_stack_haproxy
+target_path: /tmp/conductor_stack_postgres
 conductor_network: conductor-network
+init_db: True

ansible/roles/conductor/tasks/main.yaml

@@ -4,9 +4,23 @@
     src: templates/conductor-swarm.yaml.j2
     dest: "{{ target_path }}/conductor-swarm.yaml"
 
-- name: Generate conductor config from seeds
+- name: Generate conductor config from dynomite seeds
+  when: not use_postgres
   vars:
     seeds: "{{ lookup('file', '{{ target_path}}/seeds.list').splitlines() }}"
   template:
     src: "templates/{{ conductor_config_template }}"
     dest: "{{ target_path }}/{{ conductor_config }}"
+
+- name: Generate conductor config for postgres DB
+  when: use_postgres
+  template:
+    src: "templates/{{ conductor_config_template }}"
+    dest: "{{ target_path }}/{{ conductor_config }}"
+
+- name: Copy conductor SQL schema init for postgres DB
+  when: use_postgres and init_db
+  template:
+    src: "templates/conductor-db-init-postgres.sql"
+    dest: "{{ target_path }}/conductor-db-init.sql"
+

ansible/roles/conductor/templates/conductor-db-init-postgres.sql

@@ -0,0 +1,188 @@
+-- V1__initial_schema.sql--
+-- --------------------------------------------------------------------------------------------------------------
+-- SCHEMA FOR METADATA DAO
+-- --------------------------------------------------------------------------------------------------------------
+
+CREATE TABLE meta_event_handler (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  name varchar(255) NOT NULL,
+  event varchar(255) NOT NULL,
+  active boolean NOT NULL,
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE INDEX event_handler_name_index ON meta_event_handler (name);
+CREATE INDEX event_handler_event_index ON meta_event_handler (event);
+
+CREATE TABLE meta_task_def (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  name varchar(255) NOT NULL,
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_task_def_name ON meta_task_def (name);
+
+CREATE TABLE meta_workflow_def (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  name varchar(255) NOT NULL,
+  version int NOT NULL,
+  latest_version int NOT NULL DEFAULT 0,
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_name_version ON meta_workflow_def (name,version);
+CREATE INDEX workflow_def_name_index ON meta_workflow_def (name);
+
+-- --------------------------------------------------------------------------------------------------------------
+-- SCHEMA FOR EXECUTION DAO
+-- --------------------------------------------------------------------------------------------------------------
+
+CREATE TABLE event_execution (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  event_handler_name varchar(255) NOT NULL,
+  event_name varchar(255) NOT NULL,
+  message_id varchar(255) NOT NULL,
+  execution_id varchar(255) NOT NULL,
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_event_execution ON event_execution (event_handler_name,event_name,message_id);
+
+CREATE TABLE poll_data (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  queue_name varchar(255) NOT NULL,
+  domain varchar(255) NOT NULL,
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_poll_data ON poll_data (queue_name,domain);
+CREATE INDEX ON poll_data (queue_name);
+
+CREATE TABLE task_scheduled (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  workflow_id varchar(255) NOT NULL,
+  task_key varchar(255) NOT NULL,
+  task_id varchar(255) NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_workflow_id_task_key ON task_scheduled (workflow_id,task_key);
+
+CREATE TABLE task_in_progress (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  task_def_name varchar(255) NOT NULL,
+  task_id varchar(255) NOT NULL,
+  workflow_id varchar(255) NOT NULL,
+  in_progress_status boolean NOT NULL DEFAULT false,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_task_def_task_id1 ON task_in_progress (task_def_name,task_id);
+
+CREATE TABLE task (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  task_id varchar(255) NOT NULL,
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_task_id ON task (task_id);
+
+CREATE TABLE workflow (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  workflow_id varchar(255) NOT NULL,
+  correlation_id varchar(255),
+  json_data TEXT NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_workflow_id ON workflow (workflow_id);
+
+CREATE TABLE workflow_def_to_workflow (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  workflow_def varchar(255) NOT NULL,
+  date_str varchar(60),
+  workflow_id varchar(255) NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_workflow_def_date_str ON workflow_def_to_workflow (workflow_def,date_str,workflow_id);
+
+CREATE TABLE workflow_pending (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  workflow_type varchar(255) NOT NULL,
+  workflow_id varchar(255) NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_workflow_type_workflow_id ON workflow_pending (workflow_type,workflow_id);
+CREATE INDEX workflow_type_index ON workflow_pending (workflow_type);
+
+CREATE TABLE workflow_to_task (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  modified_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  workflow_id varchar(255) NOT NULL,
+  task_id varchar(255) NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_workflow_to_task_id ON workflow_to_task (workflow_id,task_id);
+CREATE INDEX workflow_id_index ON workflow_to_task (workflow_id);
+
+-- --------------------------------------------------------------------------------------------------------------
+-- SCHEMA FOR QUEUE DAO
+-- --------------------------------------------------------------------------------------------------------------
+
+CREATE TABLE queue (
+  id SERIAL,
+  created_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  queue_name varchar(255) NOT NULL,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_queue_name ON queue (queue_name);
+
+CREATE TABLE queue_message (
+  id SERIAL,
+  created_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  deliver_on TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  queue_name varchar(255) NOT NULL,
+  message_id varchar(255) NOT NULL,
+  priority integer DEFAULT 0,
+  popped boolean DEFAULT false,
+  offset_time_seconds BIGINT,
+  payload TEXT,
+  PRIMARY KEY (id)
+);
+CREATE UNIQUE INDEX unique_queue_name_message_id ON queue_message (queue_name,message_id);
+CREATE INDEX combo_queue_message ON queue_message (queue_name,popped,deliver_on,created_on);
+
+-- V2__1009_Fix_PostgresExecutionDAO_Index.sql --
+DROP INDEX IF EXISTS unique_event_execution;
+
+CREATE UNIQUE INDEX unique_event_execution ON event_execution (event_handler_name,event_name,execution_id);
+
+-- V3__correlation_id_index.sql --
+DROP INDEX IF EXISTS workflow_corr_id_index;
+
+CREATE INDEX workflow_corr_id_index ON workflow (correlation_id);
+
+-- V4__new_qm_index_with_priority.sql --
+DROP INDEX IF EXISTS combo_queue_message;
+
+CREATE INDEX combo_queue_message ON queue_message (queue_name,priority,popped,deliver_on,created_on);

ansible/roles/conductor/templates/conductor-swarm-config.properties.j2

@@ -8,7 +8,16 @@ conductor.grpc.server.enabled=false
 # memory : The data is stored in memory and lost when the server dies.  Useful for testing or demo
 # redis : non-Dynomite based redis instance
 # dynomite : Dynomite cluster.  Use this for HA configuration.
+{% if use_postgres is defined and use_postgres %}
+db=postgres
+jdbc.url=jdbc:postgresql://{{ postgres_service_name }}:5432/{{ postgres_db }}
+jdbc.username={{ postgres_user }}
+jdbc.password={{ postgres_pass }}
+conductor.postgres.connection.pool.size.max=10
+conductor.postgres.connection.pool.idle.min=2
+flyway.enabled=false
 
+{% else %}
 db=dynomite
 
 # Dynomite Cluster details.
@@ -46,6 +55,7 @@ queues.dynomite.threads=3
 # When using redis directly, set this to the same port as redis server
 # For Dynomite, this is 22122 by default or the local redis-server port used by Dynomite.
 queues.dynomite.nonQuorum.port=22122
+{% endif %}
 
 # Elastic search instance type. Possible values are memory and external.
 # If not specified, the instance type will be embedded in memory

ansible/roles/postgres/defaults/main.yml

@@ -0,0 +1,7 @@
+---
+use_postgres: True
+postgres_service_name: 'postgresdb'
+postgres_replicas: 1
+postgres_user: conductor
+postgres_pass: password
+postgres_db: conductor

ansible/roles/postgres/tasks/main.yaml

@@ -0,0 +1,5 @@
+---
+- name: Generate postgres swarm
+  template:
+    src: templates/postgres-swarm.yaml.j2
+    dest: "{{ target_path }}/postgres-swarm.yaml"

ansible/roles/postgres/templates/postgres-swarm.yaml.j2

@@ -0,0 +1,31 @@
+version: '3.6'
+
+services:
+
+  {{ postgres_service_name }}:
+    image: postgres
+    ports:
+    - "5432:5432"
+    environment:
+      POSTGRES_USER: "{{ postgres_user }}"
+      POSTGRES_PASSWORD: "{{ postgres_pass }}"
+      POSTGRES_DB: "{{ postgres_db }}"
+{% if init_db %}
+    configs:
+      - source: db-init
+        target: "/docker-entrypoint-initdb.d/db-init.sql"
+{% endif %}
+    networks:
+      - {{ conductor_network }}
+    deploy:
+      replicas: {{ postgres_replicas }}
+      placement:
+        constraints: [node.role == worker]
+
+networks:
+   {{ conductor_network }}:
+{% if init_db %}
+configs:
+  db-init:
+    file: {{ target_path }}/conductor-db-init.sql
+{% endif %}

ansible/site-with-postgres.yaml

@@ -0,0 +1,25 @@
+---
+- hosts: localhost
+  roles:
+    - common
+    - postgres
+    - elasticsearch
+    - conductor
+  tasks:
+    - name: Start postgres and es
+      docker_stack:
+        name: conductor
+        state: present
+        compose:
+          - "{{ target_path }}/postgres-swarm.yaml"
+          - "{{ target_path }}/elasticsearch-swarm.yaml"
+    - name: Waiting for postgres and ES DBs
+      pause:
+        seconds: 10
+
+    - name: Start conductor
+      docker_stack:
+        name: conductor
+        state: present
+        compose:
+          - "{{ target_path }}/conductor-swarm.yaml"