|
| 1 | +const logger = require("../config/logger"); |
| 2 | +const models = require("../models"); |
| 3 | +const directoryMonitoring = models.directoryMonitoring; |
| 4 | +const teamsWebhooks = models.teams_hook; |
| 5 | +const notification = models.notification_queue; |
| 6 | +const hpccUtil = require("../utils/hpcc-util"); |
| 7 | +const { v4: uuidv4 } = require("uuid"); |
| 8 | +const wildCardStringMatch = require("../utils/wildCardStringMatch"); |
| 9 | +const { parentPort, workerData } = require("worker_threads"); |
| 10 | +const moment = require("moment"); |
| 11 | + |
| 12 | +(async () => { |
| 13 | + try { |
| 14 | + //grab all directory monitoring that are active |
| 15 | + |
| 16 | + const directoryMonitoringDetails = await directoryMonitoring.findOne({ |
| 17 | + where: { |
| 18 | + id: workerData.directoryMonitoring_id, |
| 19 | + active: true, |
| 20 | + }, |
| 21 | + }); |
| 22 | + |
| 23 | + if (!directoryMonitoringDetails) return; |
| 24 | + |
| 25 | + let { |
| 26 | + id: directorymonitoring_id, |
| 27 | + cluster_id, |
| 28 | + directory, |
| 29 | + metaData, |
| 30 | + metaData: { |
| 31 | + machine: Netaddr, |
| 32 | + pattern: fileNameWildCard, |
| 33 | + landingZone, |
| 34 | + lastMonitored, |
| 35 | + currentlyMonitoring, |
| 36 | + monitoringCondition: { |
| 37 | + notifyCondition, |
| 38 | + threshold, |
| 39 | + fileDetected, |
| 40 | + maximumFileCount, |
| 41 | + minimumFilecount, |
| 42 | + }, |
| 43 | + notifications, |
| 44 | + }, |
| 45 | + } = directoryMonitoringDetails; |
| 46 | + |
| 47 | + let currentTimeStamp = moment.utc().valueOf(); |
| 48 | + |
| 49 | + const Path = `/var/lib/HPCCSystems/${landingZone}/${directory}/`; |
| 50 | + |
| 51 | + const result = await hpccUtil.getDirectories({ |
| 52 | + clusterId: cluster_id, |
| 53 | + Netaddr, |
| 54 | + Path, |
| 55 | + DirectoryOnly: false, |
| 56 | + }); |
| 57 | + let files = result.filter((item) => !item.isDir); |
| 58 | + |
| 59 | + const newFilesToMonitor = []; |
| 60 | + |
| 61 | + // Notification Details |
| 62 | + let emailNotificationDetails; |
| 63 | + let teamsNotificationDetails; |
| 64 | + for (let notification of notifications) { |
| 65 | + if (notification.channel === "eMail") { |
| 66 | + emailNotificationDetails = notification; |
| 67 | + } |
| 68 | + if (notification.channel === "msTeams") { |
| 69 | + teamsNotificationDetails = notification; |
| 70 | + } |
| 71 | + } |
| 72 | + |
| 73 | + let newNotificationDetails = []; |
| 74 | + |
| 75 | + //check if number of files in the directory is within the range |
| 76 | + if (files.length < minimumFilecount) { |
| 77 | + newNotificationDetails.push({ |
| 78 | + value: "file_count_below_minimum", |
| 79 | + title: `File count below minimum in ${directory.join("/")}`, |
| 80 | + text: `Number of files in ${directory.join( |
| 81 | + "/" |
| 82 | + )} is below the minimum file count of ${minimumFilecount}`, |
| 83 | + details: { |
| 84 | + "Landing zone": landingZone, |
| 85 | + Directory: directory.join("/"), |
| 86 | + "File count": files.length, |
| 87 | + }, |
| 88 | + }); |
| 89 | + } |
| 90 | + |
| 91 | + if (files.length > maximumFileCount) { |
| 92 | + newNotificationDetails.push({ |
| 93 | + value: "file_count_above_maximum", |
| 94 | + title: `File count above maximum in ${directory.join("/")}`, |
| 95 | + text: `Number of files in ${directory.join( |
| 96 | + "/" |
| 97 | + )} is above the maximum file count of ${maximumFileCount}`, |
| 98 | + details: { |
| 99 | + "Landing zone": landingZone, |
| 100 | + Directory: directory.join("/"), |
| 101 | + "File count": files.length, |
| 102 | + }, |
| 103 | + }); |
| 104 | + } |
| 105 | + |
| 106 | + //iterate through files, check for notification parameters and update each file if necessary |
| 107 | + for (let i = 0; i < files.length; i++) { |
| 108 | + let { name: fileName, modifiedtime } = files[i]; |
| 109 | + |
| 110 | + let fileModifiedTime = moment(modifiedtime); |
| 111 | + |
| 112 | + fileModifiedTime = fileModifiedTime.utc().valueOf(); |
| 113 | + |
| 114 | + //check if file is new |
| 115 | + if ( |
| 116 | + lastMonitored < fileModifiedTime && |
| 117 | + wildCardStringMatch(fileNameWildCard, fileName) && |
| 118 | + //file name not in the currently monitoring array |
| 119 | + !currentlyMonitoring.find( |
| 120 | + (item) => |
| 121 | + item.name === fileName && item.modifiedTime === fileModifiedTime |
| 122 | + ) |
| 123 | + ) { |
| 124 | + //Check if user wants to be notified when new file arrives |
| 125 | + let notificationDetail; |
| 126 | + if (fileDetected) { |
| 127 | + notificationDetail = { |
| 128 | + name: fileName, |
| 129 | + value: "file_detected", |
| 130 | + title: `New file uploaded to ${directory}`, |
| 131 | + text: "Details about recently added file - ", |
| 132 | + details: { |
| 133 | + "File Name": fileName, |
| 134 | + "Landing zone": landingZone, |
| 135 | + Directory: directory, |
| 136 | + "File detected at": new Date(fileModifiedTime).toString(), |
| 137 | + }, |
| 138 | + }; |
| 139 | + } |
| 140 | + |
| 141 | + if (notificationDetail) { |
| 142 | + newNotificationDetails.push(notificationDetail); |
| 143 | + } |
| 144 | + |
| 145 | + // Start monitoring new file if threshold parameter is set |
| 146 | + if (notifyCondition.includes("fileNotMoving")) { |
| 147 | + newFilesToMonitor.push({ |
| 148 | + name: fileName, |
| 149 | + modifiedTime: fileModifiedTime, |
| 150 | + threshold: currentTimeStamp + threshold * 60 * 1000, |
| 151 | + notified: [], |
| 152 | + }); |
| 153 | + } |
| 154 | + } |
| 155 | + } |
| 156 | + |
| 157 | + if (currentlyMonitoring.length > 0) { |
| 158 | + //remove files that have been moved out by checking against files array |
| 159 | + currentlyMonitoring = currentlyMonitoring.filter((current) => { |
| 160 | + const { name, modifiedTime } = current; |
| 161 | + return !files.find( |
| 162 | + (file) => file.name === name && file.modifiedtime === modifiedTime |
| 163 | + ); |
| 164 | + }); |
| 165 | + } |
| 166 | + |
| 167 | + // check for threshold |
| 168 | + currentlyMonitoring.forEach((current) => { |
| 169 | + const { notified } = current; |
| 170 | + |
| 171 | + const pastExpectedMoveTime = current.threshold < currentTimeStamp; |
| 172 | + |
| 173 | + if (pastExpectedMoveTime && !notified.length) { |
| 174 | + newNotificationDetails.push({ |
| 175 | + name: current.name, |
| 176 | + value: "file_not_moving", |
| 177 | + title: `${current.name} stuck at ${directory}`, |
| 178 | + text: `${current.name} has been stuck at ${directory} longer than ${threshold} minutes`, |
| 179 | + details: { |
| 180 | + "File Name": current.name, |
| 181 | + "Landing zone": landingZone, |
| 182 | + Directory: directory, |
| 183 | + "File received at": new Date(current.modifiedTime).toString(), |
| 184 | + "Expected move time": new Date(current.threshold).toString(), |
| 185 | + }, |
| 186 | + }); |
| 187 | + } |
| 188 | + }); |
| 189 | + |
| 190 | + console.log("newNotificationDetails"); |
| 191 | + console.log(newNotificationDetails); |
| 192 | + |
| 193 | + //send notifications if necessary |
| 194 | + for (let notificationDetail of newNotificationDetails) { |
| 195 | + //send email notification |
| 196 | + if (emailNotificationDetails) { |
| 197 | + //TODO: create notification queue with proper template in next isue |
| 198 | + logger.verbose("Email notification sent: " + newNotificationDetails); |
| 199 | + |
| 200 | + //once notification is sent, update currently monitoring notified field |
| 201 | + currentlyMonitoring = currentlyMonitoring.map((item) => { |
| 202 | + if (notificationDetail.name === item.name) { |
| 203 | + return { |
| 204 | + ...item, |
| 205 | + notified: [ |
| 206 | + { |
| 207 | + notified: true, |
| 208 | + method: "email", |
| 209 | + dateNotified: currentTimeStamp, |
| 210 | + }, |
| 211 | + ], |
| 212 | + }; |
| 213 | + } |
| 214 | + return item; |
| 215 | + }); |
| 216 | + } |
| 217 | + |
| 218 | + //send teams notification |
| 219 | + if (teamsNotificationDetails) { |
| 220 | + //TODO: create notification queue with proper template in next isue |
| 221 | + logger.verbose("Teams notification sent: " + newNotificationDetails); |
| 222 | + |
| 223 | + //once notification is sent, update currently monitoring notified field |
| 224 | + currentlyMonitoring = currentlyMonitoring.map((item) => { |
| 225 | + if (notificationDetail.name === item.name) { |
| 226 | + return { |
| 227 | + ...item, |
| 228 | + notified: [ |
| 229 | + { |
| 230 | + notified: true, |
| 231 | + method: "msTeams", |
| 232 | + dateNotified: currentTimeStamp, |
| 233 | + }, |
| 234 | + ], |
| 235 | + }; |
| 236 | + } |
| 237 | + return item; |
| 238 | + }); |
| 239 | + } |
| 240 | + } |
| 241 | + |
| 242 | + //update directory monitoring |
| 243 | + metaData.lastMonitored = currentTimeStamp; |
| 244 | + metaData.currentlyMonitoring = [ |
| 245 | + ...currentlyMonitoring, |
| 246 | + ...newFilesToMonitor, |
| 247 | + ]; |
| 248 | + |
| 249 | + await directoryMonitoring.update( |
| 250 | + { metaData }, |
| 251 | + { where: { id: directorymonitoring_id } } |
| 252 | + ); |
| 253 | + } catch (error) { |
| 254 | + logger.error("Error while running Directory Monitoring Jobs: " + error); |
| 255 | + } |
| 256 | +})(); |
0 commit comments