[SPARK-28475][CORE] Add regex MetricFilter to GraphiteSink

core/src/main/scala/org/apache/spark/metrics/sink/GraphiteSink.scala

@@ -20,7 +20,7 @@ package org.apache.spark.metrics.sink
 import java.util.{Locale, Properties}
 import java.util.concurrent.TimeUnit
 
-import com.codahale.metrics.MetricRegistry
+import com.codahale.metrics.{Metric, MetricFilter, MetricRegistry}
 import com.codahale.metrics.graphite.{Graphite, GraphiteReporter, GraphiteUDP}
 
 import org.apache.spark.SecurityManager
@@ -38,6 +38,7 @@ private[spark] class GraphiteSink(val property: Properties, val registry: Metric
   val GRAPHITE_KEY_UNIT = "unit"
   val GRAPHITE_KEY_PREFIX = "prefix"
   val GRAPHITE_KEY_PROTOCOL = "protocol"
+  val GRAPHITE_KEY_REGEX = "regex"
 
   def propertyToOption(prop: String): Option[String] = Option(property.getProperty(prop))
 
@@ -72,10 +73,20 @@ private[spark] class GraphiteSink(val property: Properties, val registry: Metric
     case Some(p) => throw new Exception(s"Invalid Graphite protocol: $p")
   }
 
+  val filter = propertyToOption(GRAPHITE_KEY_REGEX) match {
+    case Some(pattern) => new MetricFilter() {
+      override def matches(name: String, metric: Metric): Boolean = {
+        pattern.r.findFirstMatchIn(name).isDefined
+      }
+    }
+    case None => MetricFilter.ALL
+  }
+
   val reporter: GraphiteReporter = GraphiteReporter.forRegistry(registry)
       .convertDurationsTo(TimeUnit.MILLISECONDS)
       .convertRatesTo(TimeUnit.SECONDS)
       .prefixedWith(prefix)
+      .filter(filter)
       .build(graphite)
 
   override def start() {

core/src/test/scala/org/apache/spark/metrics/sink/GraphiteSinkSuite.scala

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.metrics.sink
+
+import java.util.Properties
+
+import scala.collection.JavaConverters._
+
+import com.codahale.metrics._
+
+import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite}
+
+class GraphiteSinkSuite extends SparkFunSuite {
+
+  test("GraphiteSink with default MetricsFilter") {
+    val props = new Properties
+    props.put("host", "127.0.0.1")
+    props.put("port", "54321")
+    val registry = new MetricRegistry
+    val securityMgr = new SecurityManager(new SparkConf(false))
+
+    val sink = new GraphiteSink(props, registry, securityMgr)
+
+    val gauge = new Gauge[Double] {
+      override def getValue: Double = 1.23
+    }
+    sink.registry.register("gauge", gauge)
+    sink.registry.register("anothergauge", gauge)
+    sink.registry.register("streaminggauge", gauge)
+
+    val metricKeys = sink.registry.getGauges(sink.filter).keySet.asScala
+
+    assert(metricKeys.equals(Set("gauge", "anothergauge", "streaminggauge")),
+      "Should contain all metrics registered")
+  }
+
+  test("GraphiteSink with regex MetricsFilter") {
+    val props = new Properties
+    props.put("host", "127.0.0.1")
+    props.put("port", "54321")
+    props.put("regex", "streaming")
+    val registry = new MetricRegistry
+    val securityMgr = new SecurityManager(new SparkConf(false))
+
+    val sink = new GraphiteSink(props, registry, securityMgr)
+
+    val gauge = new Gauge[Double] {
+      override def getValue: Double = 1.23
+    }
+    sink.registry.register("gauge", gauge)
+    sink.registry.register("anothergauge", gauge)
+    sink.registry.register("streaminggauge", gauge)
+
+    val metricKeys = sink.registry.getGauges(sink.filter).keySet.asScala
+
+    assert(metricKeys.equals(Set("streaminggauge")),
+      "Should contain only metrics matches regex filter")
+  }
+}