(hudi) 02/02: Fix missing datadog configuration metrics on mdt

[si...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

sivabalan pushed a commit to branch release-0.14.1
in repository https://gitbox.apache.org/repos/asf/hudi.git

commit 420ad9026cf70dc1625dd2b02363aaf189bf2369
Author: Nicolas Paris <ni...@adevinta.com>
AuthorDate: Mon Dec 25 17:26:58 2023 +0100

    Fix missing datadog configuration metrics on mdt
---
 .../hudi/metadata/HoodieMetadataWriteUtils.java    | 17 ++++
 .../hudi/functional/TestMetricsReporter.scala      | 98 ++++++++++++++++++++++
 2 files changed, 115 insertions(+)

diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
index e73f6fb7bc3..7c42ccf5016 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java
@@ -38,6 +38,7 @@ import org.apache.hudi.config.metrics.HoodieMetricsConfig;
 import org.apache.hudi.config.metrics.HoodieMetricsGraphiteConfig;
 import org.apache.hudi.config.metrics.HoodieMetricsJmxConfig;
 import org.apache.hudi.config.metrics.HoodieMetricsPrometheusConfig;
+import org.apache.hudi.config.metrics.HoodieMetricsDatadogConfig;
 import org.apache.hudi.exception.HoodieMetadataException;
 import org.apache.hudi.table.action.compact.strategy.UnBoundedCompactionStrategy;
 
@@ -182,6 +183,22 @@ public class HoodieMetadataWriteUtils {
           builder.withProperties(prometheusConfig.getProps());
           break;
         case DATADOG:
+          HoodieMetricsDatadogConfig.Builder datadogConfig = HoodieMetricsDatadogConfig.newBuilder()
+                  .withDatadogApiKey(writeConfig.getDatadogApiKey())
+                  .withDatadogApiKeySkipValidation(writeConfig.getDatadogApiKeySkipValidation())
+                  .withDatadogPrefix(writeConfig.getDatadogMetricPrefix())
+                  .withDatadogReportPeriodSeconds(writeConfig.getDatadogReportPeriodSeconds())
+                  .withDatadogTags(String.join(",", writeConfig.getDatadogMetricTags()))
+                  .withDatadogApiTimeoutSeconds(writeConfig.getDatadogApiTimeoutSeconds());
+          if (writeConfig.getDatadogMetricHost() != null) {
+            datadogConfig = datadogConfig.withDatadogHost(writeConfig.getDatadogMetricHost());
+          }
+          if (writeConfig.getDatadogApiSite() != null) {
+            datadogConfig = datadogConfig.withDatadogApiSite(writeConfig.getDatadogApiSite().name());
+          }
+
+          builder.withProperties(datadogConfig.build().getProps());
+          break;
         case PROMETHEUS:
         case CONSOLE:
         case INMEMORY:
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMetricsReporter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMetricsReporter.scala
new file mode 100644
index 00000000000..99f74870d87
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMetricsReporter.scala
@@ -0,0 +1,98 @@
+/*
+ * 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.hudi.functional
+
+import org.apache.hudi.HoodieConversionUtils.toJavaOption
+import org.apache.hudi.common.config.HoodieMetadataConfig
+import org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings
+import org.apache.hudi.common.util
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.config.metrics.{HoodieMetricsConfig, HoodieMetricsDatadogConfig}
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+import org.apache.hudi.util.JFunction
+import org.apache.hudi.{DataSourceWriteOptions, SparkDatasetMixin}
+import org.apache.spark.sql._
+import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
+import org.junit.jupiter.api.function.Executable
+import org.junit.jupiter.api.{AfterEach, Assertions, BeforeEach, Test}
+import org.slf4j.LoggerFactory
+
+import java.util.function.Consumer
+import scala.collection.JavaConverters._
+
+/**
+ * Tests on Spark DataSource for MOR table.
+ */
+class TestMetricsReporter extends HoodieSparkClientTestBase with SparkDatasetMixin {
+  var spark: SparkSession = null
+  private val log = LoggerFactory.getLogger(classOf[TestMORDataSource])
+  val commonOpts = Map(
+    "hoodie.insert.shuffle.parallelism" -> "4",
+    "hoodie.upsert.shuffle.parallelism" -> "4",
+    DataSourceWriteOptions.RECORDKEY_FIELD.key -> "_row_key",
+    DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+    DataSourceWriteOptions.PRECOMBINE_FIELD.key -> "timestamp",
+    HoodieWriteConfig.TBL_NAME.key -> "hoodie_test"
+  )
+
+  @BeforeEach override def setUp() {
+    setTableName("hoodie_test")
+    initPath()
+    initSparkContexts()
+    spark = sqlContext.sparkSession
+    initTestDataGenerator()
+    initFileSystem()
+  }
+
+  @AfterEach override def tearDown() = {
+    cleanupSparkContexts()
+    cleanupTestDataGenerator()
+    cleanupFileSystem()
+  }
+
+  override def getSparkSessionExtensionsInjector: util.Option[Consumer[SparkSessionExtensions]] =
+    toJavaOption(
+      Some(
+        JFunction.toJavaConsumer((receiver: SparkSessionExtensions) => new HoodieSparkSessionExtension().apply(receiver)))
+    )
+
+  @Test
+  def testSmokeDatadogReporter() {
+    val records1 = recordsToStrings(dataGen.generateInserts("001", 100)).asScala
+    val inputDF1 = spark.read.json(spark.sparkContext.parallelize(records1, 2))
+    val writeOpts: Map[String, String] = commonOpts ++ Map(
+      DataSourceWriteOptions.OPERATION.key -> DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL,
+      DataSourceWriteOptions.TABLE_TYPE.key -> DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL,
+      HoodieMetadataConfig.ENABLE.key -> "true",
+      HoodieMetricsConfig.TURN_METRICS_ON.key -> "true",
+      HoodieMetricsConfig.METRICS_REPORTER_TYPE_VALUE.key -> "DATADOG",
+      HoodieMetricsDatadogConfig.API_KEY_SKIP_VALIDATION.key -> "true",
+      HoodieMetricsDatadogConfig.METRIC_PREFIX_VALUE.key -> "hudi",
+      HoodieMetricsDatadogConfig.API_SITE_VALUE.key -> "US",
+      HoodieMetricsDatadogConfig.API_KEY.key -> "dummykey")
+
+    Assertions.assertDoesNotThrow(new Executable {
+      override def execute(): Unit =
+        inputDF1.write.format("org.apache.hudi")
+          .options(writeOpts)
+          .mode(SaveMode.Overwrite)
+          .save(basePath)
+
+    })
+  }
+}