feat: Conduit Manifest

libraries/grpc-sdk/package.json

@@ -25,6 +25,7 @@
     "fast-jwt": "^1.6.0",
     "fs-extra": "^10.1.0",
     "ioredis": "^5.1.0",
+    "jsonschema": "^1.4.1",
     "lodash": "^4.17.21",
     "nice-grpc": "^1.2.0",
     "nice-grpc-client-middleware-retry": "^1",

libraries/grpc-sdk/src/classes/HealthCheck.ts

@@ -1,9 +1,10 @@
 import { getGrpcSignedTokenInterceptor, getModuleNameInterceptor } from '../interceptors';
 import { createChannel, createClientFactory } from 'nice-grpc';
-import { HealthCheckResponse, HealthDefinition } from '../protoUtils/grpc_health_check';
+import { HealthDefinition } from '../protoUtils/grpc_health_check';
 import { clientMiddleware } from '../metrics/clientMiddleware';
+import { HealthCheckStatus } from '../types';
 
-export async function checkModuleHealth(
+export async function checkServiceHealth(
   clientName: string,
   serviceUrl: string,
   service: string = '',
@@ -21,17 +22,13 @@ export async function checkModuleHealth(
         : getModuleNameInterceptor(clientName),
     );
   const _healthClient = clientFactory.create(HealthDefinition, channel);
-
-  let error;
-  let status = await _healthClient
+  return _healthClient
     .check({ service })
-    .then((res: HealthCheckResponse) => {
+    .then(res => {
       return res.status;
     })
-    .catch(err => (error = err));
-  channel.close();
-  if (!error) {
-    return status;
-  }
-  throw error;
+    .catch(() => {
+      channel.close();
+      return HealthCheckStatus.SERVICE_UNKNOWN;
+    }) as Promise<HealthCheckStatus>;
 }

libraries/grpc-sdk/src/classes/ManagedModule.ts

@@ -1,8 +1,13 @@
 import ConduitGrpcSdk, {
   ConduitService,
   GrpcServer,
+  GrpcRequest,
+  GrpcResponse,
   SetConfigRequest,
   SetConfigResponse,
+  Indexable,
+  ModuleActivationResponse,
+  ModuleActivationStatus,
 } from '..';
 import { ConduitServiceModule } from './ConduitServiceModule';
 import { ConfigController } from './ConfigController';
@@ -11,6 +16,7 @@ import { status } from '@grpc/grpc-js';
 import convict from 'convict';
 
 export abstract class ManagedModule<T> extends ConduitServiceModule {
+  private _activated: boolean = false;
   protected abstract readonly configSchema?: object;
   protected abstract readonly metricsSchema?: object;
   readonly config?: convict.Config<T>;
@@ -122,6 +128,10 @@ export abstract class ManagedModule<T> extends ConduitServiceModule {
       this._serviceName = this.service.protoDescription.substring(
         this.service.protoDescription.indexOf('.') + 1,
       );
+      this.service.functions['ActivateModule'] = this.activateModule.bind(this);
+      if (this.config) {
+        this.service.functions['SetConfig'] = this.setConfig.bind(this);
+      }
       await this.grpcServer.addService(
         this.service.protoPath,
         this.service.protoDescription,
@@ -133,6 +143,38 @@ export abstract class ManagedModule<T> extends ConduitServiceModule {
     }
   }
 
+  async activateModule(
+    call: GrpcRequest<null>,
+    callback: GrpcResponse<ModuleActivationResponse>,
+  ) {
+    if (this._activated) {
+      return callback(null, { status: ModuleActivationStatus.ALREADY_ACTIVATED });
+    }
+    try {
+      await this.onRegister();
+      if (this.config) {
+        const configSchema = this.config.getSchema();
+        let config: any = this.config.getProperties();
+        config = await this.preConfig(config);
+        config = await this.grpcSdk.config.configure(
+          config,
+          convictConfigParser(configSchema),
+          this.configOverride,
+        );
+        ConfigController.getInstance();
+        if (config) ConfigController.getInstance().config = config;
+        if (!config || config.active || !config.hasOwnProperty('active'))
+          await this.onConfig();
+      }
+    } catch (err) {
+      ConduitGrpcSdk.Logger.error('Failed to activate module');
+      ConduitGrpcSdk.Logger.error(err as Error);
+      process.exit(-1);
+    }
+    this._activated = true;
+    return callback(null, { status: ModuleActivationStatus.ACTIVATED });
+  }
+
   async setConfig(call: SetConfigRequest, callback: SetConfigResponse) {
     try {
       if (!this.config) {
@@ -179,3 +221,16 @@ export abstract class ManagedModule<T> extends ConduitServiceModule {
     });
   }
 }
+
+const convictConfigParser = (config: Indexable) => {
+  if (typeof config === 'object') {
+    Object.keys(config).forEach(key => {
+      if (key === '_cvtProperties') {
+        config = convictConfigParser(config._cvtProperties);
+      } else {
+        config[key] = convictConfigParser(config[key]);
+      }
+    });
+  }
+  return config;
+};

libraries/grpc-sdk/src/classes/ManifestManager.ts

@@ -0,0 +1,229 @@
+import { Schema, Validator, ValidationError } from 'jsonschema';
+import fs from 'fs-extra';
+import ConduitGrpcSdk from '../index';
+import {
+  DeploymentState_ModuleStateInfo as ModuleStateInfo,
+  RegisterModuleRequest_ConduitManifest as ConduitManifest,
+  RegisterModuleRequest_ConduitManifest_Dependency as ConduitManifestDependency,
+} from '../protoUtils/core';
+
+const jsonManifestSchema: Schema = {
+  title: 'ConduitManifest',
+  type: 'object',
+  required: true,
+  properties: {
+    version: {
+      type: 'String',
+      required: true,
+    },
+    dependencies: {
+      type: 'object',
+      required: false, // required for everyone, except Core
+      properties: {
+        conduit: {
+          type: 'object',
+          properties: {
+            version: {
+              type: 'string',
+              required: true,
+            },
+          },
+        },
+      },
+      additionalProperties: {
+        // dependency (module)
+        type: 'object',
+        properties: {
+          version: {
+            type: 'string',
+            required: true,
+          },
+        },
+      },
+    },
+  },
+};
+
+type JsonManifest = {
+  version: string;
+  dependencies: {
+    // empty for Core
+    [dep: string]: {
+      version: string;
+    };
+  };
+};
+
+enum TagComparisonOperator {
+  Equal = 0,
+  GreaterEqual = 1,
+}
+
+export class ManifestError extends Error {
+  constructor(errors: ValidationError[], jsonPath: string) {
+    let errorString = `${errors.length} Conduit manifest ${
+      errors.length > 1 ? 'errors' : 'error'
+    } detected in ${jsonPath}.\n`;
+    errors.forEach(e => (errorString = `${errorString.concat(e.stack)}.\n`));
+    super(errorString);
+  }
+}
+
+export class ManifestManager {
+  private static _instance: ManifestManager;
+  readonly manifest: ConduitManifest;
+
+  private constructor(
+    private readonly grpcSdk: ConduitGrpcSdk,
+    private readonly moduleName: string,
+    packageJsonPath: string,
+  ) {
+    jsonManifestSchema.properties!.dependencies.required = grpcSdk.isModule;
+    this.manifest = this.parsePackageJson(packageJsonPath, jsonManifestSchema);
+  }
+
+  get moduleVersion() {
+    return this.manifest.moduleVersion;
+  }
+
+  static getInstance(
+    grpcSdk?: ConduitGrpcSdk,
+    moduleName?: string,
+    packageJsonPath?: string,
+  ) {
+    if (ManifestManager._instance) return ManifestManager._instance;
+    if (grpcSdk === undefined) {
+      throw new Error('ConduitGrpcSdk not provided!');
+    }
+    if (!moduleName) {
+      throw new Error('moduleName not provided!');
+    }
+    if (!packageJsonPath) {
+      throw new Error('packageJsonPath not provided!');
+    }
+    ManifestManager._instance = new ManifestManager(grpcSdk, moduleName, packageJsonPath);
+    return ManifestManager._instance;
+  }
+
+  private parsePackageJson(jsonPath: string, validationSchema: Schema): ConduitManifest {
+    const packageJson = fs.readJsonSync(jsonPath);
+    if (!packageJson.conduit) {
+      throw new Error("Missing 'conduit' section in package.json");
+    }
+    // Validate Json Input
+    const v = new Validator();
+    const jsonManifest: JsonManifest = {
+      version: packageJson.version,
+      dependencies: packageJson.conduit.dependencies ?? {},
+    };
+    const res = v.validate(jsonManifest, validationSchema);
+    if (res.errors.length > 0) {
+      throw new ManifestError(res.errors, jsonPath);
+    }
+    // Convert to Protobuff-friendly format
+    const manifest: ConduitManifest = {
+      moduleName: this.moduleName,
+      moduleVersion: jsonManifest.version,
+      dependencies: [],
+    };
+    (
+      Object.entries(jsonManifest.dependencies) as [string, ConduitManifestDependency][]
+    ).forEach(([depName, depObj]) => {
+      manifest.dependencies.push({
+        name: depName,
+        version: depObj.version,
+      });
+    });
+    return manifest;
+  }
+
+  readyCheck(
+    deploymentState: Map<string, ModuleStateInfo>,
+    moduleManifest: ConduitManifest,
+  ): { issues: string[] } {
+    const issues: string[] = [];
+    moduleManifest.dependencies.forEach(dep => {
+      const depState = [...deploymentState.values()].find(
+        m => m.moduleName === dep.name && !m.pending,
+      );
+      if (!depState) {
+        issues.push(
+          `Requested dependency on '${dep.name}' could not be fulfilled. Module not available.`,
+        );
+        return;
+      }
+      try {
+        this.validateTag(dep.name, dep.version, depState.moduleVersion);
+      } catch (err) {
+        issues.push((err as Error).message);
+      }
+    });
+    return { issues };
+  }
+
+  private parseTag(tag: string): {
+    tag: string;
+    preVersionOne: boolean;
+    majorVersion: number;
+    minorVersion: number;
+    operator: TagComparisonOperator;
+  } {
+    // ex input: '0.16', '0.16.0', '^1.12'...
+    const originalTag = tag;
+    let operator = TagComparisonOperator.Equal;
+    if (isNaN(parseInt(tag[0]))) {
+      operator =
+        tag[0] === '^' ? TagComparisonOperator.GreaterEqual : TagComparisonOperator.Equal;
+      tag = tag.slice(1);
+    }
+    const rcIndex = tag.indexOf('-');
+    if (rcIndex !== -1) {
+      tag = tag.slice(0, rcIndex);
+    }
+    const preVersionOne = parseInt(tag[0]) === 0;
+    const splitTag = tag.split('.');
+    const majorVersion = parseInt(
+      (preVersionOne ? splitTag.slice(1, 2) : splitTag.slice(0, 1))[0],
+    );
+    let minorVersion = parseInt(
+      (preVersionOne ? splitTag.slice(2, 3) : splitTag.slice(1, 2))[0],
+    );
+    if (isNaN(majorVersion)) {
+      throw new Error(`Invalid version tag '${originalTag}' format.`);
+    }
+    if (isNaN(minorVersion)) {
+      minorVersion = 0;
+    }
+    return {
+      tag,
+      preVersionOne,
+      majorVersion,
+      minorVersion,
+      operator,
+    };
+  }
+
+  private validateTag(depName: string, requestedTag: string, availableTag: string) {
+    // Minor versions don't break compatibility (v0.16.1 is a valid target for ^v0.16)
+    // RC information is stripped (v0.16.0-rc1 is a valid target for v0.16)
+    const requested = this.parseTag(requestedTag);
+    const available = this.parseTag(availableTag);
+    if (
+      requested.preVersionOne === available.preVersionOne &&
+      requested.operator === TagComparisonOperator.Equal &&
+      requested.tag === availableTag
+    )
+      return;
+    else if (
+      requested.preVersionOne === available.preVersionOne &&
+      requested.operator === TagComparisonOperator.GreaterEqual &&
+      requested.majorVersion === available.majorVersion &&
+      requested.minorVersion <= available.minorVersion
+    )
+      return;
+    throw new Error(
+      `Requested dependency on '${depName}@${requestedTag}' could not be fulfilled. ` +
+        `Target version mismatch (found: ${availableTag}).`,
+    );
+  }
+}