Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Feature][Zeta] Added other metrics info of multi-table #7338

Merged
merged 7 commits into from
Aug 14, 2024
Merged

[Feature][Zeta] Added other metrics info of multi-table #7338

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
2de971b
[Feature][Zeta] Added other metrics info of multi-table
corgy-w Aug 7, 2024
fbb3ee9
Merge branch 'refs/heads/dev' into multi-table-metrics
corgy-w Aug 8, 2024
b0ef6dd
[Feature][Zeta] Added other metrics restApi
corgy-w Aug 8, 2024
f9723fe
[Feature][Zeta] Added other metrics restApi test
corgy-w Aug 9, 2024
3e6cb8d
[Feature][Zeta] update rest-api doc
corgy-w Aug 9, 2024
4446216
[Improve][Connector-V2] optimize code
corgy-w Aug 12, 2024
9e8d982
[Improve][Connector-V2] optimize code
corgy-w Aug 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions docs/en/seatunnel-engine/rest-api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,8 +192,22 @@ This API has been deprecated, please use /hazelcast/rest/maps/job-info/:jobId in
]
},
"metrics": {
"sourceReceivedCount": "",
"sinkWriteCount": ""
"SourceReceivedCount": "",
"SourceReceivedQPS": "",
"SourceReceivedBytes": "",
"SourceReceivedBytesPerSeconds": "",
"SinkWriteCount": "",
"SinkWriteQPS": "",
"SinkWriteBytes": "",
"SinkWriteBytesPerSeconds": "",
"TableSourceReceivedCount": {},
"TableSourceReceivedBytes": {},
"TableSourceReceivedBytesPerSeconds": {},
"TableSourceReceivedQPS": {},
"TableSinkWriteCount": {},
"TableSinkWriteQPS": {},
"TableSinkWriteBytes": {},
"TableSinkWriteBytesPerSeconds": {}
},
"finishedTime": "",
"errorMsg": null,
Expand Down
18 changes: 16 additions & 2 deletions docs/zh/seatunnel-engine/rest-api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,8 +134,22 @@ network:
]
},
"metrics": {
"sourceReceivedCount": "",
"sinkWriteCount": ""
"SourceReceivedCount": "",
"SourceReceivedQPS": "",
"SourceReceivedBytes": "",
"SourceReceivedBytesPerSeconds": "",
"SinkWriteCount": "",
"SinkWriteQPS": "",
"SinkWriteBytes": "",
"SinkWriteBytesPerSeconds": "",
"TableSourceReceivedCount": {},
"TableSourceReceivedBytes": {},
"TableSourceReceivedBytesPerSeconds": {},
"TableSourceReceivedQPS": {},
"TableSinkWriteCount": {},
"TableSinkWriteQPS": {},
"TableSinkWriteBytes": {},
"TableSinkWriteBytesPerSeconds": {}
},
"finishedTime": "",
"errorMsg": null,
Expand Down
101 changes: 85 additions & 16 deletions ...seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/MultiTableMetricsIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@

import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.instance.impl.HazelcastInstanceImpl;
import io.restassured.response.Response;

