Merge pull request #219 from jembi/PLAT-669-update-kafka

PLAT-669 message-bus-kafka package uses new kafka with kraft in replicated fashion

.env.cluster

@@ -58,10 +58,10 @@ JS_REPORT_DEV_MOUNT=false
 JS_REPORT_PACKAGE_PATH=
 
 # Message Bus Kafka
-KAFKA_INSTANCES=3
 # Topics should comma seperated, optional include partion and repliction values
 # e.g. <topic>:<partions>:<replicationFactor> -> test:3:2 (defaults to <topics>:3:1)
 KAFKA_TOPICS=2xx,reprocess,3xx,metrics:3:3
+KAFKA_HOSTS=kafka-01:9092,kafka-02:9092,kafka-03:9092
 
 # MPI MEDIATOR
 MPI_MEDIATOR_INSTANCES=3
@@ -73,7 +73,6 @@ HAPI_PROXY_INSTANCES=3
 OPENHIM_MEMORY_LIMIT=4G
 ES_MEMORY_LIMIT=20G
 LOGSTASH_MEMORY_LIMIT=8G
-ZOOKEEPER_MEMORY_LIMIT=4G
 KAFKA_MEMORY_LIMIT=8G
 KAFDROP_MEMORY_LIMIT=500M
 

.env.remote

@@ -25,8 +25,6 @@ OPENHIM_MONGO_URL=mongodb://mongo-1:27017,mongo-2:27017,mongo-3:27017/openhim?re
 OPENHIM_MONGO_ATNAURL=mongodb://mongo-1:27017,mongo-2:27017,mongo-3:27017/openhim?replicaSet=mongo-set
 
 # Message Bus - Kafka
-
-KAFKA_INSTANCES=3
 # Topics should comma seperated, optional include partion and repliction values
 # e.g. <topic>:<partions>:<replicationFactor> -> test:3:2 (defaults to <topics>:3:1)
 KAFKA_TOPICS=2xx,reprocess,3xx,metrics:3:1

.github/workflows/run-tests.sh

@@ -19,7 +19,7 @@ for package in "${CHANGED_FILES[@]}"; do
  else
  IFS='/' read -r -a split_string <<<"$package"
  changed_packages["${split_string[0]}"]="true"
- 
+
  fi
 done
 
@@ -41,7 +41,7 @@ else
  DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":elk
  elif [[ $folder_name == *"kafka"* ]] || [[ $folder_name == *"monitoring"* ]]; then
  DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":kafka
- elif [[ $folder_name == *"openhim"* ]] || [[ $folder_name == *"client-registry-jempi"* ]]; then
+ elif [[ $folder_name == *"openhim"* ]]; then
  DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":openhim
  elif [[ $folder_name == *"reverse-proxy"* ]]; then
  DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":nginx
@@ -51,6 +51,8 @@ else
  DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":sante
  elif [[ $folder_name == *"monitoring"* ]]; then
  DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":monitoring
+ elif [[ $folder_name == *"jempi"* ]] && [[ $NODE_MODE == "single" ]]; then
+ DOCKER_HOST=ssh://ubuntu@$GITHUB_RUN_ID.jembi.cloud yarn test:"$NODE_MODE":jempi
  fi
  done
 fi

client-registry-jempi/docker-compose.api.yml

@@ -10,6 +10,7 @@ services:
  KC_API_URL: ${KC_API_URL}
  KC_JEMPI_ROOT_URL: ${KC_JEMPI_ROOT_URL}
  JEMPI_SESSION_SECRET: ${JEMPI_SESSION_SECRET}
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:

client-registry-jempi/docker-compose.combined.yml

@@ -3,6 +3,8 @@ version: '3.9'
 services:
  jempi-async-receiver:
  image: jembi/jempi-async-receiver:${JEMPI_ASYNC_RECEIVER_IMAGE_TAG}
+ environment:
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:
@@ -13,6 +15,8 @@ services:
 
  jempi-sync-receiver:
  image: jembi/jempi-sync-receiver:${JEMPI_SYNC_RECEIVER_IMAGE_TAG}
+ environment:
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:
@@ -23,6 +27,8 @@ services:
 
  jempi-pre-processor:
  image: jembi/jempi-pre-processor:${JEMPI_PRE_PROCESSOR_IMAGE_TAG}
