Feat/datastax metrics config (#1228)

* feat: add metrics configuration options to datastax

* feat: update datastax pureconfig readers

cassandra-datastax-driver-pureconfig/src/main/scala-2/com/avast/sst/datastax/pureconfig/ConfigReaders.scala

@@ -48,9 +48,15 @@ trait ConfigReaders {
   implicit val cassandraDatastaxDriverProtocolConfigReader: ConfigReader[ProtocolConfig] = deriveReader[ProtocolConfig]
   implicit val cassandraDatastaxDriverMetricsConfigReader: ConfigReader[MetricsConfig] = deriveReader[MetricsConfig]
   implicit val cassandraDatastaxDriverSessionConfigReader: ConfigReader[SessionConfig] = deriveReader[SessionConfig]
+  implicit val cassandraDatastaxDriverMetricsFactoryConfigReader: ConfigReader[MetricsFactoryConfig] = deriveReader[MetricsFactoryConfig]
+  implicit val cassandraDatastaxDriverIdGeneratorConfigReader: ConfigReader[IdGeneratorConfig] = deriveReader[IdGeneratorConfig]
+  implicit val cassandraDatastaxDriverContinuousRequestsConfigReader: ConfigReader[ContinuousCqlRequests] =
+    deriveReader[ContinuousCqlRequests]
+  implicit val cassandraDatastaxDriverGraphRequestsConfigReader: ConfigReader[GraphRequests] = deriveReader[GraphRequests]
   implicit val cassandraDatastaxDriverCqlRequestsConfigReader: ConfigReader[CqlRequestsConfig] = deriveReader[CqlRequestsConfig]
   implicit val cassandraDatastaxDriverThrottlingConfigReader: ConfigReader[ThrottlingConfig] = deriveReader[ThrottlingConfig]
   implicit val cassandraDatastaxDriverDelayConfigReader: ConfigReader[DelayConfig] = deriveReader[DelayConfig]
+  implicit val cassandraDatastaxDriverGraphMessagesConfigReader: ConfigReader[GraphMessagesConfig] = deriveReader[GraphMessagesConfig]
   implicit val cassandraDatastaxDriverNodeConfigReader: ConfigReader[NodeConfig] = deriveReader[NodeConfig]
   implicit val cassandraDatastaxDriverCqlMessagesConfigReader: ConfigReader[CqlMessagesConfig] = deriveReader[CqlMessagesConfig]
   implicit val cassandraDatastaxDriverSocketConfigReader: ConfigReader[SocketConfig] = deriveReader[SocketConfig]

cassandra-datastax-driver-pureconfig/src/main/scala-3/com/avast/sst/datastax/pureconfig/ConfigReaders.scala

@@ -49,9 +49,14 @@ trait ConfigReaders {
   implicit val cassandraDatastaxDriverProtocolConfigReader: ConfigReader[ProtocolConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverMetricsConfigReader: ConfigReader[MetricsConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverSessionConfigReader: ConfigReader[SessionConfig] = ConfigReader.derived
+  implicit val cassandraDatastaxDriverMetricsFactoryConfigReader: ConfigReader[MetricsFactoryConfig] = ConfigReader.derived
+  implicit val cassandraDatastaxDriverIdGeneratorConfigReader: ConfigReader[IdGeneratorConfig] = ConfigReader.derived
+  implicit val cassandraDatastaxDriverContinuousRequestsConfigReader: ConfigReader[ContinuousCqlRequests] = ConfigReader.derived
+  implicit val cassandraDatastaxDriverGraphRequestsConfigReader: ConfigReader[GraphRequests] = ConfigReader.derived
   implicit val cassandraDatastaxDriverCqlRequestsConfigReader: ConfigReader[CqlRequestsConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverThrottlingConfigReader: ConfigReader[ThrottlingConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverDelayConfigReader: ConfigReader[DelayConfig] = ConfigReader.derived
+  implicit val cassandraDatastaxDriverGraphMessagesConfigReader: ConfigReader[GraphMessagesConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverNodeConfigReader: ConfigReader[NodeConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverCqlMessagesConfigReader: ConfigReader[CqlMessagesConfig] = ConfigReader.derived
   implicit val cassandraDatastaxDriverSocketConfigReader: ConfigReader[SocketConfig] = ConfigReader.derived

cassandra-datastax-driver/src/main/scala/com/avast/sst/datastax/CassandraDatastaxDriverModule.scala

@@ -3,16 +3,21 @@ package com.avast.sst.datastax
 import cats.effect.{Resource, Sync}
 import com.avast.sst.datastax.DatastaxHelper.*
 import com.avast.sst.datastax.config.CassandraDatastaxDriverConfig
-import com.datastax.oss.driver.api.core.CqlSession
+import com.datastax.oss.driver.api.core.{CqlSession, CqlSessionBuilder}
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption.*
+import com.datastax.dse.driver.api.core.config.DseDriverOption.*
 import com.datastax.oss.driver.api.core.config.{DriverConfigLoader, ProgrammaticDriverConfigLoaderBuilder as DriverBuilder}
 
 import javax.net.ssl.SSLContext
 
 object CassandraDatastaxDriverModule {
 
   /** Makes [[com.datastax.oss.driver.api.core.CqlSession]] initialized with the given config. */
-  def make[F[_]: Sync](cfg: CassandraDatastaxDriverConfig, ssl: Option[SSLContext] = None): Resource[F, CqlSession] = {
+  def make[F[_]: Sync](
+      cfg: CassandraDatastaxDriverConfig,
+      ssl: Option[SSLContext] = None,
+      customSessionBuilderOptions: CqlSessionBuilder => CqlSessionBuilder = identity
+  ): Resource[F, CqlSession] = {
 
     val acquire = Sync[F].delay {
 
@@ -78,11 +83,27 @@ object CassandraDatastaxDriverModule {
         durationProperty(REQUEST_TRACE_INTERVAL)(cfg.advanced.request.trace.interval),
         stringProperty(REQUEST_TRACE_CONSISTENCY)(cfg.advanced.request.trace.consistency.toStringRepr),
         booleanProperty(REQUEST_LOG_WARNINGS)(cfg.advanced.request.logWarnings),
-        optional(intListProperty(METRICS_SESSION_ENABLED), cfg.advanced.metrics.session.map(_.enabled)),
+        optional(stringListProperty(METRICS_SESSION_ENABLED), cfg.advanced.metrics.session.map(_.enabled)),
+        optional(
+          stringProperty(METRICS_FACTORY_CLASS),
+          cfg.advanced.metrics.factory.map(_.`class`)
+        ),
+        optional(
+          stringProperty(METRICS_ID_GENERATOR_CLASS),
+          cfg.advanced.metrics.idGenerator.map(_.`class`)
+        ),
+        optional(
+          stringProperty(METRICS_ID_GENERATOR_PREFIX),
+          cfg.advanced.metrics.idGenerator.flatMap(_.prefix)
+        ),
         optional(
           durationProperty(METRICS_SESSION_CQL_REQUESTS_HIGHEST),
           cfg.advanced.metrics.session.flatMap(_.cqlRequests.map(_.highestLatency))
         ),
+        optional(
+          durationProperty(METRICS_SESSION_CQL_REQUESTS_LOWEST),
+          cfg.advanced.metrics.session.flatMap(_.cqlRequests.map(_.lowestLatency))
+        ),
         optional(
           durationProperty(METRICS_SESSION_CQL_REQUESTS_INTERVAL),
           cfg.advanced.metrics.session.flatMap(_.cqlRequests.map(_.refreshInterval))
@@ -95,6 +116,10 @@ object CassandraDatastaxDriverModule {
           durationProperty(METRICS_SESSION_THROTTLING_HIGHEST),
           cfg.advanced.metrics.session.flatMap(_.throttling.flatMap(_.delay.map(_.highestLatency)))
         ),
+        optional(
+          durationProperty(METRICS_SESSION_THROTTLING_LOWEST),
+          cfg.advanced.metrics.session.flatMap(_.throttling.flatMap(_.delay.map(_.lowestLatency)))
+        ),
         optional(
           durationProperty(METRICS_SESSION_THROTTLING_INTERVAL),
           cfg.advanced.metrics.session.flatMap(_.throttling.flatMap(_.delay.map(_.refreshInterval)))
@@ -103,15 +128,71 @@ object CassandraDatastaxDriverModule {
           intProperty(METRICS_SESSION_THROTTLING_DIGITS),
           cfg.advanced.metrics.session.flatMap(_.throttling.flatMap(_.delay.map(_.significantDigits)))
         ),
-        optional(intListProperty(METRICS_NODE_ENABLED), cfg.advanced.metrics.node.map(_.enabled)),
+        optional(
+          durationProperty(CONTINUOUS_PAGING_METRICS_SESSION_CQL_REQUESTS_HIGHEST),
+          cfg.advanced.metrics.session.flatMap(_.continuousCqlRequests.map(_.highestLatency))
+        ),
+        optional(
+          durationProperty(CONTINUOUS_PAGING_METRICS_SESSION_CQL_REQUESTS_LOWEST),
+          cfg.advanced.metrics.session.flatMap(_.continuousCqlRequests.map(_.lowestLatency))
+        ),
+        optional(
+          intProperty(CONTINUOUS_PAGING_METRICS_SESSION_CQL_REQUESTS_DIGITS),
+          cfg.advanced.metrics.session.flatMap(_.continuousCqlRequests.map(_.significantDigits))
+        ),
+        optional(
+          durationProperty(CONTINUOUS_PAGING_METRICS_SESSION_CQL_REQUESTS_INTERVAL),
+          cfg.advanced.metrics.session.flatMap(_.continuousCqlRequests.map(_.refreshInterval))
+        ),
+        optional(
+          durationProperty(METRICS_SESSION_GRAPH_REQUESTS_HIGHEST),
+          cfg.advanced.metrics.session.flatMap(_.graphRequests.map(_.highestLatency))
+        ),
+        optional(
+          durationProperty(METRICS_SESSION_GRAPH_REQUESTS_LOWEST),
+          cfg.advanced.metrics.session.flatMap(_.graphRequests.map(_.lowestLatency))
+        ),
+        optional(
+          intProperty(METRICS_SESSION_GRAPH_REQUESTS_DIGITS),
+          cfg.advanced.metrics.session.flatMap(_.graphRequests.map(_.significantDigits))
+        ),
+        optional(
+          durationProperty(METRICS_SESSION_GRAPH_REQUESTS_INTERVAL),
+          cfg.advanced.metrics.session.flatMap(_.graphRequests.map(_.refreshInterval))
+        ),
+        optional(stringListProperty(METRICS_NODE_ENABLED), cfg.advanced.metrics.node.map(_.enabled)),
         optional(
           durationProperty(METRICS_NODE_CQL_MESSAGES_HIGHEST),
-          cfg.advanced.metrics.node.flatMap(_.cqlRequests.map(_.highestLatency))
+          cfg.advanced.metrics.node.flatMap(_.cqlMessages.map(_.highestLatency))
         ),
-        optional(intProperty(METRICS_NODE_CQL_MESSAGES_DIGITS), cfg.advanced.metrics.node.flatMap(_.cqlRequests.map(_.significantDigits))),
+        optional(
+          durationProperty(METRICS_NODE_CQL_MESSAGES_LOWEST),
+          cfg.advanced.metrics.node.flatMap(_.cqlMessages.map(_.lowestLatency))
+        ),
+        optional(intProperty(METRICS_NODE_CQL_MESSAGES_DIGITS), cfg.advanced.metrics.node.flatMap(_.cqlMessages.map(_.significantDigits))),
         optional(
           durationProperty(METRICS_NODE_CQL_MESSAGES_INTERVAL),
-          cfg.advanced.metrics.node.flatMap(_.cqlRequests.map(_.refreshInterval))
+          cfg.advanced.metrics.node.flatMap(_.cqlMessages.map(_.refreshInterval))
+        ),
+        optional(
+          durationProperty(METRICS_NODE_GRAPH_MESSAGES_HIGHEST),
+          cfg.advanced.metrics.node.flatMap(_.graphMessages.map(_.highestLatency))
+        ),
+        optional(
+          durationProperty(METRICS_NODE_GRAPH_MESSAGES_LOWEST),
+          cfg.advanced.metrics.node.flatMap(_.graphMessages.map(_.lowestLatency))
+        ),
+        optional(
+          intProperty(METRICS_NODE_GRAPH_MESSAGES_DIGITS),
+          cfg.advanced.metrics.node.flatMap(_.graphMessages.map(_.significantDigits))
+        ),
+        optional(
+          durationProperty(METRICS_NODE_GRAPH_MESSAGES_INTERVAL),
+          cfg.advanced.metrics.node.flatMap(_.graphMessages.map(_.refreshInterval))
+        ),
+        optional(
+          durationProperty(METRICS_NODE_EXPIRE_AFTER),
+          cfg.advanced.metrics.node.map(_.expireAfter)
         ),
         durationProperty(HEARTBEAT_INTERVAL)(cfg.advanced.heartbeat.interval),
         durationProperty(HEARTBEAT_TIMEOUT)(cfg.advanced.heartbeat.timeout),
@@ -172,8 +253,8 @@ object CassandraDatastaxDriverModule {
         .build()
 
       ssl match {
-        case Some(ssl) => CqlSession.builder().withConfigLoader(loader).withSslContext(ssl).build()
-        case None      => CqlSession.builder().withConfigLoader(loader).build()
+        case Some(ssl) => customSessionBuilderOptions(CqlSession.builder().withConfigLoader(loader).withSslContext(ssl)).build()
+        case None      => customSessionBuilderOptions(CqlSession.builder().withConfigLoader(loader)).build()
       }
     }
 

cassandra-datastax-driver/src/main/scala/com/avast/sst/datastax/config/advanced.scala

@@ -490,44 +490,95 @@ object TraceConfig {
 
 /** Metrics configuration
   *
+  * @param factory
+  *   Metrics Factory configuration.
+  * @param idGenerator
+  *   This section configures how metric ids are generated. A metric id is a unique combination of a metric name and metric tags.
   * @param session
   *   The session-level metrics (all disabled by default).
   * @param node
   *   The node-level metrics (all disabled by default).
   */
-final case class MetricsConfig(session: Option[SessionConfig], node: Option[NodeConfig])
+final case class MetricsConfig(
+    factory: Option[MetricsFactoryConfig],
+    idGenerator: Option[IdGeneratorConfig],
+    session: Option[SessionConfig],
+    node: Option[NodeConfig]
+)
 
 object MetricsConfig {
-  val Default: MetricsConfig = MetricsConfig(None, None)
+  val Default: MetricsConfig = MetricsConfig(None, None, None, None)
 }
 
+/** Metrics Factory configuration.
+  *
+  * @param `class`
+  *   The class for the metrics factory.
+  *
+  * Note: specifying a metrics factory is not enough to enable metrics; for the driver to actually start collecting metrics, you also need
+  * to specify which metrics to collect. See the following options for more information:
+  *   - advanced.metrics.session.enabled
+  *   - advanced.metrics.node.enabled
+  */
+final case class MetricsFactoryConfig(`class`: String = "DefaultMetricsFactory")
+
+/** Metric ID generator configuration.
+  *
+  * The driver ships with two built-in implementations:
+  *   - DefaultMetricIdGenerator: generates identifiers composed solely of (unique) metric names; It is mostly suitable for use with metrics
+  *     libraries that do not support tags, like Dropwizard.
+  *   - TaggingMetricIdGenerator: generates identifiers composed of name and tags. It is mostly suitable for use with metrics libraries that
+  *     support tags, like Micrometer or MicroProfile Metrics.
+  *
+  * @param `class`
+  *   The class name of a component implementing `MetricIdGenerator`. If it is not qualified, the driver assumes that it resides in the
+  *   `package com.datastax.oss.driver.internal.core.metrics`.
+  * @param prefix
+  *   An optional prefix to prepend to each generated metric name. The prefix should not start nor end with a dot or any other path
+  *   separator; the following are two valid examples: "cassandra" or "myapp.prod.cassandra".
+  */
+final case class IdGeneratorConfig(`class`: String = "DefaultMetricIdGenerator", prefix: Option[String] = None)
+
 /** The session-level metrics (all disabled by default).
   *
   * @param enabled
   *   The session-level metrics (all disabled by default).
   * @param cqlRequests
   *   Extra configuration (for the metrics that need it). Required if the 'cql-requests' metric is enabled
   * @param throttling
-  *   Configures request throttling metrics..
+  *   Configures request throttling metrics.
+  * @param continuousCqlRequests
+  *   Required: if the 'continuous-cql-requests' metric is enabled, and Dropwizard or Micrometer is used.
+  * @param graphRequests
+  *   Required: if the 'graph-requests' metric is enabled, and Dropwizard or Micrometer is used.
   */
 final case class SessionConfig(
-    enabled: List[Int] = List.empty,
+    enabled: List[String] = List.empty,
     cqlRequests: Option[CqlRequestsConfig],
-    throttling: Option[ThrottlingConfig]
+    throttling: Option[ThrottlingConfig],
+    continuousCqlRequests: Option[ContinuousCqlRequests],
+    graphRequests: Option[GraphRequests]
 )
 
 /** Extra metrics configuration
   *
   * @param highestLatency
-  *   The largest latency that we expect to record.
+  *   The largest latency that we expect to record.\
+  * @param lowestLatency
+  *   The lowest latency that we expect to record.
   * @param significantDigits
   *   The number of significant decimal digits to which internal structures will maintain value resolution and separation (for example, 3
   *   means that recordings up to 1 second will be recorded with a resolution of 1 millisecond or better). This must be between 0 and 5. If
   *   the value is out of range, it defaults to 3 and a warning is logged.
   * @param refreshInterval
   *   The interval at which percentile data is refreshed.
   */
-final case class CqlRequestsConfig(highestLatency: Duration = 3.seconds, significantDigits: Int = 3, refreshInterval: Duration = 5.minutes)
+final case class CqlRequestsConfig(
+    highestLatency: Duration = 3.seconds,
+    lowestLatency: Duration = 1.millisecond,
+    significantDigits: Int = 3,
+    refreshInterval: Duration = 5.minutes
+)
 
 /** How long requests are being throttled
   *
@@ -538,18 +589,59 @@ final case class CqlRequestsConfig(highestLatency: Duration = 3.seconds, signifi
 final case class ThrottlingConfig(delay: Option[DelayConfig])
 
 /** Throttling delay metric. */
-final case class DelayConfig(highestLatency: Duration = 3.seconds, significantDigits: Int = 3, refreshInterval: Duration = 5.minutes)
+final case class DelayConfig(
+    highestLatency: Duration = 3.seconds,
+    lowestLatency: Duration = 1.millisecond,
+    significantDigits: Int = 3,
+    refreshInterval: Duration = 5.minutes
+)
+
+final case class ContinuousCqlRequests(
+    highestLatency: Duration = 120.seconds,
+    lowestLatency: Duration = 10.millisecond,
+    significantDigits: Int = 3,
+    refreshInterval: Duration = 5.minutes
+)
+
+final case class GraphRequests(
+    highestLatency: Duration = 12.seconds,
+    lowestLatency: Duration = 1.millisecond,
+    significantDigits: Int = 3,
+    refreshInterval: Duration = 5.minutes
+)
 
 /** Node-level metric.
   *
   * @param enabled
   *   node-level metrics
-  * @param cqlRequests
+  * @param cqlMessages
   *   Required: if the 'cql-messages' metric is enabled
-  */
-final case class NodeConfig(enabled: List[Int], cqlRequests: Option[CqlMessagesConfig])
+  * @param graphMessages
+  *   Required: if the 'graph-messages' metric is enabled, and Dropwizard or Micrometer is used.
+  * @param expireAfter
+  *   The time after which the node level metrics will be evicted. The lowest allowed value is 5 minutes. If you try to set it lower, the
+  *   driver will log a warning and use 5 minutes.
+  */
+final case class NodeConfig(
+    enabled: List[String],
+    cqlMessages: Option[CqlMessagesConfig],
+    graphMessages: Option[GraphMessagesConfig],
+    expireAfter: Duration = 1.hour
+)
 
-final case class CqlMessagesConfig(highestLatency: Duration = 3.seconds, significantDigits: Int = 3, refreshInterval: Duration = 5.minutes)
+final case class CqlMessagesConfig(
+    highestLatency: Duration = 3.seconds,
+    lowestLatency: Duration = 1.millisecond,
+    significantDigits: Int = 3,
+    refreshInterval: Duration = 5.minutes
+)
+
+final case class GraphMessagesConfig(
+    highestLatency: Duration = 3.seconds,
+    lowestLatency: Duration = 1.millisecond,
+    significantDigits: Int = 3,
+    refreshInterval: Duration = 5.minutes
+)
 
 /** Socket configuration.
   *