Dataset Observability

api-service/package.json

@@ -42,6 +42,7 @@
     "busboy": "^1.6.0",
     "compression": "^1.7.4",
     "dateformat": "2.0.0",
+    "dayjs": "^1.11.13",
     "express": "^5.0.0-beta.3",
     "http-errors": "^2.0.0",
     "http-status": "^1.5.3",

api-service/src/configs/Config.ts

@@ -119,5 +119,10 @@ export const config = {
     "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}'
+  "storage_types": process.env.storage_types || '{"lake_house":true,"realtime_store":true}',
+  "data_observability": {
+    "default_freshness_threshold": process.env.default_freshness_threshold ? parseInt(process.env.default_freshness_threshold) : 5, // in minutes
+    "data_out_query_time_period": process.env.data_out_query_time_period || "1d",
+    "default_query_time_period": process.env.default_query_time_period ? parseInt(process.env.default_query_time_period) : 7, // in days
+  }
 }

api-service/src/controllers/DataExhaust/DataExhaustController.ts

@@ -11,7 +11,7 @@ import { cloudProvider } from "../../services/CloudServices";
 
 export const dataExhaust = async (req: Request, res: Response) => {
     const { params } = req;
-    const { datasetId } = params;
+    const { dataset_id } = params;
     const { type }: any = req.query;
     const momentFormat = "YYYY-MM-DD";
 
@@ -28,12 +28,12 @@ export const dataExhaust = async (req: Request, res: Response) => {
         return resData || {};
     }
 
-    if (type && config.cloud_config.exclude_exhaust_types.includes(datasetId)) {
+    if (type && config.cloud_config.exclude_exhaust_types.includes(dataset_id)) {
         return ResponseHandler.errorResponse({ statusCode: 404, message: "Record not found", errCode: httpStatus["404_NAME"] }, req, res)
     }
-    const datasetRecord = await verifyDatasetExists(datasetId);
+    const datasetRecord = await verifyDatasetExists(dataset_id);
     if (datasetRecord === null) {
-        logger.error(`Dataset with ${datasetId} not found in live table`)
+        logger.error(`Dataset with ${dataset_id} not found in live table`)
         return ResponseHandler.errorResponse({ statusCode: 404, message: "Record not found", errCode: httpStatus["404_NAME"] }, req, res)
     }
     const dateRange = getDateRange(req);
@@ -47,7 +47,7 @@ export const dataExhaust = async (req: Request, res: Response) => {
         return ResponseHandler.errorResponse({ statusCode: 400, message: `Invalid date range! make sure your range cannot be more than ${config.cloud_config.maxQueryDateRange} days`, errCode: "BAD_REQUEST" }, req, res)
     }
 
-    const resData: any = await getFromStorage(type, dateRange, datasetId);
+    const resData: any = await getFromStorage(type, dateRange, dataset_id);
     if (_.isEmpty(resData.files)) {
         logger.error("Date range provided does not have any backup files")
         return ResponseHandler.errorResponse({ statusCode: 404, message: "Date range provided does not have any backup files", errCode: "NOT_FOUND" }, req, res);

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

@@ -27,7 +27,7 @@ const apiId = "api.data.in";
 const dataIn = async (req: Request, res: Response) => {
 
     const requestBody = req.body;
-        const datasetId = req.params.datasetId.trim();
+        const datasetId = req.params.dataset_id.trim();
 
         const isValidSchema = schemaValidation(requestBody, validationSchema)
         if (!isValidSchema?.isValid) {

api-service/src/controllers/DataMetrics/DataMetricsController.ts

@@ -0,0 +1,38 @@
+import { Request, Response } from "express";
+import _ from "lodash";
+import { executeNativeQuery } from "../../connections/druidConnection";
+import { ResponseHandler } from "../../helpers/ResponseHandler";
+import vaidationSchema from "./DataMetricsValidationSchema.json"
+import { schemaValidation } from "../../services/ValidationService";
+import logger from "../../logger";
+import { obsrvError } from "../../types/ObsrvError";
+import axios from "axios";
+import { config } from "../../configs/Config";
+
+const getBaseUrl = (url: string) => {
+    if (_.startsWith(url, "/prom")) return config.query_api.prometheus.url + _.replace(url, "/prom", "")
+}
+
+const dataMetrics = async (req: Request, res: Response) => {
+    const isValidSchema = schemaValidation(req.body, vaidationSchema);
+    if (!isValidSchema?.isValid) {
+        logger.error({ message: isValidSchema?.message, code: "INVALID_QUERY" })
+        throw obsrvError("", "INVALID_QUERY", isValidSchema.message, "BAD_REQUEST", 400)
+    }
+    const { query } = req.body || {};
+    const endpoint = query.url;
+    if (_.startsWith(endpoint, "/prom")) {
+        query.url = getBaseUrl(endpoint)
+        const { url, method, headers = {}, body = {}, params = {}, ...rest } = query;
+        const apiResponse = await axios.request({ url, method, headers, params, data: body, ...rest })
+        const data = _.get(apiResponse, "data");
+        return res.json(data);
+    }
+    else {
+        const query = _.get(req, ["body", "query", "body", "query"]);
+        const response = await executeNativeQuery(query);
+        ResponseHandler.successResponse(req, res, { status: 200, data: _.get(response, "data") });
+    }
+}
+
+export default dataMetrics;

api-service/src/controllers/DataOut/DataOutController.ts

@@ -9,7 +9,7 @@ import { executeNativeQuery, executeSqlQuery } from "../../connections/druidConn
 
 export const apiId = "api.data.out";
 const dataOut = async (req: Request, res: Response) => {
-    const datasetId = req.params?.datasetId;
+    const datasetId = req.params?.dataset_id;
     const requestBody = req.body;
     const msgid = _.get(req, "body.params.msgid");
     const isValidSchema = schemaValidation(requestBody, validationSchema);

api-service/src/controllers/DatasetMetrics/DatasetMetrics.json

@@ -0,0 +1,61 @@
+{
+    "type": "object",
+    "properties": {
+        "id": {
+            "type": "string",
+            "enum": [
+                "api.dataset.metrics"
+            ]
+        },
+        "ver": {
+            "type": "string"
+        },
+        "ts": {
+            "type": "string"
+        },
+        "params": {
+            "type": "object",
+            "properties": {
+                "msgid": {
+                    "type": "string"
+                }
+            },
+            "required": [
+                "msgid"
+            ],
+            "additionalProperties": false
+        },
+        "request": {
+            "type": "object",
+            "properties": {
+                "dataset_id": {
+                    "type": "string"
+                },
+                "category": {
+                    "type": "array",
+                    "items": {
+                        "type": "string",
+                        "enum": [
+                            "data_freshness",
+                            "data_observability",
+                            "data_volume",
+                            "data_lineage",
+                            "connectors",
+                            "data_quality"
+                        ]
+                    },
+                    "minItems": 1
+                },
+                "query_time_period":{
+                    "type": "integer",
+                    "minimum": 1
+                }
+            },
+            "required": [
+                "category",
+                "dataset_id"
+            ]
+        }
+    },
+    "required": ["id", "ver", "ts", "params", "request"]
+}

api-service/src/controllers/DatasetMetrics/DatasetMetricsController.ts

@@ -1,38 +1,81 @@
 import { Request, Response } from "express";
-import _ from "lodash";
-import { executeNativeQuery } from "../../connections/druidConnection";
+import * as _ from "lodash"
 import { ResponseHandler } from "../../helpers/ResponseHandler";
-import vaidationSchema from "./DatasetMetricsValidationSchema.json"
-import { schemaValidation } from "../../services/ValidationService";
+import dayjs from 'dayjs';
 import logger from "../../logger";
-import { obsrvError } from "../../types/ObsrvError";
-import axios from "axios";
+import { schemaValidation } from "../../services/ValidationService";
+import validationSchema from "./DatasetMetrics.json";
 import { config } from "../../configs/Config";
+import { datasetService } from "../../services/DatasetService";
+import { getConnectors, getDataFreshness, getDataLineage, getDataObservability, getDataQuality, getDataVolume } from "../../services/DatasetMetricsService";
 
-const getBaseUrl = (url: string) => {
-    if (_.startsWith(url, "/prom")) return config.query_api.prometheus.url + _.replace(url, "/prom", "")
-}
-
+const apiId = "api.dataset.metrics";
 const datasetMetrics = async (req: Request, res: Response) => {
-    const isValidSchema = schemaValidation(req.body, vaidationSchema);
+    const msgid = _.get(req, "body.params.msgid");
+    const requestBody = req.body;
+    const dataset_id = _.get(req, "body.request.dataset_id");
+    const timePeriod = _.get(req, "body.request.query_time_period") || config?.data_observability?.default_query_time_period;
+
+    const { category }: any = req.body.request;
+    const defaultThreshold = (typeof config?.data_observability?.default_freshness_threshold === 'number' ? config?.data_observability?.default_freshness_threshold : 5) * 60 * 1000; // 5 minutes in milliseconds
+    const dateFormat = 'YYYY-MM-DDTHH:mm:ss';
+    const endDate = dayjs().add(1, 'day').format(dateFormat);
+    const startDate = dayjs(endDate).subtract(timePeriod, 'day').format(dateFormat);
+    const intervals = `${startDate}/${endDate}`;
+    const isValidSchema = schemaValidation(requestBody, validationSchema);
+    const results = [];
+
     if (!isValidSchema?.isValid) {
-        logger.error({ message: isValidSchema?.message, code: "INVALID_QUERY" })
-        throw obsrvError("", "INVALID_QUERY", isValidSchema.message, "BAD_REQUEST", 400)
+        logger.error({ apiId, datasetId: dataset_id, msgid, requestBody, message: isValidSchema?.message, code: "DATA_OUT_INVALID_INPUT" })
+        return ResponseHandler.errorResponse({ message: isValidSchema?.message, statusCode: 400, errCode: "BAD_REQUEST", code: "DATA_OUT_INVALID_INPUT" }, req, res);
     }
-    const { query } = req.body || {};
-    const endpoint = query.url;
-    if (_.startsWith(endpoint, "/prom")) {
-        query.url = getBaseUrl(endpoint)
-        const { url, method, headers = {}, body = {}, params = {}, ...rest } = query;
-        const apiResponse = await axios.request({ url, method, headers, params, data: body, ...rest })
-        const data = _.get(apiResponse, "data");
-        return res.json(data);
+
+    const dataset = await datasetService.getDataset(dataset_id, ["id"], true)
+    if (!dataset) {
+        logger.error({ apiId, message: `Dataset with id ${dataset_id} not found in live table`, code: "DATASET_NOT_FOUND" })
+        return ResponseHandler.errorResponse({ message: `Dataset with id ${dataset_id} not found in live table`, code: "DATASET_NOT_FOUND", statusCode: 404, errCode: "NOT_FOUND" }, req, res);
     }
-    else {
-        const query = _.get(req, ["body", "query", "body", "query"]);
-        const response = await executeNativeQuery(query);
-        ResponseHandler.successResponse(req, res, { status: 200, data: _.get(response, "data") });
+
+    try {
+        if (!category || category.includes("data_freshness")) {
+            const dataFreshnessResult = await getDataFreshness(dataset_id, intervals, defaultThreshold);
+            results.push(dataFreshnessResult);
+        }
+
+        if (!category || category.includes("data_observability")) {
+            const dataObservabilityResult = await getDataObservability(dataset_id, intervals);
+            results.push(dataObservabilityResult);
+        }
+
+        if (!category || category.includes("data_volume")) {
+            const dataVolumeResult = await getDataVolume(dataset_id, timePeriod, dateFormat);
+            results.push(dataVolumeResult);
+        }
+
+        if (!category || category.includes("data_lineage")) {
+            const dataLineageResult = await getDataLineage(dataset_id, intervals);
+            results.push(dataLineageResult);
+        }
+
+        if (!category || category.includes("connectors")) {
+            const connectorsResult = await getConnectors(dataset_id, intervals);
+            results.push(connectorsResult);
+        }
+
+        if (!category || category.includes("data_quality")) {
+            const connectorsResult = await getDataQuality(dataset_id, intervals);
+            results.push(connectorsResult);
+        }
+
+        logger.info({ apiId, msgid, requestBody, datasetId: dataset_id, message: "Metrics fetched successfully" })
+        return ResponseHandler.successResponse(req, res, { status: 200, data: results });
+
     }
+    catch (error: any) {
+        logger.error({ apiId, msgid, requestBody: req?.body, datasetId: dataset_id, message: "Error while fetching metrics", code: "FAILED_TO_FETCH_METRICS", error });
+        return ResponseHandler.errorResponse({ message: "Error while fetching metrics", statusCode: 500, errCode: "FAILED", code: "FAILED_TO_FETCH_METRICS" }, req, res);
+    }
+
 }
 
 export default datasetMetrics;