+ environment:
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:
@@ -33,6 +39,8 @@ services:
 
  jempi-controller:
  image: jembi/jempi-controller:${JEMPI_CONTROLLER_IMAGE_TAG}
+ environment:
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:
@@ -43,6 +51,8 @@ services:
 
  jempi-em-calculator:
  image: jembi/jempi-em-calculator:${JEMPI_EM_CALCULATOR_IMAGE_TAG}
+ environment:
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:
@@ -53,6 +63,8 @@ services:
 
  jempi-linker:
  image: jembi/jempi-linker:${JEMPI_LINKER_IMAGE_TAG}
+ environment:
+ kafka.bootstrap.servers: ${KAFKA_HOSTS}
  deploy:
  replicas: 1
  resources:

client-registry-jempi/importer/docker-compose.config.yml

@@ -7,10 +7,10 @@ services:
  restart_policy:
  condition: none
  environment:
- KAFDROP_HOST: jempi-kafdrop
+ KAFDROP_HOST: kafdrop
  KAFDROP_PORT: 9013
  KAFKA_TOPICS: ${JEMPI_KAFKA_TOPICS}
- command: sh -c "wait-on -t 60000 http-get://jempi-kafdrop:9013/topic && node /kafka.js && echo 'success';"
+ command: sh -c "wait-on -t 60000 http-get://kafdrop:9013/topic && node /kafka.js && echo 'success';"
  configs:
  - source: jempi-kafka.js
  target: /kafka.js

client-registry-jempi/importer/kafka.js

@@ -2,7 +2,7 @@
 
 const http = require("http");
 
-const KAFDROP_HOST = process.env.KAFDROP_HOST || "jempi-kafdrop";
+const KAFDROP_HOST = process.env.KAFDROP_HOST || "kafdrop";
 const KAFDROP_PORT = process.env.KAFDROP_PORT || "9013";
 const KAFKA_TOPICS = process.env.KAFKA_TOPICS;
 

client-registry-jempi/package-metadata.json

@@ -4,18 +4,9 @@
  "description": "A patient matching and deduplicater for the platform",
  "type": "infrastructure",
  "version": "0.0.1",
