From c12d4feac440cbbe0eded32110b61f6f5b01c18a Mon Sep 17 00:00:00 2001 From: Terence Date: Wed, 9 Dec 2020 16:41:10 +0800 Subject: [PATCH 1/6] Add deadletter metrics to batch and stream Signed-off-by: Terence --- .../scala/feast/ingestion/BatchPipeline.scala | 14 +++++- .../feast/ingestion/StreamingPipeline.scala | 15 ++++++- .../deadletters/DeadLetterMetrics.scala | 41 +++++++++++++++++ .../source/DeadLetterSinkMetricSource.scala | 45 +++++++++++++++++++ 4 files changed, 112 insertions(+), 3 deletions(-) create mode 100644 spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala create mode 100644 spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala index 15d129416e9..eaa55033c1c 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala @@ -18,10 +18,12 @@ package feast.ingestion import feast.ingestion.sources.bq.BigQueryReader import feast.ingestion.sources.file.FileReader +import feast.ingestion.stores.deadletters.DeadLetterMetrics import feast.ingestion.validation.{RowValidator, TypeCheck} import org.apache.commons.lang.StringUtils import org.apache.spark.SparkEnv -import org.apache.spark.sql.{SaveMode, SparkSession} +import org.apache.spark.sql.catalyst.encoders.RowEncoder +import org.apache.spark.sql.{Encoder, Row, SaveMode, SparkSession} /** * Batch Ingestion Flow: @@ -77,8 +79,18 @@ object BatchPipeline extends BasePipeline { config.deadLetterPath match { case Some(path) => + implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) + projected .filter(!rowValidator.allChecks) + .mapPartitions(iter => { + val res = iter + .map(row => { + DeadLetterMetrics.writeMetrics + row + }) + res + }) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala index e32c3cd8b11..5af9ac3631a 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala @@ -20,8 +20,9 @@ import java.io.File import java.util.concurrent.TimeUnit import feast.ingestion.registry.proto.ProtoRegistryFactory -import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession} +import org.apache.spark.sql.{DataFrame, Encoder, Row, SaveMode, SparkSession} import org.apache.spark.sql.functions.{expr, struct, udf} +import feast.ingestion.stores.deadletters.DeadLetterMetrics import feast.ingestion.utils.ProtoReflection import feast.ingestion.utils.testing.MemoryStreamingSource import feast.ingestion.validation.{RowValidator, TypeCheck} @@ -33,8 +34,8 @@ import org.apache.spark.sql.streaming.StreamingQuery import org.apache.spark.sql.avro._ import org.apache.spark.sql.execution.python.UserDefinedPythonFunction import org.apache.spark.sql.execution.streaming.ProcessingTimeTrigger -import org.apache.spark.sql.expressions.UserDefinedFunction import org.apache.spark.sql.types.BooleanType +import org.apache.spark.sql.catalyst.encoders.RowEncoder /** * Streaming pipeline (currently in micro-batches mode only, since we need to have multiple sinks: redis & deadletters). @@ -117,8 +118,18 @@ object StreamingPipeline extends BasePipeline with Serializable { config.deadLetterPath match { case Some(path) => + implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) + rowsAfterValidation .filter("!_isValid") + .mapPartitions(iter => { + val res = iter + .map(row => { + DeadLetterMetrics.writeMetrics + row + }) + res + }) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala new file mode 100644 index 00000000000..d8e2556a6df --- /dev/null +++ b/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala @@ -0,0 +1,41 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed 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 + * + * https://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 feast.ingestion.stores.deadletters + +import org.apache.spark.SparkEnv +import org.apache.spark.metrics.source.DeadLetterSinkMetricSource + +object DeadLetterMetrics { + def writeMetrics() = { + metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc() + } + + private lazy val metricSource: Option[DeadLetterSinkMetricSource] = { + this.synchronized { + if ( + SparkEnv.get.metricsSystem.getSourcesByName(DeadLetterSinkMetricSource.sourceName).isEmpty + ) { + SparkEnv.get.metricsSystem.registerSource(new DeadLetterSinkMetricSource) + } + } + + SparkEnv.get.metricsSystem.getSourcesByName(DeadLetterSinkMetricSource.sourceName) match { + case Seq(head) => Some(head.asInstanceOf[DeadLetterSinkMetricSource]) + case _ => None + } + } +} diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala new file mode 100644 index 00000000000..ff8f0200e57 --- /dev/null +++ b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala @@ -0,0 +1,45 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed 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 + * + * https://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.source + +import com.codahale.metrics.MetricRegistry +import org.apache.spark.SparkEnv + +class DeadLetterSinkMetricSource extends Source { + override val sourceName: String = DeadLetterSinkMetricSource.sourceName + + override val metricRegistry: MetricRegistry = new MetricRegistry + + private val sparkConfig = SparkEnv.get.conf + + private val metricLabels = sparkConfig.get("spark.metrics.labels", "") + + private def counterWithLabels(name: String) = { + if (metricLabels.isEmpty) { + name + } else { + s"$name#$metricLabels" + } + } + + val METRIC_DEADLETTER_ROWS_INSERTED = + metricRegistry.counter(counterWithLabels("feast_ingestion_deadletter_count")) +} + +object DeadLetterSinkMetricSource { + val sourceName = "deadletter_sink" +} From a498dae5dec991ec2a33a535363b4e8a5af631a2 Mon Sep 17 00:00:00 2001 From: Terence Date: Fri, 11 Dec 2020 10:40:09 +0800 Subject: [PATCH 2/6] Address comments Signed-off-by: Terence --- .../scala/feast/ingestion/BatchPipeline.scala | 12 +---- .../feast/ingestion/StreamingPipeline.scala | 11 +--- .../deadletters/DeadLetterMetrics.scala | 10 +++- .../metrics/source/BaseMetricSource.scala | 50 +++++++++++++++++++ .../source/DeadLetterSinkMetricSource.scala | 19 +------ .../source/RedisSinkMetricSource.scala | 31 +----------- 6 files changed, 64 insertions(+), 69 deletions(-) create mode 100644 spark/ingestion/src/main/scala/org/apache/spark/metrics/source/BaseMetricSource.scala diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala index eaa55033c1c..c79a50952ea 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala @@ -77,20 +77,12 @@ object BatchPipeline extends BasePipeline { .option("max_age", config.featureTable.maxAge.getOrElse(0L)) .save() + implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) config.deadLetterPath match { case Some(path) => - implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) - projected .filter(!rowValidator.allChecks) - .mapPartitions(iter => { - val res = iter - .map(row => { - DeadLetterMetrics.writeMetrics - row - }) - res - }) + .mapPartitions(iter => DeadLetterMetrics.incrementCount(iter)) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala index 5af9ac3631a..9f88cf96cfc 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala @@ -116,20 +116,13 @@ object StreamingPipeline extends BasePipeline with Serializable { .option("max_age", config.featureTable.maxAge.getOrElse(0L)) .save() + implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) config.deadLetterPath match { case Some(path) => - implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) rowsAfterValidation .filter("!_isValid") - .mapPartitions(iter => { - val res = iter - .map(row => { - DeadLetterMetrics.writeMetrics - row - }) - res - }) + .mapPartitions(iter => DeadLetterMetrics.incrementCount(iter)) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala index d8e2556a6df..4d28032e097 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala @@ -18,10 +18,16 @@ package feast.ingestion.stores.deadletters import org.apache.spark.SparkEnv import org.apache.spark.metrics.source.DeadLetterSinkMetricSource +import org.apache.spark.sql.Row object DeadLetterMetrics { - def writeMetrics() = { - metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc() + def incrementCount(rowIterator: Iterator[Row]) = { + val res = rowIterator + .map(row => { + metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc() + row + }) + res } private lazy val metricSource: Option[DeadLetterSinkMetricSource] = { diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/BaseMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/BaseMetricSource.scala new file mode 100644 index 00000000000..fd5c232a1c1 --- /dev/null +++ b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/BaseMetricSource.scala @@ -0,0 +1,50 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed 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 + * + * https://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.source + +import com.codahale.metrics.MetricRegistry +import org.apache.spark.SparkEnv + +class BaseMetricSource extends Source { + override val sourceName: String = "" + + override val metricRegistry: MetricRegistry = new MetricRegistry + + private val sparkConfig = SparkEnv.get.conf + + private val metricLabels = sparkConfig.get("spark.metrics.labels", "") + + private val appId = sparkConfig.get("spark.app.id", "") + + private val executorId = sparkConfig.get("spark.executor.id", "") + + protected def metricWithLabels(name: String) = { + if (metricLabels.isEmpty) { + name + } else { + s"$name#$metricLabels,job_id=$appId-$executorId" + } + } + + protected def counterWithLabels(name: String) = { + if (metricLabels.isEmpty) { + name + } else { + s"$name#$metricLabels" + } + } +} diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala index ff8f0200e57..49cbeee713f 100644 --- a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala +++ b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala @@ -16,26 +16,9 @@ */ package org.apache.spark.metrics.source -import com.codahale.metrics.MetricRegistry -import org.apache.spark.SparkEnv - -class DeadLetterSinkMetricSource extends Source { +class DeadLetterSinkMetricSource extends BaseMetricSource { override val sourceName: String = DeadLetterSinkMetricSource.sourceName - override val metricRegistry: MetricRegistry = new MetricRegistry - - private val sparkConfig = SparkEnv.get.conf - - private val metricLabels = sparkConfig.get("spark.metrics.labels", "") - - private def counterWithLabels(name: String) = { - if (metricLabels.isEmpty) { - name - } else { - s"$name#$metricLabels" - } - } - val METRIC_DEADLETTER_ROWS_INSERTED = metricRegistry.counter(counterWithLabels("feast_ingestion_deadletter_count")) } diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala index 4d122cd6bf4..09fddd892a2 100644 --- a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala +++ b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala @@ -16,38 +16,9 @@ */ package org.apache.spark.metrics.source -import com.codahale.metrics.MetricRegistry -import org.apache.spark.SparkEnv - -class RedisSinkMetricSource extends Source { +class RedisSinkMetricSource extends BaseMetricSource { override val sourceName: String = RedisSinkMetricSource.sourceName - override val metricRegistry: MetricRegistry = new MetricRegistry - - private val sparkConfig = SparkEnv.get.conf - - private val metricLabels = sparkConfig.get("spark.metrics.labels", "") - - private val appId = sparkConfig.get("spark.app.id", "") - - private val executorId = sparkConfig.get("spark.executor.id", "") - - private def metricWithLabels(name: String) = { - if (metricLabels.isEmpty) { - name - } else { - s"$name#$metricLabels,job_id=$appId-$executorId" - } - } - - private def counterWithLabels(name: String) = { - if (metricLabels.isEmpty) { - name - } else { - s"$name#$metricLabels" - } - } - val METRIC_TOTAL_ROWS_INSERTED = metricRegistry.counter(counterWithLabels("feast_ingestion_feature_row_ingested_count")) From 158a8878c38b83afe692d32e292ee11f16a8e19b Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Wed, 23 Dec 2020 16:47:33 +0800 Subject: [PATCH 3/6] generalize ingestion pipeline metrics Signed-off-by: Oleksii Moskalenko --- .../scala/feast/ingestion/BasePipeline.scala | 2 +- .../scala/feast/ingestion/BatchPipeline.scala | 8 +-- .../feast/ingestion/StreamingPipeline.scala | 7 +-- .../metrics/IngestionPipelineMetrics.scala | 53 +++++++++++++++++++ .../deadletters/DeadLetterMetrics.scala | 47 ---------------- ...la => IngestionPipelineMetricSource.scala} | 13 +++-- .../source/RedisSinkMetricSource.scala | 4 +- .../feast/ingestion/BatchPipelineIT.scala | 5 +- .../feast/ingestion/metrics/StatsDStub.scala | 31 +++++++++++ .../ingestion/metrics/StatsReporterSpec.scala | 31 +---------- 10 files changed, 109 insertions(+), 92 deletions(-) create mode 100644 spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala delete mode 100644 spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala rename spark/ingestion/src/main/scala/org/apache/spark/metrics/source/{DeadLetterSinkMetricSource.scala => IngestionPipelineMetricSource.scala} (64%) create mode 100644 spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala index e1f5a59285d..13dd51f280c 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala @@ -55,7 +55,7 @@ trait BasePipeline { .set("spark.metrics.conf.*.sink.statsd.port", c.port.toString) .set("spark.metrics.conf.*.sink.statsd.period", "30") .set("spark.metrics.conf.*.sink.statsd.unit", "seconds") - .set("spark.metrics.namespace", jobConfig.mode.toString.toLowerCase) + .set("spark.metrics.namespace", s"feast_${jobConfig.mode.toString.toLowerCase}") // until proto parser udf will be fixed, we have to use this .set("spark.sql.legacy.allowUntypedScalaUDF", "true") case None => () diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala index c79a50952ea..3b8675f8ec0 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala @@ -16,9 +16,9 @@ */ package feast.ingestion +import feast.ingestion.metrics.IngestionPipelineMetrics import feast.ingestion.sources.bq.BigQueryReader import feast.ingestion.sources.file.FileReader -import feast.ingestion.stores.deadletters.DeadLetterMetrics import feast.ingestion.validation.{RowValidator, TypeCheck} import org.apache.commons.lang.StringUtils import org.apache.spark.SparkEnv @@ -59,6 +59,8 @@ object BatchPipeline extends BasePipeline { val projected = input.select(projection: _*).cache() + implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) + TypeCheck.allTypesMatch(projected.schema, featureTable) match { case Some(error) => throw new RuntimeException(s"Dataframe columns don't match expected feature types: $error") @@ -66,6 +68,7 @@ object BatchPipeline extends BasePipeline { } val validRows = projected + .mapPartitions(IngestionPipelineMetrics.incrementRead) .filter(rowValidator.allChecks) validRows.write @@ -77,12 +80,11 @@ object BatchPipeline extends BasePipeline { .option("max_age", config.featureTable.maxAge.getOrElse(0L)) .save() - implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) config.deadLetterPath match { case Some(path) => projected .filter(!rowValidator.allChecks) - .mapPartitions(iter => DeadLetterMetrics.incrementCount(iter)) + .mapPartitions(IngestionPipelineMetrics.incrementDeadletters) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala index 9f88cf96cfc..c6ed87df85e 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala @@ -19,10 +19,10 @@ package feast.ingestion import java.io.File import java.util.concurrent.TimeUnit +import feast.ingestion.metrics.IngestionPipelineMetrics import feast.ingestion.registry.proto.ProtoRegistryFactory import org.apache.spark.sql.{DataFrame, Encoder, Row, SaveMode, SparkSession} import org.apache.spark.sql.functions.{expr, struct, udf} -import feast.ingestion.stores.deadletters.DeadLetterMetrics import feast.ingestion.utils.ProtoReflection import feast.ingestion.utils.testing.MemoryStreamingSource import feast.ingestion.validation.{RowValidator, TypeCheck} @@ -104,8 +104,10 @@ object StreamingPipeline extends BasePipeline with Serializable { batchDF.withColumn("_isValid", rowValidator.allChecks) } rowsAfterValidation.persist() + implicit def rowEncoder: Encoder[Row] = RowEncoder(rowsAfterValidation.schema) rowsAfterValidation + .mapPartitions(IngestionPipelineMetrics.incrementRead) .filter(if (config.doNotIngestInvalidRows) expr("_isValid") else rowValidator.allChecks) .write .format("feast.ingestion.stores.redis") @@ -116,13 +118,12 @@ object StreamingPipeline extends BasePipeline with Serializable { .option("max_age", config.featureTable.maxAge.getOrElse(0L)) .save() - implicit def rowEncoder: Encoder[Row] = RowEncoder(projected.schema) config.deadLetterPath match { case Some(path) => rowsAfterValidation .filter("!_isValid") - .mapPartitions(iter => DeadLetterMetrics.incrementCount(iter)) + .mapPartitions(IngestionPipelineMetrics.incrementDeadletters) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala b/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala new file mode 100644 index 00000000000..060547e611d --- /dev/null +++ b/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala @@ -0,0 +1,53 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed 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 + * + * https://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 feast.ingestion.metrics + +import org.apache.spark.SparkEnv +import org.apache.spark.metrics.source.IngestionPipelineMetricSource + +object IngestionPipelineMetrics { + def incrementDeadletters[A](rowIterator: Iterator[A]): Iterator[A] = { + if (metricSource.nonEmpty) + metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc(rowIterator.length) + + rowIterator + } + + def incrementRead[A](rowIterator: Iterator[A]): Iterator[A] = { + if (metricSource.nonEmpty) + metricSource.get.METRIC_ROWS_READ_FROM_SOURCE.inc(rowIterator.length) + + rowIterator + } + + private lazy val metricSource: Option[IngestionPipelineMetricSource] = { + this.synchronized { + if ( + SparkEnv.get.metricsSystem + .getSourcesByName(IngestionPipelineMetricSource.sourceName) + .isEmpty + ) { + SparkEnv.get.metricsSystem.registerSource(new IngestionPipelineMetricSource) + } + } + + SparkEnv.get.metricsSystem.getSourcesByName(IngestionPipelineMetricSource.sourceName) match { + case Seq(head) => Some(head.asInstanceOf[IngestionPipelineMetricSource]) + case _ => None + } + } +} diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala deleted file mode 100644 index 4d28032e097..00000000000 --- a/spark/ingestion/src/main/scala/feast/ingestion/stores/deadletters/DeadLetterMetrics.scala +++ /dev/null @@ -1,47 +0,0 @@ -/* - * SPDX-License-Identifier: Apache-2.0 - * Copyright 2018-2020 The Feast Authors - * - * Licensed 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 - * - * https://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 feast.ingestion.stores.deadletters - -import org.apache.spark.SparkEnv -import org.apache.spark.metrics.source.DeadLetterSinkMetricSource -import org.apache.spark.sql.Row - -object DeadLetterMetrics { - def incrementCount(rowIterator: Iterator[Row]) = { - val res = rowIterator - .map(row => { - metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc() - row - }) - res - } - - private lazy val metricSource: Option[DeadLetterSinkMetricSource] = { - this.synchronized { - if ( - SparkEnv.get.metricsSystem.getSourcesByName(DeadLetterSinkMetricSource.sourceName).isEmpty - ) { - SparkEnv.get.metricsSystem.registerSource(new DeadLetterSinkMetricSource) - } - } - - SparkEnv.get.metricsSystem.getSourcesByName(DeadLetterSinkMetricSource.sourceName) match { - case Seq(head) => Some(head.asInstanceOf[DeadLetterSinkMetricSource]) - case _ => None - } - } -} diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/IngestionPipelineMetricSource.scala similarity index 64% rename from spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala rename to spark/ingestion/src/main/scala/org/apache/spark/metrics/source/IngestionPipelineMetricSource.scala index 49cbeee713f..6710619f4c7 100644 --- a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/DeadLetterSinkMetricSource.scala +++ b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/IngestionPipelineMetricSource.scala @@ -16,13 +16,16 @@ */ package org.apache.spark.metrics.source -class DeadLetterSinkMetricSource extends BaseMetricSource { - override val sourceName: String = DeadLetterSinkMetricSource.sourceName +class IngestionPipelineMetricSource extends BaseMetricSource { + override val sourceName: String = IngestionPipelineMetricSource.sourceName val METRIC_DEADLETTER_ROWS_INSERTED = - metricRegistry.counter(counterWithLabels("feast_ingestion_deadletter_count")) + metricRegistry.counter(counterWithLabels("deadletter_count")) + + val METRIC_ROWS_READ_FROM_SOURCE = + metricRegistry.counter(counterWithLabels("read_from_source_count")) } -object DeadLetterSinkMetricSource { - val sourceName = "deadletter_sink" +object IngestionPipelineMetricSource { + val sourceName = "ingestion_pipeline" } diff --git a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala index 09fddd892a2..e5949d47bb4 100644 --- a/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala +++ b/spark/ingestion/src/main/scala/org/apache/spark/metrics/source/RedisSinkMetricSource.scala @@ -20,10 +20,10 @@ class RedisSinkMetricSource extends BaseMetricSource { override val sourceName: String = RedisSinkMetricSource.sourceName val METRIC_TOTAL_ROWS_INSERTED = - metricRegistry.counter(counterWithLabels("feast_ingestion_feature_row_ingested_count")) + metricRegistry.counter(counterWithLabels("feature_row_ingested_count")) val METRIC_ROWS_LAG = - metricRegistry.histogram(metricWithLabels("feast_ingestion_feature_row_lag_ms")) + metricRegistry.histogram(metricWithLabels("feature_row_lag_ms")) } object RedisSinkMetricSource { diff --git a/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala b/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala index 85db6494334..f3054c04237 100644 --- a/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala +++ b/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala @@ -30,6 +30,7 @@ import org.scalatest._ import redis.clients.jedis.Jedis import feast.ingestion.helpers.RedisStorageHelper._ import feast.ingestion.helpers.DataHelper._ +import feast.ingestion.metrics.StatsDStub import feast.proto.storage.RedisProto.RedisKeyV2 import feast.proto.types.ValueProto import org.apache.spark.sql.Encoder @@ -55,6 +56,7 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer { jedis.flushAll() implicit def testRowEncoder: Encoder[TestRow] = ExpressionEncoder() + val statsDStub = new StatsDStub def rowGenerator(start: DateTime, end: DateTime, customerGen: Option[Gen[String]] = None) = for { @@ -95,7 +97,8 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer { ) ), startTime = DateTime.parse("2020-08-01"), - endTime = DateTime.parse("2020-09-01") + endTime = DateTime.parse("2020-09-01"), + metrics = Some(StatsDConfig(host="localhost", port=statsDStub.port)) ) } diff --git a/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala b/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala new file mode 100644 index 00000000000..29ddcb75d7c --- /dev/null +++ b/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala @@ -0,0 +1,31 @@ +package feast.ingestion.metrics + +import java.net.{DatagramPacket, DatagramSocket, SocketTimeoutException} + +import scala.collection.mutable.ArrayBuffer + +class StatsDStub { + val socket = new DatagramSocket() + socket.setSoTimeout(100) + + def port: Int = socket.getLocalPort + + def receive: Array[String] = { + val messages: ArrayBuffer[String] = ArrayBuffer() + var finished = false + + do { + val buf = new Array[Byte](65535) + val p = new DatagramPacket(buf, buf.length) + try { + socket.receive(p) + } catch { + case _: SocketTimeoutException => + finished = true + } + messages += new String(p.getData, 0, p.getLength) + } while (!finished) + + messages.toArray + } +} diff --git a/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsReporterSpec.scala b/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsReporterSpec.scala index 3b674de8b7a..b531b87a40a 100644 --- a/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsReporterSpec.scala +++ b/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsReporterSpec.scala @@ -16,46 +16,17 @@ */ package feast.ingestion.metrics -import java.net.{DatagramPacket, DatagramSocket, SocketTimeoutException} import java.util import java.util.Collections import com.codahale.metrics.{Gauge, Histogram, MetricRegistry, UniformReservoir} import feast.ingestion.UnitSpec -import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters._ class StatsReporterSpec extends UnitSpec { - class SimpleServer { - val socket = new DatagramSocket() - socket.setSoTimeout(100) - - def port: Int = socket.getLocalPort - - def receive: Array[String] = { - val messages: ArrayBuffer[String] = ArrayBuffer() - var finished = false - - do { - val buf = new Array[Byte](65535) - val p = new DatagramPacket(buf, buf.length) - try { - socket.receive(p) - } catch { - case _: SocketTimeoutException => { - finished = true - } - } - messages += new String(p.getData, 0, p.getLength) - } while (!finished) - - messages.toArray - } - } - trait Scope { - val server = new SimpleServer + val server = new StatsDStub val reporter = new StatsdReporterWithTags( new MetricRegistry, "127.0.0.1", From 372bd69f65085ee8311ac88a009b024ed87eacf0 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Thu, 24 Dec 2020 14:24:00 +0800 Subject: [PATCH 4/6] tests passing Signed-off-by: Oleksii Moskalenko --- spark/ingestion/pom.xml | 4 +-- .../scala/feast/ingestion/BatchPipeline.scala | 5 +-- .../feast/ingestion/StreamingPipeline.scala | 9 +++-- .../metrics/IngestionPipelineMetrics.scala | 33 ++++++++++-------- .../metrics/StatsdReporterWithTags.scala | 9 +++-- .../feast/ingestion/BatchPipelineIT.scala | 26 +++++++++++--- .../scala/feast/ingestion/SparkSpec.scala | 5 ++- .../feast/ingestion/metrics/StatsDStub.scala | 34 +++++++++++++++++-- 8 files changed, 92 insertions(+), 33 deletions(-) diff --git a/spark/ingestion/pom.xml b/spark/ingestion/pom.xml index cda054efe9e..0e0bf8385c5 100644 --- a/spark/ingestion/pom.xml +++ b/spark/ingestion/pom.xml @@ -189,14 +189,14 @@ com.dimafeng testcontainers-scala-scalatest_${scala.version} - 0.38.6 + 0.38.8 test com.dimafeng testcontainers-scala-kafka_${scala.version} - 0.38.6 + 0.38.8 test diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala index 3b8675f8ec0..14e73569430 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/BatchPipeline.scala @@ -39,6 +39,7 @@ object BatchPipeline extends BasePipeline { val projection = inputProjection(config.source, featureTable.features, featureTable.entities) val rowValidator = new RowValidator(featureTable, config.source.eventTimestampColumn) + val metrics = new IngestionPipelineMetrics val input = config.source match { case source: BQSource => @@ -68,7 +69,7 @@ object BatchPipeline extends BasePipeline { } val validRows = projected - .mapPartitions(IngestionPipelineMetrics.incrementRead) + .mapPartitions(metrics.incrementRead) .filter(rowValidator.allChecks) validRows.write @@ -84,7 +85,7 @@ object BatchPipeline extends BasePipeline { case Some(path) => projected .filter(!rowValidator.allChecks) - .mapPartitions(IngestionPipelineMetrics.incrementDeadletters) + .mapPartitions(metrics.incrementDeadLetters) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala index c6ed87df85e..974c7d921da 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/StreamingPipeline.scala @@ -56,8 +56,8 @@ object StreamingPipeline extends BasePipeline with Serializable { val featureTable = config.featureTable val projection = inputProjection(config.source, featureTable.features, featureTable.entities) - val rowValidator = new RowValidator(featureTable, config.source.eventTimestampColumn) - + val rowValidator = new RowValidator(featureTable, config.source.eventTimestampColumn) + val metrics = new IngestionPipelineMetrics val validationUDF = createValidationUDF(sparkSession, config) val input = config.source match { @@ -107,7 +107,7 @@ object StreamingPipeline extends BasePipeline with Serializable { implicit def rowEncoder: Encoder[Row] = RowEncoder(rowsAfterValidation.schema) rowsAfterValidation - .mapPartitions(IngestionPipelineMetrics.incrementRead) + .mapPartitions(metrics.incrementRead) .filter(if (config.doNotIngestInvalidRows) expr("_isValid") else rowValidator.allChecks) .write .format("feast.ingestion.stores.redis") @@ -120,10 +120,9 @@ object StreamingPipeline extends BasePipeline with Serializable { config.deadLetterPath match { case Some(path) => - rowsAfterValidation .filter("!_isValid") - .mapPartitions(IngestionPipelineMetrics.incrementDeadletters) + .mapPartitions(metrics.incrementDeadLetters) .write .format("parquet") .mode(SaveMode.Append) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala b/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala index 060547e611d..03d41eddd52 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/metrics/IngestionPipelineMetrics.scala @@ -18,36 +18,39 @@ package feast.ingestion.metrics import org.apache.spark.SparkEnv import org.apache.spark.metrics.source.IngestionPipelineMetricSource +import org.apache.spark.sql.Row -object IngestionPipelineMetrics { - def incrementDeadletters[A](rowIterator: Iterator[A]): Iterator[A] = { +class IngestionPipelineMetrics extends Serializable { + + def incrementDeadLetters(rowIterator: Iterator[Row]): Iterator[Row] = { + val materialized = rowIterator.toArray if (metricSource.nonEmpty) - metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc(rowIterator.length) + metricSource.get.METRIC_DEADLETTER_ROWS_INSERTED.inc(materialized.length) - rowIterator + materialized.toIterator } - def incrementRead[A](rowIterator: Iterator[A]): Iterator[A] = { + def incrementRead(rowIterator: Iterator[Row]): Iterator[Row] = { + val materialized = rowIterator.toArray if (metricSource.nonEmpty) - metricSource.get.METRIC_ROWS_READ_FROM_SOURCE.inc(rowIterator.length) + metricSource.get.METRIC_ROWS_READ_FROM_SOURCE.inc(materialized.length) - rowIterator + materialized.toIterator } private lazy val metricSource: Option[IngestionPipelineMetricSource] = { - this.synchronized { - if ( - SparkEnv.get.metricsSystem - .getSourcesByName(IngestionPipelineMetricSource.sourceName) - .isEmpty - ) { - SparkEnv.get.metricsSystem.registerSource(new IngestionPipelineMetricSource) + val metricsSystem = SparkEnv.get.metricsSystem + IngestionPipelineMetricsLock.synchronized { + if (metricsSystem.getSourcesByName(IngestionPipelineMetricSource.sourceName).isEmpty) { + metricsSystem.registerSource(new IngestionPipelineMetricSource) } } - SparkEnv.get.metricsSystem.getSourcesByName(IngestionPipelineMetricSource.sourceName) match { + metricsSystem.getSourcesByName(IngestionPipelineMetricSource.sourceName) match { case Seq(head) => Some(head.asInstanceOf[IngestionPipelineMetricSource]) case _ => None } } } + +private object IngestionPipelineMetricsLock diff --git a/spark/ingestion/src/main/scala/feast/ingestion/metrics/StatsdReporterWithTags.scala b/spark/ingestion/src/main/scala/feast/ingestion/metrics/StatsdReporterWithTags.scala index 894014b6fdf..880fc2b9829 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/metrics/StatsdReporterWithTags.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/metrics/StatsdReporterWithTags.scala @@ -125,8 +125,13 @@ class StatsdReporterWithTags( private def reportGauge(name: String, gauge: Gauge[_])(implicit socket: DatagramSocket): Unit = formatAny(gauge.getValue).foreach(v => send(fullName(name), v, GAUGE)) - private def reportCounter(name: String, counter: Counter)(implicit socket: DatagramSocket): Unit = - send(fullName(name), format(counter.getCount), COUNTER) + private def reportCounter(name: String, counter: Counter)(implicit + socket: DatagramSocket + ): Unit = { + val snapshot = counter.getCount + send(fullName(name), format(snapshot), COUNTER) + counter.dec(snapshot) // reset counter + } private def reportHistogram(name: String, histogram: Histogram)(implicit socket: DatagramSocket diff --git a/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala b/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala index f3054c04237..0e9a96eb422 100644 --- a/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala +++ b/spark/ingestion/src/test/scala/feast/ingestion/BatchPipelineIT.scala @@ -23,7 +23,7 @@ import collection.JavaConverters._ import com.dimafeng.testcontainers.{ForAllTestContainer, GenericContainer} import com.google.protobuf.util.Timestamps import feast.proto.types.ValueProto.ValueType -import org.apache.spark.SparkConf +import org.apache.spark.{SparkConf, SparkEnv} import org.joda.time.{DateTime, Seconds} import org.scalacheck._ import org.scalatest._ @@ -46,17 +46,20 @@ case class TestRow( class BatchPipelineIT extends SparkSpec with ForAllTestContainer { override val container = GenericContainer("redis:6.0.8", exposedPorts = Seq(6379)) + val statsDStub = new StatsDStub override def withSparkConfOverrides(conf: SparkConf): SparkConf = conf .set("spark.redis.host", container.host) .set("spark.redis.port", container.mappedPort(6379).toString) + .set("spark.metrics.conf.*.sink.statsd.port", statsDStub.port.toString) trait Scope { val jedis = new Jedis("localhost", container.mappedPort(6379)) jedis.flushAll() + statsDStub.receivedMetrics // clean the buffer + implicit def testRowEncoder: Encoder[TestRow] = ExpressionEncoder() - val statsDStub = new StatsDStub def rowGenerator(start: DateTime, end: DateTime, customerGen: Option[Gen[String]] = None) = for { @@ -64,7 +67,7 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer { feature1 <- Gen.choose(0, 100) feature2 <- Gen.choose[Float](0, 1) eventTimestamp <- Gen - .choose(0, Seconds.secondsBetween(start, end).getSeconds) + .choose(0, Seconds.secondsBetween(start, end).getSeconds - 1) .map(start.withMillisOfSecond(0).plusSeconds) } yield TestRow( customer, @@ -98,7 +101,7 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer { ), startTime = DateTime.parse("2020-08-01"), endTime = DateTime.parse("2020-09-01"), - metrics = Some(StatsDConfig(host="localhost", port=statsDStub.port)) + metrics = Some(StatsDConfig(host = "localhost", port = statsDStub.port)) ) } @@ -129,6 +132,14 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer { keyTTL shouldEqual -1 }) + + SparkEnv.get.metricsSystem.report() + statsDStub.receivedMetrics should contain.allElementsOf( + Map( + "driver.ingestion_pipeline.read_from_source_count" -> rows.length, + "driver.redis_sink.feature_row_ingested_count" -> rows.length + ) + ) } "Parquet source file" should "be ingested in redis with expiry time equal to the largest of (event_timestamp + max_age) for" + @@ -466,6 +477,13 @@ class BatchPipelineIT extends SparkSpec with ForAllTestContainer { .toString ) .count() should be(rows.length) + + SparkEnv.get.metricsSystem.report() + statsDStub.receivedMetrics should contain.allElementsOf( + Map( + "driver.ingestion_pipeline.deadletter_count" -> rows.length + ) + ) } "Columns from source" should "be mapped according to configuration" in new Scope { diff --git a/spark/ingestion/src/test/scala/feast/ingestion/SparkSpec.scala b/spark/ingestion/src/test/scala/feast/ingestion/SparkSpec.scala index 1a030733348..025a3b8be1d 100644 --- a/spark/ingestion/src/test/scala/feast/ingestion/SparkSpec.scala +++ b/spark/ingestion/src/test/scala/feast/ingestion/SparkSpec.scala @@ -36,7 +36,10 @@ class SparkSpec extends UnitSpec with BeforeAndAfter { "org.apache.spark.metrics.sink.StatsdSinkWithTags" ) .set("spark.metrics.conf.*.sink.statsd.host", "localhost") - .set("spark.metrics.conf.*.sink.statsd.port", "8125") + .set("spark.metrics.conf.*.sink.statsd.period", "999") // disable scheduled reporting + .set("spark.metrics.conf.*.sink.statsd.unit", "minutes") + .set("spark.metrics.labels", "job_id=test") + .set("spark.metrics.namespace", "") .set("spark.sql.legacy.allowUntypedScalaUDF", "true") .set("spark.sql.execution.arrow.maxRecordsPerBatch", "50000") diff --git a/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala b/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala index 29ddcb75d7c..62a6d5a8e30 100644 --- a/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala +++ b/spark/ingestion/src/test/scala/feast/ingestion/metrics/StatsDStub.scala @@ -1,3 +1,19 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Copyright 2018-2020 The Feast Authors + * + * Licensed 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 + * + * https://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 feast.ingestion.metrics import java.net.{DatagramPacket, DatagramSocket, SocketTimeoutException} @@ -12,11 +28,11 @@ class StatsDStub { def receive: Array[String] = { val messages: ArrayBuffer[String] = ArrayBuffer() - var finished = false + var finished = false do { val buf = new Array[Byte](65535) - val p = new DatagramPacket(buf, buf.length) + val p = new DatagramPacket(buf, buf.length) try { socket.receive(p) } catch { @@ -28,4 +44,18 @@ class StatsDStub { messages.toArray } + + private val metricLine = """(.+):(.+)\|(.+)#(.+)""".r + + def receivedMetrics: Map[String, Float] = { + receive + .flatMap { + case metricLine(name, value, type_, tags) => + Seq(name -> value.toFloat) + case s: String => + Seq() + } + .groupBy(_._1) + .mapValues(_.map(_._2).sum) + } } From e7c3f12d165a92e827710809074243cb0e6386bc Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Thu, 24 Dec 2020 14:53:38 +0800 Subject: [PATCH 5/6] increase grpc connection timeout Signed-off-by: Oleksii Moskalenko --- tests/e2e/fixtures/client.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/e2e/fixtures/client.py b/tests/e2e/fixtures/client.py index 0227ec79307..3ee189250c0 100644 --- a/tests/e2e/fixtures/client.py +++ b/tests/e2e/fixtures/client.py @@ -63,6 +63,7 @@ def feast_client( local_staging_path, "historical_output" ), ingestion_drop_invalid_rows=True, + grpc_connection_timeout=30, **job_service_env, ) elif pytestconfig.getoption("env") == "aws": From b35790771924e5dc072864cc0cce7d3531d79373 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Thu, 24 Dec 2020 15:32:25 +0800 Subject: [PATCH 6/6] mvn retry in e2e tests Signed-off-by: Oleksii Moskalenko --- infra/scripts/test-end-to-end-gcp.sh | 2 +- infra/scripts/test-end-to-end.sh | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/infra/scripts/test-end-to-end-gcp.sh b/infra/scripts/test-end-to-end-gcp.sh index 4679cbb9b1f..cc54d9c8f88 100755 --- a/infra/scripts/test-end-to-end-gcp.sh +++ b/infra/scripts/test-end-to-end-gcp.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash export DISABLE_SERVICE_FIXTURES=1 -export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false" +export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=3 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false" export MAVEN_CACHE="gs://feast-templocation-kf-feast/.m2.2020-11-17.tar" infra/scripts/download-maven-cache.sh --archive-uri ${MAVEN_CACHE} --output-dir /tmp diff --git a/infra/scripts/test-end-to-end.sh b/infra/scripts/test-end-to-end.sh index 03beb22ab4d..c4853dcbcf2 100755 --- a/infra/scripts/test-end-to-end.sh +++ b/infra/scripts/test-end-to-end.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash -export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false" +export MAVEN_OPTS="-Dmaven.repo.local=/tmp/.m2/repository -DdependencyLocationsEnabled=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=3 -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false" export MAVEN_CACHE="gs://feast-templocation-kf-feast/.m2.2020-11-17.tar" infra/scripts/download-maven-cache.sh --archive-uri ${MAVEN_CACHE} --output-dir /tmp