Documentation Update

api-service/.dockerignore

@@ -1 +1,3 @@
-node_modules
+node_modules
+api-service/postman-collection
+postman-collection

api-service/Dockerfile

@@ -1,7 +1,8 @@
-FROM --platform=linux/amd64 node:20-alpine
+FROM --platform=linux/amd64 node:23.4-alpine
 RUN mkdir -p /opt/api-service
 COPY ./api-service  ./opt/api-service
 WORKDIR /opt/api-service
+RUN rm -rf postman-collection
 RUN npm install
 COPY . .
 EXPOSE 3000

api-service/package.json

@@ -22,6 +22,17 @@
     "@azure/storage-blob": "^12.17.0",
     "@google-cloud/storage": "^7.9.0",
     "@jsonhero/schema-infer": "^0.1.5",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/exporter-logs-otlp-http": "^0.53.0",
+    "@opentelemetry/exporter-metrics-otlp-http": "^0.53.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.55.0",
+    "@opentelemetry/resources": "^1.28.0",
+    "@opentelemetry/sdk-logs": "^0.53.0",
+    "@opentelemetry/sdk-metrics": "^1.26.0",
+    "@opentelemetry/sdk-node": "^0.53.0",
+    "@opentelemetry/sdk-trace-base": "^1.28.0",
+    "@opentelemetry/sdk-trace-node": "^1.28.0",
+    "@opentelemetry/semantic-conventions": "^1.28.0",
     "@project-sunbird/logger": "^0.0.9",
     "ajv": "^8.11.2",
     "ajv-formats": "^2.1.1",

api-service/src/app.ts

@@ -1,17 +1,23 @@
 import express, { Application } from "express";
-import {router as v2Router} from "./routes/Router"
-import { metricRouter } from "./routes/MetricRouter"
-import { druidProxyRouter } from "./routes/DruidProxyRouter"
+import { druidProxyRouter } from "./routes/DruidProxyRouter";
+import { metricRouter } from "./routes/MetricRouter";
+import { router as v2Router } from "./routes/Router";
 
 import bodyParser from "body-parser";
-import { errorHandler, obsrvErrorHandler } from "./middlewares/errors";
-import { ResponseHandler } from "./helpers/ResponseHandler";
 import { config } from "./configs/Config";
+import { ResponseHandler } from "./helpers/ResponseHandler";
+import { errorHandler, obsrvErrorHandler } from "./middlewares/errors";
+import { OTelService } from "./services/otel/OTelService";
 import { alertsRouter } from "./routes/AlertsRouter";
 import { interceptAuditEvents } from "./services/telemetry";
+import _ from "lodash";
+
+
 
 const app: Application = express();
-
+// Initialisation of Open telemetry Service.
+(config.otel && _.toLower(config?.otel?.enable) === "true") ? OTelService.init() : console.info("OpenTelemetry Service is Disabled"); 
+
 app.use(bodyParser.json({ limit: config.body_parser_limit}));
 app.use(express.text());
 app.use(express.json());

api-service/src/configs/Config.ts

@@ -42,8 +42,8 @@ export const config = {
         "connectionTimeout": process.env.kafka_connection_timeout ? parseInt(process.env.kafka_connection_timeout) : 5000
       },
       "topics": {  // Default Kafka topics depend on type of dataset.
-        "createDataset": `${process.env.system_env || "local"}.ingest`,
-        "createMasterDataset": `${process.env.system_env || "local"}.masterdata.ingest`
+        "createDataset": `ingest`,
+        "createMasterDataset": `masterdata.ingest`
       }
     }
   },
@@ -115,4 +115,9 @@ export const config = {
   },
   "user_token_public_key": process.env.user_token_public_key || "",
   "is_RBAC_enabled": process.env.is_rbac_enabled || "false",
+  "otel": {
+    "enable": process.env.otel_enable || "false",
+    "collector_endpoint": process.env.otel_collector_endpoint || "http://localhost:4318"
+  },
+  "storage_types": process.env.storage_types || '{"lake_house":true,"realtime_store":true}'
 }

api-service/src/configs/DatasetConfigDefault.ts

@@ -17,7 +17,7 @@ export const defaultDatasetConfig = {
         }
     },
     "dedup_config": {
-        "drop_duplicates": true,
+        "drop_duplicates": false,
         "dedup_key": "id",
         "dedup_period": 604800, // 7 days
     },

api-service/src/configs/IngestionConfig.ts

