Skip to content

Commit

Permalink
Add flint opensearch metrics
Browse files Browse the repository at this point in the history 
Signed-off-by: Vamsi Manohar <[email protected]>
  • Loading branch information
@vmmusings
vmmusings committed Jan 8, 2024
1 parent 9061eb9 commit 791220c
Show file tree
Hide file tree
Showing 4 changed files with 86 additions and 6 deletions.
55 changes: 52 additions & 3 deletions ...c/main/java/org/opensearch/flint/core/metrics/reporter/DimensionedCloudWatchReporter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
import com.amazonaws.services.cloudwatch.model.PutMetricDataResult;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.cloudwatch.model.StatisticSet;
import com.amazonaws.util.StringUtils;
import com.codahale.metrics.Clock;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Counting;
Expand All @@ -36,6 +35,7 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
Expand All @@ -47,6 +47,9 @@
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import java.util.stream.Stream;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -83,6 +86,16 @@ public class DimensionedCloudWatchReporter extends ScheduledReporter {
// Visible for testing
public static final String DIMENSION_SNAPSHOT_STD_DEV = "snapshot-std-dev";

public static final String DIMENSION_JOB_ID = "jobId";

public static final String DIMENSION_APPLICATION_ID = "applicationId";

public static final String DIMENSION_DOMAIN_ID = "domainId";

public static final String DIMENSION_INSTANCE_TYPE = "instance";

public static final String UNKNOWN = "unknown";

/**
* Amazon CloudWatch rejects values that are either too small or too large.
* Values must be in the range of 8.515920e-109 to 1.174271e+108 (Base 10) or 2e-360 to 2e360 (Base 2).
Expand Down Expand Up @@ -135,7 +148,6 @@ public void report(final SortedMap<String, Gauge> gauges,
if (builder.withDryRun) {
LOGGER.warn("** Reporter is running in 'DRY RUN' mode **");
}

try {
final List<MetricDatum> metricData = new ArrayList<>(
gauges.size() + counters.size() + 10 * histograms.size() + 10 * timers.size());
Expand Down Expand Up @@ -339,6 +351,8 @@ private void stageMetricDatum(final boolean metricConfigured,
if (metricConfigured && (builder.withZeroValuesSubmission || metricValue > 0)) {
final DimensionedName dimensionedName = DimensionedName.decode(metricName);
final Set<Dimension> dimensions = new LinkedHashSet<>(builder.globalDimensions);
Pair<String, Set<Dimension>> finalNameAndDefaultDimensions = getFinalMetricNameAndDefaultDimensions(dimensionedName);
dimensions.addAll(finalNameAndDefaultDimensions.getRight());
dimensions.addAll(dimensionedName.getDimensions());
if (shouldAppendDropwizardTypeDimension) {
dimensions.add(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(dimensionValue));
Expand All @@ -347,12 +361,47 @@ private void stageMetricDatum(final boolean metricConfigured,
metricData.add(new MetricDatum()
.withTimestamp(new Date(builder.clock.getTime()))
.withValue(cleanMetricValue(metricValue))
.withMetricName(dimensionedName.getName())
.withMetricName(finalNameAndDefaultDimensions.getLeft())
.withDimensions(dimensions)
.withUnit(standardUnit));
}
}

@NotNull
private Pair<String, Set<Dimension>> getFinalMetricNameAndDefaultDimensions(DimensionedName dimensionedName) {
final String jobId = System.getenv().getOrDefault("SERVERLESS_EMR_JOB_ID", UNKNOWN);
final String applicationId = System.getenv().getOrDefault("SERVERLESS_EMR_VIRTUAL_CLUSTER_ID", UNKNOWN);
final String domainId = System.getenv().getOrDefault("FLINT_CLUSTER_NAME", UNKNOWN);
final Dimension jobDimension = new Dimension().withName(DIMENSION_JOB_ID).withValue(jobId);
final Dimension applicationDimension = new Dimension().withName(DIMENSION_APPLICATION_ID).withValue(applicationId);
final Dimension domainIdDimension = new Dimension().withName(DIMENSION_DOMAIN_ID).withValue(domainId);
Dimension instanceDimension = new Dimension().withName(DIMENSION_INSTANCE_TYPE).withValue(UNKNOWN);
String name = dimensionedName.getName();
String finalMetricName = name;
String[] parts = name.split("\\.");
if (doesNameConsistsOfMetricNameSpace(parts, jobId)) {
finalMetricName = Stream.of(parts).skip(2).collect(Collectors.joining("."));
//For executors only id is added to the metric name, thats why the numeric check.
//If it is not numeric then the instance is driver.
if (StringUtils.isNumeric(parts[1])) {
instanceDimension = new Dimension().withName(DIMENSION_INSTANCE_TYPE).withValue("executor" + parts[1]);
}
else {
instanceDimension = new Dimension().withName(DIMENSION_INSTANCE_TYPE).withValue(parts[1]);
}
}
Set<Dimension> dimensions = new HashSet<>();
dimensions.add(jobDimension);
dimensions.add(applicationDimension);
dimensions.add(instanceDimension);
dimensions.add(domainIdDimension);
return Pair.of(finalMetricName, dimensions);
}

private boolean doesNameConsistsOfMetricNameSpace(String[] metricNameParts, String jobId) {
return metricNameParts[0].equals(jobId);
}

private void stageMetricDatumWithConvertedSnapshot(final boolean metricConfigured,
final String metricName,
final Snapshot snapshot,
Expand Down
5 changes: 3 additions & 2 deletions flint-core/src/main/java/org/opensearch/flint/core/metrics/reporter/DimensionedName.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@ public static DimensionedName decode(final String encodedDimensionedName) {
if (matcher.find() && matcher.groupCount() == 2) {
final DimensionedNameBuilder builder = new DimensionedNameBuilder(matcher.group(1).trim());
for (String t : matcher.group(2).split(",")) {
final String[] keyAndValue = t.split(":");
//## acts as a distinct separator.
final String[] keyAndValue = t.split("##");
builder.withDimension(keyAndValue[0].trim(), keyAndValue[1].trim());
}
return builder.build();
Expand Down Expand Up @@ -59,7 +60,7 @@ public synchronized String encode() {
final StringBuilder sb = new StringBuilder(this.name);
sb.append('[');
sb.append(this.dimensions.values().stream()
.map(dimension -> dimension.getName() + ":" + dimension.getValue())
.map(dimension -> dimension.getName() + "##" + dimension.getValue())
.collect(Collectors.joining(",")));
sb.append(']');

Expand Down
12 changes: 12 additions & 0 deletions flint-core/src/main/scala/org/apache/spark/metrics/source/FlintMetricSource.scala
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
/*
* Copyright OpenSearch Contributors
* SPDX-License-Identifier: Apache-2.0
*/

package org.apache.spark.metrics.source

import com.codahale.metrics.MetricRegistry

class FlintMetricSource(val sourceName: String) extends Source {
override val metricRegistry: MetricRegistry = new MetricRegistry
}
20 changes: 19 additions & 1 deletion ...c/test/java/opensearch/flint/core/metrics/reporter/DimensionedCloudWatchReporterTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,12 +47,17 @@
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_APPLICATION_ID;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_COUNT;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_DOMAIN_ID;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_GAUGE;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_INSTANCE_TYPE;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_JOB_ID;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_NAME_TYPE;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_SNAPSHOT_MEAN;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_SNAPSHOT_STD_DEV;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.DIMENSION_SNAPSHOT_SUMMARY;
import static org.opensearch.flint.core.metrics.reporter.DimensionedCloudWatchReporter.UNKNOWN;

@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
Expand Down Expand Up @@ -104,10 +109,12 @@ public void shouldReportWithoutGlobalDimensionsWhenGlobalDimensionsNotConfigured

final List<Dimension> dimensions = firstMetricDatumDimensionsFromCapturedRequest();

assertThat(dimensions).hasSize(1);
assertThat(dimensions).hasSize(5);
assertThat(dimensions).contains(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(DIMENSION_COUNT));
assertDefaultDimensionsWithUnknownValue(dimensions);
}


