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.data
10 
11 import com.github.benmanes.caffeine.cache.{CacheLoader, Caffeine}
12 import com.typesafe.scalalogging.LazyLogging
13 import org.apache.hadoop.conf.Configuration
14 import org.apache.hadoop.fs.{FileContext, Path}
15 import org.locationtech.geomesa.fs.storage.api._
16 import org.locationtech.geomesa.fs.storage.common.utils.PathCache
17 import org.locationtech.geomesa.utils.io.CloseQuietly
18 import org.locationtech.geomesa.utils.stats.MethodProfiling
19 
20 import java.util.concurrent.ConcurrentHashMap
21 import scala.util.control.NonFatal
22 
23 /**
24   * Manages the storages and associated simple feature types underneath a given path
25   *
26   * @param fc file context
27   * @param conf configuration
28   * @param root root path for the data store
29   */
30 class FileSystemStorageManager private (fc: FileContext, conf: Configuration, root: Path, namespace: Option[String])
31     extends MethodProfiling with LazyLogging {
32 
33   import scala.collection.JavaConverters._
34 
35   private val cache = new ConcurrentHashMap[String, (Path, FileSystemStorage)]().asScala
36 
37   /**
38     * Gets the storage associated with the given simple feature type, if any
39     *
40     * @param typeName simple feature type name
41     * @return
42     */
43   def storage(typeName: String): Option[FileSystemStorage] = {
44     cache.get(typeName).map(_._2) // check cached values
45         .orElse(Some(defaultPath(typeName)).filter(PathCache.exists(fc, _)).flatMap(loadPath)) // check expected (default) path
46         .orElse(loadAll().find(_.metadata.sft.getTypeName == typeName)) // check other paths until we find it
47   }
48 
49   /**
50     * Gets all storages under the root path
51     *
52     * @return
53     */
54   def storages(): Seq[FileSystemStorage] = {
55     loadAll().foreach(_ => ()) // force loading of everything
56     cache.map { case (_, (_, storage)) => storage }.toSeq
57   }
58 
59   /**
60     * Caches a storage instance for future use. Avoids loading it a second time if referenced later.
61     *
62     * @param path path for the storage
63     * @param storage storage instance
64     */
65   def register(path: Path, storage: FileSystemStorage): Unit =
66     cache.put(storage.metadata.sft.getTypeName, (path, storage))
67 
68   /**
69     * Default path for a given simple feature type name. Generally the simple feature type will go under
70     * a folder with the type name, but this is not required
71     *
72     * @param typeName simple feature type name
73     * @return
74     */
75   def defaultPath(typeName: String): Path = new Path(root, typeName)
76 
77   /**
78     * Loads all storages under this root (if they aren't already loaded)
79     *
80     * @return
81     */
82   private def loadAll(): Iterator[FileSystemStorage] = {
83     if (!PathCache.exists(fc, root)) { Iterator.empty } else {
84       val dirs = PathCache.list(fc, root).filter(_.isDirectory).map(_.getPath)
85       dirs.filterNot(path => cache.exists { case (_, (p, _)) => p == path }).flatMap(loadPath)
86     }
87   }
88 
89   /**
90     * Attempt to load a storage under the given root path. Requires an appropriate storage implementation
91     * to be available on the classpath.
92     *
93     * @param path storage root path
94     * @return
95     */
96   private def loadPath(path: Path): Option[FileSystemStorage] = {
97 
98     def complete(storage: Option[FileSystemStorage], time: Long): Unit =
99       logger.debug(s"${ if (storage.isDefined) "Loaded" else "No" } storage at path '$path' in ${time}ms")
100 
101     profile(complete _) {
102       val context = FileSystemContext(fc, conf, path, namespace)
103       StorageMetadataFactory.load(context).map { meta =>
104         try {
105           val storage = FileSystemStorageFactory(context, meta)
106           register(path, storage)
107           storage
108         } catch {
109           case NonFatal(e) => CloseQuietly(meta).foreach(e.addSuppressed); throw e
110         }
111       }
112     }
113   }
114 }
115 
116 object FileSystemStorageManager {
117 
118   private val cache = Caffeine.newBuilder().build(
119     new CacheLoader[(FileContext, Configuration, Path, Option[String]), FileSystemStorageManager]() {
120       override def load(key: (FileContext, Configuration, Path, Option[String])): FileSystemStorageManager =
121         new FileSystemStorageManager(key._1, key._2, key._3, key._4)
122     }
123   )
124 
125   /**
126     * Load a cached storage manager instance
127     *
128     * @param fc file context
129     * @param conf configuration
130     * @param root data store root path
131     * @return
132     */
133   def apply(fc: FileContext, conf: Configuration, root: Path, namespace: Option[String]): FileSystemStorageManager =
134     cache.get((fc, conf, root, namespace))
135 }