Merge pull request #358 from Sanketika-Obsrv/release-1.9.0

Release 1.9.0 to main

Author: manjudr

api-service/src/configs/Config.ts

@@ -144,5 +144,8 @@ export const config = {
   },
   "alerts_rules": {
     "config_path": process.env.alerts_config_path
-  }
+  },
+  "dataset_filter_config": {
+    "status_filter_limit": process.env.status_filter_limit ? parseInt(process.env.status_filter_limit) : 10 // Maximum number of filters allowed in a dataset
+  } 
 }

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

@@ -5,16 +5,16 @@ import logger from "../../logger";
 import { schemaValidation } from "../../services/ValidationService";
 import DatasetCreate from "./DatasetListValidationSchema.json";
 import { ResponseHandler } from "../../helpers/ResponseHandler";
-import { datasetService } from "../../services/DatasetService";
+import { attachDraftConnectors, attachLiveConnectors, datasetService } from "../../services/DatasetService";
 import { obsrvError } from "../../types/ObsrvError";
-import { Dataset } from "../../models/Dataset";
-import { Datasource } from "../../models/Datasource";
+import { config } from "../../configs/Config";
 
 export const apiId = "api.datasets.list"
 export const errorCode = "DATASET_LIST_FAILURE"
 const liveDatasetStatus = ["Live", "Retired", "Purged"]
 const draftDatasetStatus = ["Draft", "ReadyToPublish"]
 const defaultFields = ["dataset_id", "name", "type", "status", "tags", "version", "api_version", "dataset_config", "created_date", "updated_date"]
+const MAX_STATUS_ARRAY_SIZE = config.dataset_filter_config.status_filter_limit || 10;
 
 const datasetList = async (req: Request, res: Response) => {
 
@@ -35,11 +35,19 @@ const listDatasets = async (request: Record<string, any>): Promise<Record<string
 
     const { filters = {} } = request || {};
     const datasetStatus = _.get(filters, "status");
+    const connectorFilter = _.get(filters, "connectors");
     const status = _.isArray(datasetStatus) ? datasetStatus : _.compact([datasetStatus])
-    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.getLiveDatasets(liveFilters, defaultFields)
-    const draftDatasetList = await datasetService.findDraftDatasets(draftFilters, [...defaultFields, "data_schema", "validation_config", "dedup_config", "denorm_config", "connectors_config", "version_key"], [["updated_date", "DESC"]]);
+    if (status.length > MAX_STATUS_ARRAY_SIZE) {
+        throw obsrvError("", "DATASET_LIST_INPUT_INVALID", "Status filter array length exceeds the allowed limit", "BAD_REQUEST", 400);
+    }
+    const draftFilters = _.omit(_.set(_.cloneDeep(filters), "status", _.isEmpty(status) ? draftDatasetStatus : _.intersection(status, draftDatasetStatus)), "connectors");
+    const liveFilters = _.omit(_.set(_.cloneDeep(filters), "status", _.isEmpty(status) ? liveDatasetStatus : _.intersection(status, liveDatasetStatus)), "connectors");
+    let liveDatasetList = await datasetService.getLiveDatasets(liveFilters, defaultFields)
+    let draftDatasetList = await datasetService.findDraftDatasets(draftFilters, [...defaultFields, "data_schema", "validation_config", "dedup_config", "denorm_config", "connectors_config", "version_key"], [["updated_date", "DESC"]]);
+    if(connectorFilter && !_.isEmpty(connectorFilter)) {
+        liveDatasetList = await attachLiveConnectors(liveDatasetList, connectorFilter);
+        draftDatasetList = await attachDraftConnectors(draftDatasetList, connectorFilter);
+    }
     return _.compact(_.concat(liveDatasetList, draftDatasetList));
 }
 

api-service/src/controllers/DatasetList/DatasetListValidationSchema.json

@@ -45,6 +45,20 @@
             "type": {
               "type": "string",
               "enum": ["event", "transaction", "master"]
+            },
+            "connectors": {
+              "anyOf": [
+                {
+                  "type": "array",
+                  "items": {
+                    "type": "string"
+                  }
+                },
+                {
+                  "type": "string",
+                  "default": "all"
+                }
+              ]
             }
           },
           "additionalProperties": false

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

@@ -82,7 +82,7 @@ const readDataset = async (datasetId: string, attributes: string[]): Promise<any
     const api_version = _.get(dataset, "api_version")
     const datasetConfigs: any = {}
     const transformations_config = await datasetService.getTransformations(datasetId, ["field_key", "transformation_function", "mode", "metadata"])
