|
|
|
@@ -22,7 +22,7 @@ import { User, RequestUser } from '../../decorators/user.decorator';
|
|
|
|
|
import { PlatformApiService } from './platform-api.service';
|
|
|
|
|
import { PERMISSIONS_GROUPS } from '../../authentication/permissions.enum';
|
|
|
|
|
import { ElasticsearchService } from '../../services/elasticsearch';
|
|
|
|
|
import { DynamoDBService, ReferenceColumn } from '../../services/dynamodb';
|
|
|
|
|
import { DynamoDBService } from '../../services/dynamodb';
|
|
|
|
|
import { CustomersService } from '../customers/customers.service';
|
|
|
|
|
import { validateCronAgainstScheduleLimit } from '../../utils/cron-validation';
|
|
|
|
|
|
|
|
|
@@ -155,7 +155,7 @@ export class PlatformApiController {
|
|
|
|
|
* Extract and transform tables from jobs for DynamoDB input.
|
|
|
|
|
* Maps connector-specific fields to a common table format.
|
|
|
|
|
*
|
|
|
|
|
* - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name/type (reference_column object)
|
|
|
|
|
* - JDBC: load_type, table_name, column_include_list (columns), incremental_column_name (reference_column)
|
|
|
|
|
* - Singer: type maps replication_method (FULL_TABLE -> full_load, INCREMENTAL -> incremental), no columns
|
|
|
|
|
* - S3: same mapping as Singer, no columns
|
|
|
|
|
*/
|
|
|
|
@@ -163,7 +163,7 @@ export class PlatformApiController {
|
|
|
|
|
name: string;
|
|
|
|
|
type: string;
|
|
|
|
|
columns?: string[];
|
|
|
|
|
reference_column?: ReferenceColumn;
|
|
|
|
|
reference_column?: string;
|
|
|
|
|
}> {
|
|
|
|
|
if (!jobs || jobs.length === 0) return [];
|
|
|
|
|
|
|
|
|
@@ -171,7 +171,7 @@ export class PlatformApiController {
|
|
|
|
|
name: string;
|
|
|
|
|
type: string;
|
|
|
|
|
columns?: string[];
|
|
|
|
|
reference_column?: ReferenceColumn;
|
|
|
|
|
reference_column?: string;
|
|
|
|
|
}> = [];
|
|
|
|
|
|
|
|
|
|
for (const job of jobs) {
|
|
|
|
@@ -179,12 +179,12 @@ export class PlatformApiController {
|
|
|
|
|
if (!input) continue;
|
|
|
|
|
|
|
|
|
|
if (connector === 'jdbc') {
|
|
|
|
|
// JDBC: table_name, load_type, column_include_list, incremental_column_name/type
|
|
|
|
|
// JDBC: table_name, load_type, column_include_list, incremental_column_name
|
|
|
|
|
const table: {
|
|
|
|
|
name: string;
|
|
|
|
|
type: string;
|
|
|
|
|
columns?: string[];
|
|
|
|
|
reference_column?: ReferenceColumn;
|
|
|
|
|
reference_column?: string;
|
|
|
|
|
} = {
|
|
|
|
|
name: input.table_name || '',
|
|
|
|
|
type: input.load_type || 'full_load',
|
|
|
|
@@ -193,11 +193,8 @@ export class PlatformApiController {
|
|
|
|
|
table.columns = input.column_include_list;
|
|
|
|
|
}
|
|
|
|
|
if (input.incremental_column_name) {
|
|
|
|
|
// reference_column is stored as an object with name and type
|
|
|
|
|
table.reference_column = {
|
|
|
|
|
name: input.incremental_column_name,
|
|
|
|
|
type: input.incremental_column_type || 'unknown',
|
|
|
|
|
};
|
|
|
|
|
// reference_column is stored as a string (column name)
|
|
|
|
|
table.reference_column = input.incremental_column_name;
|
|
|
|
|
}
|
|
|
|
|
tables.push(table);
|
|
|
|
|
} else if (connector === 'singer' || connector === 's3') {
|
|
|
|
@@ -320,11 +317,11 @@ export class PlatformApiController {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Build changes for DynamoDB table entry
|
|
|
|
|
// reference_column is stored as an object with name and type
|
|
|
|
|
// reference_column is stored as a string (column name), not an object
|
|
|
|
|
const changes: {
|
|
|
|
|
type?: string;
|
|
|
|
|
columns?: string[];
|
|
|
|
|
reference_column?: ReferenceColumn | null;
|
|
|
|
|
reference_column?: string | null;
|
|
|
|
|
} = {};
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -335,15 +332,8 @@ export class PlatformApiController {
|
|
|
|
|
changes.columns = body.column_include_list;
|
|
|
|
|
}
|
|
|
|
|
if ('incremental_column_name' in body) {
|
|
|
|
|
// reference_column is stored as an object with name and type
|
|
|
|
|
if (body.incremental_column_name) {
|
|
|
|
|
changes.reference_column = {
|
|
|
|
|
name: body.incremental_column_name,
|
|
|
|
|
type: body.incremental_column_type || 'unknown',
|
|
|
|
|
};
|
|
|
|
|
} else {
|
|
|
|
|
changes.reference_column = null;
|
|
|
|
|
}
|
|
|
|
|
// reference_column is just the column name as a string
|
|
|
|
|
changes.reference_column = body.incremental_column_name || null;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Update DynamoDB if there are changes
|
|
|
|
@@ -476,13 +466,11 @@ export class PlatformApiController {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const pipelineType = this.mapConnectorToDynamoType(connectorType);
|
|
|
|
|
this.logger.info('Syncing pipeline to Elasticsearch', {
|
|
|
|
|
customerName: user.customer_name,
|
|
|
|
|
pipelineId: body.id,
|
|
|
|
|
plugin,
|
|
|
|
|
connector: connectorType,
|
|
|
|
|
type: pipelineType,
|
|
|
|
|
properties,
|
|
|
|
|
inputId,
|
|
|
|
|
});
|
|
|
|
@@ -502,7 +490,6 @@ export class PlatformApiController {
|
|
|
|
|
cron: body.cron,
|
|
|
|
|
tables: inputId,
|
|
|
|
|
properties,
|
|
|
|
|
type: pipelineType,
|
|
|
|
|
},
|
|
|
|
|
connector,
|
|
|
|
|
);
|
|
|
|
@@ -630,39 +617,21 @@ export class PlatformApiController {
|
|
|
|
|
@ApiOperation({ summary: 'Execute a pipeline' })
|
|
|
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
|
|
|
async executePipeline(@Body() body: any, @User() user: RequestUser) {
|
|
|
|
|
// Inject customer_id (actually customer_name) into body for Platform-API
|
|
|
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
|
|
|
const enrichedBody = {
|
|
|
|
|
...body,
|
|
|
|
|
customer_id: user.customer_name,
|
|
|
|
|
};
|
|
|
|
|
return this.platformApiService.proxy('POST', '/pipeline/execute', user, enrichedBody);
|
|
|
|
|
return this.platformApiService.proxy('POST', '/pipeline/execute', user, body);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Post('pipeline/pause')
|
|
|
|
|
@ApiOperation({ summary: 'Pause a pipeline' })
|
|
|
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
|
|
|
async pausePipeline(@Body() body: any, @User() user: RequestUser) {
|
|
|
|
|
// Inject customer_id (actually customer_name) into body for Platform-API
|
|
|
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
|
|
|
const enrichedBody = {
|
|
|
|
|
...body,
|
|
|
|
|
customer_id: user.customer_name,
|
|
|
|
|
};
|
|
|
|
|
return this.platformApiService.proxy('POST', '/pipeline/pause', user, enrichedBody);
|
|
|
|
|
return this.platformApiService.proxy('POST', '/pipeline/pause', user, body);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Post('pipeline/unpause')
|
|
|
|
|
@ApiOperation({ summary: 'Unpause a pipeline' })
|
|
|
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
|
|
|
async unpausePipeline(@Body() body: any, @User() user: RequestUser) {
|
|
|
|
|
// Inject customer_id (actually customer_name) into body for Platform-API
|
|
|
|
|
// Note: Platform-API was created before customer_id existed, so it expects customer_name in the customer_id field
|
|
|
|
|
const enrichedBody = {
|
|
|
|
|
...body,
|
|
|
|
|
customer_id: user.customer_name,
|
|
|
|
|
};
|
|
|
|
|
return this.platformApiService.proxy('POST', '/pipeline/unpause', user, enrichedBody);
|
|
|
|
|
return this.platformApiService.proxy('POST', '/pipeline/unpause', user, body);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Put('pipeline/:pipelineId/memory')
|
|
|
|
@@ -722,6 +691,18 @@ export class PlatformApiController {
|
|
|
|
|
|
|
|
|
|
// ==================== PIPELINE RUN ROUTES ====================
|
|
|
|
|
|
|
|
|
|
@Post('pipeline/pipeline_run')
|
|
|
|
|
@ApiOperation({ summary: 'Create a pipeline run' })
|
|
|
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.UPDATE)
|
|
|
|
|
async createPipelineRun(@Body() body: any, @User() user: RequestUser) {
|
|
|
|
|
return this.platformApiService.proxy(
|
|
|
|
|
'POST',
|
|
|
|
|
'/pipeline/pipeline_run',
|
|
|
|
|
user,
|
|
|
|
|
body,
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Get('pipeline/:pipelineId/pipeline_run')
|
|
|
|
|
@ApiOperation({ summary: 'Get pipeline runs for a pipeline' })
|
|
|
|
|
@RequireAllPermissions(PERMISSIONS_GROUPS.PIPELINE.permissions.GET)
|
|
|
|
@@ -730,10 +711,9 @@ export class PlatformApiController {
|
|
|
|
|
@User() user: RequestUser,
|
|
|
|
|
@Query() query: Record<string, string>,
|
|
|
|
|
) {
|
|
|
|
|
const normalizedId = this.normalizePipelineId(pipelineId);
|
|
|
|
|
return this.platformApiService.proxy(
|
|
|
|
|
'GET',
|
|
|
|
|
`/pipeline/${normalizedId}/pipeline_run`,
|
|
|
|
|
`/pipeline/${pipelineId}/pipeline_run`,
|
|
|
|
|
user,
|
|
|
|
|
undefined,
|
|
|
|
|
query,
|
|
|
|
@@ -748,11 +728,9 @@ export class PlatformApiController {
|
|
|
|
|
@Param('runId') runId: string,
|
|
|
|
|
@User() user: RequestUser,
|
|
|
|
|
) {
|
|
|
|
|
const normalizedPipelineId = this.normalizePipelineId(pipelineId);
|
|
|
|
|
const normalizedRunId = this.normalizePipelineId(runId);
|
|
|
|
|
return this.platformApiService.proxy(
|
|
|
|
|
'GET',
|
|
|
|
|
`/pipeline/${normalizedPipelineId}/pipeline_run/${normalizedRunId}`,
|
|
|
|
|
`/pipeline/${pipelineId}/pipeline_run/${runId}`,
|
|
|
|
|
user,
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
@@ -765,10 +743,9 @@ export class PlatformApiController {
|
|
|
|
|
@User() user: RequestUser,
|
|
|
|
|
@Query() query: Record<string, string>,
|
|
|
|
|
) {
|
|
|
|
|
const normalizedRunId = this.normalizePipelineId(runId);
|
|
|
|
|
return this.platformApiService.proxy(
|
|
|
|
|
'GET',
|
|
|
|
|
`/pipeline/pipeline_run/${normalizedRunId}/logs`,
|
|
|
|
|
`/pipeline/pipeline_run/${runId}/logs`,
|
|
|
|
|
user,
|
|
|
|
|
undefined,
|
|
|
|
|
query,
|
|
|
|
|
Rafael Santana