- "dependencies": ["interoperability-layer-openhim"],
+ "dependencies": ["interoperability-layer-openhim", "message-bus-kafka"],
  "environmentVariables": {
- "JEMPI_KAFKA_01_MEMORY_LIMIT": "3G",
- "JEMPI_KAFKA_01_MEMORY_RESERVE": "500M",
- "JEMPI_KAFKA_02_MEMORY_LIMIT": "3G",
- "JEMPI_KAFKA_02_MEMORY_RESERVE": "500M",
- "JEMPI_KAFKA_03_MEMORY_LIMIT": "3G",
- "JEMPI_KAFKA_03_MEMORY_RESERVE": "500M",
- "JEMPI_KAFDROP_CPU_LIMIT": "0",
- "JEMPI_KAFDROP_CPU_RESERVE": "0.05",
- "JEMPI_KAFDROP_MEMORY_LIMIT": "3G",
- "JEMPI_KAFDROP_MEMORY_RESERVE": "500M",
+ "KAFKA_HOSTS": "kafka-01:9092",
  "JEMPI_ZERO_01_MEMORY_LIMIT": "3G",
  "JEMPI_ZERO_01_MEMORY_RESERVE": "500M",
  "JEMPI_ALPHA_01_MEMORY_LIMIT": "3G",
@@ -41,7 +32,6 @@
  "JEMPI_API_MEMORY_LIMIT": "3G",
  "JEMPI_API_MEMORY_RESERVE": "500M",
  "JEMPI_KAFKA_TOPICS": "JeMPI-async-preprocessor,JeMPI-patient-controller,JeMPI-patient-em,JeMPI-patient-linker,JeMPI-mu-linker,JeMPI-notifications",
- "JEMPI_KAFKA_DEBUG": false,
  "JEMPI_ASYNC_RECEIVER_IMAGE_TAG": "0.1.1",
  "JEMPI_SYNC_RECEIVER_IMAGE_TAG": "0.1.1",
  "JEMPI_PRE_PROCESSOR_IMAGE_TAG": "0.1.1",

client-registry-jempi/swarm.sh

@@ -4,7 +4,6 @@ declare ACTION=""
 declare MODE=""
 declare COMPOSE_FILE_PATH=""
 declare UTILS_PATH=""
-declare KAFKA_SERVICES=()
 declare DGRAPH_SERVICES=()
 declare COMBINED_SERVICES=()
 declare SERVICE_NAMES=()
@@ -25,17 +24,13 @@ function init_vars() {
  VOLUME_NAMES=("jempi-zero-01-data")
 
  for i in {1..3}; do
- KAFKA_SERVICES=(
- "${KAFKA_SERVICES[@]}"
- "jempi-kafka-0$i"
- )
  DGRAPH_SERVICES=(
  "${DGRAPH_SERVICES[@]}"
  "jempi-alpha-0$i"
  )
+
  VOLUME_NAMES=(
  "${VOLUME_NAMES[@]}"
- "jempi-kafka-0$i-data"
  "jempi-alpha-0$i-data"
  )
  done
@@ -50,10 +45,8 @@ function init_vars() {
  )
 
  SERVICE_NAMES=(
- "${KAFKA_SERVICES[@]}"
  "${DGRAPH_SERVICES[@]}"
  "${COMBINED_SERVICES[@]}"
- "jempi-kafdrop"
  "jempi-zero-01"
  "jempi-api"
  )
@@ -62,7 +55,6 @@ function init_vars() {
  readonly MODE
  readonly COMPOSE_FILE_PATH
  readonly UTILS_PATH
- readonly KAFKA_SERVICES
  readonly DGRAPH_SERVICES
  readonly COMBINED_SERVICES
  readonly SERVICE_NAMES
@@ -77,7 +69,6 @@ function import_sources() {
 }
 
 function initialize_package() {
- local kafdrop_dev_compose_param=""
  local dgraph_dev_compose_param=""
  local dgraph_zero_dev_compose_param=""
  local combined_dev_compose_param=""
@@ -87,7 +78,6 @@ function initialize_package() {
 
  if [[ "$MODE" == "dev" ]]; then
  log info "Running package in DEV mode"
- kafdrop_dev_compose_param="docker-compose.kafdrop-dev.yml"
  dgraph_dev_compose_param="docker-compose.dgraph-dev.yml"
  dgraph_zero_dev_compose_param="docker-compose.dgraph-zero-dev.yml"
  combined_dev_compose_param="docker-compose.combined-dev.yml"
@@ -102,14 +92,7 @@ function initialize_package() {
  fi
 
  (
- log info "Deploy Kafka"
- docker::deploy_service "${COMPOSE_FILE_PATH}" "docker-compose.kafka.yml"
- docker::deploy_sanity "${KAFKA_SERVICES[@]}"
-
- log info "Deploy Kafdrop"
- docker::deploy_service "${COMPOSE_FILE_PATH}" "docker-compose.kafdrop.yml" "$kafdrop_dev_compose_param"
- docker::deploy_sanity "jempi-kafdrop"
-
+ log info "Importing JeMPI Kafka topics"
  docker::deploy_config_importer "$COMPOSE_FILE_PATH/importer/docker-compose.config.yml" "jempi-kafka-config-importer" "jempi-kafka"
 
  log info "Deploy Dgraph"

data-mapper-logstash/docker-compose.yml

@@ -13,6 +13,8 @@ services:
  ES_HOSTS: ${ES_HOSTS}
  ES_ELASTIC: ${ES_ELASTIC}
  LS_JAVA_OPTS: ${LS_JAVA_OPTS}
+ # Kafka as an input (required in the implementations)
+ KAFKA_HOSTS: ${KAFKA_HOSTS}
  volumes:
  - logstash-data:/usr/share/logstash/data
  deploy:

data-mapper-logstash/package-metadata.json

@@ -14,6 +14,7 @@
  "LOGSTASH_MEMORY_LIMIT": "3G",
  "LOGSTASH_MEMORY_RESERVE": "500M",
  "ES_ELASTIC": "dev_password_only",
- "ES_HOSTS": "\"analytics-datastore-elastic-search:9200\""
+ "ES_HOSTS": "\"analytics-datastore-elastic-search:9200\"",
+ "KAFKA_HOSTS": "kafka-01:9092"
  }
 }