Merge pull request #3517 from RedisInsight/be/feature/RI-5750_update_rdi

Be/feature/ri 5750 update rdi

Author: AmirAllayarovSofteq

redisinsight/api/src/modules/rdi/client/api.rdi.client.ts

@@ -1,7 +1,6 @@
 import axios, { AxiosInstance } from 'axios';
 import { plainToClass } from 'class-transformer';
 import { decode } from 'jsonwebtoken';
-import { Request } from 'express';
 
 import { RdiClient } from 'src/modules/rdi/client/rdi.client';
 import {
@@ -14,10 +13,10 @@ import {
 import {
   RdiDryRunJobDto,
   RdiDryRunJobResponseDto,
+  RdiTemplateResponseDto,
   RdiTestConnectionsResponseDto,
 } from 'src/modules/rdi/dto';
 import {
-  RdiPipelineDeployFailedException,
   RdiPipelineInternalServerErrorException,
   wrapRdiPipelineError,
 } from 'src/modules/rdi/exceptions';
@@ -29,8 +28,7 @@ import {
 } from 'src/modules/rdi/models';
 import { convertKeysToCamelCase } from 'src/utils/base.helper';
 import { RdiPipelineTimeoutException } from 'src/modules/rdi/exceptions/rdi-pipeline.timeout-error.exception';
-
-const RDI_DEPLOY_FAILED_STATUS = 'failed';
+import * as https from 'https';
 
 export class ApiRdiClient extends RdiClient {
   protected readonly client: AxiosInstance;
@@ -42,6 +40,9 @@ export class ApiRdiClient extends RdiClient {
     this.client = axios.create({
       baseURL: rdi.url,
       timeout: RDI_TIMEOUT,
+      httpsAgent: new https.Agent({
+        rejectUnauthorized: false,
+      }),
     });
   }
 
@@ -72,25 +73,32 @@ export class ApiRdiClient extends RdiClient {
     }
   }
 
-  async getTemplate(options: object): Promise<object> {
+  async getConfigTemplate(pipelineType: string, dbType: string): Promise<RdiTemplateResponseDto> {
     try {
-      const response = await this.client.get(RdiUrl.GetTemplate, { params: options });
+      const response = await this.client.get(`${RdiUrl.GetConfigTemplate}/${pipelineType}/${dbType}`);
       return response.data;
     } catch (error) {
       throw wrapRdiPipelineError(error);
     }
   }
 
-  async deploy(pipeline: RdiPipeline): Promise<void> {
-    let response;
+  async getJobTemplate(pipelineType: string): Promise<RdiTemplateResponseDto> {
     try {
-      response = await this.client.post(RdiUrl.Deploy, { ...pipeline });
+      const response = await this.client.get(`${RdiUrl.GetJobTemplate}/${pipelineType}`);
+      return response.data;
     } catch (error) {
-      throw wrapRdiPipelineError(error, error.response.data.message);
+      throw wrapRdiPipelineError(error);
     }
+  }
 
-    if (response.data?.status === RDI_DEPLOY_FAILED_STATUS) {
-      throw new RdiPipelineDeployFailedException(undefined, { error: response.data?.error });
+  async deploy(pipeline: RdiPipeline): Promise<void> {
+    try {
+      const response = await this.client.post(RdiUrl.Deploy, { ...pipeline });
+      const actionId = response.data.action_id;
+
+      return this.pollActionStatus(actionId);
+    } catch (error) {
+      throw wrapRdiPipelineError(error, error.response.data.message);
     }
   }
 
@@ -103,21 +111,14 @@ export class ApiRdiClient extends RdiClient {
     }
   }
 
-  async testConnections(config: string, req: Request): Promise<RdiTestConnectionsResponseDto> {
+  async testConnections(config: object): Promise<RdiTestConnectionsResponseDto> {
     try {
-      const abortController = new AbortController();
-      req.socket.on('close', () => {
-        abortController.abort();
-      });
       const response = await this.client.post(
         RdiUrl.TestConnections,
         config,
-        { signal: abortController.signal },
       );
 
-      const actionId = response.data.action_id;
-
-      return this.pollActionStatus(actionId, abortController.signal);
+      return response.data;
     } catch (e) {
       throw wrapRdiPipelineError(e);
     }
@@ -147,7 +148,7 @@ export class ApiRdiClient extends RdiClient {
 
   async getJobFunctions(): Promise<object> {
     try {
-      const response = await this.client.post(RdiUrl.JobFunctions);
+      const response = await this.client.get(RdiUrl.JobFunctions);
       return response.data;
     } catch (e) {
       throw wrapRdiPipelineError(e);
@@ -178,10 +179,10 @@ export class ApiRdiClient extends RdiClient {
     }
   }
 
-  private async pollActionStatus(actionId: string, abortSignal: AbortSignal): Promise<any> {
+  private async pollActionStatus(actionId: string, abortSignal?: AbortSignal): Promise<any> {
     const startTime = Date.now();
     while (true) {
-      if (abortSignal.aborted) {
+      if (abortSignal?.aborted) {
         throw new RdiPipelineInternalServerErrorException();
       }
       if (Date.now() - startTime > MAX_POLLING_TIME) {

redisinsight/api/src/modules/rdi/client/rdi.client.ts

@@ -1,10 +1,9 @@
-import { Request } from 'express';
 import {
   Rdi,
   RdiClientMetadata, RdiPipeline, RdiStatisticsResult,
 } from 'src/modules/rdi/models';
 import {
-  RdiDryRunJobDto, RdiDryRunJobResponseDto, RdiTestConnectionsResponseDto,
+  RdiDryRunJobDto, RdiDryRunJobResponseDto, RdiTemplateResponseDto, RdiTestConnectionsResponseDto,
 } from 'src/modules/rdi/dto';
 import { IDLE_TRESHOLD } from 'src/modules/rdi/constants';
 
@@ -28,16 +27,17 @@ export abstract class RdiClient {
 
   abstract getPipeline(): Promise<RdiPipeline>;
 
-  // TODO validate options and response
-  abstract getTemplate(options: object): Promise<unknown>;
+  abstract getConfigTemplate(pipelineType: string, dbType: string): Promise<RdiTemplateResponseDto>;
+
+  abstract getJobTemplate(pipelineType: string): Promise<RdiTemplateResponseDto>;
 
   abstract getStrategies(): Promise<object>;
 
   abstract deploy(pipeline: RdiPipeline): Promise<void>;
 
   abstract dryRunJob(data: RdiDryRunJobDto): Promise<RdiDryRunJobResponseDto>;
 
-  abstract testConnections(config: string, req: Request): Promise<RdiTestConnectionsResponseDto>;
+  abstract testConnections(config: object): Promise<RdiTestConnectionsResponseDto>;
 
   abstract getStatistics(sections?: string): Promise<RdiStatisticsResult>;
 

redisinsight/api/src/modules/rdi/constants/index.ts

@@ -2,7 +2,8 @@ export enum RdiUrl {
   GetSchema = 'api/v1/schemas',
   GetPipeline = 'api/v1/pipelines',
   GetStrategies = 'api/v1/pipelines/strategies',
-  GetTemplate = 'deployments/templates',
+  GetConfigTemplate = 'api/v1/pipelines/config/templates',
+  GetJobTemplate = 'api/v1/pipelines/jobs/templates',
   DryRunJob = 'api/v1/pipelines/jobs/dry-run',
   JobFunctions = '/api/v1/pipelines/jobs/functions',
   Deploy = 'api/v1/pipelines',

redisinsight/api/src/modules/rdi/dto/index.ts

@@ -3,3 +3,4 @@ export * from './update.rdi.dto';
 export * from './rdi.dry-run.job.dto';
 export * from './rdi.dry-run.job.response.dto';
 export * from './rdi-test-connections.response.dto';
+export * from './rdi-template.response.dto';

redisinsight/api/src/modules/rdi/dto/rdi-template.response.dto.ts

@@ -0,0 +1,11 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { Expose } from 'class-transformer';
+
+export class RdiTemplateResponseDto {
+  @ApiProperty({
+    description: 'Template for rdi file',
+    type: String,
+  })
+  @Expose()
+  template: string;
+}

redisinsight/api/src/modules/rdi/rdi-pipeline.controller.ts

@@ -1,15 +1,15 @@
 import {
   Body,
   ClassSerializerInterceptor, Controller, Get, Post, UseInterceptors, UsePipes, ValidationPipe,
-  Query, Req,
+  Req, Param,
 } from '@nestjs/common';
 import { Rdi, RdiPipeline, RdiClientMetadata } from 'src/modules/rdi/models';
 import { ApiTags } from '@nestjs/swagger';
 import { Request } from 'express';
 import { ApiEndpoint } from 'src/decorators/api-endpoint.decorator';
 import { RdiPipelineService } from 'src/modules/rdi/rdi-pipeline.service';
 import { RequestRdiClientMetadata } from 'src/modules/rdi/decorators';
-import { RdiDryRunJobDto, RdiTestConnectionsResponseDto } from 'src/modules/rdi/dto';
+import { RdiDryRunJobDto, RdiTemplateResponseDto, RdiTestConnectionsResponseDto } from 'src/modules/rdi/dto';
 import { RdiDryRunJobResponseDto } from 'src/modules/rdi/dto/rdi.dry-run.job.response.dto';
 
 @ApiTags('RDI')
@@ -73,11 +73,10 @@ export class RdiPipelineController {
     responses: [{ status: 200, type: RdiTestConnectionsResponseDto }],
   })
   async testConnections(
-    @Req() req: Request,
-      @RequestRdiClientMetadata() rdiClientMetadata: RdiClientMetadata,
-      @Body() config: string,
+    @RequestRdiClientMetadata() rdiClientMetadata: RdiClientMetadata,
+      @Body() config: object,
   ): Promise<RdiTestConnectionsResponseDto> {
-    return this.rdiPipelineService.testConnections(rdiClientMetadata, config, req);
+    return this.rdiPipelineService.testConnections(rdiClientMetadata, config);
   }
 
   @Get('/strategies')
@@ -91,16 +90,29 @@ export class RdiPipelineController {
     return this.rdiPipelineService.getStrategies(rdiClientMetadata);
   }
 
-  @Get('/template')
+  @Get('/job/template/:pipelineType')
   @ApiEndpoint({
-    description: 'Get pipeline template for selected pipeline type',
-    responses: [{ status: 200, type: Rdi }],
+    description: 'Get job template for selected pipeline type',
+    responses: [{ status: 200, type: RdiTemplateResponseDto }],
   })
-  async getTemplate(
+  async getJobTemplate(
     @RequestRdiClientMetadata() rdiClientMetadata: RdiClientMetadata,
-      @Query() options: object,
-  ): Promise<unknown> {
-    return this.rdiPipelineService.getTemplate(rdiClientMetadata, options);
+      @Param('pipelineType') pipelineType: string,
+  ): Promise<RdiTemplateResponseDto> {
+    return this.rdiPipelineService.getJobTemplate(rdiClientMetadata, pipelineType);
+  }
+
+  @Get('/config/template/:pipelineType/:dbType')
+  @ApiEndpoint({
+    description: 'Get config template for selected pipeline and db types',
+    responses: [{ status: 200, type: RdiTemplateResponseDto }],
+  })
+  async getConfigTemplate(
+    @RequestRdiClientMetadata() rdiClientMetadata: RdiClientMetadata,
+      @Param('pipelineType') pipelineType: string,
+      @Param('dbType') dbType: string,
+  ): Promise<RdiTemplateResponseDto> {
+    return this.rdiPipelineService.getConfigTemplate(rdiClientMetadata, pipelineType, dbType);
   }
 
   @Get('/status')

redisinsight/api/src/modules/rdi/rdi-pipeline.service.ts

@@ -2,7 +2,7 @@ import { Injectable, Logger } from '@nestjs/common';
 import { Request } from 'express';
 import { RdiClientMetadata, RdiPipeline } from 'src/modules/rdi/models';
 import { RdiClientProvider } from 'src/modules/rdi/providers/rdi.client.provider';
-import { RdiDryRunJobDto, RdiTestConnectionsResponseDto } from 'src/modules/rdi/dto';
+import { RdiDryRunJobDto, RdiTemplateResponseDto, RdiTestConnectionsResponseDto } from 'src/modules/rdi/dto';
 import { RdiDryRunJobResponseDto } from 'src/modules/rdi/dto/rdi.dry-run.job.response.dto';
 import { RdiPipelineAnalytics } from 'src/modules/rdi/rdi-pipeline.analytics';
 import { wrapHttpError } from 'src/common/utils';
@@ -70,12 +70,12 @@ export class RdiPipelineService {
     }
   }
 
-  async testConnections(rdiClientMetadata: RdiClientMetadata, config: string, req: Request): Promise<RdiTestConnectionsResponseDto> {
+  async testConnections(rdiClientMetadata: RdiClientMetadata, config: object): Promise<RdiTestConnectionsResponseDto> {
     this.logger.log('Trying to test connections');
 
     const client = await this.rdiClientProvider.getOrCreate(rdiClientMetadata);
 
-    return await client.testConnections(config, req);
+    return await client.testConnections(config);
   }
 
   async getStrategies(rdiClientMetadata: RdiClientMetadata): Promise<object> {
@@ -85,12 +85,20 @@ export class RdiPipelineService {
     return await client.getStrategies();
   }
 
-  async getTemplate(rdiClientMetadata: RdiClientMetadata, options: object): Promise<unknown> {
-    this.logger.log('Getting RDI pipeline template');
+  async getConfigTemplate(rdiClientMetadata: RdiClientMetadata, pipelineType: string, dbType: string): Promise<RdiTemplateResponseDto> {
+    this.logger.log('Getting RDI config template');
 
     const client = await this.rdiClientProvider.getOrCreate(rdiClientMetadata);
 
-    return await client.getTemplate(options);
+    return await client.getConfigTemplate(pipelineType, dbType);
+  }
+
+  async getJobTemplate(rdiClientMetadata: RdiClientMetadata, pipelineType: string): Promise<RdiTemplateResponseDto> {
+    this.logger.log('Getting RDI job template');
+
+    const client = await this.rdiClientProvider.getOrCreate(rdiClientMetadata);
+
+    return await client.getJobTemplate(pipelineType);
   }
 
   async getPipelineStatus(rdiClientMetadata: RdiClientMetadata): Promise<unknown> {

redisinsight/ui/src/constants/api.ts

@@ -143,7 +143,8 @@ enum ApiEndpoints {
   RDI_DEPLOY_PIPELINE = 'pipeline/deploy',
   RDI_TEST_CONNECTIONS = 'pipeline/test-connections',
   RDI_PIPELINE_STRATEGIES = 'pipeline/strategies',
-  RDI_PIPELINE_TEMPLATE = 'pipeline/template',
+  RDI_JOB_TEMPLATE = 'pipeline/job/template',
+  RDI_CONFIG_TEMPLATE = 'pipeline/config/template',
   RDI_PIPELINE_JOB_FUNCTIONS = 'pipeline/job-functions',
   RDI_STATISTICS = 'statistics',
   RDI_PIPELINE_STATUS = 'pipeline/status',

redisinsight/ui/src/pages/rdi/pipeline-management/components/template-form/TemplateForm.tsx

@@ -13,7 +13,12 @@ import { useDispatch, useSelector } from 'react-redux'
 import { useParams } from 'react-router-dom'
 import cx from 'classnames'
 
-import { fetchPipelineStrategies, fetchPipelineTemplate, rdiPipelineStrategiesSelector } from 'uiSrc/slices/rdi/pipeline'
+import {
+  fetchPipelineStrategies,
+  fetchJobTemplate,
+  fetchConfigTemplate,
+  rdiPipelineStrategiesSelector,
+} from 'uiSrc/slices/rdi/pipeline'
 import { RdiPipelineTabs } from 'uiSrc/slices/interfaces/rdi'
 import { sendEventTelemetry, TelemetryEvent } from 'uiSrc/telemetry'
 import { NO_TEMPLATE_VALUE, NO_OPTIONS, INGEST_OPTION } from './constants'
@@ -69,11 +74,12 @@ const TemplateForm = (props: Props) => {
   }
 
   const handleApply = () => {
-    const values = source === RdiPipelineTabs.Config
-      ? { dbType: selectedDbType, pipelineType: selectedPipelineType }
-      : { pipelineType: selectedPipelineType }
-
-    dispatch(fetchPipelineTemplate(rdiInstanceId, values, onSuccess))
+    if (source === RdiPipelineTabs.Config) {
+      dispatch(fetchConfigTemplate(rdiInstanceId, selectedPipelineType, selectedDbType, onSuccess))
+    }
+    if (source === RdiPipelineTabs.Jobs) {
+      dispatch(fetchJobTemplate(rdiInstanceId, selectedPipelineType, onSuccess))
+    }
     sendEventTelemetry({
       event: TelemetryEvent.RDI_TEMPLATE_CLICKED,
       eventData: {