feat(extension-driver-ksqldb): add testing for ksqldb

Author: andreashimin

packages/extension-driver-ksqldb/README.md

@@ -1,11 +1,36 @@
 # extension-driver-ksqldb
 
-This library was generated with [Nx](https://nx.dev).
+This is the KSqlDb driver for VulcanSQL, provided by [Canner](https://cannerdata.com/).
 
-## Building
+## Installation
 
-Run `nx build extension-driver-ksqldb` to build the library.
+1. Install the package:
 
-## Running unit tests
+   ```bash
+   npm i @vulcan-sql/extension-driver-ksqldb
+   ```
 
-Run `nx test extension-driver-ksqldb` to execute the unit tests via [Jest](https://jestjs.io).
+2. Update your `vulcan.yaml` file to enable the extension:
+
+   ```yaml
+   extensions:
+     ksqldb: '@vulcan-sql/extension-driver-ksqldb'
+   ```
+
+3. Create a new profile in your `profiles.yaml` file or in the designated profile paths. For more information, please refer to the [KsqlDb documentation](https://ksqldb.io/) for the available arguments.
+
+   ```yaml
+   - name: ksql # Profile name
+     type: ksqldb
+     connection:
+       # Optional: KSqlDb instance URL. Default is http://localhost:8088.
+       host: 'www.example.com:8088'
+   ```
+
+## Testing
+
+To run tests for the `extension-driver-ksqldb` module, use the following command:
+
+```bash
+nx test extension-driver-ksqldb
+```

packages/extension-driver-ksqldb/jest.config.ts

@@ -1,6 +1,9 @@
 module.exports = {
   displayName: 'extension-driver-ksqldb',
   preset: '../../jest.preset.ts',
+  // Use node environment to avoid facing "TypeError: The "listener" argument must be of type function. Received an instance of Object" error
+  // when using ksqldb client executing query in the jest environment.
+  testEnvironment: 'node',
   globals: {
     'ts-jest': {
       tsconfig: '<rootDir>/tsconfig.spec.json',

packages/extension-driver-ksqldb/project.json

@@ -25,7 +25,13 @@
       "options": {
         "jestConfig": "packages/extension-driver-ksqldb/jest.config.ts",
         "passWithNoTests": true
-      }
+      },
+      "dependsOn": [
+        {
+          "projects": "self",
+          "target": "install-dependencies"
+        }
+      ]
     }
   },
   "tags": []

packages/extension-driver-ksqldb/src/lib/ksqldbDataSource.ts

@@ -39,7 +39,11 @@ export class KSQLDBDataSource extends DataSource<any, any> {
       this.clientMapping.set(profile.name, { client, options });
 
       // Testing connection
-      await client.checkConnection();
+      const isRunning = await client.checkConnectionRunning();
+      if (!isRunning) {
+        throw new Error('KsqlDb server is not running');
+      }
+
       this.logger.debug(`Profile ${profile.name} initialized`);
     }
   }

packages/extension-driver-ksqldb/src/lib/restfulClient.ts

@@ -6,7 +6,7 @@ const BASIC_HEADERS = {
 
 const RESTFUL_API = {
   INFO: '/info',
-  KQL: '/ksql',
+  KSQL: '/ksql',
   QUERY: '/query',
 };
 
@@ -96,32 +96,68 @@ export class RestfulClient {
     });
   }
 
-  public async checkConnection() {
-    const res = await this.request<KsqlInfoResponse>(RESTFUL_API.INFO, 'GET');
-    if (res.KsqlServerInfo['serverStatus'] !== 'RUNNING') {
-      throw new Error('KSQLDB server is not running');
+  public async checkConnection(): Promise<
+    KsqlInfoResponse['KsqlServerInfo']['serverStatus']
+  > {
+    try {
+      const res = await this.request<KsqlInfoResponse>(RESTFUL_API.INFO, 'GET');
+      return res.KsqlServerInfo['serverStatus'];
+    } catch (e) {
+      throw new Error('KsqlDb server is not ready');
+    }
+  }
+
+  public async checkConnectionRunning(): Promise<boolean> {
+    try {
+      const status = await this.checkConnection();
+      const isRunning = status === 'RUNNING';
+      return isRunning;
+    } catch (e) {
+      return false;
     }
-    return res;
   }
 
   public async query({
     query,
     query_params = {},
   }: {
     query: string;
-    query_params: Record<string, any>;
+    query_params?: Record<string, any>;
   }) {
+    // bind query parameters
+    const ksql = this.bindParams(query, query_params);
+
+    const buffer = Buffer.from(JSON.stringify({ ksql }));
+    const res = await this.request(RESTFUL_API.QUERY, 'POST', buffer);
+    return res;
+  }
+
+  public async exec({
+    query,
+    query_params = {},
+  }: {
+    query: string;
+    query_params?: Record<string, any>;
+  }) {
+    // bind query parameters
+    const ksql = this.bindParams(query, query_params);
+
+    const buffer = Buffer.from(JSON.stringify({ ksql }));
+    const res = await this.request(RESTFUL_API.KSQL, 'POST', buffer);
+    return res;
+  }
+
+  private bindParams(query: string, query_params: Record<string, any>) {
     const values = Object.values(query_params);
 
     // replace the parameterized placeholder to values
     const ksql = query.replace(/\$(\d+)/g, (_, index) => {
       const valueIndex = parseInt(index) - 1;
-      return values[valueIndex];
+      const paramValue = values[valueIndex];
+      return typeof paramValue === 'string' ? `'${paramValue}'` : paramValue;
     });
 
-    const buffer = Buffer.from(JSON.stringify({ ksql }));
-    const res = await this.request(RESTFUL_API.QUERY, 'POST', buffer);
-    return res;
+    return ksql;
   }
 
   private async request<R = QueryResponse[]>(
@@ -151,11 +187,17 @@ export class RestfulClient {
       });
 
       req.on('end', () => {
-        const jsonData = JSON.parse(data);
+        let responseData: any = data;
+        try {
+          responseData = JSON.parse(data);
+        } catch (e) {
+          responseData = data;
+        }
+
         if (status === 200) {
-          resolve(jsonData);
+          resolve(responseData);
         } else {
-          reject(jsonData);
+          reject(responseData);
         }
         this.close();
       });

packages/extension-driver-ksqldb/test/docker/compose.ts

@@ -0,0 +1,100 @@
+import * as Docker from 'dockerode';
+import * as path from 'path';
+import * as jsYaml from 'js-yaml';
+import { promises as fs } from 'fs';
+
+const docker = new Docker();
+
+interface ComposeConfig {
+  image: string;
+  hostname: string;
+  ports: string[];
+  environment: Record<string, any>;
+  expose: string[];
+  entrypoint: string | string[];
+  tty: boolean;
+  networks: string[];
+}
+
+interface ComposeJson {
+  services: Record<string, ComposeConfig>;
+  networks: Record<string, any>;
+}
+
+export class Compose {
+  public network?: Docker.Network;
+  public containers = new Map<string, Docker.Container>();
+  public ksqldbName = 'ksqldb-server';
+  public ksqldbContainer?: Docker.Container;
+
+  public async up() {
+    const composeJson = await this.getComposeJson();
+    await docker.pruneContainers();
+    await docker.pruneNetworks();
+    const network = Object.keys(composeJson.networks)[0];
+    this.network = await docker.createNetwork({
+      Name: network,
+    });
+
+    for (const serviceName in composeJson.services) {
+      const service = composeJson.services[serviceName];
+      const pullStream = await docker.pull(service.image);
+      // https://github.com/apocas/dockerode/issues/647
+      await new Promise((res) => docker.modem.followProgress(pullStream, res));
+
+      const containerConfig: Docker.ContainerCreateOptions =
+        this.convertContainerConfig(service, network);
+
+      this.containers.set(
+        serviceName,
+        await docker.createContainer(containerConfig)
+      );
+
+      await this.containers.get(serviceName)!.start();
+    }
+  }
+
+  public async down() {
+    for (const container of this.containers.values()) {
+      await container.stop();
+    }
+    await docker.pruneContainers();
+    await docker.pruneNetworks();
+  }
+
+  private async getComposeJson(): Promise<ComposeJson> {
+    const content = await fs.readFile(
+      path.resolve(__dirname, 'docker-compose.yml'),
+      'utf-8'
+    );
+    const composeJson = jsYaml.load(content) as ComposeJson;
+    return composeJson;
+  }
+
+  private convertContainerConfig(service: ComposeConfig, network: string) {
+    return {
+      Image: service['image'],
+      name: service['hostname'],
+      Hostname: service['hostname'],
+      HostConfig: {
+        PortBindings: (service['ports'] || []).reduce((acc, port) => {
+          const [hostPort, containerPort] = port.split(':');
+          return {
+            ...acc,
+            [`${containerPort}/tcp`]: [{ HostPort: hostPort }],
+          };
+        }, {}),
+        NetworkMode: network,
+      },
+      Env: Object.keys(service['environment'] || {}).map(
+        (key) => `${key}=${service['environment'][key]}`
+      ),
+      ExposedPorts: (service['expose'] || []).reduce(
+        (acc, port) => ({ ...acc, [`${port}/tcp`]: {} }),
+        {}
+      ),
+      Entrypoint: service['entrypoint'] || [],
+      Tty: service['tty'] || false,
+    };
+  }
+}

packages/extension-driver-ksqldb/test/docker/docker-compose.yml

@@ -0,0 +1,67 @@
+version: '2'
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.4.0
+    hostname: zookeeper
+    ports:
+      - "2181:2181"
+    expose:
+      - "2181"
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    networks:
+      - ksqldb_default
+
+  broker:
+    image: confluentinc/cp-kafka:7.4.0
+    hostname: broker
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+    expose:
+      - "29092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+    networks:
+      - ksqldb_default
+
+  ksqldb-server:
+    image: confluentinc/ksqldb-server:0.29.0
+    hostname: ksqldb-server
+    depends_on:
+      - broker
+    ports:
+      - "8088:8088"
+    expose:
+      - "8088"
+    environment:
+      KSQL_LISTENERS: http://0.0.0.0:8088
+      KSQL_BOOTSTRAP_SERVERS: broker:9092
+      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
+      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
+    networks:
+      - ksqldb_default
+
+  ksqldb-cli:
+    image: confluentinc/ksqldb-cli:0.29.0
+    hostname: ksqldb-cli
+    depends_on:
+      - broker
+      - ksqldb-server
+    entrypoint: /bin/sh
+    tty: true
+    networks:
+      - ksqldb_default
+
+networks:
+  ksqldb_default:
+    driver: bridge