Skip to content

Commit 57030dd

Browse files
michaelkoepfmichaelkoepf
committed
Remove HTTP exporter (there was a bug which was not caught by the IT test setup)
1 parent 351dcff commit 57030dd

File tree

8 files changed

+41
-144
lines changed

8 files changed

+41
-144
lines changed

‎fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java‎

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1617,32 +1617,13 @@ public class ConfigOptions {
16171617
// ------------------------------------------------------------------------
16181618
// ConfigOptions for OpenTelemetry reporter
16191619
// ------------------------------------------------------------------------
1620-
/** OpenTelemetry protocol. */
1621-
public enum OpenTelemetryExporter {
1622-
GRPC,
1623-
HTTP
1624-
}
1625-
16261620
public static final ConfigOption<String> METRICS_REPORTER_OPENTELEMETRY_ENDPOINT =
16271621
key("metrics.reporter.opentelemetry.endpoint")
16281622
.stringType()
16291623
.noDefaultValue()
16301624
.withDescription(
16311625
"Target to which the OpenTelemetry metric reporter is going to send metrics to.");
16321626

1633-
public static final ConfigOption<OpenTelemetryExporter>
1634-
METRICS_REPORTER_OPENTELEMETRY_EXPORTER =
1635-
key("metrics.reporter.opentelemetry.exporter")
1636-
.enumType(OpenTelemetryExporter.class)
1637-
.defaultValue(OpenTelemetryExporter.GRPC)
1638-
.withDescription(
1639-
"The type of exporter that is used by the OpenTelemetry metric exporter to send metrics to the configured endpoint. "
1640-
+ "The endpoint must accept connections for the given exporter type. Supported exporters: "
1641-
+ OpenTelemetryExporter.GRPC.name()
1642-
+ ", "
1643-
+ OpenTelemetryExporter.HTTP.name()
1644-
+ ".");
1645-
16461627
public static final ConfigOption<Duration> METRICS_REPORTER_OPENTELEMETRY_EXPORT_INTERVAL =
16471628
key("metrics.reporter.opentelemetry.export-interval")
16481629
.durationType()

‎fluss-metrics/fluss-metrics-opentelemetry/src/main/java/org/apache/fluss/metrics/opentelemetry/OpenTelemetryReporter.java‎

Lines changed: 6 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,7 @@
1818
package org.apache.fluss.metrics.opentelemetry;
1919

2020
import org.apache.fluss.annotation.VisibleForTesting;
21-
import org.apache.fluss.config.ConfigOptions;
2221
import org.apache.fluss.config.Configuration;
23-
import org.apache.fluss.exception.FlussRuntimeException;
2422
import org.apache.fluss.metrics.CharacterFilter;
2523
import org.apache.fluss.metrics.Counter;
2624
import org.apache.fluss.metrics.Gauge;
@@ -32,8 +30,6 @@
3230
import org.apache.fluss.metrics.reporter.ScheduledMetricReporter;
3331

3432
import io.opentelemetry.api.common.Attributes;
35-
import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter;
36-
import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder;
3733
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
3834
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder;
3935
import io.opentelemetry.sdk.common.CompletableResultCode;
@@ -87,36 +83,21 @@ public class OpenTelemetryReporter implements MetricReporter, ScheduledMetricRep
8783

8884
OpenTelemetryReporter(
8985
String endpoint,
90-
ConfigOptions.OpenTelemetryExporter exporterType,
9186
Duration interval,
9287
Duration timeout,
9388
String serviceName,
9489
String serviceVersion) {
95-
this.exporter = buildMetricExporter(exporterType, endpoint, timeout);
90+
this.exporter = buildMetricExporter(endpoint, timeout);
9691
this.resource = buildResource(serviceName, serviceVersion);
9792
this.interval = interval;
9893
this.clock = Clock.systemUTC();
9994
}
10095

101-
private static MetricExporter buildMetricExporter(
102-
ConfigOptions.OpenTelemetryExporter exporterType, String endpoint, Duration timeout) {
103-
switch (exporterType) {
104-
case GRPC:
105-
OtlpGrpcMetricExporterBuilder grpcExporterBuilder =
106-
OtlpGrpcMetricExporter.builder();
107-
grpcExporterBuilder.setEndpoint(endpoint);
108-
grpcExporterBuilder.setTimeout(timeout);
109-
return grpcExporterBuilder.build();
110-
case HTTP:
111-
OtlpHttpMetricExporterBuilder httpExporterBuilder =
112-
OtlpHttpMetricExporter.builder();
113-
httpExporterBuilder.setEndpoint(endpoint);
114-
httpExporterBuilder.setTimeout(timeout);
115-
return httpExporterBuilder.build();
116-
default:
117-
LOG.error("Unsupported OpenTelemetry exporter type: {}", exporterType);
118-
throw new FlussRuntimeException("OpenTelemetry exporter type: " + exporterType);
119-
}
96+
private static MetricExporter buildMetricExporter(String endpoint, Duration timeout) {
97+
OtlpGrpcMetricExporterBuilder grpcExporterBuilder = OtlpGrpcMetricExporter.builder();
98+
grpcExporterBuilder.setEndpoint(endpoint);
99+
grpcExporterBuilder.setTimeout(timeout);
100+
return grpcExporterBuilder.build();
120101
}
121102

122103
private static Resource buildResource(String serviceName, String serviceVersion) {

‎fluss-metrics/fluss-metrics-opentelemetry/src/main/java/org/apache/fluss/metrics/opentelemetry/OpenTelemetryReporterPlugin.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,6 @@ public MetricReporter createMetricReporter(Configuration configuration) {
4242
throw new IllegalConfigurationException(
4343
ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_ENDPOINT.key() + " must be set.");
4444
}
45-
ConfigOptions.OpenTelemetryExporter exporterType =
46-
configuration.get(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_EXPORTER);
47-
if (exporterType == null) {
48-
throw new IllegalConfigurationException(
49-
ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_EXPORTER.key() + " must be set.");
50-
}
5145
Duration interval =
5246
configuration.get(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_EXPORT_INTERVAL);
5347
Duration timeout =
@@ -56,8 +50,7 @@ public MetricReporter createMetricReporter(Configuration configuration) {
5650
configuration.get(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_SERVICE_NAME);
5751
String serviceVersion =
5852
configuration.get(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_SERVICE_VERSION);
59-
return new OpenTelemetryReporter(
60-
endpoint, exporterType, interval, timeout, serviceName, serviceVersion);
53+
return new OpenTelemetryReporter(endpoint, interval, timeout, serviceName, serviceVersion);
6154
}
6255

6356
@Override

‎fluss-metrics/fluss-metrics-opentelemetry/src/test/java/org/apache/fluss/metrics/opentelemetry/OpenTelemetryReporterITCase.java‎

Lines changed: 24 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
package org.apache.fluss.metrics.opentelemetry;
1919

20-
import org.apache.fluss.config.ConfigOptions;
2120
import org.apache.fluss.metrics.Gauge;
2221
import org.apache.fluss.metrics.Histogram;
2322
import org.apache.fluss.metrics.MeterView;
@@ -29,9 +28,8 @@
2928

3029
import org.assertj.core.data.Percentage;
3130
import org.junit.jupiter.api.BeforeEach;
31+
import org.junit.jupiter.api.Test;
3232
import org.junit.jupiter.api.extension.ExtendWith;
33-
import org.junit.jupiter.params.ParameterizedTest;
34-
import org.junit.jupiter.params.provider.EnumSource;
3533

3634
import java.time.Duration;
3735
import java.util.HashMap;
@@ -58,10 +56,9 @@ public void setUpEach() {
5856
group = TestUtils.createTestMetricGroup(LOGICAL_SCOPE, new HashMap<>());
5957
}
6058

61-
@ParameterizedTest
62-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
63-
public void testReport(ConfigOptions.OpenTelemetryExporter exporterType) throws Exception {
64-
OpenTelemetryReporter reporter = createReporter(exporterType);
59+
@Test
60+
public void testReport() throws Exception {
61+
OpenTelemetryReporter reporter = createReporter();
6562

6663
SimpleCounter counter = new SimpleCounter();
6764
reporter.notifyOfAddedMetric(counter, "foo.counter", group);
@@ -117,11 +114,9 @@ private static void assertMetrics(JsonNode metric) {
117114
}
118115
}
119116

120-
@ParameterizedTest
121-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
122-
public void testReportAfterUnregister(ConfigOptions.OpenTelemetryExporter exporterType)
123-
throws Exception {
124-
OpenTelemetryReporter reporter = createReporter(exporterType);
117+
@Test
118+
public void testReportAfterUnregister() throws Exception {
119+
OpenTelemetryReporter reporter = createReporter();
125120

126121
SimpleCounter counter1 = new SimpleCounter();
127122
SimpleCounter counter2 = new SimpleCounter();
@@ -145,11 +140,9 @@ public void testReportAfterUnregister(ConfigOptions.OpenTelemetryExporter export
145140
});
146141
}
147142

148-
@ParameterizedTest
149-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
150-
public void testCounterDelta(ConfigOptions.OpenTelemetryExporter exporterType)
151-
throws Exception {
152-
OpenTelemetryReporter reporter = createReporter(exporterType);
143+
@Test
144+
public void testCounterDelta() throws Exception {
145+
OpenTelemetryReporter reporter = createReporter();
153146

154147
SimpleCounter counter = new SimpleCounter();
155148
reporter.notifyOfAddedMetric(counter, "foo.counter", group);
@@ -199,13 +192,11 @@ public void testCounterDelta(ConfigOptions.OpenTelemetryExporter exporterType)
199192
});
200193
}
201194

202-
@ParameterizedTest
203-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
204-
public void testOpenTelemetryAttributes(ConfigOptions.OpenTelemetryExporter exporterType)
205-
throws Exception {
195+
@Test
196+
public void testOpenTelemetryAttributes() throws Exception {
206197
String serviceName = "flink-bar";
207198
String serviceVersion = "v42";
208-
OpenTelemetryReporter reporter = createReporter(exporterType, serviceName, serviceVersion);
199+
OpenTelemetryReporter reporter = createReporter(serviceName, serviceVersion);
209200

210201
SimpleCounter counter = new SimpleCounter();
211202
reporter.notifyOfAddedMetric(counter, "foo.counter", group);
@@ -246,39 +237,19 @@ public void testOpenTelemetryAttributes(ConfigOptions.OpenTelemetryExporter expo
246237
});
247238
}
248239

249-
private static OpenTelemetryReporter createReporter(
250-
ConfigOptions.OpenTelemetryExporter exporterType,
251-
String serviceName,
252-
String serviceVersion) {
253-
String endpoint;
254-
switch (exporterType) {
255-
case GRPC:
256-
endpoint =
257-
OpenTelemetryReporterITCaseBase.getOpenTelemetryContainer()
258-
.getGrpcEndpoint();
259-
break;
260-
case HTTP:
261-
endpoint =
262-
OpenTelemetryReporterITCaseBase.getOpenTelemetryContainer()
263-
.getHttpEndpoint();
264-
break;
265-
default:
266-
throw new IllegalStateException("OpenTelemetry exporter type: " + exporterType);
267-
}
240+
private static OpenTelemetryReporter createReporter(String serviceName, String serviceVersion) {
241+
String endpoint =
242+
OpenTelemetryReporterITCaseBase.getOpenTelemetryContainer().getGrpcEndpoint();
268243

269-
OpenTelemetryReporter reporter =
270-
new OpenTelemetryReporter(
271-
endpoint,
272-
exporterType,
273-
Duration.ofSeconds(10),
274-
Duration.ofSeconds(10),
275-
serviceName,
276-
serviceVersion);
277-
return reporter;
244+
return new OpenTelemetryReporter(
245+
endpoint,
246+
Duration.ofSeconds(10),
247+
Duration.ofSeconds(10),
248+
serviceName,
249+
serviceVersion);
278250
}
279251

280-
private static OpenTelemetryReporter createReporter(
281-
ConfigOptions.OpenTelemetryExporter exporterType) {
282-
return createReporter(exporterType, null, null);
252+
private static OpenTelemetryReporter createReporter() {
253+
return createReporter(null, null);
283254
}
284255
}

‎fluss-metrics/fluss-metrics-opentelemetry/src/test/java/org/apache/fluss/metrics/opentelemetry/OpenTelemetryReporterPluginTest.java‎

Lines changed: 6 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,7 @@
2121
import org.apache.fluss.config.Configuration;
2222
import org.apache.fluss.exception.IllegalConfigurationException;
2323

24-
import org.junit.jupiter.params.ParameterizedTest;
25-
import org.junit.jupiter.params.provider.EnumSource;
24+
import org.junit.jupiter.api.Test;
2625

2726
import java.time.Duration;
2827

@@ -35,17 +34,13 @@ public class OpenTelemetryReporterPluginTest {
3534
private final OpenTelemetryReporterPlugin openTelemetryReporterPlugin =
3635
new OpenTelemetryReporterPlugin();
3736

38-
@ParameterizedTest
39-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
40-
void testValidConfiguration(ConfigOptions.OpenTelemetryExporter exporterType) {
37+
@Test
38+
void testValidConfiguration() {
4139
// mandatory options
4240
Configuration configuration = new Configuration();
4341
configuration.setString(
4442
ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_ENDPOINT,
4543
"http://opentelemetry-metric-collector:4317");
46-
configuration.set(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_EXPORTER, exporterType);
47-
assertThatCode(() -> openTelemetryReporterPlugin.createMetricReporter(configuration))
48-
.doesNotThrowAnyException();
4944

5045
// optional options
5146
configuration.set(
@@ -62,11 +57,10 @@ void testValidConfiguration(ConfigOptions.OpenTelemetryExporter exporterType) {
6257
.doesNotThrowAnyException();
6358
}
6459

65-
@ParameterizedTest
66-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
67-
void testInvalidConfiguration(ConfigOptions.OpenTelemetryExporter exporterType) {
60+
@Test
61+
void testInvalidConfiguration() {
6862
Configuration configuration = new Configuration();
69-
// invalid endpoint and no exporter type
63+
7064
assertThatThrownBy(() -> openTelemetryReporterPlugin.createMetricReporter(configuration))
7165
.isInstanceOf(IllegalConfigurationException.class);
7266

@@ -77,10 +71,5 @@ void testInvalidConfiguration(ConfigOptions.OpenTelemetryExporter exporterType)
7771
configuration.setString(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_ENDPOINT, " ");
7872
assertThatThrownBy(() -> openTelemetryReporterPlugin.createMetricReporter(configuration))
7973
.isInstanceOf(IllegalConfigurationException.class);
80-
81-
// endpoint is still invalid
82-
configuration.set(ConfigOptions.METRICS_REPORTER_OPENTELEMETRY_EXPORTER, exporterType);
83-
assertThatThrownBy(() -> openTelemetryReporterPlugin.createMetricReporter(configuration))
84-
.isInstanceOf(IllegalConfigurationException.class);
8574
}
8675
}

‎fluss-metrics/fluss-metrics-opentelemetry/src/test/java/org/apache/fluss/metrics/opentelemetry/OpenTelemetryReporterTest.java‎

Lines changed: 4 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717

1818
package org.apache.fluss.metrics.opentelemetry;
1919

20-
import org.apache.fluss.config.ConfigOptions;
2120
import org.apache.fluss.metrics.Counter;
2221
import org.apache.fluss.metrics.Gauge;
2322
import org.apache.fluss.metrics.Histogram;
@@ -30,8 +29,6 @@
3029
import io.opentelemetry.semconv.ServiceAttributes;
3130
import org.junit.jupiter.api.BeforeEach;
3231
import org.junit.jupiter.api.Test;
33-
import org.junit.jupiter.params.ParameterizedTest;
34-
import org.junit.jupiter.params.provider.EnumSource;
3532

3633
import java.time.Duration;
3734
import java.util.AbstractMap;
@@ -64,14 +61,12 @@ void setupReporter() {
6461
metricGroup = TestUtils.createTestMetricGroup(LOGICAL_SCOPE, labels);
6562
}
6663

67-
@ParameterizedTest
68-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
69-
void testInvalidEndpoint(ConfigOptions.OpenTelemetryExporter exporterType) {
64+
@Test
65+
void testInvalidEndpoint() {
7066
assertThatThrownBy(
7167
() ->
7268
new OpenTelemetryReporter(
7369
"endpoint-with-missing-protocol",
74-
exporterType,
7570
Duration.ofSeconds(5),
7671
Duration.ofSeconds(5),
7772
null,
@@ -82,22 +77,18 @@ void testInvalidEndpoint(ConfigOptions.OpenTelemetryExporter exporterType) {
8277
() ->
8378
new OpenTelemetryReporter(
8479
"invalid://protocol",
85-
exporterType,
8680
Duration.ofSeconds(5),
8781
Duration.ofSeconds(5),
8882
null,
8983
null))
9084
.isInstanceOf(IllegalArgumentException.class);
9185
}
9286

93-
@ParameterizedTest
94-
@EnumSource(ConfigOptions.OpenTelemetryExporter.class)
95-
void testOpenTelemetryResourceIsConstructedCorrectly(
96-
ConfigOptions.OpenTelemetryExporter exporterType) {
87+
@Test
88+
void testOpenTelemetryResourceIsConstructedCorrectly() {
9789
OpenTelemetryReporter reporter =
9890
new OpenTelemetryReporter(
9991
"http://opentelemetry-collector:4317",
100-
exporterType,
10192
Duration.ofSeconds(5),
10293
Duration.ofSeconds(5),
10394
"fluss",
@@ -162,7 +153,6 @@ void testRemoveEnclosingAngleBrackets() {
162153
private OpenTelemetryReporter createReporter() {
163154
return new OpenTelemetryReporter(
164155
"http://endpoint-must-not-be-called-in-unit-tests",
165-
ConfigOptions.OpenTelemetryExporter.GRPC,
166156
Duration.ofSeconds(5),
167157
Duration.ofSeconds(5),
168158
null,

‎fluss-metrics/fluss-metrics-opentelemetry/src/test/java/org/apache/fluss/metrics/opentelemetry/OpenTelemetryTestContainer.java‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@ public class OpenTelemetryTestContainer extends GenericContainer<OpenTelemetryTe
3838
private static final String ALPINE_DOCKER_IMAGE_TAG = "3.22.0";
3939
private static final String OPENTELEMETRY_COLLECTOR_DOCKER_IMAGE_TAG = "0.128.0";
4040

41-
private static final int DEFAULT_HTTP_PORT = 4318;
4241
private static final int DEFAULT_GRPC_PORT = 4317;
4342

4443
// must be kept in sync with opentelemetry-config.yaml
@@ -55,7 +54,6 @@ public OpenTelemetryTestContainer() {
5554
.withDockerfileFromBuilder(
5655
OpenTelemetryTestContainer::buildOpenTelemetryCollectorImage));
5756
withNetworkAliases(randomString("opentelemetry-collector", 6));
58-
addExposedPort(DEFAULT_HTTP_PORT);
5957
addExposedPort(DEFAULT_GRPC_PORT);
6058
withCopyFileToContainer(
6159
MountableFile.forHostPath(CONFIG_PATH.toString()), "opentelemetry-config.yaml");
@@ -93,10 +91,6 @@ private static String randomString(String prefix, int length) {
9391
return String.format("%s-%s", prefix, Base58.randomString(length).toLowerCase(Locale.ROOT));
9492
}
9593

96-
public String getHttpEndpoint() {
97-
return String.format("http://%s:%s", getHost(), getMappedPort(DEFAULT_HTTP_PORT));
98-
}
99-
10094
public String getGrpcEndpoint() {
10195
return String.format("http://%s:%s", getHost(), getMappedPort(DEFAULT_GRPC_PORT));
10296
}

0 commit comments

Comments
 (0)