async-schema-fetching v2 (#72)

* async-schema-fetching v2
- moved kafka.send to happen after actual schema registration (in case registration fails and exception is thrown, it shouldn't reach kafka)
- removed globalThis.kafkaProducer
- removed kafka clientId & topic from config, because it should be compliant with consul API (only host, port & credentials, no other properties)
- changed config.brokers to host & port, same thing with env vars
- updated README

* linter + updated README

Author: tot-ra

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+
+## [2.2.0] - 2021-11-23
+
+### Added
+- New feature - asynchroneous notification of gateway via kafka to avoid polling & reduce schema inconsistency time. See [examples/gateway_service_kafka_notification] for gateway. Use env vars to configure schema-registry
+
 ## [2.1.1] - 2021-11-08
 
 ### Updated

README.md

@@ -17,6 +17,7 @@ Graphql schema storage as dockerized on-premise service for federated graphql ga
 - Provides UI for developers to see stored schema & its history diff
 - Stores & shows in UI persisted queries passed by the gateway for debugging
 - Stores service urls emulating managed federation: you no longer need to hardcode the services in your gateway's constructor, or rely on an additonal service (etcd, consul) for service discovery
+- Async schema registration of new schema with events to avoid polling (`schema-registry -> kafka -> gateway`)
 
 <img width="1309" alt="Screenshot 2020-08-31 at 15 40 43" src="https://user-images.githubusercontent.com/445122/91720806-65985c00-eba0-11ea-8763-986b9f3f166b.png">
 
@@ -27,7 +28,6 @@ Graphql schema storage as dockerized on-premise service for federated graphql ga
 - client tracking (for breaking changes)
 - schema usage tracking (for breaking changes)
 - separate ephemeral automatic PQs, registered by frontend (use cache only with TTL) from true PQs backend-registered persisted queries (use DB only)
-- async schema registration of new schema with events to avoid polling. `schema-registry -> kafka -> gateway`
 - integrate [inspector](https://graphql-inspector.com/docs/essentials/diff)
 
 ## Configuration
@@ -54,10 +54,13 @@ The following are the different environment variables that are looked up that al
 | REDIS_SECRET          | Password used to connect to Redis                                             | Empty                     |
 | ASSETS_URL            | Controls the url that web assets are served from                              | localhost:6001            |
 | NODE_ENV              | Specifies the environment. Use _production_ to load js/css from `dist/assets` | Empty                     |
-| ASYNC_SCHEMA_UPDATES  | Specifies if experimental Async Schema Updates is Enabled                  | false                     |
+| ASYNC_SCHEMA_UPDATES  | Specifies if async achema updates is enabled                                  | false                     |
+| KAFKA_BROKER_HOST     | Host name of the Kafka broker, used if ASYNC_SCHEMA_UPDATES = true            | gql-schema-registry-kafka |
+| KAFKA_BROKER_PORT     | Port used when connecting to Kafka, used if ASYNC_SCHEMA_UPDATES = true       | 9092                      |
 
 For development we rely on docker network and use hostnames from `docker-compose.yml`.
-For dynamic service discovery, see `app/config.js`. Node service uses to connect to mysql & redis and change it if you install it with own setup. If you use dynamic service discovery (consul/etcd), edit `diplomat.js`
+Node service uses to connect to mysql & redis and change it if you install it with own setup.
+For dynamic service discovery (if you need multiple hosts for scaling), override `app/config.js` and `diplomat.js`
 
 ## Installation
 
@@ -81,7 +84,6 @@ docker run -e DB_HOST=localhost -e DB_USERNAME=root pipedrive/graphql-schema-reg
 
 ### Docker-compose
 
-
 ```
 git clone https://github.com/pipedrive/graphql-schema-registry.git && cd graphql-schema-registry
 docker-compose up

app/config.js

@@ -20,12 +20,8 @@ module.exports = {
 			secret: process.env.REDIS_SECRET || '',
 		},
 		'gql-schema-registry-kafka': {
-			clientId:
-				process.env.KAFKA_CLIENT || 'graphql-schema-registry-server',
-			brokers: process.env.KAFKA_BROKERS
-				? process.env.KAFKA_BROKERS.split(',')
-				: ['gql-schema-registry-kafka:9092'],
-			topic: process.env.KAFKA_TOPIC || 'test-topic',
+			host: process.env.KAFKA_BROKER_HOST || 'gql-schema-registry-kafka',
+			port: process.env.KAFKA_BROKER_PORT || '9092',
 		},
 	},
 	asyncSchemaUpdates: Boolean(process.env.ASYNC_SCHEMA_UPDATES || 'false'),

app/index.js

@@ -2,7 +2,7 @@ const express = require('express');
 const logger = require('./logger');
 const { get } = require('lodash');
 const initGraphql = require('./graphql');
-const { producer } = require('./kafka/producer');
+const kafka = require('./kafka');
 const config = require('./config');
 
 const app = express();
@@ -105,12 +105,9 @@ exports.init = async () => {
 	if (server) {
 		return server;
 	}
+
 	if (config.asyncSchemaUpdates) {
-		producer()
-			.then((p) => {
-				globalThis.kafkaProducer = p;
-			})
-			.catch(console.error);
+		kafka.init();
 	}
 
 	server = app.listen(config.port, () => {

app/kafka/index.js

@@ -0,0 +1,32 @@
+const { Kafka } = require('kafkajs');
+const diplomat = require('../diplomat');
+
+const KAFKA_SCHEMA_REGISTRY =
+	process.env.KAFKA_SCHEMA_REGISTRY || 'gql-schema-registry-kafka';
+
+let producer = null;
+
+exports.init = async () => {
+	const { host, port } = diplomat.getServiceInstance(KAFKA_SCHEMA_REGISTRY);
+
+	const kafka = new Kafka({
+		clientId: process.env.KAFKA_CLIENT || 'graphql-schema-registry-server',
+		brokers: [`${host}:${port}`],
+	});
+
+	producer = kafka.producer();
+	await producer.connect();
+};
+
+exports.send = (data) => {
+	if (!producer) {
+		throw new Error('Kafka producer not initialized');
+	}
+
+	producer.send({
+		topic: process.env.KAFKA_TOPIC || 'test-topic',
+		messages: [{ value: JSON.stringify(data) }],
+	});
+};
+
+exports.KAFKA_SCHEMA_REGISTRY = KAFKA_SCHEMA_REGISTRY;

app/kafka/producer.js

@@ -6,11 +6,8 @@ const {
 	deactivateSchema,
 	diffSchemas,
 } = require('../controller/schema');
-const diplomat = require('../diplomat');
 const config = require('../config');
-const { KAFKA_SCHEMA_REGISTRY } = require('../kafka/producer');
-
-const { topic } = diplomat.getServiceInstance(KAFKA_SCHEMA_REGISTRY);
+const kafka = require('../kafka');
 
 exports.composeLatest = async (req, res) => {
 	const schema = await getAndValidateSchema();
@@ -58,16 +55,15 @@ exports.push = async (req, res) => {
 		})
 	);
 
+	const data = await pushAndValidateSchema({ service });
+
 	if (config.asyncSchemaUpdates) {
-		await globalThis.kafkaProducer.send({
-			topic,
-			messages: [{ value: 'Schema Updated!' }],
-		});
+		await kafka.send(data);
 	}
 
 	return res.json({
 		success: true,
-		data: await pushAndValidateSchema({ service }),
+		data,
 	});
 };
 

app/router/schema.js

@@ -40,7 +40,8 @@ services:
       - DB_SCHEMA_REGISTRY=gql-schema-registry-db
       - ASYNC_SCHEMA_UPDATES=true
       - KAFKA_CLIENT=graphql-schema-registry-server
-      - KAFKA_BROKERS=gql-schema-registry-kafka:9092
+      - KAFKA_BROKER_HOST=gql-schema-registry-kafka
+      - KAFKA_BROKER_PORT=9092
       - KAFKA_TOPIC=test-topic
     volumes:
       - .:/app/

docker-compose.with-kafka-dev.yml

@@ -1,6 +1,6 @@
 {
   "name": "graphql-schema-registry",
-  "version": "2.1.1",
+  "version": "2.2.0",
   "description": "Graphql schema registry",
   "main": "schema-registry.js",
   "scripts": {