@@ -84,12 +84,12 @@ export const rawIngestionSpecDefaults = {
     "flattenSpec": [
         {
             "type": "path",
-            "expr": "$.obsrv_meta.source.connector",
+            "expr": "$.obsrv_meta.source.entry_source",
             "name": "obsrv.meta.source.connector"
         },
         {
             "type": "path",
-            "expr": "$.obsrv_meta.source.connectorInstance",
+            "expr": "$.obsrv_meta.source.id",
             "name": "obsrv.meta.source.id"
         }
     ],

api-service/src/controllers/DataIngestion/DataIngestionController.ts

@@ -39,34 +39,34 @@ const dataIn = async (req: Request, res: Response) => {
             logger.error({ apiId, message: `Dataset with id ${datasetId} not found in live table`, code: "DATASET_NOT_FOUND" })
             return ResponseHandler.errorResponse(errorObject.datasetNotFound, req, res);
         }
-        const { entry_topic, dataset_config, api_version } = dataset
+        const { entry_topic, dataset_config, extraction_config, api_version } = dataset
         const entryTopic = api_version !== "v2" ? _.get(dataset_config, "entry_topic") : entry_topic
         if (!entryTopic) {
             logger.error({ apiId, message: "Entry topic not found", code: "TOPIC_NOT_FOUND" })
             return ResponseHandler.errorResponse(errorObject.topicNotFound, req, res);
         }
-        await send(addMetadataToEvents(datasetId, requestBody), entryTopic)
+        await send(addMetadataToEvents(datasetId, requestBody, extraction_config), entryTopic)
         ResponseHandler.successResponse(req, res, { status: 200, data: { message: "Data ingested successfully" } });
 
 }
 
