forked from opensearch-project/opensearch-spark
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add read/write bytes metrics (opensearch-project#803)
* Add read/write bytes metrics Signed-off-by: Tomoyuki Morita <[email protected]> * Add unit test Signed-off-by: Tomoyuki Morita <[email protected]> * Address comments Signed-off-by: Tomoyuki Morita <[email protected]> --------- Signed-off-by: Tomoyuki Morita <[email protected]>
- Loading branch information
Showing
11 changed files
with
337 additions
and
23 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
flint-core/src/main/java/org/opensearch/flint/core/metrics/HistoricGauge.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.core.metrics; | ||
|
|
||
| import com.codahale.metrics.Gauge; | ||
| import java.util.ArrayList; | ||
| import java.util.Collections; | ||
| import java.util.LinkedList; | ||
| import java.util.List; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Value; | ||
|
|
||
| /** | ||
| * Gauge which stores historic data points with timestamps. | ||
| * This is used for emitting separate data points per request, instead of single aggregated metrics. | ||
| */ | ||
| public class HistoricGauge implements Gauge<Long> { | ||
| @AllArgsConstructor | ||
| @Value | ||
| public static class DataPoint { | ||
| Long value; | ||
| long timestamp; | ||
| } | ||
|
|
||
| private final List<DataPoint> dataPoints = Collections.synchronizedList(new LinkedList<>()); | ||
|
|
||
| /** | ||
| * This method will just return first value. | ||
| * @return first value | ||
| */ | ||
| @Override | ||
| public Long getValue() { | ||
| if (!dataPoints.isEmpty()) { | ||
| return dataPoints.get(0).value; | ||
| } else { | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Add new data point. Current time stamp will be attached to the data point. | ||
| * @param value metric value | ||
| */ | ||
| public void addDataPoint(Long value) { | ||
| dataPoints.add(new DataPoint(value, System.currentTimeMillis())); | ||
| } | ||
|
|
||
| /** | ||
| * Return copy of dataPoints and remove them from internal list | ||
| * @return copy of the data points | ||
| */ | ||
| public List<DataPoint> pollDataPoints() { | ||
| int size = dataPoints.size(); | ||
| List<DataPoint> result = new ArrayList<>(dataPoints.subList(0, size)); | ||
| if (size > 0) { | ||
| dataPoints.subList(0, size).clear(); | ||
| } | ||
| return result; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
58 changes: 58 additions & 0 deletions
58
...t-core/src/main/scala/org/opensearch/flint/core/metrics/ReadWriteBytesSparkListener.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.core.metrics | ||
|
|
||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} | ||
| import org.apache.spark.sql.SparkSession | ||
|
|
||
| /** | ||
| * Collect and emit bytesRead/Written and recordsRead/Written metrics | ||
| */ | ||
| class ReadWriteBytesSparkListener extends SparkListener with Logging { | ||
| var bytesRead: Long = 0 | ||
| var recordsRead: Long = 0 | ||
| var bytesWritten: Long = 0 | ||
| var recordsWritten: Long = 0 | ||
|
|
||
| override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = { | ||
| val inputMetrics = taskEnd.taskMetrics.inputMetrics | ||
| val outputMetrics = taskEnd.taskMetrics.outputMetrics | ||
| val ids = s"(${taskEnd.taskInfo.taskId}, ${taskEnd.taskInfo.partitionId})" | ||
| logInfo( | ||
| s"${ids} Input: bytesRead=${inputMetrics.bytesRead}, recordsRead=${inputMetrics.recordsRead}") | ||
| logInfo( | ||
| s"${ids} Output: bytesWritten=${outputMetrics.bytesWritten}, recordsWritten=${outputMetrics.recordsWritten}") | ||
|
|
||
| bytesRead += inputMetrics.bytesRead | ||
| recordsRead += inputMetrics.recordsRead | ||
| bytesWritten += outputMetrics.bytesWritten | ||
| recordsWritten += outputMetrics.recordsWritten | ||
| } | ||
|
|
||
| def emitMetrics(): Unit = { | ||
| logInfo(s"Input: totalBytesRead=${bytesRead}, totalRecordsRead=${recordsRead}") | ||
| logInfo(s"Output: totalBytesWritten=${bytesWritten}, totalRecordsWritten=${recordsWritten}") | ||
| MetricsUtil.addHistoricGauge(MetricConstants.INPUT_TOTAL_BYTES_READ, bytesRead) | ||
| MetricsUtil.addHistoricGauge(MetricConstants.INPUT_TOTAL_RECORDS_READ, recordsRead) | ||
| MetricsUtil.addHistoricGauge(MetricConstants.OUTPUT_TOTAL_BYTES_WRITTEN, bytesWritten) | ||
| MetricsUtil.addHistoricGauge(MetricConstants.OUTPUT_TOTAL_RECORDS_WRITTEN, recordsWritten) | ||
| } | ||
| } | ||
|
|
||
| object ReadWriteBytesSparkListener { | ||
| def withMetrics[T](spark: SparkSession, lambda: () => T): T = { | ||
| val listener = new ReadWriteBytesSparkListener() | ||
| spark.sparkContext.addSparkListener(listener) | ||
|
|
||
| val result = lambda() | ||
|
|
||
| spark.sparkContext.removeSparkListener(listener) | ||
| listener.emitMetrics() | ||
|
|
||
| result | ||
| } | ||
| } |
79 changes: 79 additions & 0 deletions
79
flint-core/src/test/java/org/opensearch/flint/core/metrics/HistoricGaugeTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| /* | ||
| * Copyright OpenSearch Contributors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package org.opensearch.flint.core.metrics; | ||
|
|
||
| import org.junit.Test; | ||
| import static org.junit.Assert.*; | ||
| import org.opensearch.flint.core.metrics.HistoricGauge.DataPoint; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public class HistoricGaugeTest { | ||
|
|
||
| @Test | ||
| public void testGetValue_EmptyGauge_ShouldReturnNull() { | ||
| HistoricGauge gauge= new HistoricGauge(); | ||
| assertNull(gauge.getValue()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testGetValue_WithSingleDataPoint_ShouldReturnFirstValue() { | ||
| HistoricGauge gauge= new HistoricGauge(); | ||
| Long value = 100L; | ||
| gauge.addDataPoint(value); | ||
|
|
||
| assertEquals(value, gauge.getValue()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testGetValue_WithMultipleDataPoints_ShouldReturnFirstValue() { | ||
| HistoricGauge gauge= new HistoricGauge(); | ||
| Long firstValue = 100L; | ||
| Long secondValue = 200L; | ||
| gauge.addDataPoint(firstValue); | ||
| gauge.addDataPoint(secondValue); | ||
|
|
||
| assertEquals(firstValue, gauge.getValue()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testPollDataPoints_WithMultipleDataPoints_ShouldReturnAndClearDataPoints() { | ||
| HistoricGauge gauge= new HistoricGauge(); | ||
| gauge.addDataPoint(100L); | ||
| gauge.addDataPoint(200L); | ||
| gauge.addDataPoint(300L); | ||
|
|
||
| List<DataPoint> dataPoints = gauge.pollDataPoints(); | ||
|
|
||
| assertEquals(3, dataPoints.size()); | ||
| assertEquals(Long.valueOf(100L), dataPoints.get(0).getValue()); | ||
| assertEquals(Long.valueOf(200L), dataPoints.get(1).getValue()); | ||
| assertEquals(Long.valueOf(300L), dataPoints.get(2).getValue()); | ||
|
|
||
| assertTrue(gauge.pollDataPoints().isEmpty()); | ||
| } | ||
|
|
||
| @Test | ||
| public void testAddDataPoint_ShouldAddDataPointWithCorrectValueAndTimestamp() { | ||
| HistoricGauge gauge= new HistoricGauge(); | ||
| Long value = 100L; | ||
| gauge.addDataPoint(value); | ||
|
|
||
| List<DataPoint> dataPoints = gauge.pollDataPoints(); | ||
|
|
||
| assertEquals(1, dataPoints.size()); | ||
| assertEquals(value, dataPoints.get(0).getValue()); | ||
| assertTrue(dataPoints.get(0).getTimestamp() > 0); | ||
| } | ||
|
|
||
| @Test | ||
| public void testPollDataPoints_EmptyGauge_ShouldReturnEmptyList() { | ||
| HistoricGauge gauge= new HistoricGauge(); | ||
| List<DataPoint> dataPoints = gauge.pollDataPoints(); | ||
|
|
||
| assertTrue(dataPoints.isEmpty()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.