refactor: introduce grpc client params to enable LB, fix state storin…

…g inconsistencies. (#1157) * fix(authentication, authorization): teamMembers count, remove default pagination from rpc routes (#1161) * fix(authentication): count members users instead of relations * fix(authorization): remove rpc default limit * fix(admin,core,router): state publishing inconsistency * feat(grpc-sdk,hermes): round robin lb for grpc connections --------- Co-authored-by: Rena Tsagka <[email protected]>
ConduitPlatform · Sep 19, 2024 · feb2911 · feb2911
1 parent 6ff5cd8
commit feb2911
Show file tree

Hide file tree

Showing 5 changed files with 23 additions and 9 deletions.
diff --git a/libraries/grpc-sdk/src/classes/ConduitModule.ts b/libraries/grpc-sdk/src/classes/ConduitModule.ts
@@ -73,7 +73,9 @@ export class ConduitModule<T extends CompatServiceDefinition> {
     this.channel = createChannel(this._serviceUrl, undefined, {
       'grpc.max_receive_message_length': 1024 * 1024 * 100,
       'grpc.max_send_message_length': 1024 * 1024 * 100,
+      'grpc.service_config': '{"loadBalancingConfig":[{"round_robin":{}}]}',
     });
+
     let clientFactory = createClientFactory()
       .use(
         this._grpcToken

diff --git a/libraries/hermes/src/utils/GrpcConverter.ts b/libraries/hermes/src/utils/GrpcConverter.ts
@@ -67,6 +67,7 @@ export function grpcToConduitRoute(
     {
       'grpc.max_receive_message_length': 1024 * 1024 * 100,
       'grpc.max_send_message_length': 1024 * 1024 * 100,
+      'grpc.service_config': '{"loadBalancingConfig":[{"round_robin":{}}]}',
     },
   );
 

diff --git a/modules/router/src/Router.ts b/modules/router/src/Router.ts
@@ -44,6 +44,7 @@ import * as adminRoutes from './admin/routes/index.js';
 import metricsSchema from './metrics/index.js';
 import { ConfigController, ManagedModule } from '@conduitplatform/module-tools';
 import { fileURLToPath } from 'node:url';
+
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = path.dirname(__filename);
 
@@ -161,9 +162,9 @@ export default class ConduitDefaultRouter extends ManagedModule<Config> {
       const messageParsed = JSON.parse(message);
       try {
         this.internalRegisterRoute(
-          messageParsed.protofile,
           messageParsed.routes,
           messageParsed.url,
+          messageParsed.moduleName,
         );
       } catch (err) {
         ConduitGrpcSdk.Logger.error(err as Error);
@@ -172,7 +173,11 @@ export default class ConduitDefaultRouter extends ManagedModule<Config> {
     this.scheduleMiddlewareApply();
   }
 
-  updateState(routes: RegisterConduitRouteRequest_PathDefinition[], url: string) {
+  updateState(
+    routes: RegisterConduitRouteRequest_PathDefinition[],
+    url: string,
+    moduleName: string,
+  ) {
     this.grpcSdk
       .state!.modifyState(async (existingState: Indexable) => {
         const state = existingState ?? {};
@@ -189,12 +194,13 @@ export default class ConduitDefaultRouter extends ManagedModule<Config> {
           state.routes.push({
             routes,
             url,
+            moduleName,
           });
         }
         return state;
       })
       .then(() => {
-        this.publishRouteData(routes, url);
+        this.publishRouteData(routes, url, moduleName);
         ConduitGrpcSdk.Logger.log('Updated routes state');
       })
       .catch(e => {
@@ -203,12 +209,17 @@ export default class ConduitDefaultRouter extends ManagedModule<Config> {
       });
   }
 
-  publishRouteData(routes: RegisterConduitRouteRequest_PathDefinition[], url: string) {
+  publishRouteData(
+    routes: RegisterConduitRouteRequest_PathDefinition[],
+    url: string,
+    moduleName: string,
+  ) {
     this.grpcSdk.bus!.publish(
       'router',
       JSON.stringify({
         routes,
         url,
+        moduleName,
       }),
     );
   }
@@ -253,7 +264,7 @@ export default class ConduitDefaultRouter extends ManagedModule<Config> {
         moduleName as string,
       );
 
-      this.updateState(call.request.routes, call.request.routerUrl);
+      this.updateState(call.request.routes, call.request.routerUrl, moduleName as string);
       this.scheduleMiddlewareApply();
     } catch (err) {
       ConduitGrpcSdk.Logger.error(err as Error);

diff --git a/packages/admin/src/index.ts b/packages/admin/src/index.ts
@@ -383,7 +383,7 @@ export default class AdminModule extends IConduitAdmin {
                   }
                 });
             }
-            return this.internalRegisterRoute(r.protofile, r.routes, r.url);
+            return this.internalRegisterRoute(r.routes, r.url);
           } catch (err) {
             ConduitGrpcSdk.Logger.error(err as Error);
           }
@@ -418,9 +418,9 @@ export default class AdminModule extends IConduitAdmin {
       const messageParsed = JSON.parse(message);
       try {
         this.internalRegisterRoute(
-          messageParsed.protofile,
           messageParsed.routes,
           messageParsed.url,
+          messageParsed.moduleName,
         );
       } catch (err) {
         ConduitGrpcSdk.Logger.error(err as Error);

diff --git a/packages/core/src/config-manager/service-discovery/index.ts b/packages/core/src/config-manager/service-discovery/index.ts
@@ -77,7 +77,7 @@ export class ServiceDiscovery {
           });
       }
     }
-    this.grpcSdk.bus!.subscribe('config', (message: string) => {
+    this.grpcSdk.bus!.subscribe('service-discover', (message: string) => {
       const parsedMessage = JSON.parse(message);
       if (parsedMessage.type === 'module-health') {
         this._serviceMonitor.updateModuleHealth(
@@ -233,7 +233,7 @@ export class ServiceDiscovery {
     status?: HealthCheckStatus,
   ) {
     this.grpcSdk.bus!.publish(
-      'config',
+      'service-discover',
       JSON.stringify({
         type,
         name,