import java.util.Collections;
import java.util.concurrent.TimeUnit;
Expand Down Expand Up @@ -82,33 +83,101 @@ public void multiTableMetrics() {
Collections.singletonList(node1)
.forEach(
instance -> {
given().get(
HOST
+ instance.getCluster()
.getLocalMember()
.getAddress()
.getPort()
+ RestConstant.JOB_INFO_URL
+ "/"
+ batchJobProxy.getJobId())
.then()
Response response =
given().get(
HOST
+ instance.getCluster()
.getLocalMember()
.getAddress()
.getPort()
+ RestConstant.JOB_INFO_URL
+ "/"
+ batchJobProxy.getJobId());
// In the test example, the data size of a single [3, "C", 100] is 13
int dataSize = 13;
response.prettyPrint();
response.then()
.statusCode(200)
.body("jobName", equalTo("batch_fake_multi_table_to_console"))
.body("jobStatus", equalTo("FINISHED"))
.body("metrics.SourceReceivedCount", equalTo("50"))
.body("metrics.SinkWriteCount", equalTo("50"))
.body("metrics.SourceReceivedCount", equalTo("15"))
.body("metrics.SinkWriteCount", equalTo("15"))
.body(
"metrics.TableSourceReceivedCount.'fake.table1'",
equalTo("20"))
equalTo("10"))
.body(
"metrics.TableSourceReceivedCount.'fake.public.table2'",
equalTo("30"))
equalTo("5"))
.body(
"metrics.TableSinkWriteCount.'fake.table1'",
equalTo("20"))
equalTo("10"))
.body(
"metrics.TableSinkWriteCount.'fake.public.table2'",
equalTo("30"));
equalTo("5"))
.body(
"metrics.SourceReceivedBytes",
equalTo(String.valueOf(dataSize * 15)))
.body(
"metrics.SinkWriteBytes",
equalTo(String.valueOf(dataSize * 15)))
.body(
"metrics.TableSourceReceivedBytes.'fake.table1'",
equalTo(String.valueOf(dataSize * 10)))
.body(
"metrics.TableSourceReceivedBytes.'fake.public.table2'",
equalTo(String.valueOf(dataSize * 5)))
.body(
"metrics.TableSinkWriteBytes.'fake.table1'",
equalTo(String.valueOf(dataSize * 10)))
.body(
"metrics.TableSinkWriteBytes.'fake.public.table2'",
equalTo(String.valueOf(dataSize * 5)));
Assertions.assertTrue(
Double.parseDouble(response.path("metrics.SourceReceivedQPS"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSourceReceivedQPS.'fake.table1'"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSourceReceivedQPS.'fake.public.table2'"))
> 0
&& Double.parseDouble(
response.path("metrics.SinkWriteQPS"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSinkWriteQPS.'fake.table1'"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSinkWriteQPS.'fake.public.table2'"))
> 0
&& Double.parseDouble(
response.path(
"metrics.SourceReceivedBytesPerSeconds"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSourceReceivedBytesPerSeconds.'fake.table1'"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSourceReceivedBytesPerSeconds.'fake.public.table2'"))
> 0
&& Double.parseDouble(
response.path(
"metrics.SinkWriteBytesPerSeconds"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSinkWriteBytesPerSeconds.'fake.table1'"))
> 0
&& Double.parseDouble(
response.path(
"metrics.TableSinkWriteBytesPerSeconds.'fake.public.table2'"))
> 0);
});
}