-    const datasourceConfig = await Datasource.findOne({ where: { dataset_id: datasetId, is_primary: true }, attributes: ["datasource"], raw: true })
+    const datasourceConfig = await Datasource.findOne({ where: { dataset_id: datasetId, is_primary: true, type: "druid" }, attributes: ["datasource"], raw: true })
     datasetConfigs["alias"] = _.get(datasourceConfig, "datasource")
     if (api_version !== "v2") {
         datasetConfigs["transformations_config"] = _.map(transformations_config, (config) => {

api-service/src/models/ConnectorInstances.ts

@@ -12,10 +12,6 @@ export const ConnectorInstances = sequelize.define("connector_instances", {
     connector_id: {
         type: DataTypes.STRING
     },
-    data_format: {
-        type: DataTypes.STRING,
-        defaultValue: "json"
-    },
     connector_config: {
         type: DataTypes.STRING
     },

api-service/src/services/DatasetService.ts

@@ -22,6 +22,7 @@ import { config } from "../configs/Config";
 import { Op } from "sequelize";
 import TableDraft from "../models/Table";
 import { alertService } from "./AlertManagerService";
+import { ConnectorRegistry } from "../models/ConnectorRegistry";
 
 class DatasetService {
 
@@ -144,7 +145,7 @@ class DatasetService {
                 {
                     model: Datasource,
                     attributes: ['datasource'],
-                    where: { is_primary: true },
+                    where: { is_primary: true, type: "druid" },
                     required: false
                 },
             ], raw: true, where: filters, attributes, order: [["updated_date", "DESC"]]
@@ -510,4 +511,102 @@ export const validateStorageSupport = (dataset: Record<string, any>) => {
     }
 }
 
+export const attachDraftConnectors = async (
+    draftDatasetList: Record<string, any>[],
+    connectorFilter: string | string[]
+): Promise<Record<string, any>[]> => {
+    if (_.isEmpty(draftDatasetList)) {
+        return [];
+    }
+
+    const connectorIds = _.uniq(
+        _.flatMap(draftDatasetList, dataset =>
+            _.map(dataset.connectors_config, 'connector_id')
+        )
+    );
+
+    if (_.isEmpty(connectorIds)) {
+        return draftDatasetList.map(dataset => ({
+            ...dataset,
+            connectors_config: []
+        }));
+    }
+
+    const connectorRegistry: any = await ConnectorRegistry.findAll({
+        where: { id: connectorIds },
+        raw: true,
+        attributes: ['id', 'name', 'category', 'type']
+    });
+
+    return draftDatasetList.map(dataset => {
+        let filteredConnectors = dataset.connectors_config;
+        if (connectorFilter !== 'all') {
+            const filterArray = _.castArray(connectorFilter); // Ensure it's an array
+            filteredConnectors = _.filter(filteredConnectors, connector =>
+                filterArray.includes(String(connector.connector_id))
+            );
+        }
+
+        const enrichedConnectors = filteredConnectors.map((connector: any) => {
+            const registryDetails = _.find(connectorRegistry, {
+                id: connector.connector_id
+            });
+            return {
+                ...connector,
+                name: registryDetails?.name || null,
+                category: registryDetails?.category || null,
+                source: registryDetails?.type || null
+            };
+        });
+
+        return {
+            ...dataset,
+            connectors_config: enrichedConnectors
+        };
+    });
+};
+
+export const attachLiveConnectors = async (
+    liveDatasetList: Record<string, any>,
+    connectorFilter: string | string[]
+): Promise<Record<string, any>[]> => {
+    if (_.isEmpty(liveDatasetList)) {
+        return [];
+    }
+
+    ConnectorRegistry.hasMany(ConnectorInstances, { foreignKey: 'connector_id' });
+    const connectorRegistry = await ConnectorRegistry.findAll({
+        include: [{
+            model: ConnectorInstances,
+            attributes: ['dataset_id', 'connector_id'],
+            required: true
+        }],
+        raw: true,
+        attributes: ['id', 'name', 'category', 'type']
+    });
+
+    const filterArray = connectorFilter === 'all' ? null : _.castArray(connectorFilter);
+
+    return liveDatasetList.map((dataset: Record<string, any>) => {
+        const datasetId = dataset.dataset_id;
+
+        const filteredConnectors = connectorRegistry.filter((connector: any) =>
+            connector['connector_instances.dataset_id'] === datasetId &&
+            (!filterArray || filterArray.includes(String(connector.id)))
+        );
+
+        const enrichedConnectors = filteredConnectors.map((connector: any) => ({
+            connector_id: connector.id,
+            name: connector.name,
+            category: connector.category,
+            type: connector.type
+        }));
+
+        return {
+            ...dataset,
+            connectors_config: enrichedConnectors
+        };
+    });
+};
+
 export const datasetService = new DatasetService();

api-service/src/services/TableGenerator.ts

@@ -171,6 +171,7 @@ class TableGenerator extends BaseTableGenerator {
             case "boolean": return "string";
             case "array": return "json";
             case "string": return "string";
+            case "double": return "double";
             default: return "auto";
         }
     }