diff --git a/package-lock.json b/package-lock.json index eba19a9..fc5d425 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,7 +17,7 @@ "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", "@dadosfera/protospack": "2.5.3", - "@dadosfera/protospack-v2": "3.38.0-beta.31", + "@dadosfera/protospack-v2": "3.40.0-beta.2", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", @@ -1744,9 +1744,9 @@ } }, "node_modules/@dadosfera/protospack-v2": { - "version": "3.38.0-beta.31", - "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.38.0-beta.31.tgz", - "integrity": "sha512-IxnVEaVvQNumN/PQsnF+9CsFvmn6/0YyfXauoPQ1bccvqU8Lo+hKZv4qTBrR+urd05OWyg+dJBv/QtW45b04XQ==", + "version": "3.40.0-beta.2", + "resolved": "https://dadosfera-611330257153.d.codeartifact.us-east-1.amazonaws.com/npm/dadosfera-npm/@dadosfera/protospack-v2/-/protospack-v2-3.40.0-beta.2.tgz", + "integrity": "sha512-WMoL9OhKJ05qDojd7cPcP7x45bHHnnQttfVoTWrc+lOkcVHygu0ob52ArfD8RRmpAXezBh9vDSbxz45n/lxbKg==", "license": "ISC", "dependencies": { "@grpc/grpc-js": "^1.9.3", diff --git a/package.json b/package.json index 7da4765..059f151 100644 --- a/package.json +++ b/package.json @@ -35,7 +35,7 @@ "@aws-sdk/signature-v4": "^3.370.0", "@dadosfera/dadosfera-logs": "^1.0.0-beta.4", "@dadosfera/protospack": "2.5.3", - "@dadosfera/protospack-v2": "3.38.0-beta.31", + "@dadosfera/protospack-v2": "3.40.0-beta.2", "@grpc/grpc-js": "^1.9.3", "@grpc/proto-loader": "^0.7.9", "@nestjs/cli": "^9.5.0", diff --git a/src/modules/inputs/inputs.service.ts b/src/modules/inputs/inputs.service.ts index 44c7ddd..46ba2a4 100644 --- a/src/modules/inputs/inputs.service.ts +++ b/src/modules/inputs/inputs.service.ts @@ -18,6 +18,7 @@ import { InputCreateS3Request, InputNewCreateRequest, InputUpdateResponse, + RollbackInputRequest, TestConnectionRequest, } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; @@ -209,7 +210,8 @@ export class InputsService { // this.validateCron({ ...data, info }); try { const { - nimbusUpdates, + tablesUpdate, + dataAssetUpdate, input } = await this.OLD_inputClient.update({ id, @@ -222,13 +224,29 @@ export class InputsService { ); return { input: updateInputResponse, - nimbusUpdates + tablesUpdate, + dataAssetUpdate }; } catch (err) { throw new HttpException(err.message, HttpStatus.NOT_FOUND); } } + async rollbackUpdate( + data: RollbackInputRequest + ) { + this.logger.info('PipelinesClientService - rollbackUpdate'); + this.logger.info('Rolling back input update with data: ' + JSON.stringify(data)); + const updatePipelineResponse = await lastValueFrom( + this.inputWriteService.RollbackInputUpdate( + data + ), + ); + this.logger.info('Done'); + + return updatePipelineResponse; + } + async remove(idRequest: IIdRequest) { return lastValueFrom(this.inputWriteService.InputRemove(idRequest)); } diff --git a/src/modules/pipelinesV2/pipelines.service.ts b/src/modules/pipelinesV2/pipelines.service.ts index db04131..17eb41b 100644 --- a/src/modules/pipelinesV2/pipelines.service.ts +++ b/src/modules/pipelinesV2/pipelines.service.ts @@ -28,11 +28,13 @@ import { getObjValueFromPath, objHasPath } from 'src/utils/ObjValueFromPath'; import ErrorCodes from 'src/utils/errorCodes'; import ErrorBuilder from 'src/utils/ErrorBuilder'; import { PlatformApiService } from '../platform-api/platform-api.service'; -import { Info, NewTable } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; -import { NimbusUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; -import axios from 'axios'; +import { Info } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/entities'; +import { TableUpdate } from '@dadosfera/protospack-v2/dist/lib/Input/interfaces/messages'; +import { AxiosError } from 'axios'; import { NimbusService } from 'src/services/nimbus/nimbus.service'; +type RollbackPromise = () => Promise; + export class PipelinesService implements OnModuleInit { logger: DadosferaLogger; pipelineWriteService: WriteService.PipelineV2WriteService; @@ -352,33 +354,116 @@ export class PipelinesService implements OnModuleInit { async updatePipelineInput(pipelineId: string, inputId: string, updateInputDTO: UpdatePlatformInputRequest, info: Info, user: RequestUser, metadata: Metadata) { this.logger.info('InputClientService - Update'); - this.logger.info('Update Dynamo Reference'); + const { + input: oldInput + } = await this.inputsService.findOne({ + id: inputId, + info: info + }); + + this.logger.info('Update Dynamo Reference :' + JSON.stringify(oldInput)); const pipelineIdFormat = pipelineId.split('-').join('_'); - console.log("updatePipelineInput", info, updateInputDTO) + const rollback: RollbackPromise[] = []; + const updateInputResponse = await this.inputsService.update( inputId, updateInputDTO, info - ) - - this.logger.info(JSON.stringify(updateInputResponse)) - await this.updateNimbus(info.customer, updateInputResponse.nimbusUpdates); - - await this.updatePlatformJobs( - pipelineIdFormat, - updateInputDTO, - user ); + const inputRollback = () => { + this.logger.info("exec rollback to input: " + JSON.stringify(oldInput)); + return this.inputsService.rollbackUpdate( + { + id: inputId, + dataAssetUpdate: updateInputResponse.dataAssetUpdate, + tables: oldInput.tables, + info + } + ) as Promise; + } + + rollback.push(inputRollback); + + this.logger.info("Input Update Response: " + JSON.stringify(updateInputResponse)) + + const nimbusUpdates = updateInputResponse?.tablesUpdate || []; + + nimbusUpdates.forEach(update => { + const nimbusRollback = () => { + return this.nimbusService.renameTable( + info.customer, + update.database, + { + table_name: update.table_name, + table_schema: update.table_schema + }, + { + table_name: update.old_table_name, + table_schema: update.old_table_schema + } + ); + } + rollback.push(nimbusRollback); + }); + + try { + await this.updateNimbus(info.customer, nimbusUpdates); + } catch (error) { + this.logger.error(error); + if (error instanceof AxiosError) { + this.logger.error(JSON.stringify(error.response.data)); + } + await this.executeRenameRollback(rollback); + + throw new Error("Error Nimbus updating tables"); + } + + try { + await this.updatePlatformJobs( + pipelineIdFormat, + updateInputDTO, + user + ); + } catch (error) { + this.logger.error(error); + await this.executeRenameRollback(rollback) + throw new Error("Error Platform API updating jobs"); + } + return updateInputResponse; } - private async updateNimbus(customer: string, changes: NimbusUpdate[]) { + private async executeRenameRollback(request: RollbackPromise[]) { + this.logger.info('rollback steps: ' + request.length) + const result = await Promise.allSettled(request.map(func => func())); + result.forEach(promise => { + this.logger.info("Promise finish with status: " + promise.status) + + if (promise.status === "rejected") { + this.logger.error("reject with: " + JSON.stringify(promise.reason || {})) + } + + if (promise.status === "fulfilled") { + this.logger.info("success with: " + JSON.stringify(promise.value || {})) + } + }); + + } + + private async updateNimbus(customer: string, changes: TableUpdate[]) { + // throw new Error("teste error nimbus"); this.logger.info('Nimbus Changes: ' + JSON.stringify(changes)); if(!changes || changes.length === 0) return; const requests = changes.map(change => { - return this.nimbusService.renameTable(customer, change.database, change.old, change.new); + return this.nimbusService.renameTable(customer, change.database, { + table_name: change.old_table_name, + table_schema: change.old_table_schema + }, { + table_name: change.table_name, + table_schema: change.table_schema + }); }) const values = await Promise.allSettled(requests); @@ -386,6 +471,20 @@ export class PipelinesService implements OnModuleInit { const success = values.map(request => request.status === "fulfilled") this.logger.info("Updates with succes: " + success.length); + + values.forEach(promise => { + this.logger.info("Promise finish with status: " + promise.status) + + if (promise.status === "rejected") { + this.logger.error("reject with: " + JSON.stringify(promise.reason || {})); + throw new Error(promise.reason ); + } + + if (promise.status === "fulfilled") { + this.logger.info("success with: " + JSON.stringify(promise.value || {})); + } + }); + } async updatePlatformJobs(pipelineId: string, updateInputDTO: UpdatePlatformInputRequest, user: RequestUser) { @@ -396,6 +495,10 @@ export class PipelinesService implements OnModuleInit { job_id: `${pipelineId}_${index}`, } + if (table.type !== "incremental_with_qualify") { + delete table.destinations?.qualify; + } + if (table.memory) { jobUpdate["memory"] = { amount: table.memory * 1000 diff --git a/src/services/nimbus/nimbus.service.ts b/src/services/nimbus/nimbus.service.ts index 031aae7..03789ed 100644 --- a/src/services/nimbus/nimbus.service.ts +++ b/src/services/nimbus/nimbus.service.ts @@ -34,7 +34,7 @@ export class NimbusService { const path = `/api/catalog/rename-tables/?database_name=${encodeURIComponent(database)}&table_name=${encodeURIComponent(old.table_name)}&table_schema=${encodeURIComponent(old.table_schema)}`; try { - this.logger.info("Request for PATCH" + nimbusUrl + path); + this.logger.info("Request for PATCH " + nimbusUrl + path); this.logger.info("Payload: " + JSON.stringify(update)); const { data } = await axios.patch(nimbusUrl + path, { table_name: update.table_name,