-const addMetadataToEvents = (datasetId: string, payload: any) => {
+const addMetadataToEvents = (datasetId: string, payload: any, extraction_config: any) => {
     const validData = _.get(payload, "data");
     const now = Date.now();
     const mid = _.get(payload, "params.msgid");
     const source = { id: "api.data.in", version: config?.version, entry_source: "api" };
     const obsrvMeta = { syncts: now, flags: {}, timespans: {}, error: {}, source: source };
     if (Array.isArray(validData)) {
-        const payloadRef = validData.map((event: any) => {
-            const payload = {
-                event,
+        const extraction_key: string = _.get(extraction_config, "extraction_key", 'events');
+        const dedup_key: string = _.get(extraction_config, "dedup_config.dedup_key", 'id');
+            const payload: any = {
                 "obsrv_meta": obsrvMeta,
                 "dataset": datasetId,
                 "msgid": mid
-            }
+            };
+            payload[extraction_key] = validData;
+            payload[dedup_key] = mid
             return payload;
-        })
-        return payloadRef;
     }
     else {
         return ({

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

@@ -4,7 +4,7 @@ import { ResponseHandler } from "../../helpers/ResponseHandler";
 import * as _ from "lodash";
 import { schemaValidation } from "../../services/ValidationService";
 import validationSchema from "./RequestValidationSchema.json";
-import { datasetService, getLiveDatasetConfigs } from "../../services/DatasetService";
+import { datasetService, getLiveDatasetConfigs, validateStorageSupport } from "../../services/DatasetService";
 import { updateRecords } from "./DatasetCopyHelper";
 import { obsrvError } from "../../types/ObsrvError";
 
@@ -40,6 +40,7 @@ const datasetCopy = async (req: Request, res: Response) => {
     validateRequest(req);
     const newDatasetId = _.get(req, "body.request.destination.datasetId");
     const dataset = await fetchDataset(req);
+    validateStorageSupport(dataset);
     const userID = (req as any)?.userID;
     _.set(dataset, "created_by", userID);
     _.set(dataset, "updated_by", userID);

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

@@ -1,7 +1,7 @@
 import _ from "lodash";
 import { Request, Response } from "express";
 import httpStatus from "http-status";
-import { datasetService } from "../../services/DatasetService";
+import { datasetService, validateStorageSupport } from "../../services/DatasetService";
 import DatasetCreate from "./DatasetCreateValidationSchema.json";
 import { schemaValidation } from "../../services/ValidationService";
 import { ResponseHandler } from "../../helpers/ResponseHandler";
@@ -28,6 +28,7 @@ const validateRequest = async (req: Request) => {
         throw obsrvError(datasetId, "DATASET_DUPLICATE_DENORM_KEY", "Duplicate denorm output fields found.", "BAD_REQUEST", 400, undefined, {duplicateKeys: duplicateDenormKeys})
     }
 
+    validateStorageSupport(_.get(req, ["body", "request"]))
 }
 
 const datasetCreate = async (req: Request, res: Response) => {

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

@@ -2,7 +2,7 @@ import { Request, Response } from "express";
 import { ResponseHandler } from "../../helpers/ResponseHandler";
 import httpStatus from "http-status";
 import _ from "lodash";
-import { datasetService } from "../../services/DatasetService";
+import { datasetService, validateStorageSupport } from "../../services/DatasetService";
 import { datasetImportValidation, migrateExportedDatasetV1 } from "./DatasetImportHelper";
 import { obsrvError } from "../../types/ObsrvError";
 
@@ -15,18 +15,23 @@ const datasetImport = async (req: Request, res: Response) => {
         const migratedConfigs = migrateExportedDatasetV1(requestBody)
         datasetPayload = migratedConfigs;
     }
+    const userID = (req as any)?.userID;
+    _.set(datasetPayload, "created_by", userID);
+    _.set(datasetPayload, "updated_by", userID);
     const { updatedDataset, ignoredFields } = await datasetImportValidation({ ...requestBody, "request": datasetPayload })
     const { successMsg, partialIgnored } = getResponseData(ignoredFields)
 
-    const dataset = await importDataset(updatedDataset, overwrite);
+    validateStorageSupport(updatedDataset);
+    const dataset = await importDataset(updatedDataset, overwrite, userID);
     ResponseHandler.successResponse(req, res, { status: httpStatus.OK, data: { message: successMsg, data: dataset, ...(!_.isEmpty(partialIgnored) && { ignoredFields: partialIgnored }) } });
 }
 
-const importDataset = async (dataset: Record<string, any>, overwrite: string | any) => {
+const importDataset = async (dataset: Record<string, any>, overwrite: string | any, userID : string) => {
     const dataset_id = _.get(dataset,"dataset_id")
     const response = await datasetService.createDraftDataset(dataset).catch(err => { return err })
     if (response?.name === "SequelizeUniqueConstraintError") {
         if (overwrite === "true") {
+            _.set(dataset, "updated_by", userID);
             const overwriteRes = await datasetService.updateDraftDataset(dataset).catch(()=>{
                 throw obsrvError(dataset_id, "DATASET_IMPORT_FAILURE", `Failed to import dataset: ${dataset_id} as overwrite failed`, "INTERNAL_SERVER_ERROR", 500);
             })

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

@@ -37,7 +37,7 @@ const listDatasets = async (request: Record<string, any>): Promise<Record<string
     const draftFilters = _.set(_.cloneDeep(filters), "status", _.isEmpty(status) ? draftDatasetStatus : _.intersection(status, draftDatasetStatus));
     const liveFilters = _.set(_.cloneDeep(filters), "status", _.isEmpty(status) ? liveDatasetStatus : _.intersection(status, liveDatasetStatus));
     const liveDatasetList = await datasetService.findDatasets(liveFilters, defaultFields, [["updated_date", "DESC"]]);
-    const draftDatasetList = await datasetService.findDraftDatasets(draftFilters, defaultFields, [["updated_date", "DESC"]]);
+    const draftDatasetList = await datasetService.findDraftDatasets(draftFilters, [...defaultFields, "data_schema", "validation_config", "dedup_config", "denorm_config", "connectors_config", "version_key"], [["updated_date", "DESC"]]);
     return _.compact(_.concat(liveDatasetList, draftDatasetList));
 }
 

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

@@ -1,7 +1,7 @@
 import { Request, Response } from "express";
 import _ from "lodash";
 import { ResponseHandler } from "../../helpers/ResponseHandler";
-import { datasetService } from "../../services/DatasetService";
+import { datasetService, validateStorageSupport } from "../../services/DatasetService";
 import { schemaValidation } from "../../services/ValidationService";
 import StatusTransitionSchema from "./RequestValidationSchema.json";
 import ReadyToPublishSchema from "./ReadyToPublishSchema.json"
@@ -88,6 +88,7 @@ const deleteDataset = async (dataset: Record<string, any>) => {
 const readyForPublish = async (dataset: Record<string, any>, updated_by: any) => {
 
     const draftDataset: any = await datasetService.getDraftDataset(dataset.dataset_id)
+    validateStorageSupport(draftDataset);
     let defaultConfigs: any = _.cloneDeep(defaultDatasetConfig)
     defaultConfigs = _.omit(defaultConfigs, ["router_config"])
     defaultConfigs = _.omit(defaultConfigs, "dedup_config.dedup_key");
@@ -136,6 +137,7 @@ const readyForPublish = async (dataset: Record<string, any>, updated_by: any) =>
 const publishDataset = async (dataset: Record<string, any>, userID: any) => {
 
     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);

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

@@ -5,7 +5,7 @@ import Model from "sequelize/types/model";
 import { DatasetStatus } from "../../types/DatasetModels";
 import { ResponseHandler } from "../../helpers/ResponseHandler";
 import { cipherService } from "../../services/CipherService";
-import { datasetService } from "../../services/DatasetService";
+import { datasetService, validateStorageSupport } from "../../services/DatasetService";
 import { schemaValidation } from "../../services/ValidationService";
 import DatasetUpdate from "./DatasetUpdateValidationSchema.json";
 import { obsrvError } from "../../types/ObsrvError";
@@ -30,6 +30,7 @@ const validateRequest = async (req: Request) => {
         throw obsrvError(datasetId, "DATASET_UPDATE_NO_FIELDS", "Provide atleast one field in addition to the dataset_id to update the dataset", "BAD_REQUEST", 400)
     }
 
+    validateStorageSupport(_.get(req, ["body", "request"]))
 }
 
 const validateDataset = (dataset: Record<string, any> | null, req: Request) => {

api-service/src/controllers/QueryWrapper/SqlQueryWrapper.ts

@@ -5,6 +5,8 @@ import logger from "../../logger";
 import { ResponseHandler } from "../../helpers/ResponseHandler";
 import { ErrorObject } from "../../types/ResponseModel";
 import { druidHttpService } from "../../connections/druidConnection";
+import { getDatasourceList } from "../../services/DatasourceService";
+import { AxiosResponse } from "axios";
 
 const apiId = "api.obsrv.data.sql-query";
 const errorCode = "SQL_QUERY_FAILURE"
@@ -25,10 +27,16 @@ export const sqlQuery = async (req: Request, res: Response) => {
             } as ErrorObject, req, res);
         }
 
-        const result = await druidHttpService.post(`${config.query_api.druid.sql_query_path}`, req.body, {
-            headers: { Authorization: authorization },
-        });
-
+        const query = req.body.query as string;
+        let result: AxiosResponse;
+        if (isTableSchemaQuery(query)) {
+            const dataSources = await fetchDruidDataSources();
+            result = createMockAxiosResponse(dataSources);
+        } else {
+            result = await druidHttpService.post(`${config.query_api.druid.sql_query_path}`, req.body, {
+                headers: { Authorization: authorization },
+            });
+        }
         logger.info({ messsge: "Successfully fetched data using sql query", apiId, resmsgid })
         ResponseHandler.flatResponse(req, res, result)
     } catch (error: any) {
@@ -37,4 +45,37 @@ export const sqlQuery = async (req: Request, res: Response) => {
         logger.error(error, apiId, code, resmsgid)
         ResponseHandler.errorResponse(errorMessage, req, res);
     }
-}
+}
+
+const fetchDruidDataSources = async (): Promise<{ TABLE_NAME: string }[]> => {
+    try {
+        const dataSources = await getDatasourceList();
+        return dataSources
+            .filter((ds: any) => ds.type === "druid")
+            .map((ds: any) => ({ TABLE_NAME: ds.dataValues.datasource_ref }));
+    } catch (error) {
+        logger.error({ message: "Failed to fetch Druid data sources", error });
+        throw new Error("Failed to fetch Druid data sources");
+    }
+};
+
+const isTableSchemaQuery = (sqlQuery?: string): boolean => {
+    return (
+      sqlQuery
+        ?.trim()
+        .replace(/\s+/g, " ")
+        .toUpperCase() ===
+      "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'DRUID'"
+    );
+  };
+
+
+const createMockAxiosResponse = (data: any): AxiosResponse => {
+    return {
+        data,
+        status: 200,
+        statusText: "OK",
+        headers: {},
+        config: {},
+    } as AxiosResponse;
+};

api-service/src/middlewares/RBAC_middleware.ts

@@ -77,7 +77,7 @@ const checkAccess = (decoded: any, action: string, req: Request, res: Response)
 
 const basicToken = (token: string, req: Request, res: Response, next: NextFunction) => {
   try {
-    const decoded = jwt.verify(token, config.user_token_public_key);
+    const decoded = jwt.verify(token, config.user_token_public_key, { algorithms: ['RS256'] });
 
     if (!decoded || !_.isObject(decoded)) {
       return errorHandler(401, "Token verification failed or invalid token", req, res);
@@ -125,8 +125,8 @@ export default {
         (req as any).userID = "SYSTEM";
         return next();
       }
-
-      const token = req.get("x-user-token");
+      const authHeader = req.headers['authorization'];
+      const token = authHeader && authHeader.split(' ')[1];
       if (!token) {
         return errorHandler(401, "No token provided", req, res);
       }

api-service/src/middlewares/userPermissions.json

@@ -4,6 +4,7 @@
             "api.datasets.list",
             "api.datasets.read",
             "api.datasets.export",
+            "api.dataset.health",
             "api.data.out",
             "api.data.exhaust",
             "api.alert.list",
@@ -46,7 +47,6 @@
             "api.datasets.update",
             "api.datasets.import",
             "api.datasets.copy",
-            "api.dataset.health",
             "api.datasets.dataschema"
         ],
         "data": [
@@ -116,7 +116,11 @@
             "file",
             "connector",
             "sqlQuery",
-            "restricted_dataset_api"
+            "restricted_dataset_api",
+            "alert",
+            "metric",
+            "silence",
+            "notificationChannel"
         ],
         "operations_admin": [
             "alert",