feat(consul): register to consul server and lookup service with balan…

…cer (#949)

packages/consul/README.md

@@ -0,0 +1,12 @@
+# midway-consul
+
+[![Package Quality](http://npm.packagequality.com/shield/midway-core.svg)](http://packagequality.com/#?package=midway-core)
+[![PRs Welcome](https://img.shields.io/badge/PRs-welcome-brightgreen.svg)](https://github.com/midwayjs/midway/pulls)
+
+this is a sub package for midway.
+
+Document: [https://midwayjs.org/midway](https://midwayjs.org/midway)
+
+## License
+
+[MIT]((http://github.com/midwayjs/midway/blob/master/LICENSE))

packages/consul/jest.config.js

@@ -0,0 +1,6 @@
+module.exports = {
+  preset: 'ts-jest',
+  testPathIgnorePatterns: ['<rootDir>/test/fixtures'],
+  coveragePathIgnorePatterns: ['<rootDir>/test/'],
+  setupFilesAfterEnv: ['./jest.setup.js']
+};

packages/consul/jest.setup.js

@@ -0,0 +1 @@
+jest.setTimeout(30000);

packages/consul/package.json

@@ -0,0 +1,40 @@
+{
+  "name": "@midwayjs/consul",
+  "description": "midway consul",
+  "version": "1.0.0",
+  "main": "dist/index",
+  "typings": "dist/index.d.ts",
+  "files": [
+    "dist/**/*.js",
+    "dist/**/*.d.ts"
+  ],
+  "devDependencies": {
+    "@midwayjs/cli": "^1.2.36",
+    "@midwayjs/core": "^2.9.1",
+    "@midwayjs/decorator": "^2.9.1",
+    "@midwayjs/koa": "^2.9.1",
+    "@midwayjs/mock": "^2.9.1",
+    "@types/consul": "^0.23.34",
+    "@types/sinon": "^9.0.11",
+    "nock": "^13.0.11"
+  },
+  "dependencies": {
+    "consul": "^0.40.0"
+  },
+  "keywords": [
+    "consul"
+  ],
+  "author": "Wang Bo <1982wb@gmail.com>",
+  "license": "MIT",
+  "scripts": {
+    "build": "midway-bin build -c",
+    "test": "midway-bin test --ts",
+    "cov": "midway-bin cov --ts",
+    "ci": "npm run test",
+    "lint": "mwts check"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/midwayjs/midway.git"
+  }
+}

packages/consul/src/config/config.default.ts

@@ -0,0 +1,17 @@
+export default {
+  consul: {
+    provider: {
+      register: false,
+      host: '127.0.0.1',
+      port: 8500,
+      strategy: 'random',
+    },
+    service: {
+      id: null,
+      name: null,
+      tags: [],
+      address: null,
+      port: 7001,
+    },
+  },
+};

packages/consul/src/configuration.ts

@@ -0,0 +1,106 @@
+import {
+  Config,
+  Configuration,
+  MidwayFrameworkType,
+} from '@midwayjs/decorator';
+import { join } from 'path';
+import {
+  ILifeCycle,
+  IMidwayApplication,
+  IMidwayContainer,
+} from '@midwayjs/core';
+import {
+  IConsulProviderInfoOptions,
+  IConsulRegisterInfoOptions,
+} from './interface';
+import { ConsulProvider } from './lib/provider';
+
+@Configuration({
+  namespace: 'consul',
+  importConfigs: [join(__dirname, 'config')],
+})
+export class AutoConfiguration implements ILifeCycle {
+  /**
+   * 有关 consul server 的配置
+   */
+  @Config('consul.provider')
+  consulProviderConfig: IConsulProviderInfoOptions;
+
+  /**
+   * 有关 service registry 注册的信息
+   */
+  @Config('consul.service')
+  consulRegisterConfig: IConsulRegisterInfoOptions;
+
+  get consulProvider(): ConsulProvider {
+    const symbol = Symbol('consulProvider');
+    this[symbol] =
+      this[symbol] || new ConsulProvider(this.consulProviderConfig);
+    return this[symbol];
+  }
+
+  /**
+   * 注册自己的条件
+   * 由于环境的复杂性(多网卡、自动端口冲突) address 和 port 必须提供
+   */
+  get shouldBeRegisterMe(): boolean {
+    const { address, port } = this.consulRegisterConfig;
+    return this.consulProviderConfig.register && address.length > 0 && port > 0;
+  }
+
+  /**
+   * 注册 consul 服务
+   * @param container 容器 IoC
+   * @param app 应用 App
+   */
+  async registerConsul(
+    container: IMidwayContainer,
+    app: IMidwayApplication
+  ): Promise<void> {
+    const config = this.consulRegisterConfig;
+    const { address, port } = this.consulRegisterConfig;
+
+    if (this.shouldBeRegisterMe) {
+      config.name = config.name || app.getProjectName();
+      config.id = config.id || `${config.name}:${address}:${port}`;
+
+      config.check =
+        config.check ||
+        (app.getFrameworkType() === MidwayFrameworkType.WEB
+          ? {
+              http: `http://${address}:${port}/consul/health/self/check`,
+              interval: '3s',
+            }
+          : {
+              tcp: `${address}:${port}`,
+              interval: '3s',
+            });
+
+      Object.assign(this.consulRegisterConfig, config);
+
+      // 把原始的 consul 对象注入到容器
+      container.registerObject('consul', this.consulProvider.getConsul());
+      await this.consulProvider.registerService(this.consulRegisterConfig);
+    }
+  }
+
+  async onReady(
+    container: IMidwayContainer,
+    app?: IMidwayApplication
+  ): Promise<void> {
+    await this.registerConsul(container, app);
+  }
+
+  async onStop(
+    container: IMidwayContainer,
+    app?: IMidwayApplication
+  ): Promise<void> {
+    if (
+      this.consulProviderConfig.register &&
+      this.consulProviderConfig.deregister
+    ) {
+      const { id } = this.consulRegisterConfig;
+      await this.consulProvider.deregisterService({ id });
+    }
+  }
+}

packages/consul/src/controller/consul.ts

@@ -0,0 +1,12 @@
+import { Controller, Get, Provide } from '@midwayjs/decorator';
+
+@Provide()
+@Controller('/consul')
+export class ConsulController {
+  @Get('/health/self/check')
+  async healthCheck(): Promise<any> {
+    return {
+      status: 'success',
+    };
+  }
+}

packages/consul/src/index.ts

@@ -0,0 +1,4 @@
+export { AutoConfiguration as Configuration } from './configuration';
+export * from './controller/consul';
+export * from './service/balancer';
+export * from './interface';

packages/consul/src/interface.ts

@@ -0,0 +1,78 @@
+import * as Consul from "consul";
+import { ConsulOptions } from "consul";
+import RegisterOptions = Consul.Agent.Service.RegisterOptions;
+
+export interface IServiceBalancer {
+  /**
+   * 根据服务名称选择实例
+   * @param serviceName 注册的服务名称
+   * @param passingOnly 只返回通过健康检查的实例，默认为 true
+   */
+  select(serviceName: string, passingOnly?: boolean): any | never;
+}
+
+export interface IConsulBalancer {
+  /**
+   * 根绝策略返回负载均衡器
+   * @param strategy 负载均衡策略
+   */
+  getServiceBalancer(strategy?: string): IServiceBalancer;
+}
+
+export interface IConsulProviderInfoOptions extends ConsulOptions {
+  /**
+   * 本服务是否注册到 consul 服务器，默认是 NO 不会执行注册
+   */
+  register?: boolean;
+
+  /**
+   * 应用正常关闭的额时候自动反注册，默认是 YES 会执行反注册
+   * 如果 register=false 改参数无效
+   */
+  deregister?: boolean;
+
+  /**
+   * 调用服务负载均衡的策略(default、random)，默认是 random 随机
+   */
+  strategy?: string
+}
+
+export interface IConsulRegisterInfoOptions extends RegisterOptions {
+  /**
+   * 注册 id 标识，默认是 name:address:port 的组合
+   */
+  id?: string;
+
+  /**
+   * 服务名称
+   */
+  name: string;
+
+  /**
+   * 服务地址
+   */
+  address: string;
+
+  /**
+   * 服务端口
+   */
+  port: number;
+
+  /**
+   * 服务标签
+   */
+  tags?: string[];
+
+  /**
+   * 健康检查配置，组件默认会配置一个(检查间隔是3秒)
+   */
+  check?: {
+    tcp?: string;
+    http?: string;
+    script?: string;
+    interval?: string;
+    ttl?: string;
+    notes?: string;
+    status?: string;
+  }
+}

packages/consul/src/lib/balancer.ts

@@ -0,0 +1,84 @@
+import * as Consul from 'consul';
+import { IServiceBalancer, IConsulBalancer } from '../interface';
+
+abstract class Balancer implements IServiceBalancer {
+  protected constructor(protected consul: Consul.Consul) {
+    //
+  }
+
+  async select(serviceName: string, passingOnly = true): Promise<any | never> {
+    // throw new Error('not implemented');
+  }
+
+  async loadServices(serviceName: string, passingOnly = true): Promise<any[]> {
+    if (passingOnly) return await this.loadPassing(serviceName);
+    return await this.loadAll(serviceName);
+  }
+
+  private async loadAll(serviceName: string): Promise<any[]> {
+    return (await this.consul.catalog.service.nodes(serviceName)) as any[];
+  }
+
+  private async loadPassing(serviceName: string): Promise<any[]> {
+    const passingIds = [];
+    const checks = (await this.consul.health.checks(serviceName)) as any[];
+
+    // 健康检查通过的
+    checks.forEach(check => {
+      if (check.Status === 'passing') {
+        passingIds.push(check.ServiceID);
+      }
+    });
+
+    const instances = await this.loadAll(serviceName);
+    return instances.filter(item => passingIds.includes(item.ServiceID));
+  }
+}
+
+class RandomBalancer extends Balancer {
+  constructor(consul: Consul.Consul) {
+    super(consul);
+  }
+
+  async select(serviceName: string, passingOnly = true): Promise<any | never> {
+    let instances = await this.loadServices(serviceName, passingOnly);
+    if (instances.length > 0) {
+      instances = this.shuffle(instances);
+      return instances[Math.floor(Math.random() * instances.length)];
+    }
+
+    throw new Error('no available instance named ' + serviceName);
+  }
+
+  /**
+   * shuff by fisher-yates
+   */
+  shuffle(instances: Array<any>): Array<any> {
+    const result = [];
+
+    for (let i = 0; i < instances.length; i++) {
+      const j = Math.floor(Math.random() * (i + 1));
+      if (j !== i) {
+        result[i] = result[j];
+      }
+      result[j] = instances[i];
+    }
+
+    return result;
+  }
+}
+
+export class ConsulBalancer implements IConsulBalancer {
+  constructor(private consul: Consul.Consul) {
+    //
+  }
+
+  getServiceBalancer(strategy?: string): IServiceBalancer {
+    switch (strategy) {
+      case 'random':
+        return new RandomBalancer(this.consul);
+    }
+
+    throw new Error('invalid strategy balancer');
+  }
+}

packages/consul/src/lib/provider.ts

@@ -0,0 +1,29 @@
+import {
+  IConsulProviderInfoOptions,
+  IConsulRegisterInfoOptions,
+} from '../interface';
+import * as Consul from 'consul';
+
+export class ConsulProvider {
+  private readonly consul: Consul.Consul;
+
+  constructor(providerOptions: IConsulProviderInfoOptions) {
+    // should be, ignore config
+    providerOptions.promisify = true;
+    this.consul = Consul(providerOptions);
+  }
+
+  getConsul(): Consul.Consul {
+    return this.consul;
+  }
+
+  async registerService(
+    registerOptions: IConsulRegisterInfoOptions
+  ): Promise<void> {
+    await this.consul.agent.service.register(registerOptions);
+  }
+
+  async deregisterService(deregisterOptions: { id: string }): Promise<void> {
+    await this.consul.agent.service.deregister(deregisterOptions);
+  }
+}