feat #OBS-I501 : data analyse PII API added

Author: HarishGangula

api-service/src/configs/Config.ts

@@ -98,7 +98,7 @@ export const config = {
   "command_service_config": {
     "host": process.env.command_service_host || "http://localhost",
     "port": parseInt(process.env.command_service_port || "8000"),
-    "path": process.env.command_service_path || "/system/v1/dataset/command"
+    "paths": JSON.parse(process.env.command_service_paths || '{"dataset":"/system/v1/dataset/command","connector":"/connector/v1/register","analyzePII":"/system/data/v1/analyze/pii"}')
   },
   "flink_job_configs": {
     "pipeline_merged_job_manager_url": process.env.pipeline_merged_job_manager_url || "http://localhost:8081",

api-service/src/connections/commandServiceConnection.ts

@@ -5,21 +5,29 @@ import { v4 } from "uuid";
 
 const commandHost = _.get(config, ["command_service_config", "host"])
 const commandPort = _.get(config, ["command_service_config", "port"])
-const commandPath = _.get(config, ["command_service_config", "path"])
+const commandPaths = _.get(config, ["command_service_config", "paths"])
+const datasetPath = _.get(commandPaths, ["dataset"])
+const connectorRegisterPath = _.get(commandPaths, ["connector"])
+const analyzePIIPath = _.get(commandPaths, ["analyzePII"])
 
 export const commandHttpService = axios.create({ baseURL: `${commandHost}:${commandPort}`, headers: { "Content-Type": "application/json" } });
 
-export const executeCommand = async (id: string, command: string) => {
+export const executeCommand = async (id: string, command: string, userToken: string) => {
     const payload = {
         "id": v4(),
         "data": {
             "dataset_id": id,
             "command": command
         }
     }
-    return commandHttpService.post(commandPath, payload)
+    return commandHttpService.post(datasetPath, payload, { headers: { Authorization: userToken }})
 }
 
-export const registerConnector = async (requestBody: any) => {
-    return commandHttpService.post("/connector/v1/register", requestBody)
+export const registerConnector = async (requestBody: any, userToken: string) => {
+    return commandHttpService.post(connectorRegisterPath, requestBody, { headers: { Authorization: userToken }})
+}
+
+export const detectPII = async (requestBody: any, userToken: string) => {
+    console.log(`analyzePIIPath : ${analyzePIIPath}`)
+    return commandHttpService.post(analyzePIIPath, requestBody, { headers: { Authorization: userToken }})
 }

api-service/src/controllers/ConnectorRegister/ConnectorRegisterController.ts

@@ -22,7 +22,8 @@ const connectorRegisterController = async (req: Request, res: Response) => {
             relative_path: uploadStreamResponse[0]
         }
         logger.info({ apiId, resmsgid, message: `File uploaded to cloud provider successfully` })
-        const registryResponse = await registerConnector(payload);
+        const userToken = req.get('authorization') as string;
+        const registryResponse = await registerConnector(payload, userToken);
         logger.info({ apiId, resmsgid, message: `Connector registered successfully` })
         ResponseHandler.successResponse(req, res, { status: httpStatus.OK, data: { message: registryResponse?.data?.message } })
     } catch (error: any) {

api-service/src/controllers/DataAnalyzePII/DataAnalyzePIIController.ts

@@ -0,0 +1,26 @@
+import { Request, Response } from "express";
+import { ResponseHandler } from "../../helpers/ResponseHandler";
+import httpStatus from "http-status";
+import * as _ from "lodash";
+import logger from "../../logger";
+import { detectPII } from "../../connections/commandServiceConnection";
+
+const code = "FAILED_TO_DETECT_PII";
+export const dataAnalyzePII = async (req: Request, res: Response) => {
+    const apiId =  _.get(req, 'id')
+    try {
+        const userToken = req.get('authorization') as string;
+        const piiSuggestionsResponse = await detectPII(_.get(req, ['body', 'data']), userToken);
+        logger.info({apiId , message: `Detected PII successfully` })
+        ResponseHandler.successResponse(req, res, { status: httpStatus.OK, data: piiSuggestionsResponse?.data})
+    } catch (error: any) {
+        const errMessage = _.get(error, "response.data.detail")
+        logger.error(error, apiId, code);
+        let errorMessage = error;
+        const statusCode = _.get(error, "status")
+        if (!statusCode || statusCode == 500) {
+            errorMessage = { code, message: errMessage || "Failed to detect pii" }
+        }
+        ResponseHandler.errorResponse(errorMessage, req, res);
+    }
+}

api-service/src/controllers/DatasetReset/DatasetReset.ts

@@ -29,12 +29,13 @@ const datasetReset = async (req: Request, res: Response) => {
 
     const category = _.get(req, ["body", "request", "category"]);
     const datasetId = _.get(req, ["params"," datasetId"]);
+    const userToken = req.get('authorization') as string;
 
     await validateRequest(req);
     if (category == "processing") {
         const pipeLineStatus = await getFlinkHealthStatus()
         if (pipeLineStatus == HealthStatus.UnHealthy) {
-            await restartPipeline({ "dataset": { "dataset_id": datasetId } })
+            await restartPipeline({ "dataset": { "dataset_id": datasetId } }, userToken)
         }
     } else if (category == "query") {
         const datasources = await datasetService.findDatasources({"dataset_id": datasetId})

api-service/src/controllers/DatasetStatusTransition/DatasetStatusTransition.ts

@@ -50,6 +50,7 @@ const validateDataset = (dataset: any, datasetId: any, action: string) => {
 const datasetStatusTransition = async (req: Request, res: Response) => {
 
     const { dataset_id, status } = _.get(req.body, "request");
+    const userToken = req.get('authorization') as string;
     validateRequest(req, dataset_id);
 
     const dataset: Record<string, any> = (_.includes(liveDatasetActions, status)) ? await datasetService.getDataset(dataset_id, ["id", "status", "type", "api_version", "name"], true) : await datasetService.getDraftDataset(dataset_id, ["id", "dataset_id", "status", "type", "api_version"])
@@ -64,10 +65,10 @@ const datasetStatusTransition = async (req: Request, res: Response) => {
             await readyForPublish(dataset, userID);
             break;
         case "Live":
-            await publishDataset(dataset, userID);
+            await publishDataset(dataset, userID, userToken);
             break;
         case "Retire":
-            await retireDataset(dataset, userID);
+            await retireDataset(dataset, userID, userToken);
             break;
         default:
             throw obsrvError(dataset.id, "UNKNOWN_STATUS_TRANSITION", "Unknown status transition requested", "BAD_REQUEST", 400)
@@ -134,15 +135,15 @@ const readyForPublish = async (dataset: Record<string, any>, updated_by: any) =>
  * 
  * @param dataset 
  */
-const publishDataset = async (dataset: Record<string, any>, userID: any) => {
+const publishDataset = async (dataset: Record<string, any>, userID: any, userToken: string) => {
 
     const draftDataset: Record<string, any> = await datasetService.getDraftDataset(dataset.dataset_id) as unknown as Record<string, any>
     validateStorageSupport(draftDataset);
     _.set(draftDataset, ["created_by"], userID);
     _.set(draftDataset, ["updated_by"], userID);
     await validateAndUpdateDenormConfig(draftDataset);
     await updateMasterDataConfig(draftDataset)
-    await datasetService.publishDataset(draftDataset)
+    await datasetService.publishDataset(draftDataset, userToken);
 }
 
 const validateAndUpdateDenormConfig = async (draftDataset: Record<string, any>) => {
@@ -227,11 +228,11 @@ const updateMasterDataConfig = async (draftDataset: Record<string, any>) => {
     }
 }
 
-const retireDataset = async (dataset: Record<string, any>, updated_by: any) => {
+const retireDataset = async (dataset: Record<string, any>, updated_by: any, userToken: string) => {
 
     await canRetireIfMasterDataset(dataset);
     await datasetService.retireDataset(dataset, updated_by);
-    await restartPipeline(dataset);
+    await restartPipeline(dataset, userToken);
 }
 
 
@@ -254,8 +255,8 @@ const canRetireIfMasterDataset = async (dataset: Record<string, any>) => {
     }
 }
 
-export const restartPipeline = async (dataset: Record<string, any>) => {
-    return executeCommand(dataset.id, "RESTART_PIPELINE")
+export const restartPipeline = async (dataset: Record<string, any>, userToken: string) => {
+    return executeCommand(dataset.id, "RESTART_PIPELINE", userToken)
 }
 
 export default datasetStatusTransition;

api-service/src/middlewares/userPermissions.json

@@ -50,7 +50,8 @@
             "api.datasets.dataschema"
         ],
         "data": [
-            "api.data.in"
+            "api.data.in",
+            "api.data.analyze.pii"
         ],
         "queryTemplate": [
             "api.query.template.create",
@@ -92,7 +93,8 @@
             "schema",
             "file",
             "connector",
-            "sqlQuery"
+            "sqlQuery",
+            "data"
         ],
         "dataset_manager": [
             "general_access",
@@ -104,7 +106,8 @@
             "file",
             "connector",
             "sqlQuery",
-            "restricted_dataset_api"
+            "restricted_dataset_api",
+            "data"
         ],
         "admin": [
             "general_access",
@@ -120,7 +123,8 @@
             "alert",
             "metric",
             "silence",
-            "notificationChannel"
+            "notificationChannel",
+            "data"
         ],
         "operations_admin": [
             "alert",

api-service/src/routes/Router.ts

@@ -32,6 +32,7 @@ import telemetryActions from "../telemetry/telemetryActions";
 import datasetMetrics from "../controllers/DatasetMetrics/DatasetMetricsController";
 import checkRBAC from "../middlewares/RBAC_middleware";
 import connectorRegisterController from "../controllers/ConnectorRegister/ConnectorRegisterController";
+import { dataAnalyzePII } from "../controllers/DataAnalyzePII/DataAnalyzePIIController";
 
 export const router = express.Router();
 
@@ -60,6 +61,7 @@ router.post("/connectors/list", setDataToRequestObject("api.connectors.list"), o
 router.get("/connectors/read/:id", setDataToRequestObject("api.connectors.read"), onRequest({entity: Entity.Management }), telemetryAuditStart({action: telemetryActions.readConnectors, operationType: OperationType.GET}), checkRBAC.handler(), ConnectorsRead);
 router.post("/datasets/import", setDataToRequestObject("api.datasets.import"), onRequest({ entity: Entity.Management }), checkRBAC.handler(), DatasetImport);
 router.post("/connector/register", setDataToRequestObject("api.connector.register"), onRequest({ entity: Entity.Management }), connectorRegisterController);
+router.post("/data/analyze/pii", setDataToRequestObject("api.data.analyze.pii"), onRequest({ entity: Entity.Management }),checkRBAC.handler(), dataAnalyzePII);
 //Wrapper Service
 router.post("/obsrv/data/sql-query", setDataToRequestObject("api.obsrv.data.sql-query"), onRequest({ entity: Entity.Data_out }), checkRBAC.handler(), sqlQuery);
 router.post("/data/metrics", setDataToRequestObject("api.data.metrics"), onRequest({ entity: Entity.Data_out }), datasetMetrics)

api-service/src/services/DatasetService.ts

@@ -303,7 +303,7 @@ class DatasetService {
         }
     }
 
-    publishDataset = async (draftDataset: Record<string, any>) => {
+    publishDataset = async (draftDataset: Record<string, any>, userToken: string) => {
 
         const indexingConfig = draftDataset.dataset_config.indexing_config;
         const transaction = await sequelize.transaction()
@@ -326,7 +326,7 @@ class DatasetService {
             await transaction.rollback()
             throw obsrvError(draftDataset.id, "FAILED_TO_PUBLISH_DATASET", err.message, "SERVER_ERROR", 500, err);
         }
-        await executeCommand(draftDataset.dataset_id, "PUBLISH_DATASET");
+        await executeCommand(draftDataset.dataset_id, "PUBLISH_DATASET", userToken);
 
     }