1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.fs.storage.common.utils
10 
11 import com.github.benmanes.caffeine.cache.{CacheLoader, Caffeine}
12 import org.apache.hadoop.fs.{FileContext, FileStatus, Path, RemoteIterator}
13 import org.locationtech.geomesa.utils.conf.GeoMesaSystemProperties.SystemProperty
14 
15 import java.util.concurrent.TimeUnit
16 
17 /**
18   * Caches file statuses to avoid repeated file system operations. Status expires after a
19   * configurable period, by default 10 minutes.
20   */
21 object PathCache {
22 
23   val CacheDurationProperty: SystemProperty = SystemProperty("geomesa.fs.file.cache.duration", "15 minutes")
24 
25   private val duration = CacheDurationProperty.toDuration.get.toMillis
26 
27   // cache for checking existence of files
28   private val pathCache =
29     Caffeine.newBuilder().expireAfterWrite(duration, TimeUnit.MILLISECONDS).build(
30       new CacheLoader[(FileContext, Path), java.lang.Boolean]() {
31         override def load(key: (FileContext, Path)): java.lang.Boolean = key._1.util.exists(key._2)
32       }
33     )
34 
35   // cache for individual file status
36   private val statusCache =
37     Caffeine.newBuilder().expireAfterWrite(duration, TimeUnit.MILLISECONDS).build(
38       new CacheLoader[(FileContext, Path), FileStatus]() {
39         override def load(key: (FileContext, Path)): FileStatus = key._1.getFileStatus(key._2)
40       }
41     )
42 
43   // cache for checking directory contents
44   private val listCache =
45     Caffeine.newBuilder().expireAfterWrite(duration, TimeUnit.MILLISECONDS).build(
46       new CacheLoader[(FileContext, Path), Stream[FileStatus]]() {
47         override def load(key: (FileContext, Path)): Stream[FileStatus] =
48           RemoteIterator(key._1.listStatus(key._2)).toStream
49       }
50     )
51 
52   /**
53     * * Register a path as existing
54     *
55     * @param fc file context
56     * @param path path
57     * @param status file status, if available
58     * @param list directory contents, if available
59     */
60   def register(
61       fc: FileContext,
62       path: Path,
63       status: Option[FileStatus] = None,
64       list: Option[Stream[FileStatus]] = None): Unit = {
65     pathCache.put((fc, path), java.lang.Boolean.TRUE)
66     status.foreach(statusCache.put((fc, path), _))
67     list.foreach(listCache.put((fc, path), _))
68   }
69 
70   /**
71     * Check to see if a path exists
72     *
73     * @param fc file context
74     * @param path path
75     * @param reload reload the file status from the underlying file system before checking
76     * @return
77     */
78   def exists(fc: FileContext, path: Path, reload: Boolean = false): Boolean = {
79     if (reload) {
80       invalidate(fc, path)
81     }
82     pathCache.get((fc, path)).booleanValue()
83   }
84 
85   /**
86     * Gets the file status for a path
87     *
88     * @param fc file context
89     * @param path path
90     * @return
91     */
92   def status(fc: FileContext, path: Path, reload: Boolean = false): FileStatus = {
93     if (reload) {
94       invalidate(fc, path)
95     }
96     statusCache.get((fc, path))
97   }
98 
99   /**
100     * List the children of a path
101     *
102     * @param fc file context
103     * @param dir directory path
104     * @return
105     */
106   def list(fc: FileContext, dir: Path, reload: Boolean = false): Iterator[FileStatus] = {
107     if (reload) {
108       invalidate(fc, dir)
109     }
110     listCache.get((fc, dir)).iterator
111   }
112 
113   /**
114     * Invalidate any cached values for the path - they will be re-loaded on next access
115     *
116     * @param fc file context
117     * @param path path
118     */
119   def invalidate(fc: FileContext, path: Path): Unit =
120     Seq(pathCache, statusCache, listCache).foreach(_.invalidate((fc, path)))
121 
122   object RemoteIterator {
123     def apply[T](iter: RemoteIterator[T]): Iterator[T] = new Iterator[T] {
124       override def hasNext: Boolean = iter.hasNext
125       override def next(): T = iter.next
126     }
127   }
128 }