Skip to content

Commit cc50db5

Browse files
xwm1992xwm1992
authored
[ISSUE #5137] update connector runtime v2 module (#5138)
* [ISSUE #5137] update connector runtime v2 module * fix checkStyle error
1 parent cf1c07f commit cc50db5

File tree

13 files changed

+872
-141
lines changed

13 files changed

+872
-141
lines changed

‎eventmesh-common/src/main/java/org/apache/eventmesh/common/remote/request/ReportMonitorRequest.java

+38
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.eventmesh.common.remote.request;
19+
20+
import lombok.Data;
21+
import lombok.EqualsAndHashCode;
22+
import lombok.ToString;
23+
24+
@Data
25+
@EqualsAndHashCode(callSuper = true)
26+
@ToString
27+
public class ReportMonitorRequest extends BaseRemoteRequest {
28+
private String taskID;
29+
private String jobID;
30+
private String address;
31+
private String connectorStage;
32+
private String transportType;
33+
private long totalReqNum;
34+
private long totalTimeCost;
35+
private long maxTimeCost;
36+
private long avgTimeCost;
37+
private double tps;
38+
}

‎eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/monitor/AbstractConnectorMonitor.java

+80
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.eventmesh.openconnect.api.monitor;
19+
20+
import java.util.concurrent.atomic.AtomicLong;
21+
import java.util.concurrent.atomic.LongAdder;
22+
23+
import lombok.Getter;
24+
import lombok.extern.slf4j.Slf4j;
25+
26+
@Slf4j
27+
@Getter
28+
public abstract class AbstractConnectorMonitor implements Monitor {
29+
30+
private final String taskId;
31+
private final String jobId;
32+
private final String ip;
33+
private final LongAdder totalRecordNum;
34+
private final LongAdder totalTimeCost;
35+
protected final AtomicLong startTime;
36+
private final AtomicLong maxTimeCost;
37+
private long averageTime = 0;
38+
private double tps = 0;
39+
40+
public AbstractConnectorMonitor(String taskId, String jobId, String ip) {
41+
this.taskId = taskId;
42+
this.jobId = jobId;
43+
this.ip = ip;
44+
this.totalRecordNum = new LongAdder();
45+
this.totalTimeCost = new LongAdder();
46+
this.startTime = new AtomicLong(System.currentTimeMillis());
47+
this.maxTimeCost = new AtomicLong();
48+
}
49+
50+
@Override
51+
public synchronized void recordProcess(long timeCost) {
52+
totalRecordNum.increment();
53+
totalTimeCost.add(timeCost);
54+
maxTimeCost.updateAndGet(max -> Math.max(max, timeCost));
55+
}
56+
57+
@Override
58+
public synchronized void recordProcess(int recordCount, long timeCost) {
59+
totalRecordNum.add(recordCount);
60+
totalTimeCost.add(timeCost);
61+
maxTimeCost.updateAndGet(max -> Math.max(max, timeCost));
62+
}
63+
64+
@Override
65+
public synchronized void printMetrics() {
66+
long totalRecords = totalRecordNum.sum();
67+
long totalCost = totalTimeCost.sum();
68+
averageTime = totalRecords > 0 ? totalCost / totalRecords : 0;
69+
long elapsedTime = (System.currentTimeMillis() - startTime.get()) / 1000; // in seconds
70+
tps = elapsedTime > 0 ? (double) totalRecords / elapsedTime : 0;
71+
72+
log.info("========== Metrics ==========");
73+
log.info("TaskId: {}|JobId: {}|ip: {}", taskId, jobId, ip);
74+
log.info("Total records: {}", totalRecordNum);
75+
log.info("Total time (ms): {}", totalTimeCost);
76+
log.info("Max time per record (ms): {}", maxTimeCost);
77+
log.info("Average time per record (ms): {}", averageTime);
78+
log.info("TPS: {}", tps);
79+
}
80+
}

‎eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/monitor/Monitor.java

+30
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.eventmesh.openconnect.api.monitor;
19+
20+
/**
21+
* Monitor Interface.
22+
* All monitors should implement this interface.
23+
*/
24+
public interface Monitor {
25+
void recordProcess(long timeCost);
26+
27+
void recordProcess(int recordCount, long timeCost);
28+
29+
void printMetrics();
30+
}

‎eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/api/monitor/MonitorRegistry.java

+34
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.eventmesh.openconnect.api.monitor;
19+
20+
import java.util.ArrayList;
21+
import java.util.List;
22+
23+
import lombok.Getter;
24+
25+
public class MonitorRegistry {
26+
27+
@Getter
28+
private static final List<Monitor> monitors = new ArrayList<>();
29+
30+
public static void registerMonitor(Monitor monitor) {
31+
monitors.add(monitor);
32+
}
33+
34+
}

‎eventmesh-runtime-v2/src/main/java/org/apache/eventmesh/runtime/boot/RuntimeInstanceStarter.java

-1
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,6 @@ public static void main(String[] args) {
4040
long start = System.currentTimeMillis();
4141
runtimeInstance.shutdown();
4242
long end = System.currentTimeMillis();
43-
4443
log.info("runtime shutdown cost {}ms", end - start);
4544
} catch (Exception e) {
4645
log.error("exception when shutdown {}", e.getMessage(), e);

0 commit comments

Comments
 (0)