[airavata-data-lake] branch master updated: extend Directory monitor

[isjarana]

This is an automated email from the ASF dual-hosted git repository.

isjarana pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-data-lake.git


The following commit(s) were added to refs/heads/master by this push:
     new a9ccbe1  extend Directory monitor
     new 613764b  Merge pull request #194 from isururanawaka/mysql_conversion
a9ccbe1 is described below

commit a9ccbe181c71d98b596804d8f42f6de075f40782
Author: Isuru Ranawaka <irjan...@gmail.com>
AuthorDate: Wed Nov 30 18:57:25 2022 -0500

    extend Directory monitor
---
 .../src/main/dist/conf/application.properties                |  3 ++-
 .../apache/airavata/datalake/dmonitor/DirectoryMonitor.java  |  5 ++++-
 .../apache/airavata/datalake/dmonitor/SaturationGauge.java   | 12 +++++++++++-
 3 files changed, 17 insertions(+), 3 deletions(-)

diff --git 
a/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/dist/conf/application.properties
 
b/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/dist/conf/application.properties
index 67b7778..5a36588 100644
--- 
a/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/dist/conf/application.properties
+++ 
b/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/dist/conf/application.properties
@@ -25,4 +25,5 @@ event.type=MODIFY
 auth.token=
 kafka.url=
 kafka.publisher.id=evenpub
-kafka.event.topic=data-orchestrator-file-events
\ No newline at end of file
+kafka.event.topic=data-orchestrator-file-events
+saturation.count=3
\ No newline at end of file
diff --git 
a/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/DirectoryMonitor.java
 
b/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/DirectoryMonitor.java
index fe8a7cb..13d4bd8 100644
--- 
a/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/DirectoryMonitor.java
+++ 
b/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/DirectoryMonitor.java
@@ -65,6 +65,9 @@ public class DirectoryMonitor implements CommandLineRunner {
     @org.springframework.beans.factory.annotation.Value("${kafka.event.topic}")
     private String kafkaEventTopic;
 
+    @org.springframework.beans.factory.annotation.Value("${saturation.count}")
+    private int saturationCount;
+
     private Map<WatchKey, Path> watchKeyPathMap = new HashMap<>();
 
     public static void main(String args[]) {
@@ -104,7 +107,7 @@ public class DirectoryMonitor implements CommandLineRunner {
 
         SaturationGauge saturationGauge = new SaturationGauge();
         saturationGauge.start(eventNotifier);
-
+        saturationGauge.setSaturationCount(saturationCount);
         WatchService watchService = FileSystems.getDefault().newWatchService();
         Path base = Paths.get(basePath);
         watchKeyPathMap.put(base.register(watchService, 
StandardWatchEventKinds.ENTRY_CREATE), base);
diff --git 
a/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/SaturationGauge.java
 
b/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/SaturationGauge.java
index b409d17..342693d 100644
--- 
a/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/SaturationGauge.java
+++ 
b/data-orchestrator/data-orchestrator-clients/directory-monitor/src/main/java/org/apache/airavata/datalake/dmonitor/SaturationGauge.java
@@ -35,6 +35,8 @@ public class SaturationGauge {
     final ScheduledExecutorService scheduledExecutorService = 
Executors.newSingleThreadScheduledExecutor();
     final ExecutorService monitoringService = Executors.newFixedThreadPool(10);
 
+    private int saturationCount = 3;
+
     public void start(EventNotifier eventNotifier) {
         scheduledExecutorService.scheduleWithFixedDelay(() -> {
             List<Future<Boolean>> submitFutures = new ArrayList<>();
@@ -49,7 +51,7 @@ public class SaturationGauge {
                         directorySizes.put(key, newSize);
                         logger.info("Directory : " + key + " Size : " + 
newSize + " Scan count : " + monitorCount.get(key));
 
-                        if (oldSize == newSize && monitorCount.get(key) > 3) {
+                        if (oldSize == newSize && monitorCount.get(key) > 
saturationCount) {
                             logger.info("Directory " + key + " is saturated. 
Final size " + oldSize);
                             monitorCount.remove(key);
                             directorySizes.remove(key);
@@ -108,4 +110,12 @@ public class SaturationGauge {
         }
         return length;
     }
+
+    public int getSaturationCount() {
+        return saturationCount;
+    }
+
+    public void setSaturationCount(int saturationCount) {
+        this.saturationCount = saturationCount;
+    }
 }