@Test
public void reportedCounterShouldContainExpectedDimension() throws Exception {
metricRegistry.counter(ARBITRARY_COUNTER_NAME).inc();
Expand All @@ -116,6 +123,7 @@ public void reportedCounterShouldContainExpectedDimension() throws Exception {
final List<Dimension> dimensions = firstMetricDatumDimensionsFromCapturedRequest();

assertThat(dimensions).contains(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(DIMENSION_COUNT));
assertDefaultDimensionsWithUnknownValue(dimensions);
}

@Test
Expand All @@ -126,6 +134,7 @@ public void reportedGaugeShouldContainExpectedDimension() throws Exception {
final List<Dimension> dimensions = firstMetricDatumDimensionsFromCapturedRequest();

assertThat(dimensions).contains(new Dimension().withName(DIMENSION_NAME_TYPE).withValue(DIMENSION_GAUGE));
assertDefaultDimensionsWithUnknownValue(dimensions);
}

@Test
Expand Down Expand Up @@ -475,6 +484,15 @@ public void shouldReportExpectedGlobalAndCustomDimensions() throws Exception {
assertThat(dimensions).contains(new Dimension().withName("key2").withValue("value2"));
}


private void assertDefaultDimensionsWithUnknownValue(List<Dimension> dimensions) {
assertThat(dimensions).contains(new Dimension().withName(DIMENSION_JOB_ID).withValue(UNKNOWN));
assertThat(dimensions).contains(new Dimension().withName(DIMENSION_INSTANCE_TYPE).withValue(UNKNOWN));
assertThat(dimensions).contains(new Dimension().withName(DIMENSION_DOMAIN_ID).withValue(UNKNOWN));
assertThat(dimensions).contains(new Dimension().withName(DIMENSION_APPLICATION_ID).withValue(UNKNOWN));
}


private MetricDatum metricDatumByDimensionFromCapturedRequest(final String dimensionValue) {
final PutMetricDataRequest putMetricDataRequest = metricDataRequestCaptor.getValue();
final List<MetricDatum> metricData = putMetricDataRequest.getMetricData();
Expand Down

0 comments on commit 791220c

Please sign in to comment.