azkaban-aplcache

diff --git a/azkaban-exec-server/src/main/java/azkaban/execapp/ExecMetrics.java b/azkaban-exec-server/src/main/java/azkaban/execapp/ExecMetrics.java
index 58aba26..c5185be 100644
--- a/azkaban-exec-server/src/main/java/azkaban/execapp/ExecMetrics.java
+++ b/azkaban-exec-server/src/main/java/azkaban/execapp/ExecMetrics.java
@@ -43,5 +43,7 @@ public class ExecMetrics {
         .addGauge("EXEC-NumRunningFlows", flowRunnerManager::getNumRunningFlows);
     this.metricsManager
         .addGauge("EXEC-NumQueuedFlows", flowRunnerManager::getNumQueuedFlows);
+    this.metricsManager
+        .addGauge("EXEC-ProjectDirCacheHitRatio", flowRunnerManager::getProjectDirCacheHitRatio);
   }
 }

diff --git a/azkaban-exec-server/src/main/java/azkaban/execapp/FlowPreparer.java b/azkaban-exec-server/src/main/java/azkaban/execapp/FlowPreparer.java
index 0e943c9..e79278e 100644
--- a/azkaban-exec-server/src/main/java/azkaban/execapp/FlowPreparer.java
+++ b/azkaban-exec-server/src/main/java/azkaban/execapp/FlowPreparer.java
@@ -59,6 +59,30 @@ public class FlowPreparer {
   private final File projectsDir;
   private final StorageManager storageManager;
   private final ProjectCacheDirCleaner projectDirCleaner;
+  private final ProjectsDirCacheMetrics cacheMetrics;
+
+  @VisibleForTesting
+  static class ProjectsDirCacheMetrics {
+
+    private long cacheHit;
+    private long cacheMiss;
+
+    /**
+     * @return hit ratio of project dirs cache
+     */
+    synchronized double getHitRatio() {
+      final long total = this.cacheHit + this.cacheMiss;
+      return total == 0 ? 0 : this.cacheHit * 1.0 / total;
+    }
+
+    synchronized void incrementCacheHit() {
+      this.cacheHit++;
+    }
+
+    synchronized void incrementCacheMiss() {
+      this.cacheMiss++;
+    }
+  }
 
   public FlowPreparer(final StorageManager storageManager, final File executionsDir,
       final File projectsDir, final Long projectDirMaxSizeInMb) {
@@ -66,7 +90,12 @@ public class FlowPreparer {
     this.executionsDir = executionsDir;
     this.projectsDir = projectsDir;
     this.projectDirCleaner = new ProjectCacheDirCleaner(projectDirMaxSizeInMb);
+    this.cacheMetrics = new ProjectsDirCacheMetrics();
+  }
+
 
+  public double getProjectDirCacheHitRatio() {
+    return this.cacheMetrics.getHitRatio();
   }
 
   /**
@@ -206,6 +235,7 @@ public class FlowPreparer {
     // If directory exists. Assume its prepared and skip.
     if (pv.getInstalledDir().exists()) {
       log.info("Project already cached. Skipping download. " + pv);
+      this.cacheMetrics.incrementCacheHit();
       touchIfExists(
           Paths.get(pv.getInstalledDir().getPath(), PROJECT_DIR_SIZE_FILE_NAME));
       return;
@@ -222,6 +252,7 @@ public class FlowPreparer {
     ProjectFileHandler projectFileHandler = null;
     try {
       log.info(String.format("Downloading zip file for Project Version {%s}", pv));
+      this.cacheMetrics.incrementCacheMiss();
       projectFileHandler = requireNonNull(
           this.storageManager.getProjectFile(pv.getProjectId(), pv.getVersion()));
       checkState("zip".equals(projectFileHandler.getFileType()));

diff --git a/azkaban-exec-server/src/main/java/azkaban/execapp/FlowRunnerManager.java b/azkaban-exec-server/src/main/java/azkaban/execapp/FlowRunnerManager.java
index 5b7bd5c..3111c3f 100644
--- a/azkaban-exec-server/src/main/java/azkaban/execapp/FlowRunnerManager.java
+++ b/azkaban-exec-server/src/main/java/azkaban/execapp/FlowRunnerManager.java
@@ -202,6 +202,10 @@ public class FlowRunnerManager implements EventListener,
     this.cleanerThread.start();
   }
 
+  public double getProjectDirCacheHitRatio() {
+    return this.flowPreparer.getProjectDirCacheHitRatio();
+  }
+
   /**
    * Setting the gid bit on the execution directory forces all files/directories created within the
    * directory to be a part of the group associated with the azkaban process. Then, when users

diff --git a/azkaban-exec-server/src/test/java/azkaban/execapp/FlowPreparerTest.java b/azkaban-exec-server/src/test/java/azkaban/execapp/FlowPreparerTest.java
index 640e203..ca47350 100644
--- a/azkaban-exec-server/src/test/java/azkaban/execapp/FlowPreparerTest.java
+++ b/azkaban-exec-server/src/test/java/azkaban/execapp/FlowPreparerTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import azkaban.execapp.FlowPreparer.ProjectsDirCacheMetrics;
 import azkaban.executor.ExecutableFlow;
 import azkaban.project.ProjectFileHandler;
 import azkaban.storage.StorageManager;
@@ -183,4 +184,34 @@ public class FlowPreparerTest {
     assertThat(this.projectsDir.listFiles()).containsExactlyInAnyOrder(expectedRemainingFiles
         .toArray(new File[expectedRemainingFiles.size()]));
   }
+
+  @Test
+  public void testProjectsCacheMetricsZeroHit() {
+    //given
+    final FlowPreparer.ProjectsDirCacheMetrics cacheMetrics = new ProjectsDirCacheMetrics();
+
+    //when zero hit and zero miss then
+    assertThat(cacheMetrics.getHitRatio()).isEqualTo(0);
+
+    //when
+    cacheMetrics.incrementCacheMiss();
+    //then
+    assertThat(cacheMetrics.getHitRatio()).isEqualTo(0);
+  }
+
+  @Test
+  public void testProjectsCacheMetricsHit() {
+    //given
+    final FlowPreparer.ProjectsDirCacheMetrics cacheMetrics = new ProjectsDirCacheMetrics();
+
+    //when one hit
+    cacheMetrics.incrementCacheHit();
+    //then
+    assertThat(cacheMetrics.getHitRatio()).isEqualTo(1);
+
+    //when one miss
+    cacheMetrics.incrementCacheMiss();
+    //then
+    assertThat(cacheMetrics.getHitRatio()).isEqualTo(0.5);
+  }
 }