Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support connection pool #261

Merged
merged 10 commits into from
Dec 14, 2023
Merged

Support connection pool #261

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0ef4944
part1
shanghaikid Dec 8, 2023
8c0f29e
update poole
shanghaikid Dec 10, 2023
a5b0696
finish connection pool
shanghaikid Dec 11, 2023
791ee74
adjust
shanghaikid Dec 11, 2023
fcf36ff
fix test
shanghaikid Dec 11, 2023
7718ed6
add more test
shanghaikid Dec 12, 2023
ccf2155
adjust jest config
shanghaikid Dec 12, 2023
1b1ce31
more coverage
shanghaikid Dec 12, 2023
cea32b3
using generic-pool's options
shanghaikid Dec 12, 2023
40ca6d2
fix clientConfig
shanghaikid Dec 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions milvus/const/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@ export enum METADATA {

export enum CONNECT_STATUS {
NOT_CONNECTED,
CONNECTING,
CONNECTED,
CONNECTING = 0, // GRPC channel state connecting
CONNECTED = 1, // GRPC channel state ready
UNIMPLEMENTED,
SHUTDOWN = 5, // GRPC channel state shutdown
Comment on lines 7 to +12
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we set NOT_CONNECTED and UNIMPLEMENTED more clearly？

}

export enum TLS_MODE {
Expand Down
3 changes: 3 additions & 0 deletions milvus/const/defaults.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,3 +16,6 @@ export const DEFAULT_DYNAMIC_FIELD = '$meta';
export const DEFAULT_COUNT_QUERY_STRING = 'count(*)';
export const DEFAULT_HTTP_TIMEOUT = 60000; // 60s
export const DEFAULT_HTTP_ENDPOINT_VERSION = 'v1'; // api version, default v1

export const DEFAULT_POOL_MAX = 10; // default max pool client number
export const DEFAULT_POOL_MIN = 2; // default min pool client number
68 changes: 62 additions & 6 deletions milvus/grpc/BaseClient.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,14 @@
import path from 'path';
import crypto from 'crypto';
import protobuf, { Root, Type } from 'protobufjs';
import { Client, ChannelOptions } from '@grpc/grpc-js';
import { readFileSync } from 'fs';
import {
Client,
ChannelOptions,
credentials,
ChannelCredentials,
} from '@grpc/grpc-js';
import { Pool } from 'generic-pool';
import {
ERROR_REASONS,
ClientConfig,
Expand All @@ -26,11 +33,15 @@
* Base gRPC client, setup all configuration here
*/
export class BaseClient {
// channel pool
public channelPool!: Pool<Client>;
// ChannelCredentials object used for authenticating the client on the gRPC channel.
protected creds!: ChannelCredentials;
// Client ID
clientId: string = `${crypto.randomUUID()}`;
public clientId: string = `${crypto.randomUUID()}`;
// flags to indicate that if the connection is established and its state
connectStatus = CONNECT_STATUS.NOT_CONNECTED;
connectPromise = Promise.resolve();
public connectStatus = CONNECT_STATUS.NOT_CONNECTED;
public connectPromise = Promise.resolve();
// metadata
protected metadata: Map<string, string> = new Map<string, string>();
// The path to the Milvus protobuf file.
Expand Down Expand Up @@ -62,8 +73,8 @@
public readonly channelOptions: ChannelOptions;
// server info
public serverInfo: ServerInfo = {};
// The gRPC client instance.
public client: Client | undefined;
// // The gRPC client instance.
// public client!: Promise<Client>;
// The timeout for connecting to the Milvus service.
public timeout: number = DEFAULT_CONNECT_TIMEOUT;

Expand Down Expand Up @@ -163,6 +174,51 @@
this.config.tls.serverName;
}

// Switch based on the TLS mode
switch (this.tlsMode) {
case TLS_MODE.ONE_WAY:

Check warning on line 179 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L179 

Added line #L179 was not covered by tests
// Create SSL credentials with empty parameters for one-way authentication
this.creds = credentials.createSsl();
break;
case TLS_MODE.TWO_WAY:

Check warning on line 183 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L181-L183 

Added lines #L181 - L183 were not covered by tests
// Extract paths for root certificate, private key, certificate chain, and verify options from the client configuration
const { rootCertPath, privateKeyPath, certChainPath, verifyOptions } =
this.config.tls!;

Check warning on line 186 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L186 

Added line #L186 was not covered by tests

// Initialize buffers for root certificate, private key, and certificate chain
let rootCertBuff: Buffer | null = null;
let privateKeyBuff: Buffer | null = null;
let certChainBuff: Buffer | null = null;

Check warning on line 191 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L189-L191 

Added lines #L189 - L191 were not covered by tests

// Read root certificate file if path is provided
if (rootCertPath) {
rootCertBuff = readFileSync(rootCertPath);

Check warning on line 195 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L195 

Added line #L195 was not covered by tests
}

// Read private key file if path is provided
if (privateKeyPath) {
privateKeyBuff = readFileSync(privateKeyPath);

Check warning on line 200 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L200 

Added line #L200 was not covered by tests
}

// Read certificate chain file if path is provided
if (certChainPath) {
certChainBuff = readFileSync(certChainPath);

Check warning on line 205 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L205 

Added line #L205 was not covered by tests
}

// Create SSL credentials with the read files and verify options for two-way authentication
this.creds = credentials.createSsl(

Check warning on line 209 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L209 

Added line #L209 was not covered by tests
rootCertBuff,
privateKeyBuff,
certChainBuff,
verifyOptions
);
break;

Check warning on line 215 in milvus/grpc/BaseClient.ts

View check run for this annotation

Codecov / codecov/patch

milvus/grpc/BaseClient.ts#L215 

Added line #L215 was not covered by tests
default:
// Create insecure credentials if no TLS mode is specified
this.creds = credentials.createInsecure();
break;
}

// Set up the timeout for connecting to the Milvus service.
this.timeout =
typeof config.timeout === 'string'
Expand Down
40 changes: 20 additions & 20 deletions milvus/grpc/Collection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,7 +153,7 @@ export class Collection extends Database {

// Call the promisify function to create the collection.
const createPromise = await promisify(
this.client,
this.channelPool,
'CreateCollection',
{
...data,
Expand Down Expand Up @@ -201,7 +201,7 @@ export class Collection extends Database {

// avoid to call describe collection, because it has cache
const res = await promisify(
this.client,
this.channelPool,
'DescribeCollection',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -242,7 +242,7 @@ export class Collection extends Database {
data?: ShowCollectionsReq
): Promise<ShowCollectionsResponse> {
const promise = await promisify(
this.client,
this.channelPool,
'ShowCollections',
{
type: data ? data.type : ShowCollectionsType.All,
Expand Down Expand Up @@ -290,7 +290,7 @@ export class Collection extends Database {
async alterCollection(data: AlterCollectionReq): Promise<ResStatus> {
checkCollectionName(data);
const promise = await promisify(
this.client,
this.channelPool,
'AlterCollection',
{
collection_name: data.collection_name,
Expand Down Expand Up @@ -346,7 +346,7 @@ export class Collection extends Database {

// get new data
const promise = await promisify(
this.client,
this.channelPool,
'DescribeCollection',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -391,7 +391,7 @@ export class Collection extends Database {
checkCollectionName(data);

const promise = await promisify(
this.client,
this.channelPool,
'GetCollectionStatistics',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -432,7 +432,7 @@ export class Collection extends Database {
checkCollectionName(data);

const promise = await promisify(
this.client,
this.channelPool,
'LoadCollection',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -470,7 +470,7 @@ export class Collection extends Database {
checkCollectionName(data);

const promise = await promisify(
this.client,
this.channelPool,
'LoadCollection',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -529,7 +529,7 @@ export class Collection extends Database {
checkCollectionName(data);

const promise = await promisify(
this.client,
this.channelPool,
'ReleaseCollection',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -564,7 +564,7 @@ export class Collection extends Database {
*/
async renameCollection(data: RenameCollectionReq): Promise<ResStatus> {
const promise = await promisify(
this.client,
this.channelPool,
'RenameCollection',
{
oldName: data.collection_name,
Expand Down Expand Up @@ -602,7 +602,7 @@ export class Collection extends Database {
checkCollectionName(data);

const promise = await promisify(
this.client,
this.channelPool,
'DropCollection',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -649,7 +649,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.ALIAS_NAME_IS_REQUIRED);
}
const promise = await promisify(
this.client,
this.channelPool,
'CreateAlias',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -688,7 +688,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.ALIAS_NAME_IS_REQUIRED);
}
const promise = await promisify(
this.client,
this.channelPool,
'DropAlias',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -728,7 +728,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.ALIAS_NAME_IS_REQUIRED);
}
const promise = await promisify(
this.client,
this.channelPool,
'AlterAlias',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -763,7 +763,7 @@ export class Collection extends Database {
checkCollectionName(data);
const collectionInfo = await this.describeCollection(data);
const res = await promisify(
this.client,
this.channelPool,
'ManualCompaction',
{
collectionID: collectionInfo.collectionID,
Expand Down Expand Up @@ -803,7 +803,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.COMPACTION_ID_IS_REQUIRED);
}
const res = await promisify(
this.client,
this.channelPool,
'GetCompactionState',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -841,7 +841,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.COMPACTION_ID_IS_REQUIRED);
}
const res = await promisify(
this.client,
this.channelPool,
'GetCompactionStateWithPlans',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -894,7 +894,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.COLLECTION_ID_IS_REQUIRED);
}
const res = await promisify(
this.client,
this.channelPool,
'GetReplicas',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -935,7 +935,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.COLLECTION_NAME_IS_REQUIRED);
}
const res = await promisify(
this.client,
this.channelPool,
'GetLoadingProgress',
data,
data.timeout || this.timeout
Expand Down Expand Up @@ -973,7 +973,7 @@ export class Collection extends Database {
throw new Error(ERROR_REASONS.COLLECTION_NAME_IS_REQUIRED);
}
const res = await promisify(
this.client,
this.channelPool,
'GetLoadState',
data,
data.timeout || this.timeout
Expand Down
Loading