multiple node with data thresholding (#912)

* fix build error

* single endpoint

* fix GQL type name collisions

* move data mask threshold env. ensure data masking applies even if hits.total is not queried

* change belowThreshold to relation object

* surface api errors to axios responses

* add relation logic and agg merging for network search

* read data mask threshold min from config

* check for valid env val

* remove unused func

* fix TS build errors

* fix schema merge error

---------

Co-authored-by: Ciaran Schutte <[email protected]>

Author: ciaranschutte

modules/server/.env.schema

@@ -1,6 +1,6 @@
 ALLOW_CUSTOM_MAX_DOWNLOAD_ROWS=false
 CONFIG_PATH=./configs
-DATA_MASK_THRESHOLD=
+DATA_MASK_MIN_THRESHOLD=
 DEBUG=false
 DOCUMENT_TYPE=''
 DOWNLOAD_STREAM_BUFFER_SIZE=2000

modules/server/src/app.js

@@ -14,7 +14,7 @@ export default async function (rootPath = '') {
 	 * @param {boolean} enableAdmin
 	 * @param {boolean} enabledDocumentHits - enables including "hits" property in the GQL response
 	 */
-	return Arranger({
+	return arranger({
 		enableAdmin: ENV_CONFIG.ENABLE_ADMIN,
 		enableDocumentHits: ENV_CONFIG.ENABLE_DOCUMENT_HITS,
 	}).then((router) => {

modules/server/src/config/constants.ts

@@ -4,7 +4,8 @@ export const ALLOW_CUSTOM_MAX_DOWNLOAD_ROWS = stringToBool(
 	process.env.ALLOW_CUSTOM_MAX_DOWNLOAD_ROWS,
 );
 export const CONFIG_FILES_PATH = process.env.CONFIG_PATH || './configs';
-export const DATA_MASK_THRESHOLD = process.env.DATA_MASK_THRESHOLD || Number.MAX_SAFE_INTEGER;
+export const DATA_MASK_MIN_THRESHOLD =
+	process.env.DATA_MASK_MIN_THRESHOLD || Number.MAX_SAFE_INTEGER;
 export const DEBUG_MODE = stringToBool(process.env.DEBUG);
 export const DOCUMENT_TYPE = process.env.DOCUMENT_TYPE || '';
 export const DOWNLOAD_STREAM_BUFFER_SIZE =

modules/server/src/graphqlRoutes.js

@@ -3,12 +3,13 @@ import { ApolloServer } from 'apollo-server-express';
 import { Router } from 'express';
 import expressPlayground from 'graphql-playground-middleware-express';
 
-import getConfigObject, { initializeSets } from './config';
+import { mergeSchemas } from '@graphql-tools/schema';
+import getConfigObject, { ENV_CONFIG, initializeSets } from './config';
 import { DEBUG_MODE, ES_PASS, ES_USER } from './config/constants';
 import { ConfigProperties } from './config/types';
 import { addMappingsToTypes, extendFields, fetchMapping } from './mapping';
 import { extendColumns, extendFacets, flattenMappingToFields } from './mapping/extendMapping';
-import { createSchemaFromNetworkConfig, mergeSchemas } from './network';
+import { createSchemaFromNetworkConfig } from './network';
 import makeSchema from './schema';
 
 const getESMapping = async (esClient, index) => {
@@ -156,32 +157,13 @@ const noSchemaHandler =
 		});
 	};
 
-const createEndpoint = async ({
-	esClient,
-	graphqlOptions = {},
-	mockSchema,
-	schema,
-	networkSchema,
-}) => {
+const createEndpoint = async ({ esClient, graphqlOptions = {}, mockSchema, schema }) => {
 	const mainPath = '/graphql';
 	const mockPath = '/mock/graphql';
 	const router = Router();
 
 	console.log('Starting GraphQL server:');
 
-	if (ENABLE_NETWORK_AGGREGATION) {
-		/**
-		 * TODO: make available on one route
-		 */
-		const networkPath = '/network';
-		const apolloNetworkServer = new ApolloServer({
-			cache: 'bounded',
-			schema: networkSchema,
-		});
-		await apolloNetworkServer.start();
-		apolloNetworkServer.applyMiddleware({ app: router, path: networkPath });
-	}
-
 	try {
 		await router.get(
 			mainPath,
@@ -261,10 +243,11 @@ const createEndpoint = async ({
 	return router;
 };
 
-export const createSchemasFromConfigs = async ({
+const createSchemasFromConfigs = async ({
 	configsSource = '',
 	enableAdmin,
 	enableDocumentHits,
+	enableNetworkAggregation,
 	esClient,
 	getServerSideFilter,
 	graphqlOptions = {},
@@ -287,28 +270,30 @@ export const createSchemasFromConfigs = async ({
 			types: typesWithMappings,
 		});
 
-		/**
+		const schemasToMerge = [schema];
+
+		/*
 		 * Federated Network Search
 		 */
-		if (ENABLE_NETWORK_AGGREGATION) {
-			const { networkSchema } = await createSchemaFromNetworkConfig({
+		if (enableNetworkAggregation) {
+			const networkSchema = await createSchemaFromNetworkConfig({
 				networkConfigs: configsFromFiles[ConfigProperties.NETWORK_AGGREGATION].map((config) => ({
 					...config,
-					/**
+					/*
 					 * part of the gql schema is generated dynamically
-					 * in the case of the "file" field, the field name and type name are the same
-					 * it's more flexible to define it here as an additional property than to confuse functions further down the pipeline
+					 * in the case of the "file" field, the field name and gql type name are the same
 					 */
 					documentName: config.documentType,
 				})),
 			});
+			schemasToMerge.push(networkSchema);
 		}
 
+		const fullSchema = mergeSchemas({ schemas: schemasToMerge });
 		return {
 			...commonFields,
 			mockSchema,
-			schema,
-			networkSchema,
+			schema: fullSchema,
 		};
 	} catch (error) {
 		const message = error?.message || error;
@@ -323,16 +308,18 @@ export default async ({
 	configsSource = '',
 	enableAdmin,
 	enableDocumentHits,
+	enableNetworkAggregation,
 	esClient,
 	getServerSideFilter,
 	graphqlOptions = {},
 }) => {
 	try {
-		const { fieldsFromMapping, mockSchema, schema, typesWithMappings, networkSchema } =
+		const { fieldsFromMapping, mockSchema, schema, typesWithMappings } =
 			await createSchemasFromConfigs({
 				configsSource,
 				enableAdmin,
 				enableDocumentHits,
+				enableNetworkAggregation,
 				esClient,
 				getServerSideFilter,
 				graphqlOptions,
@@ -343,7 +330,6 @@ export default async ({
 			graphqlOptions,
 			mockSchema,
 			schema,
-			networkSchema,
 		});
 
 		await initializeSets({ esClient });

modules/server/src/index.js

@@ -1,3 +1,3 @@
-export { default as getGraphQLRoutes, createSchemasFromConfigs } from './graphqlRoutes';
-export { default } from './server';
 export { default as App } from './app';
+export { default as getGraphQLRoutes } from './graphqlRoutes';
+export { default } from './server';

modules/server/src/mapping/createConnectionResolvers.ts

@@ -9,7 +9,6 @@ export type CreateConnectionResolversArgs = {
 	createStateResolvers?: boolean;
 	enableAdmin: boolean;
 	enableDocumentHits: boolean;
-	dataMaskThreshold: number;
 	getServerSideFilter?: GetServerSideFilterFn;
 	Parallel: any;
 	type: Record<string, any>;
@@ -20,7 +19,6 @@ const createConnectionResolvers: CreateConnectionResolversFn = ({
 	createStateResolvers = true,
 	enableAdmin,
 	enableDocumentHits,
-	dataMaskThreshold,
 	getServerSideFilter,
 	Parallel,
 	type,
@@ -30,7 +28,6 @@ const createConnectionResolvers: CreateConnectionResolversFn = ({
 		type,
 		Parallel,
 		getServerSideFilter,
-		dataMaskThreshold,
 		enableDocumentHits,
 	});
 

modules/server/src/mapping/createConnectionTypeDefs.js

@@ -12,8 +12,6 @@ export default ({ type, fields = '', createStateTypeDefs = true, showRecords })
         aggregations_filter_themselves: Boolean
       ): ${type.name}Aggregations
 
-      ${!showRecords ? 'dataMasking: DataMasking' : ''}
-
       configs: ${createStateTypeDefs ? 'ConfigsWithState' : 'ConfigsWithoutState'}
 
       hits(

modules/server/src/mapping/masking.ts

@@ -1,8 +1,15 @@
+import { ENV_CONFIG } from '@/config';
 import { Aggregation } from './types';
 
+export const Relation = {
+	eq: 'eq',
+	gte: 'gte',
+} as const;
+export type Relation = keyof typeof Relation;
+
 /**
  * This returns a total count that is less than or equal to the actual total hits in the query.
- * It is calculated by adding +1 for values under threshold and bucket.doc_count
+ * It is calculated by adding +1 for values under threshold or adding bucket.doc_count amount
  * for values greater than or equal to
  *
  * @param aggregation an aggregation with the most buckets which has data masking applied
@@ -18,7 +25,8 @@ const calculateHitsFromAggregation = ({
 		return 0;
 	}
 	return aggregation.buckets.reduce(
-		(totalAcc, bucket) => (bucket.belowThreshold ? totalAcc + 1 : totalAcc + bucket.doc_count),
+		(totalAcc, bucket) =>
+			bucket.relation === Relation.gte ? totalAcc + 1 : totalAcc + bucket.doc_count,
 		0,
 	);
 };
@@ -29,12 +37,10 @@ const calculateHitsFromAggregation = ({
  * 2) Find the agg with the most bucket count and data masking applied to be used in calculating hits.total
  *
  * @param aggregations - aggregations from query
- * @param thresholdMin - threshold value
  * @returns aggregations with data masking applied and hits total
  */
 export const applyAggregationMasking = ({
 	aggregations,
-	thresholdMin,
 }: {
 	aggregations: Record<
 		string,
@@ -43,13 +49,16 @@ export const applyAggregationMasking = ({
 			buckets: Array<{
 				doc_count: number;
 				key: string;
+				relation: Relation;
 			}>;
 		}
 	>;
-	thresholdMin: number;
 }) => {
-	// set data masked properties to one less than the configured threshold value (under threshold)
-	const THRESHOLD_REPLACEMENT_VALUE = thresholdMin - 1;
+	const thresholdMin = ENV_CONFIG.DATA_MASK_MIN_THRESHOLD;
+	if (thresholdMin < 1) {
+		throw Error('DATA_MASK_MIN_THRESHOLD environment variable has to be a positive integer.');
+	}
+	const THRESHOLD_REPLACEMENT_VALUE = 1;
 
 	const { aggsTotal: dataMaskedAggregations, totalHitsAgg } = Object.entries(aggregations).reduce<{
 		aggsTotal: Record<string, Aggregation>;
@@ -59,12 +68,13 @@ export const applyAggregationMasking = ({
 			// mask buckets if under threshold
 			const dataMaskedBuckets = aggregation.buckets.map((bucket) =>
 				bucket.doc_count < thresholdMin
-					? { ...bucket, doc_count: THRESHOLD_REPLACEMENT_VALUE, belowThreshold: true }
-					: { ...bucket, belowThreshold: false },
+					? { ...bucket, doc_count: THRESHOLD_REPLACEMENT_VALUE, relation: Relation.gte }
+					: { ...bucket, relation: Relation.eq },
 			);
 
 			// update total hits selected agg if needed
-			const bucketIsMasked = dataMaskedBuckets.some((bucket) => bucket.belowThreshold);
+			const bucketIsMasked = dataMaskedBuckets.some((bucket) => bucket.relation === Relation.gte);
+			// take aggregation with the most buckets that has masked data
 			const hitsAgg =
 				totalHitsAgg.bucketCount < aggregation.bucket_count && bucketIsMasked
 					? { key: type, bucketCount: aggregation.bucket_count }

modules/server/src/mapping/resolvers.ts

@@ -1,12 +1,11 @@
 import { ConfigProperties, ExtendedConfigsInterface } from '@/config/types';
 import { GraphQLResolveInfo } from 'graphql';
 import { get } from 'lodash';
-import { Context } from 'vm';
 import { CreateConnectionResolversArgs } from './createConnectionResolvers';
 import { applyAggregationMasking } from './masking';
 import resolveAggregations, { aggregationsToGraphql } from './resolveAggregations';
 import resolveHits from './resolveHits';
-import { Aggregation, Hits, Root } from './types';
+import { Aggregation, Context, Hits, Root } from './types';
 
 /**
  * Resolve hits from aggregations
@@ -27,7 +26,6 @@ const resolveHitsFromAggs =
 			context: Context,
 			info: GraphQLResolveInfo,
 		) => Record<string, Aggregation>,
-		dataMaskThreshold: number,
 	) =>
 	async (obj: Root, args: Hits, context: Context, info: GraphQLResolveInfo) => {
 		/*
@@ -36,7 +34,8 @@ const resolveHitsFromAggs =
 		 */
 		const aggregationsPath = 'operation.selectionSet.selections[0].selectionSet.selections';
 		const aggregationsSelectionSet = get(info, aggregationsPath, []).find(
-			(selection) => selection.kind === 'Field' && selection.name.value === 'aggregations',
+			(selection: { kind: string; name: { value: string } }) =>
+				selection.kind === 'Field' && selection.name.value === 'aggregations',
 		);
 
 		/*
@@ -49,9 +48,8 @@ const resolveHitsFromAggs =
 			// modifying the query info field inline so it can query aggregations correctly
 			// not idiomatic so doesn't line up with typings from graphql
 			const aggregations = await aggregationsQuery(obj, info.variableValues, context, modifiedInfo);
-			const { hitsTotal: total } = applyAggregationMasking({
+			const { hitsTotal: total, dataMaskedAggregations } = applyAggregationMasking({
 				aggregations,
-				thresholdMin: dataMaskThreshold,
 			});
 			return { total };
 		} else {
@@ -64,7 +62,6 @@ export const createResolvers = ({
 	type,
 	Parallel,
 	getServerSideFilter,
-	dataMaskThreshold,
 	enableDocumentHits,
 }: Omit<CreateConnectionResolversArgs, 'enableAdmin'>) => {
 	// configs
@@ -97,8 +94,15 @@ export const createResolvers = ({
 		context: Context,
 		info: GraphQLResolveInfo,
 	) => {
-		const aggs = await aggregationsQuery(obj, args, context, info);
-		return aggregationsToGraphql(aggs);
+		const aggregations = await aggregationsQuery(obj, args, context, info);
+		if (enableDocumentHits) {
+			return aggregationsToGraphql(aggregations);
+		} else {
+			const { dataMaskedAggregations } = applyAggregationMasking({
+				aggregations,
+			});
+			return aggregationsToGraphql(dataMaskedAggregations);
+		}
 	};
 
 	// hits
@@ -108,7 +112,7 @@ export const createResolvers = ({
 		: // @ts-ignore
 		  // typing resolveAggregations requires typing a lot of code down the chain
 		  // TODO: improve typing
-		  resolveHitsFromAggs(aggregationsQuery, dataMaskThreshold);
+		  resolveHitsFromAggs(aggregationsQuery);
 
 	return { hits, aggregations, configs };
 };

modules/server/src/mapping/types.ts

@@ -1,9 +1,10 @@
 import { Client } from '@elastic/elasticsearch';
+import { Relation } from './masking';
 
 export type Bucket = {
 	doc_count: number;
 	key: string;
-	belowThreshold: boolean;
+	relation: Relation;
 };
 
 export type Aggregation = {