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.metadata
10 
11 import com.github.benmanes.caffeine.cache.{CacheLoader, Caffeine, LoadingCache}
12 import com.typesafe.config._
13 import com.typesafe.scalalogging.LazyLogging
14 import org.apache.hadoop.fs.Options.CreateOpts
15 import org.apache.hadoop.fs._
16 import org.geotools.api.feature.simple.SimpleFeatureType
17 import org.locationtech.geomesa.fs.storage.api.StorageMetadata.PartitionMetadata
18 import org.locationtech.geomesa.fs.storage.api._
19 import org.locationtech.geomesa.fs.storage.common.utils.PathCache
20 import org.locationtech.geomesa.utils.concurrent.{CachedThreadPool, PhaserUtils}
21 import org.locationtech.geomesa.utils.io.WithClose
22 import org.locationtech.geomesa.utils.stats.MethodProfiling
23 import org.locationtech.geomesa.utils.text.StringSerialization
24 
25 import java.io.{FileNotFoundException, InputStreamReader}
26 import java.nio.charset.StandardCharsets
27 import java.util.UUID
28 import java.util.concurrent._
29 import java.util.function.BiFunction
30 import scala.collection.mutable.ArrayBuffer
31 import scala.runtime.BoxedUnit
32 import scala.util.control.NonFatal
33 
34 /**
35  * StorageMetadata implementation. Saves changes as a series of timestamped changelogs to allow
36  * concurrent modifications. The current state is obtained by replaying all the logs.
37  *
38  * Note that state is not read off disk until 'reload' is called.
39  *
40  * When accessed through the standard factory methods, the state will periodically reload from disk
41  * in order to pick up external changes (every 10 minutes by default).
42  *
43  * Note that modifications made to the metadata may not be immediately available, if they occur
44  * simultaneously with a reload. For example, calling `getPartition` immediately after `addPartition` may
45  * not return anything. However, the change is always persisted to disk, and will be available after the next
46  * reload. In general this does not cause problems, as reads and writes happen in different JVMs (ingest
47  * vs query).
48  *
49  * @param fc file context
50  * @param directory metadata root path
51  * @param sft simple feature type
52  * @param meta basic metadata config
53  * @param converter file converter
54  */
55 class FileBasedMetadata(
56     private val fc: FileContext,
57     val directory: Path,
58     val sft: SimpleFeatureType,
59     private val meta: Metadata,
60     private val converter: MetadataConverter
61   ) extends StorageMetadata with MethodProfiling with LazyLogging {
62 
63   import FileBasedMetadata._
64 
65   import scala.collection.JavaConverters._
66 
67   private val expiry = PathCache.CacheDurationProperty.toDuration.get.toMillis
68 
69   // cache of files associated with each partition
70   // we use a cache to provide lazy non-blocking refresh, but the cache will only ever have 1 element in it
71   private val partitions: LoadingCache[BoxedUnit, ConcurrentMap[String, PartitionFiles]] =
72     Caffeine.newBuilder().refreshAfterWrite(expiry, TimeUnit.MILLISECONDS).build(
73       new CacheLoader[BoxedUnit, ConcurrentMap[String, PartitionFiles]]() {
74         override def load(key: BoxedUnit): ConcurrentMap[String, PartitionFiles] = readPartitionFiles(8)
75       }
76     )
77 
78   // cache of parsed metadata, keyed by partition
79   private val metadata: LoadingCache[String, PartitionMetadata] =
80     Caffeine.newBuilder().refreshAfterWrite(expiry, TimeUnit.MILLISECONDS).build(
81       new CacheLoader[String, PartitionMetadata]() {
82         override def load(key: String): PartitionMetadata =
83           Option(partitions.get(BoxedUnit.UNIT).get(key)).flatMap(readPartition(_, 8)).map(_.toMetadata).orNull
84       }
85     )
86 
87   override val scheme: PartitionScheme = PartitionSchemeFactory.load(sft, meta.scheme)
88   override val encoding: String = meta.config(Metadata.Encoding)
89   override val leafStorage: Boolean = meta.config(Metadata.LeafStorage).toBoolean
90 
91   private val kvs = new ConcurrentHashMap[String, String](meta.config.asJava)
92 
93   override def get(key: String): Option[String] = Option(kvs.get(key))
94 
95   override def set(key: String, value: String): Unit = {
96     kvs.put(key, value)
97     FileBasedMetadataFactory.write(fc, directory.getParent, meta.copy(config = kvs.asScala.toMap))
98   }
99 
100   override def getPartitions(prefix: Option[String]): Seq[PartitionMetadata] = {
101     partitions.get(BoxedUnit.UNIT).asScala.toStream.flatMap { case (p, _) =>
102       if (prefix.forall(p.startsWith)) { Option(metadata.get(p)) } else { None }
103     }
104   }
105 
106   override def getPartition(name: String): Option[PartitionMetadata] = Option(metadata.get(name))
107 
108   override def addPartition(partition: PartitionMetadata): Unit = {
109     val config = {
110       val action = PartitionAction.Add
111       val envelope = partition.bounds.map(b => EnvelopeConfig(b.envelope)).getOrElse(Seq.empty)
112       PartitionConfig(partition.name, action, partition.files, partition.count, envelope, System.currentTimeMillis())
113     }
114     val path = writePartition(config)
115     // if we have already loaded the partition, merge in the new value
116     if (metadata.getIfPresent(partition.name) != null) {
117       metadata.asMap.merge(partition.name, partition, addMetadata)
118     }
119     Option(partitions.getIfPresent(BoxedUnit.UNIT)).foreach { files =>
120       files.merge(partition.name, PartitionFiles(config = Seq(config), parsed = Seq(path)), addFiles)
121     }
122   }
123 
124   override def removePartition(partition: PartitionMetadata): Unit = {
125     val config = {
126       val action = PartitionAction.Remove
127       val envelope = partition.bounds.map(b => EnvelopeConfig(b.envelope)).getOrElse(Seq.empty)
128       PartitionConfig(partition.name, action, partition.files, partition.count, envelope, System.currentTimeMillis())
129     }
130     val path = writePartition(config)
131     // if we have already loaded the partition, merge in the new value
132     if (metadata.getIfPresent(partition.name) != null) {
133       metadata.asMap.merge(partition.name, partition, removeMetadata)
134     }
135     Option(partitions.getIfPresent(BoxedUnit.UNIT)).foreach { files =>
136       files.merge(partition.name, PartitionFiles(config = Seq(config), parsed = Seq(path)), addFiles)
137     }
138   }
139 
140   override def setPartitions(partitions: Seq[PartitionMetadata]): Unit = {
141     val map = new ConcurrentHashMap[String, PartitionFiles]
142     this.partitions.put(BoxedUnit.UNIT, map)
143     this.metadata.invalidateAll()
144 
145     val configs = partitions.map { partition =>
146       val action = PartitionAction.Add
147       val envelope = partition.bounds.map(b => EnvelopeConfig(b.envelope)).getOrElse(Seq.empty)
148       val config = PartitionConfig(partition.name, action, partition.files, partition.count, envelope, System.currentTimeMillis())
149       // note: side effects in map
150       this.metadata.put(partition.name, partition)
151       map.put(partition.name, PartitionFiles(config = Seq(config)))
152       config
153     }
154 
155     writeCompactedConfig(configs)
156     delete(readPartitionFiles(8).asScala.flatMap { case (_, f) => f.unparsed ++ f.parsed }, 8)
157   }
158 
159   override def compact(partition: Option[String], fileSize: Option[Long], threads: Int): Unit = {
160     require(threads > 0, "Threads must be a positive number")
161 
162     // in normal usage, we never pass in a partition to this method
163     partition.foreach(p => logger.warn(s"Ignoring requested partition '$p' and compacting all metadata"))
164 
165     val configs = ArrayBuffer.empty[PartitionConfig]
166     val paths = ArrayBuffer.empty[Path]
167 
168     readPartitionFiles(threads).asScala.foreach { case (name, f) =>
169       val config = readPartition(f, threads).filter(_.files.nonEmpty)
170       config match {
171         case None => metadata.invalidate(name)
172         case Some(c) =>
173           metadata.put(name, c.toMetadata)
174           configs += c
175       }
176       paths ++= f.unparsed
177       paths ++= f.parsed
178     }
179 
180     writeCompactedConfig(configs.toSeq)
181     delete(paths, threads)
182 
183     partitions.invalidate(BoxedUnit.UNIT)
184   }
185 
186   override def invalidate(): Unit = {
187     partitions.invalidateAll()
188     metadata.invalidateAll()
189   }
190 
191   override def close(): Unit = {}
192 
193   /**
194    * Serialize a partition config to disk
195    *
196    * @param config config
197    * @return
198    */
199   private def writePartition(config: PartitionConfig): Path = {
200     val data = profile("Serialized partition configuration") {
201       converter.renderPartition(config)
202     }
203     profile("Persisted partition configuration") {
204       val encoded = StringSerialization.alphaNumericSafeString(config.name)
205       val name = s"$UpdatePartitionPrefix$encoded-${UUID.randomUUID()}${converter.suffix}"
206       val file = new Path(directory, name)
207       WithClose(fc.create(file, java.util.EnumSet.of(CreateFlag.CREATE), CreateOpts.createParent)) { out =>
208         out.write(data.getBytes(StandardCharsets.UTF_8))
209         out.hflush()
210         out.hsync()
211       }
212       PathCache.register(fc, file)
213       file
214     }
215   }
216 
217   /**
218    * Write metadata for a compacted set of partition operations to disk
219    *
220    * @param config partition config
221    */
222   private def writeCompactedConfig(config: Seq[PartitionConfig]): Unit = {
223     val data = profile("Serialized compacted partition configuration") {
224       converter.renderCompaction(config)
225     }
226     profile("Persisted compacted partition configuration") {
227       val file = new Path(directory, CompactedPrefix + converter.suffix)
228       val flags = java.util.EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
229       WithClose(fc.create(file, flags, CreateOpts.createParent)) { out =>
230         out.write(data.getBytes(StandardCharsets.UTF_8))
231         out.hflush()
232         out.hsync()
233       }
234       PathCache.register(fc, file)
235       // generally we overwrite the existing file but if we change rendering the name will change
236       val toRemove =
237         new Path(directory, if (converter.suffix == HoconPathSuffix) { CompactedJson } else { CompactedHocon })
238       if (PathCache.exists(fc, toRemove, reload = true)) {
239         fc.delete(toRemove, false)
240         PathCache.invalidate(fc, toRemove)
241       }
242     }
243   }
244 
245   /**
246    * Reads all the metadata files and groups them by partition, parsing them only if needed
247    * to determine the partition name
248    *
249    * @param threads threads
250    * @return
251    */
252   private def readPartitionFiles(threads: Int): ConcurrentMap[String, PartitionFiles] = {
253     val result = new ConcurrentHashMap[String, PartitionFiles]()
254 
255     // list all the metadata files on disk
256     profile("Listed metadata files") {
257       val pool = new CachedThreadPool(threads)
258       // use a phaser to track worker thread completion
259       val phaser = new Phaser(2) // 1 for the initial directory worker + 1 for this thread
260       pool.submit(new DirectoryWorker(pool, phaser, fc.listStatus(directory), result))
261       // wait for the worker threads to complete
262       try {
263         phaser.awaitAdvanceInterruptibly(phaser.arrive())
264       } finally {
265         pool.shutdown()
266       }
267     }
268 
269     result
270   }
271 
272   /**
273    * Parses and merges the config files for a given partition
274    *
275    * @param files files associated with the partition
276    * @param threads threads
277    * @return
278    */
279   private def readPartition(files: PartitionFiles, threads: Int): Option[PartitionConfig] = {
280     val updates = if (threads < 2) {
281       files.unparsed.flatMap(readPartitionConfig)
282     } else {
283       val ec = new CachedThreadPool(threads)
284       try {
285         val results = Seq.newBuilder[PartitionConfig]
286         def readOne(p: Path): Unit = {
287           readPartitionConfig(p).foreach { c =>
288             results.synchronized(results += c)
289           }
290         }
291         files.unparsed.toList.map(p => ec.submit(new Runnable() { override def run(): Unit = readOne(p)})).foreach(_.get)
292         results.result
293       } finally {
294         ec.shutdown()
295       }
296     }
297     mergePartitionConfigs(updates ++ files.config).filter(_.files.nonEmpty)
298   }
299 
300   /**
301    * Read and parse a partition metadata file
302    *
303    * @param file file path
304    * @return
305    */
306   private def readPartitionConfig(file: Path): Option[PartitionConfig] = {
307     try {
308       val config = profile("Loaded partition configuration") {
309         WithClose(new InputStreamReader(fc.open(file), StandardCharsets.UTF_8)) { in =>
310           ConfigFactory.parseReader(in, ConfigParseOptions.defaults().setSyntax(getSyntax(file.getName)))
311         }
312       }
313       profile("Parsed partition configuration") {
314         Some(converter.parsePartition(config))
315       }
316     } catch {
317       case NonFatal(e) => logger.error(s"Error reading config at path $file:", e); None
318     }
319   }
320 
321   /**
322    * Read and parse a compacted partition metadata file
323    *
324    * @param file compacted config file
325    * @return
326    */
327   private def readCompactedConfig(file: Path): Seq[PartitionConfig] = {
328     try {
329       val config = profile("Loaded compacted partition configuration") {
330         WithClose(new InputStreamReader(fc.open(file), StandardCharsets.UTF_8)) { in =>
331           ConfigFactory.parseReader(in, ConfigParseOptions.defaults().setSyntax(getSyntax(file.getName)))
332         }
333       }
334       profile("Parsed compacted partition configuration") {
335         converter.parseCompaction(config)
336       }
337     } catch {
338       case NonFatal(e) => logger.error(s"Error reading config at path $file:", e); Seq.empty
339     }
340   }
341 
342   /**
343    * Delete a seq of paths
344    *
345    * @param paths paths to delete
346    * @param threads number of threads to use
347    */
348   private def delete(paths: Iterable[Path], threads: Int): Unit = {
349     if (threads < 2) {
350       paths.foreach(fc.delete(_, false))
351     } else {
352       val ec = new CachedThreadPool(threads)
353       try {
354         paths.toList.map(p => ec.submit(new Runnable() { override def run(): Unit = fc.delete(p, false)})).foreach(_.get)
355       } finally {
356         ec.shutdown()
357       }
358     }
359   }
360 
361   private class DirectoryWorker(
362       es: ExecutorService,
363       phaser: Phaser,
364       listDirectory: => RemoteIterator[FileStatus],
365       result: ConcurrentHashMap[String, PartitionFiles]
366     ) extends Runnable {
367 
368     override def run(): Unit = {
369       try {
370         var i = phaser.getRegisteredParties + 1
371         val iter = listDirectory
372         while (iter.hasNext && i < PhaserUtils.MaxParties) {
373           val status = iter.next
374           val path = status.getPath
375           lazy val name = path.getName
376           if (status.isDirectory) {
377             i += 1
378             // use a tiered phaser on each directory avoid the limit of 65535 registered parties
379             es.submit(new DirectoryWorker(es, new Phaser(phaser, 1), fc.listStatus(path), result))
380           } else if (name.startsWith(UpdatePartitionPrefix)) {
381             // pull out the partition name but don't parse the file yet
382             val encoded = name.substring(8, name.length - 42) // strip out prefix and suffix
383             val partition = StringSerialization.decodeAlphaNumericSafeString(encoded)
384             result.merge(partition, PartitionFiles(unparsed = Seq(path)), addFiles)
385           } else if (name == CompactedHocon || name == CompactedJson) {
386             i += 1
387             phaser.register() // register the new worker thread
388             es.submit(new CompactedParser(phaser, path, result))
389           } else if (name.startsWith(UpdateFilePrefix) && name.endsWith(JsonPathSuffix)) {
390             // old update files - have to parse them to get the partition name
391             i += 1
392             phaser.register() // register the new worker thread
393             es.submit(new UpdateParser(phaser, path, result))
394           }
395         }
396         if (iter.hasNext) {
397           es.submit(new DirectoryWorker(es, new Phaser(phaser, 1), iter, result))
398         }
399       } catch {
400         case _: FileNotFoundException => // the partition dir was deleted... just return
401         case NonFatal(e) => logger.error("Error scanning metadata directory:", e)
402       } finally {
403         phaser.arrive() // notify that this thread is done
404       }
405     }
406   }
407 
408   private class CompactedParser(phaser: Phaser, path: Path, result: ConcurrentHashMap[String, PartitionFiles])
409       extends Runnable {
410 
411     override def run(): Unit = {
412       try {
413         readCompactedConfig(path).foreach { config =>
414           // note: don't track the path since it's from the compacted config file
415           result.merge(config.name, PartitionFiles(config = Seq(config)), addFiles)
416         }
417       } catch {
418         case _: FileNotFoundException => // the file was deleted... just return
419         case NonFatal(e) => logger.error("Error reading compacted metadata entry:", e)
420       } finally {
421         phaser.arrive() // notify that this thread is done
422       }
423     }
424   }
425 
426   private class UpdateParser(phaser: Phaser, path: Path, result: ConcurrentHashMap[String, PartitionFiles])
427       extends Runnable {
428 
429     override def run(): Unit = {
430       try {
431         readPartitionConfig(path).foreach { config =>
432           result.merge(config.name, PartitionFiles(config = Seq(config), parsed = Seq(path)), addFiles)
433         }
434       } catch {
435         case _: FileNotFoundException => // the file was deleted... just return
436         case NonFatal(e) => logger.error("Error reading metadata update entry:", e)
437       } finally {
438         phaser.arrive() // notify that this thread is done
439       }
440     }
441   }
442 }
443 
444 object FileBasedMetadata {
445 
446   val MetadataType = "file"
447   val DefaultOptions: NamedOptions = NamedOptions(MetadataType, Map(Config.RenderKey -> Config.RenderCompact))
448   val LegacyOptions : NamedOptions = NamedOptions(MetadataType, Map(Config.RenderKey -> Config.RenderPretty))
449 
450   object Config {
451     val RenderKey = "render"
452 
453     val RenderPretty  = "pretty"
454     val RenderCompact = "compact"
455   }
456 
457   private val CompactedPrefix       = "compacted"
458   private val UpdateFilePrefix      = "update-"
459   private val UpdatePartitionPrefix = UpdateFilePrefix + "$"
460   private val JsonPathSuffix        = RenderPretty.suffix
461   private val HoconPathSuffix       = RenderCompact.suffix
462   private val CompactedJson         = CompactedPrefix + JsonPathSuffix
463   private val CompactedHocon        = CompactedPrefix + HoconPathSuffix
464 
465   // function to add/merge an existing partition in an atomic call
466   private val addMetadata = new BiFunction[PartitionMetadata, PartitionMetadata, PartitionMetadata]() {
467     override def apply(existing: PartitionMetadata, update: PartitionMetadata): PartitionMetadata =
468       existing + update
469   }
470 
471   // function to remove/merge an existing partition in an atomic call
472   private val removeMetadata = new BiFunction[PartitionMetadata, PartitionMetadata, PartitionMetadata]() {
473     override def apply(existing: PartitionMetadata, update: PartitionMetadata): PartitionMetadata = {
474       val result = existing - update
475       if (result.files.isEmpty) { null } else { result }
476     }
477   }
478 
479   // function to merge partition files in an atomic call
480   private val addFiles = new BiFunction[PartitionFiles, PartitionFiles, PartitionFiles]() {
481     override def apply(existing: PartitionFiles, update: PartitionFiles): PartitionFiles = {
482       val config = existing.config ++ update.config
483       PartitionFiles(config, existing.parsed ++ update.parsed, existing.unparsed ++ update.unparsed)
484     }
485   }
486 
487   /**
488    * Copy a metadata instance. Discards any cached state
489    *
490    * @param m metadata
491    * @return
492    */
493   def copy(m: FileBasedMetadata): FileBasedMetadata =
494     new FileBasedMetadata(m.fc, m.directory, m.sft, m.meta, m.converter)
495 
496   private def getSyntax(file: String): ConfigSyntax = {
497     if (file.endsWith(HoconPathSuffix)) {
498       ConfigSyntax.CONF
499     } else if (file.endsWith(JsonPathSuffix)) {
500       ConfigSyntax.JSON
501     } else {
502       ConfigSyntax.JSON
503     }
504   }
505 
506   /**
507    * Holder for metadata files for a partition
508    *
509    * @param config any parsed configurations
510    * @param parsed references to the files corresponding to `config`
511    * @param unparsed unparsed config files associated with the partition
512    */
513   private case class PartitionFiles(
514       config: Seq[PartitionConfig] = Seq.empty,
515       parsed: Seq[Path] = Seq.empty,
516       unparsed: Seq[Path] = Seq.empty
517     )
518 }