Merge pull request #367 from Sanketika-Obsrv/api-svc-fixes

#OI-I3: Fix: Mandatory header required during azure blob upload

Author: manjudr

api-service/src/configs/Config.ts

@@ -14,7 +14,7 @@ const getCloudConfigs = () => {
     return parsedCloudConfig;
   }
   return {};
-} 
+}
 
 
 export const config = {
@@ -30,8 +30,8 @@ export const config = {
       "native_query_path": "/druid/v2",
       "list_datasources_path": "/druid/v2/datasources",
       "submit_ingestion": "druid/indexer/v1/supervisor",
-      "username": process.env.druid_username || "admin",  
-      "password": process.env.druid_password || "admin123" 
+      "username": process.env.druid_username || "admin",
+      "password": process.env.druid_password || "admin123"
     },
     "prometheus": {
       "url": process.env.prometheus_url || "http://localhost:9090"
@@ -102,6 +102,7 @@ export const config = {
     "maxQueryDateRange": process.env.exhaust_query_range ? parseInt(process.env.exhaust_query_range) : 31, // in days. Defines the maximum no. of days the files can be fetched
     "exclude_exhaust_types": process.env.exclude_exhaust_types ? process.env.exclude_exhaust_types.split(",") : ["system-stats", "masterdata-system-stats", "system-events",], // list of folder type names to skip exhaust service
     "telemetry_data_path": process.env.telemetry_data_path || "telemetry-data",
+    "azure_blob_type": process.env.azure_blob_type || "BlockBlob"
   },
   "template_config": {
     "template_required_variables": process.env.template_required_vars ? process.env.template_required_vars.split(",") : ["DATASET", "STARTDATE", "ENDDATE"],
@@ -149,5 +150,5 @@ export const config = {
   },
   "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/ConnectorRegister/ConnectorRegisterController.ts

@@ -9,6 +9,7 @@ import { PassThrough } from "stream";
 import { registerConnector } from "../../connections/commandServiceConnection";
 import { generatePreSignedUrl } from "../GenerateSignedURL/helper";
 import { obsrvError } from "../../types/ObsrvError";
+import { config } from "../../configs/Config";
 
 export const apiId = "api.connector.register";
 export const code = "FAILED_TO_REGISTER_CONNECTOR";
@@ -25,11 +26,11 @@ const connectorRegisterController = async (req: Request, res: Response) => {
         logger.info({ apiId, resmsgid, message: `File uploaded to cloud provider successfully` })
         const downloadUrls = await generatePreSignedUrl("read", [payload.relative_path], "connector")
         const urlPayload = {
-            download_url : _.get(downloadUrls, [0, "preSignedUrl"]),
+            download_url: _.get(downloadUrls, [0, "preSignedUrl"]),
             file_name: _.get(downloadUrls, [0, "fileName"])
         }
-        if(!urlPayload.download_url){            
-            throw obsrvError("", "SIGNED_URL_NOT_FOUND",`Failed to generate signed url for path ${payload.relative_path}`,  "BAD_REQUEST", 400)
+        if (!urlPayload.download_url) {
+            throw obsrvError("", "SIGNED_URL_NOT_FOUND", `Failed to generate signed url for path ${payload.relative_path}`, "BAD_REQUEST", 400)
         }
         const userToken = req.get('authorization') as string;
         const registryResponse = await registerConnector(urlPayload, userToken);
@@ -76,11 +77,18 @@ const uploadStream = async (req: Request) => {
                     const pass = new PassThrough();
                     file.pipe(pass);
                     const fileBuffer = await streamToBuffer(pass);
+
+                    const uploadHeaders: any = {
+                        "Content-Type": info.mimeType,
+                        "Content-Length": fileBuffer.length,
+                    };
+
+                    if (config.cloud_config.cloud_storage_provider === "azure") {
+                        uploadHeaders["x-ms-blob-type"] = config.cloud_config.azure_blob_type;
+                    }
+
                     await axios.put(preSignedUrl[0]?.preSignedUrl, fileBuffer, {
-                        headers: {
-                            "Content-Type": info.mimeType,
-                            "Content-Length": fileBuffer.length,
-                        }
+                        headers: uploadHeaders
                     });
                 }
                 catch (err) {