bug: Sidecar would not be started when getClient was called due to previous abstraction (#324)

* Ensure sidecar starts correctly

Signed-off-by: Xavier Geerinck <[email protected]>

* Linting fixes

Signed-off-by: Xavier Geerinck <[email protected]>

* Resolve some comments

Signed-off-by: Xavier Geerinck <[email protected]>

* Give more time for timeout

Signed-off-by: Xavier Geerinck <[email protected]>

* Changes

Signed-off-by: Xavier Geerinck <[email protected]>

Author: XavierGeerinck

package.json

@@ -8,10 +8,14 @@
         "test:load": "jest --runInBand --detectOpenHandles",
         "test:load:http": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- npm run test:load 'test/load'",
         "test:e2e": "jest --runInBand --detectOpenHandles",
-        "test:e2e:all": "npm run test:e2e:grpc && npm run test:e2e:http && npm run test:e2e:http:actors",
-        "test:e2e:grpc": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol grpc --app-port 50001 --dapr-grpc-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/grpc/(client).test.ts' ]",
-        "test:e2e:http": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/http/(client|server).test.ts' ]",
-        "test:e2e:http:actors": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/http/actors.test.ts' ]",
+        "test:e2e:all": "npm run test:e2e:http; npm run test:e2e:grpc",
+        "test:e2e:grpc": "npm run test:e2e:grpc:client && npm run test:e2e:grpc:server",
+        "test:e2e:grpc:client": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol grpc --app-port 50001 --dapr-grpc-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/grpc/(client).test.ts' ]",
+        "test:e2e:grpc:server": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol grpc --app-port 50001 --dapr-grpc-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/grpc/(server).test.ts' ]",
+        "test:e2e:http": "npm run test:e2e:http:client && npm run test:e2e:http:server && npm run test:e2e:http:actors",
+        "test:e2e:http:client": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/http/(client).test.ts' ]",
+        "test:e2e:http:server": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/http/(server).test.ts' ]",
+        "test:e2e:http:actors": "npm run prebuild && TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components -- jest --runInBand --detectOpenHandles --testMatch [ '**/test/e2e/http/actors.test.ts' ]",
         "test:unit": "jest --runInBand --detectOpenHandles",
         "test:unit:all": "npm run test:unit:main && npm run test:unit:actors && npm run test:unit:logger && npm run test:unit:utils",
         "test:unit:main": "NODE_ENV=test npm run test:unit 'test/unit/main/.*\\.test\\.ts'",
@@ -60,4 +64,4 @@
         "url": "https://github.com/dapr/js-sdk.git",
         "directory": ""
     }
-}
+}

src/implementation/Client/DaprClient.ts

@@ -11,8 +11,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import * as NodeJSUtils from "../../utils/NodeJS.util";
-
 import IClientBinding from '../../interfaces/Client/IClientBinding';
 import IClientPubSub from '../../interfaces/Client/IClientPubSub';
 import IClientState from '../../interfaces/Client/IClientState';
@@ -56,6 +54,7 @@ import { DaprClientOptions } from '../../types/DaprClientOptions';
 import { Settings } from '../../utils/Settings.util';
 import { Logger } from '../../logger/Logger';
 import GRPCClientProxy from "./GRPCClient/proxy";
