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.metrics.core
10 
11 import com.codahale.metrics.{MetricRegistry, ScheduledReporter}
12 import com.typesafe.config.{Config, ConfigFactory}
13 import org.locationtech.geomesa.utils.classpath.ServiceLoader
14 import pureconfig.error.{CannotConvert, ConfigReaderFailures}
15 import pureconfig.{ConfigCursor, ConfigObjectCursor, ConfigReader, ConfigSource}
16 
17 import java.util.concurrent.TimeUnit
18 import scala.concurrent.duration.Duration
19 import scala.util.control.NonFatal
20 
21 /**
22   * Factory for SPI loading reporters at runtime
23   */
24 trait ReporterFactory {
25   def apply(conf: Config, registry: MetricRegistry, rates: TimeUnit, durations: TimeUnit): Option[ScheduledReporter]
26 }
27 
28 object ReporterFactory {
29 
30   private val defaults = ConfigFactory.parseString("""{units:seconds,interval:"60 seconds"}""")
31 
32   private lazy val factories = ServiceLoader.load[ReporterFactory]()
33 
34   private implicit val reader: ConfigReader[ReporterConfig] = ReporterReader
35 
36   /**
37     * Load a reporter from the available factories
38     *
39     * @param config config
40     * @param registry registry
41     * @return
42     */
43   def apply(config: Config, registry: MetricRegistry): ScheduledReporter = {
44     val ReporterConfig(rates, durations, interval) =
45       ConfigSource.fromConfig(config.withFallback(defaults)).loadOrThrow[ReporterConfig]
46     val reporter = factories.toStream.flatMap(_.apply(config, registry, rates, durations)).headOption.getOrElse {
47       throw new IllegalArgumentException(
48         s"Could not load reporter factory using provided config:\n${config.root().render()}")
49     }
50     if (interval > 0) {
51       reporter.start(interval, TimeUnit.MILLISECONDS)
52     }
53     reporter
54   }
55 
56   case class ReporterConfig(rates: TimeUnit, durations: TimeUnit, interval: Long)
57 
58   object ReporterReader extends ConfigReader[ReporterConfig] {
59     override def from(cur: ConfigCursor): Either[ConfigReaderFailures, ReporterConfig] = {
60       for {
61         obj      <- cur.asObjectCursor.right
62         rate     <- timeUnit(obj, "rate-units", "units").right
63         duration <- timeUnit(obj, "duration-units", "units").right
64         interval <- durationMillis(obj.atKeyOrUndefined("interval")).right
65       } yield {
66         ReporterConfig(rate, duration, interval)
67       }
68     }
69 
70     private def timeUnit(cur: ConfigObjectCursor, key: String, fallback: String): Either[ConfigReaderFailures, TimeUnit] = {
71       val primary = cur.atKey(key).right.flatMap(_.asString).right.flatMap { unit =>
72         TimeUnit.values().find(_.toString.equalsIgnoreCase(unit)).toRight[ConfigReaderFailures](
73           ConfigReaderFailures(cur.failureFor(CannotConvert(cur.valueOpt.map(_.toString).orNull, "TimeUnit", "Does not match a TimeUnit")))
74         )
75       }
76       if (primary.isRight || fallback == null) { primary } else {
77         // use the fallback if it's a right, otherwise use the primary as a left
78         timeUnit(cur, fallback, null).left.flatMap(_ => primary)
79       }
80     }
81 
82     private def durationMillis(cur: ConfigCursor): Either[ConfigReaderFailures, Long] = {
83       if (cur.isUndefined) { Right(-1L) } else {
84         cur.asString.right.flatMap { d =>
85           try { Right(Duration(d).toMillis) } catch {
86             case NonFatal(e) => cur.failed(CannotConvert(cur.valueOpt.map(_.toString).orNull, "Duration", e.getMessage))
87           }
88         }
89       }
90     }
91   }
92 }