Expand Down
102 changes: 83 additions & 19 deletions ...2e/connector-seatunnel-e2e-base/src/test/resources/batch_fake_multi_table_to_console.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,29 +26,93 @@ env {
source {
# This is a example source plugin **only for test and demonstrate the feature source plugin**
FakeSource {
result_table_name = "fake1"
row.num = 20
schema = {
table = "fake.table1"
fields {
name = "string"
age = "int"
result_table_name = "fake1"
schema = {
table = "fake.table1"
fields {
id = bigint
name = string
score = int
}
}
rows = [
{
kind = INSERT
fields = [1, "A", 100]
},
{
kind = INSERT
fields = [2, "B", 100]
},
{
kind = INSERT
fields = [3, "C", 100]
},
{
kind = INSERT
fields = [3, "C", 100]
},
{
kind = INSERT
fields = [3, "C", 100]
},
{
kind = INSERT
fields = [3, "C", 100]
}
{
kind = UPDATE_BEFORE
fields = [1, "A", 100]
},
{
kind = UPDATE_AFTER
fields = [1, "A", 300]
},
{
kind = DELETE
fields = [2, "B", 100]
},
{
kind = INSERT
fields = [2, "B", 100]
}
]
}
}

FakeSource {
result_table_name = "fake2"
row.num = 30
schema = {
table = "fake.public.table2"
fields {
name = "string"
age = "int"
sex = "int"
FakeSource {
result_table_name = "fake2"
schema = {
table = "fake.public.table2"
fields {
id = bigint
name = string
score = int
}
}
rows = [
{
kind = INSERT
fields = [1, "A", 100]
},
{
kind = INSERT
fields = [2, "B", 100]
},
{
kind = DELETE
fields = [2, "B", 100]
},
{
kind = INSERT
fields = [3, "C", 100]
},
{
kind = INSERT
fields = [3, "C", 100]
}
]
}
}
}

}

transform {
Expand Down
50 changes: 46 additions & 4 deletions ...l-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,12 @@
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES_PER_SECONDS;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_COUNT;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_QPS;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES_PER_SECONDS;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_COUNT;
import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_QPS;
import static org.awaitility.Awaitility.await;
Expand Down Expand Up @@ -592,6 +596,23 @@ public void testGetMultiTableJobMetrics() {
jobMetrics.contains(SOURCE_RECEIVED_COUNT + "#fake.public.table2"));
Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake.table1"));
Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake.public.table2"));
Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_BYTES + "#fake.table1"));
Assertions.assertTrue(
jobMetrics.contains(SOURCE_RECEIVED_BYTES + "#fake.public.table2"));
Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_BYTES + "#fake.table1"));
Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_BYTES + "#fake.public.table2"));
Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_QPS + "#fake.table1"));
Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_QPS + "#fake.public.table2"));
Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_QPS + "#fake.table1"));
Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_QPS + "#fake.public.table2"));
Assertions.assertTrue(
jobMetrics.contains(SOURCE_RECEIVED_BYTES_PER_SECONDS + "#fake.table1"));
Assertions.assertTrue(
jobMetrics.contains(SOURCE_RECEIVED_BYTES_PER_SECONDS + "#fake.public.table2"));
Assertions.assertTrue(
jobMetrics.contains(SINK_WRITE_BYTES_PER_SECONDS + "#fake.table1"));
Assertions.assertTrue(
jobMetrics.contains(SINK_WRITE_BYTES_PER_SECONDS + "#fake.public.table2"));

log.info("jobMetrics : {}", jobMetrics);
JsonNode jobMetricsStr = new ObjectMapper().readTree(jobMetrics);
Expand All @@ -600,10 +621,6 @@ public void testGetMultiTableJobMetrics() {
Spliterators.spliteratorUnknownSize(
jobMetricsStr.fieldNames(), 0),
false)
.filter(
metricName ->
metricName.startsWith(SOURCE_RECEIVED_COUNT)
|| metricName.startsWith(SINK_WRITE_COUNT))
.collect(Collectors.toList());

Map<String, Long> totalCount =
Expand Down Expand Up @@ -654,6 +671,31 @@ public void testGetMultiTableJobMetrics() {
.filter(e -> e.getKey().startsWith(SINK_WRITE_COUNT))
.mapToLong(Map.Entry::getValue)
.sum());
Assertions.assertEquals(
totalCount.get(SOURCE_RECEIVED_BYTES),
tableCount.entrySet().stream()
.filter(e -> e.getKey().startsWith(SOURCE_RECEIVED_BYTES + "#"))
.mapToLong(Map.Entry::getValue)
.sum());
Assertions.assertEquals(
totalCount.get(SINK_WRITE_BYTES),
tableCount.entrySet().stream()
.filter(e -> e.getKey().startsWith(SINK_WRITE_BYTES + "#"))
.mapToLong(Map.Entry::getValue)
.sum());
// Instantaneous rates in the same direction are directly added
Assertions.assertEquals(
totalCount.get(SOURCE_RECEIVED_QPS),
tableCount.entrySet().stream()
.filter(e -> e.getKey().startsWith(SOURCE_RECEIVED_QPS + "#"))
.mapToLong(Map.Entry::getValue)
.sum());
Assertions.assertEquals(
totalCount.get(SINK_WRITE_QPS),
tableCount.entrySet().stream()
.filter(e -> e.getKey().startsWith(SINK_WRITE_QPS + "#"))
.mapToLong(Map.Entry::getValue)
.sum());

} catch (ExecutionException | InterruptedException | JsonProcessingException e) {
throw new RuntimeException(e);
Expand Down
Loading
Loading