+import * as NodeJSUtils from "../../utils/NodeJS.util";
 import { SDK_PACKAGE_NAME } from "../../version";
 
 export default class DaprClient {
@@ -144,47 +143,44 @@ export default class DaprClient {
     return new DaprClient(client.getClientHost(), client.getClientPort(), client.getClientCommunicationProtocol(), client.getOptions());
   }
 
-  async stop(): Promise<void> {
-    await this.daprClient.stop();
-  }
+  static async awaitSidecarStarted(fnIsSidecarStarted: () => Promise<boolean>): Promise<void> {
+    const logger = new Logger("DaprClient", "DaprClient");
 
-  async awaitSidecarStarted(): Promise<void> {
     // Dapr will probe every 50ms to see if we are listening on our port: https://github.com/dapr/dapr/blob/a43712c97ead550ca2f733e9f7e7769ecb195d8b/pkg/runtime/runtime.go#L1694
     // if we are using actors we will change this to 4s to let the placement tables update
-    let isHealthy = false;
-    let isHealthyRetryCount = 0;
-    const isHealthyMaxRetryCount = 60; // 1s startup delay and we try max for 60s
+    let isStarted = await fnIsSidecarStarted();
+    let isStartedRetryCount = 0;
+    const isStartedMaxRetryCount = 60; // 1s startup delay and we try max for 60s
 
-    this.logger.info(`Awaiting Sidecar to be Started`);
-    while (!isHealthy) {
-      this.logger.verbose(`Waiting for the Dapr Sidecar to start, retry count: (#${isHealthyRetryCount})`);
+    if (isStarted) {
+      return;
+    }
+
+    logger.info(`Awaiting Sidecar to be Started`);
+    while (!isStarted) {
+      logger.verbose(`Waiting for the Dapr Sidecar to start, retry count: (#${isStartedRetryCount})`);
       await NodeJSUtils.sleep(Settings.getDaprSidecarPollingDelayMs());
 
       // Implement API call manually as we need to enable calling without initialization
       // everything routes through the `execute` method
       // to check health, we just ping the /metadata endpoint and see if we get a response
-      isHealthy = await this.health.isHealthy();
+      isStarted = await fnIsSidecarStarted();
 
       // Finally, Handle the retry logic
-      isHealthyRetryCount++;
+      isStartedRetryCount++;
 
-      if (isHealthyRetryCount > isHealthyMaxRetryCount) {
+      if (isStartedRetryCount > isStartedMaxRetryCount) {
         throw new Error("DAPR_SIDECAR_COULD_NOT_BE_STARTED");
       }
     }
   }
 
-  /**
-   * Ensure the client is started, this takes care of:
-   * 1. Making sure the sidecar is started
-   * 2. Making sure the connection is established (e.g. in gRPC)
-   * 3. Making sure the client is ready to be used
-   */
+  async stop(): Promise<void> {
+    await this.daprClient.stop();
+  }
+
   async start(): Promise<void> {
-    await this.awaitSidecarStarted();
     await this.daprClient.start();
-    await this.daprClient.setIsInitialized(true);
-    this.logger.info("Sidecar Started");
   }
 
   getDaprClient(): IClient {

src/implementation/Client/GRPCClient/GRPCClient.ts

@@ -12,17 +12,19 @@ limitations under the License.
 */
 
 import * as grpc from "@grpc/grpc-js";
-import { DaprClient } from "../../../proto/dapr/proto/runtime/v1/dapr_grpc_pb"
+import { DaprClient as GrpcDaprClient } from "../../../proto/dapr/proto/runtime/v1/dapr_grpc_pb"
 import IClient from "../../../interfaces/Client/IClient";
 import CommunicationProtocolEnum from "../../../enum/CommunicationProtocol.enum";
 import { DaprClientOptions } from "../../../types/DaprClientOptions";
 import { Settings } from '../../../utils/Settings.util';
 import { Logger } from "../../../logger/Logger";
+import GRPCClientSidecar from "./sidecar";
+import DaprClient from "../DaprClient";
 
 export default class GRPCClient implements IClient {
   private isInitialized: boolean;
 
-  private readonly client: DaprClient;
+  private readonly client: GrpcDaprClient;
   private readonly clientCredentials: grpc.ChannelCredentials;
   private readonly clientHost: string;
   private readonly clientPort: string;
@@ -53,7 +55,7 @@ export default class GRPCClient implements IClient {
     return this.clientPort;
   }
 
-  async getClient(requiresInitialization = true): Promise<DaprClient> {
+  async getClient(requiresInitialization = true): Promise<GrpcDaprClient> {
     // Ensure the sidecar has been started
     if (!this.isInitialized && requiresInitialization) {
       await this.start();
@@ -70,8 +72,8 @@ export default class GRPCClient implements IClient {
     return this.clientCredentials;
   }
 
-  private generateClient(host: string, port: string, credentials: grpc.ChannelCredentials): DaprClient {
-    const client = new DaprClient(`${host}:${port}`, credentials);
+  private generateClient(host: string, port: string, credentials: grpc.ChannelCredentials): GrpcDaprClient {
+    const client = new GrpcDaprClient(`${host}:${port}`, credentials);
     return client;
   }
 
@@ -112,7 +114,18 @@ export default class GRPCClient implements IClient {
     })
   }
 
+  async _startAwaitSidecarStarted(): Promise<void> {
+    await DaprClient.awaitSidecarStarted(async () => await GRPCClientSidecar.isStarted(this));
+  }
+
+  /**
+   * Ensure the client is started, this takes care of:
+   * 1. Making sure the sidecar is started
+   * 2. Making sure the connection is established (e.g. in gRPC)
+   * 3. Making sure the client is ready to be used
+   */
   async start(): Promise<void> {
+    await this._startAwaitSidecarStarted();
     await this._startWaitForClientReady();
     this.isInitialized = true;
   }

src/implementation/Client/GRPCClient/sidecar.ts

@@ -13,6 +13,7 @@ limitations under the License.
 
 import GRPCClient from './GRPCClient';
 import IClientSidecar from "../../../interfaces/Client/IClientSidecar";
+import { GetMetadataResponse } from '../../../proto/dapr/proto/runtime/v1/dapr_pb';
 import { Empty } from "google-protobuf/google/protobuf/empty_pb";
 
 // https://docs.dapr.io/reference/api/secrets_api/
@@ -36,4 +37,22 @@ export default class GRPCClientSidecar implements IClientSidecar {
       });
     });
   }
+
+  static async isStarted(client: GRPCClient): Promise<boolean> {
+    const callClient = await client.getClient(false);
+
+    return new Promise((resolve, _reject) => {
+      try {
+        callClient.getMetadata(new Empty(), (err, _res: GetMetadataResponse) => {
+          if (err) {
+            return resolve(false);
+          }
+
+          return resolve(true);
+        });
+      } catch (_e) {
+        return resolve(false);
+      }
+    });
+  }
 }

src/implementation/Client/HTTPClient/HTTPClient.ts

@@ -12,14 +12,15 @@ limitations under the License.
 */
 
 import fetch from "node-fetch";
-import { CommunicationProtocolEnum } from "../../..";
+import { CommunicationProtocolEnum, DaprClient } from "../../..";
 import IClient from "../../../interfaces/Client/IClient";
 import http from "http";
 import https from "https";
 import { DaprClientOptions } from "../../../types/DaprClientOptions";
 import { Settings } from '../../../utils/Settings.util';
 import { THTTPExecuteParams } from "../../../types/http/THTTPExecuteParams.type"
 import { Logger } from "../../../logger/Logger";
+import HTTPClientSidecar from "./sidecar";
 
 export default class HTTPClient implements IClient {
   private isInitialized: boolean;
@@ -72,7 +73,8 @@ export default class HTTPClient implements IClient {
 
   async getClient(requiresInitialization = true): Promise<typeof fetch> {
     // Ensure the sidecar has been started
-    if (!this.isInitialized && requiresInitialization) {
+    if (requiresInitialization && !this.isInitialized) {
+      this.logger.verbose("Client is not initialized, starting sidecar and initializing");
       await this.start();
     }
 
@@ -107,12 +109,21 @@ export default class HTTPClient implements IClient {
     return this.isInitialized;
   }
 
+  async _startAwaitSidecarStarted(): Promise<void> {
+    await DaprClient.awaitSidecarStarted(async () => await HTTPClientSidecar.isStarted(this));
+  }
+
   async stop(): Promise<void> {
     this.httpAgent.destroy();
     this.httpsAgent.destroy();
   }
 
   async start(): Promise<void> {
+    await this._startAwaitSidecarStarted();
+    this.isInitialized = true;
+
+    this.logger.info("Sidecar Started");
+
     return;
   }
 

src/implementation/Client/HTTPClient/sidecar.ts

@@ -27,4 +27,13 @@ export default class HTTPClientSidecar implements IClientSidecar {
       method: 'POST'
     });
   }
+
+  static async isStarted(client: HTTPClient): Promise<boolean> {
+    try {
+      const result = await client.execute(`/metadata`, null, false);
+      return !!result;
+    } catch (_e) {
+      return false;
+    }
+  }
 }

src/implementation/Server/DaprServer.ts

@@ -113,9 +113,6 @@ export default class DaprServer {
 
     // Ensure our sidecar starts and the client is ready
     await this.client.start();
-
-    // We are initialized
-    this.logger.info("Sidecar Started");
   }
 
   async stop(): Promise<void> {

test/e2e/grpc/server.test.ts

@@ -39,6 +39,8 @@ describe('grpc/server', () => {
 
     // Start server
     await server.start();
+
+    await new Promise((resolve, _reject) => setTimeout(resolve, 2500));
   }, 10 * 1000);
 
   beforeEach(() => {