add insight cache

Signed-off-by: nameczz <zizhao.chen@zilliz.com>

client/src/components/layout/GlobalEffect.tsx

@@ -2,12 +2,15 @@ import React, { useContext } from 'react';
 import axiosInstance from '../../http/Axios';
 import { rootContext } from '../../context/Root';
 import { CODE_STATUS } from '../../consts/Http';
+import { MILVUS_ADDRESS } from '../../consts/Localstorage';
+import { authContext } from '../../context/Auth';
 
 let axiosResInterceptor: number | null = null;
 // let timer: Record<string, ReturnType<typeof setTimeout> | number>[] = [];
 // we only take side effect here, nothing else
 const GlobalEffect = (props: { children: React.ReactNode }) => {
   const { openSnackBar } = useContext(rootContext);
+  const { setAddress } = useContext(authContext);
 
   // catch axios error here
   if (axiosResInterceptor === null) {
@@ -22,7 +25,14 @@ const GlobalEffect = (props: { children: React.ReactNode }) => {
       },
       function (error: any) {
         const { response = {} } = error;
-
+        switch (response.status) {
+          case CODE_STATUS.UNAUTHORIZED:
+            setAddress('');
+            window.localStorage.removeItem(MILVUS_ADDRESS);
+            break;
+          default:
+            break;
+        }
         if (response.data) {
           const { message: errMsg } = response.data;
 

express/.prettierrc

@@ -1,7 +1,10 @@
 {
+  "printWidth": 80,
   "tabWidth": 2,
+  "useTabs": false,
   "semi": true,
   "singleQuote": true,
   "trailingComma": "es5",
-  "bracketSpacing": true
-}
+  "bracketSpacing": true,
+  "arrowParens": "avoid"
+}

express/package.json

@@ -14,6 +14,8 @@
     "express": "^4.17.1",
     "glob": "^7.2.0",
     "helmet": "^4.6.0",
+    "http-errors": "^1.8.1",
+    "lru-cache": "^6.0.0",
     "morgan": "^1.10.0",
     "node-cron": "^3.0.0",
     "rimraf": "^3.0.2",
@@ -42,7 +44,9 @@
     "@types/cors": "^2.8.12",
     "@types/express": "^4.17.13",
     "@types/glob": "^7.2.0",
+    "@types/http-errors": "^1.8.1",
     "@types/jest": "^27.0.2",
+    "@types/lru-cache": "^5.1.1",
     "@types/morgan": "^1.9.3",
     "@types/node": "^16.11.6",
     "@types/node-cron": "^3.0.0",

express/src/app.ts

@@ -21,11 +21,18 @@ import * as path from 'path';
 import chalk from 'chalk';
 import { surveSwaggerSpecification } from './swagger';
 import swaggerUi from 'swagger-ui-express';
+import LruCache from 'lru-cache';
+import { EXPIRED_TIME, INSIGHT_CACHE } from './utils/Const';
 
 const PLUGIN_DEV = process.env?.PLUGIN_DEV;
 const SRC_PLUGIN_DIR = 'src/plugins';
 const DEV_PLUGIN_DIR = '../../src/*/server';
 
+const insightCache = new LruCache({
+  maxAge: EXPIRED_TIME,
+  updateAgeOnGet: true,
+});
+
 export const app = express();
 const PORT = 3000;
 // initialize a simple http server
@@ -38,6 +45,7 @@ const io = new Server(server, {
   },
 });
 
+app.set(INSIGHT_CACHE, insightCache);
 // https://expressjs.com/en/resources/middleware/cors.html
 app.use(cors());
 // https://github.com/helmetjs/helmet
@@ -64,7 +72,7 @@ io.on('connection', (socket: Socket) => {
   socket.on('COLLECTION', (message: any) => {
     socket.emit('COLLECTION', { data: message });
   });
-  pubSub.on('ws_pubsub', (msg) => {
+  pubSub.on('ws_pubsub', msg => {
     const { event, data } = msg;
     socket.emit(event, data);
   });

express/src/middlewares/index.ts

@@ -2,18 +2,23 @@ import { Request, Response, NextFunction, Errback } from 'express';
 import morgan from 'morgan';
 import chalk from 'chalk';
 import { MilvusService } from '../milvus/milvus.service';
-
-const MILVUS_ADDRESS = 'milvus_address';
+import { INSIGHT_CACHE, MILVUS_ADDRESS } from '../utils/Const';
+import { HttpError } from 'http-errors';
 
 export const ReqHeaderMiddleware = (
   req: Request,
   res: Response,
   next: NextFunction
 ) => {
+  const insightCache = req.app.get(INSIGHT_CACHE);
   // all request need set milvus address in header.
   // server will set activeaddress in milvus service.
-  const milvusAddress = req.headers[MILVUS_ADDRESS] || '';
-  MilvusService.activeAddress = milvusAddress as string;
+  const milvusAddress = (req.headers[MILVUS_ADDRESS] as string) || '';
+  MilvusService.activeAddress = milvusAddress;
+  // insight cache will update expire time when use insightCache.get
+  MilvusService.activeMilvusClient = insightCache.get(
+    MilvusService.formatAddress(milvusAddress)
+  );
   next();
 };
 
@@ -23,7 +28,7 @@ export const TransformResMiddlerware = (
   next: NextFunction
 ) => {
   const oldSend = res.json;
-  res.json = (data) => {
+  res.json = data => {
     // console.log(data); // do something with the data
     const statusCode = data?.statusCode;
     const message = data?.message;
@@ -42,12 +47,17 @@ export const TransformResMiddlerware = (
  * Normally depend on status which from milvus service.
  */
 export const ErrorMiddleware = (
-  err: Error,
+  err: HttpError,
   req: Request,
   res: Response,
   next: NextFunction
 ) => {
-  console.log(chalk.blue.bold(req.method, req.url), chalk.red.bold(err));
+  const statusCode = err.statusCode || 500;
+  console.log(
+    chalk.blue.bold(req.method, req.url),
+    chalk.magenta.bold(statusCode),
+    chalk.red.bold(err)
+  );
   // Boolean property that indicates if the app sent HTTP headers for the response.
   // Here to prevent sending response after header has been sent.
   if (res.headersSent) {
@@ -55,8 +65,8 @@ export const ErrorMiddleware = (
   }
   if (err) {
     res
-      .status(500)
-      .json({ message: `${err}`, error: 'Bad Request', statusCode: 500 });
+      .status(statusCode)
+      .json({ message: `${err}`, error: 'Bad Request', statusCode });
   }
   next();
 };

express/src/milvus/milvus.controller.ts

@@ -1,7 +1,8 @@
-import { NextFunction, Request, Response, Router } from "express";
-import { dtoValidationMiddleware } from "../middlewares/validation";
-import { MilvusService } from "./milvus.service";
-import { ConnectMilvusDto, FlushDto } from "./dto";
+import { NextFunction, Request, Response, Router } from 'express';
+import { dtoValidationMiddleware } from '../middlewares/validation';
+import { MilvusService } from './milvus.service';
+import { ConnectMilvusDto, FlushDto } from './dto';
+import { INSIGHT_CACHE } from '../utils/Const';
 
 export class MilvusController {
   private router: Router;
@@ -18,28 +19,32 @@ export class MilvusController {
 
   generateRoutes() {
     this.router.post(
-      "/connect",
+      '/connect',
       dtoValidationMiddleware(ConnectMilvusDto),
       this.connectMilvus.bind(this)
     );
 
-    this.router.get("/check", this.checkConnect.bind(this));
+    this.router.get('/check', this.checkConnect.bind(this));
 
     this.router.put(
-      "/flush",
+      '/flush',
       dtoValidationMiddleware(FlushDto),
       this.flush.bind(this)
     );
 
-    this.router.get("/metrics", this.getMetrics.bind(this));
+    this.router.get('/metrics', this.getMetrics.bind(this));
 
     return this.router;
   }
 
   async connectMilvus(req: Request, res: Response, next: NextFunction) {
     const address = req.body?.address;
+    const insightCache = req.app.get(INSIGHT_CACHE);
     try {
-      const result = await this.milvusService.connectMilvus(address);
+      const result = await this.milvusService.connectMilvus(
+        address,
+        insightCache
+      );
 
       res.send(result);
     } catch (error) {
@@ -48,9 +53,14 @@ export class MilvusController {
   }
 
   async checkConnect(req: Request, res: Response, next: NextFunction) {
-    const address = "" + req.query?.address;
+    const address = '' + req.query?.address;
+    const insightCache = req.app.get(INSIGHT_CACHE);
+
     try {
-      const result = await this.milvusService.checkConnect(address);
+      const result = await this.milvusService.checkConnect(
+        address,
+        insightCache
+      );
       res.send(result);
     } catch (error) {
       next(error);

express/src/milvus/milvus.service.ts

@@ -3,87 +3,87 @@ import {
   FlushReq,
   GetMetricsResponse,
 } from '@zilliz/milvus2-sdk-node/dist/milvus/types';
-
+import HttpErrors from 'http-errors';
+import LruCache from 'lru-cache';
+import { HTTP_STATUS_CODE } from '../utils/Error';
 export class MilvusService {
   // Share with all instances, so activeAddress is static
   static activeAddress: string;
-  private milvusClients: { [x: string]: MilvusClient };
-
-  constructor() {
-    this.milvusClients = {};
-  }
-
-  get activeMilvusClient() {
-    // undefined means not connect yet, will throw error to client.
-    return this.milvusClients[MilvusService.activeAddress];
-  }
+  static activeMilvusClient: MilvusClient;
 
   get collectionManager() {
     this.checkMilvus();
-    return this.activeMilvusClient.collectionManager;
+    return MilvusService.activeMilvusClient.collectionManager;
   }
 
   get partitionManager() {
     this.checkMilvus();
-    return this.activeMilvusClient.partitionManager;
+    return MilvusService.activeMilvusClient.partitionManager;
   }
 
   get indexManager() {
     this.checkMilvus();
-    return this.activeMilvusClient.indexManager;
+    return MilvusService.activeMilvusClient.indexManager;
   }
 
   get dataManager() {
     this.checkMilvus();
-    return this.activeMilvusClient.dataManager;
+    return MilvusService.activeMilvusClient.dataManager;
   }
 
-  private checkMilvus() {
-    if (!this.activeMilvusClient) {
-      throw new Error('Please connect milvus first');
-    }
+  static formatAddress(address: string) {
+    return address.replace(/(http|https):\/\//, '');
   }
 
-  formatAddress(address: string) {
-    return address.replace(/(http|https):\/\//, '');
+  checkMilvus() {
+    if (!MilvusService.activeMilvusClient) {
+      throw HttpErrors(
+        HTTP_STATUS_CODE.UNAUTHORIZED,
+        'Please connect milvus first'
+      );
+      // throw new Error('Please connect milvus first');
+    }
   }
 
-  async connectMilvus(address: string) {
+  async connectMilvus(address: string, cache: LruCache<any, any>) {
     // grpc only need address without http
-    const milvusAddress = this.formatAddress(address);
+    const milvusAddress = MilvusService.formatAddress(address);
     try {
       const milvusClient = new MilvusClient(milvusAddress);
       await milvusClient.collectionManager.hasCollection({
         collection_name: 'not_exist',
       });
       MilvusService.activeAddress = address;
-      this.milvusClients[milvusAddress] = milvusClient;
+      cache.set(milvusAddress, milvusClient);
       return { address };
     } catch (error) {
       // if milvus is not working, delete connection.
-      delete this.milvusClients[milvusAddress];
-      throw new Error('Connect milvus failed, check your milvus address.');
+      cache.del(milvusAddress);
+      throw HttpErrors(
+        HTTP_STATUS_CODE.BAD_REQUEST,
+        'Connect milvus failed, check your milvus address.'
+      );
     }
   }
 
-  async checkConnect(address: string) {
-    const milvusAddress = this.formatAddress(address);
-    if (!Object.keys(this.milvusClients).includes(milvusAddress)) {
+  async checkConnect(address: string, cache: LruCache<any, any>) {
+    const milvusAddress = MilvusService.formatAddress(address);
+    if (!cache.has(milvusAddress)) {
       return { connected: false };
     }
-    const res = await this.connectMilvus(address);
+    const res = await this.connectMilvus(address, cache);
     return {
       connected: res.address ? true : false,
     };
   }
 
   async flush(data: FlushReq) {
-    const res = await this.activeMilvusClient.dataManager.flush(data);
+    const res = await MilvusService.activeMilvusClient.dataManager.flush(data);
     return res;
   }
 
   async getMetrics(): Promise<GetMetricsResponse> {
-    const res = await this.activeMilvusClient.dataManager.getMetric({
+    const res = await MilvusService.activeMilvusClient.dataManager.getMetric({
       request: { metric_type: 'system_info' },
     });
     return res;

express/src/utils/Const.ts

@@ -1,5 +1,12 @@
 export const ROW_COUNT = 'row_count';
 
+// use in req header
+export const MILVUS_ADDRESS = 'milvus_address';
+
+// for lru cache
+export const INSIGHT_CACHE = 'insight_cache';
+export const EXPIRED_TIME = 1000 * 10;
+
 export enum LOADING_STATE {
   LOADED,
   LOADING,

express/src/utils/Error.ts

@@ -8,3 +8,53 @@ export const throwErrorFromSDK = (res: ResStatus) => {
     throw res.reason;
   }
 };
+
+export enum HTTP_STATUS_CODE {
+  CONTINUE = 100,
+  SWITCHING_PROTOCOLS = 101,
+  PROCESSING = 102,
+  EARLYHINTS = 103,
+  OK = 200,
+  CREATED = 201,
+  ACCEPTED = 202,
+  NON_AUTHORITATIVE_INFORMATION = 203,
+  NO_CONTENT = 204,
+  RESET_CONTENT = 205,
+  PARTIAL_CONTENT = 206,
+  AMBIGUOUS = 300,
+  MOVED_PERMANENTLY = 301,
+  FOUND = 302,
+  SEE_OTHER = 303,
+  NOT_MODIFIED = 304,
+  TEMPORARY_REDIRECT = 307,
+  PERMANENT_REDIRECT = 308,
+  BAD_REQUEST = 400,
+  UNAUTHORIZED = 401,
+  PAYMENT_REQUIRED = 402,
+  FORBIDDEN = 403,
+  NOT_FOUND = 404,
+  METHOD_NOT_ALLOWED = 405,
+  NOT_ACCEPTABLE = 406,
+  PROXY_AUTHENTICATION_REQUIRED = 407,
+  REQUEST_TIMEOUT = 408,
+  CONFLICT = 409,
+  GONE = 410,
+  LENGTH_REQUIRED = 411,
+  PRECONDITION_FAILED = 412,
+  PAYLOAD_TOO_LARGE = 413,
+  URI_TOO_LONG = 414,
+  UNSUPPORTED_MEDIA_TYPE = 415,
+  REQUESTED_RANGE_NOT_SATISFIABLE = 416,
+  EXPECTATION_FAILED = 417,
+  I_AM_A_TEAPOT = 418,
+  MISDIRECTED = 421,
+  UNPROCESSABLE_ENTITY = 422,
+  FAILED_DEPENDENCY = 424,
+  TOO_MANY_REQUESTS = 429,
+  INTERNAL_SERVER_ERROR = 500,
+  NOT_IMPLEMENTED = 501,
+  BAD_GATEWAY = 502,
+  SERVICE_UNAVAILABLE = 503,
+  GATEWAY_TIMEOUT = 504,
+  HTTP_VERSION_NOT_SUPPORTED = 505,
+}

express/yarn.lock

@@ -877,6 +877,11 @@
   dependencies:
     "@types/node" "*"
 
+"@types/http-errors@^1.8.1":
+  version "1.8.1"
+  resolved "https://registry.yarnpkg.com/@types/http-errors/-/http-errors-1.8.1.tgz#e81ad28a60bee0328c6d2384e029aec626f1ae67"
+  integrity sha512-e+2rjEwK6KDaNOm5Aa9wNGgyS9oSZU/4pfSMMPYNOfjvFI0WVXm29+ITRFr6aKDvvKo7uU1jV68MW4ScsfDi7Q==
+
 "@types/istanbul-lib-coverage@*", "@types/istanbul-lib-coverage@^2.0.0", "@types/istanbul-lib-coverage@^2.0.1":
   version "2.0.3"
   resolved "https://registry.yarnpkg.com/@types/istanbul-lib-coverage/-/istanbul-lib-coverage-2.0.3.tgz#4ba8ddb720221f432e443bd5f9117fd22cfd4762"
@@ -914,6 +919,11 @@
   resolved "https://registry.yarnpkg.com/@types/long/-/long-4.0.1.tgz#459c65fa1867dafe6a8f322c4c51695663cc55e9"
   integrity sha512-5tXH6Bx/kNGd3MgffdmP4dy2Z+G4eaXw0SE81Tq3BNadtnMR5/ySMzX4SLEzHJzSmPNn4HIdpQsBvXMUykr58w==
 
+"@types/lru-cache@^5.1.1":
+  version "5.1.1"
+  resolved "https://registry.yarnpkg.com/@types/lru-cache/-/lru-cache-5.1.1.tgz#c48c2e27b65d2a153b19bfc1a317e30872e01eef"
+  integrity sha512-ssE3Vlrys7sdIzs5LOxCzTVMsU7i9oa/IaW92wF32JFb3CVczqOkru2xspuKczHEbG3nvmPY7IFqVmGGHdNbYw==
+
 "@types/mime@^1":
   version "1.3.2"
   resolved "https://registry.yarnpkg.com/@types/mime/-/mime-1.3.2.tgz#93e25bf9ee75fe0fd80b594bc4feb0e862111b5a"
@@ -2238,6 +2248,17 @@ http-errors@1.7.2:
     statuses ">= 1.5.0 < 2"
     toidentifier "1.0.0"
 
+http-errors@^1.8.1:
+  version "1.8.1"
+  resolved "https://registry.yarnpkg.com/http-errors/-/http-errors-1.8.1.tgz#7c3f28577cbc8a207388455dbd62295ed07bd68c"
+  integrity sha512-Kpk9Sm7NmI+RHhnj6OIWDI1d6fIoFAtFt9RLaTMRlg/8w49juAStsrBgp0Dp4OdxdVbRIeKhtCUvoi/RuAhO4g==
+  dependencies:
+    depd "~1.1.2"
+    inherits "2.0.4"
+    setprototypeof "1.2.0"
+    statuses ">= 1.5.0 < 2"
+    toidentifier "1.0.1"
+
 http-errors@~1.7.2:
   version "1.7.3"
   resolved "https://registry.yarnpkg.com/http-errors/-/http-errors-1.7.3.tgz#6c619e4f9c60308c38519498c14fbb10aacebb06"
@@ -3748,6 +3769,11 @@ setprototypeof@1.1.1:
   resolved "https://registry.yarnpkg.com/setprototypeof/-/setprototypeof-1.1.1.tgz#7e95acb24aa92f5885e0abef5ba131330d4ae683"
   integrity sha512-JvdAWfbXeIGaZ9cILp38HntZSFSo3mWg6xGcJJsd+d4aRMOqauag1C63dJfDw7OaMYwEbHMOxEZ1lqVRYP2OAw==
 
+setprototypeof@1.2.0:
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/setprototypeof/-/setprototypeof-1.2.0.tgz#66c9a24a73f9fc28cbe66b09fed3d33dcaf1b424"
+  integrity sha512-E5LDX7Wrp85Kil5bhZv46j8jOeboKq5JMmYM3gVGdGH8xFpPWXUMsNrlODCrkoxMEeNi/XZIwuRvY4XNwYMJpw==
+
 shebang-command@^2.0.0:
   version "2.0.0"
   resolved "https://registry.yarnpkg.com/shebang-command/-/shebang-command-2.0.0.tgz#ccd0af4f8835fbdc265b82461aaf0c36663f34ea"
@@ -4050,6 +4076,11 @@ toidentifier@1.0.0:
   resolved "https://registry.yarnpkg.com/toidentifier/-/toidentifier-1.0.0.tgz#7e1be3470f1e77948bc43d94a3c8f4d7752ba553"
   integrity sha512-yaOH/Pk/VEhBWWTlhI+qXxDFXlejDGcQipMlyxda9nthulaxLZUNcUqFxokp0vcYnvteJln5FNQDRrxj3YcbVw==
 
+toidentifier@1.0.1:
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/toidentifier/-/toidentifier-1.0.1.tgz#3be34321a88a820ed1bd80dfaa33e479fbb8dd35"
+  integrity sha512-o5sSPKEkg/DIQNmH43V0/uerLrpzVedkUh8tGNvaeXpfpuwjKenlSox/2O/BTlZUtEe+JG7s5YhEz608PlAHRA==
+
 touch@^3.1.0:
   version "3.1.0"
   resolved "https://registry.yarnpkg.com/touch/-/touch-3.1.0.tgz#fe365f5f75ec9ed4e56825e0bb76d24ab74af83b"