This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 1641cd6d92 [INLONG-8372][Sort] MySQL connector supports uploading
flink job delay metrics #8373
1641cd6d92 is described below
commit 1641cd6d92e6ad7ef28ded874ec935612073ee5a
Author: Liao Rui <[email protected]>
AuthorDate: Fri Jun 30 10:17:14 2023 +0800
[INLONG-8372][Sort] MySQL connector supports uploading flink job delay
metrics #8373
---
.../org/apache/inlong/sort/base/Constants.java | 5 ++
.../apache/inlong/sort/base/metric/MetricData.java | 15 ++++
.../inlong/sort/base/metric/SourceMetricData.java | 75 +++++++++++++++++++
.../base/metric/sub/SourceTableMetricData.java | 83 ++++++++++++++++++++++
.../source/metrics/MySqlSourceReaderMetrics.java | 3 +-
5 files changed, 180 insertions(+), 1 deletion(-)
diff --git
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
index 84cf2c51eb..5ac90a8077 100644
---
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
+++
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/Constants.java
@@ -62,6 +62,11 @@ public final class Constants {
public static final String NUM_BYTES_IN_PER_SECOND = "numBytesInPerSecond";
public static final String NUM_RECORDS_IN_PER_SECOND =
"numRecordsInPerSecond";
+
+ public static final String CURRENT_FETCH_EVENT_TIME_LAG =
"currentFetchEventTimeLag";
+
+ public static final String CURRENT_EMIT_EVENT_TIME_LAG =
"currentEmitEventTimeLag";
+
/**
* Timestamp when the read phase changed
*/
diff --git
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricData.java
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricData.java
index 381fa5ac4a..a67f0fdbb1 100644
---
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricData.java
+++
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/MetricData.java
@@ -18,6 +18,7 @@
package org.apache.inlong.sort.base.metric;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.MeterView;
import org.apache.flink.metrics.MetricGroup;
@@ -115,4 +116,18 @@ public interface MetricData {
return inlongMetricGroup.meter(metricName, new MeterView(counter,
TIME_SPAN_IN_SECONDS));
}
+ /**
+ * Register a gauge metric
+ *
+ * @param metricName The gauge name
+ * @return Gauge of registered
+ */
+ default Gauge registerGauge(String metricName, Gauge gauge) {
+ MetricGroup inlongMetricGroup = getMetricGroup();
+ for (Map.Entry<String, String> label : getLabels().entrySet()) {
+ inlongMetricGroup = inlongMetricGroup.addGroup(label.getKey(),
label.getValue());
+ }
+ return inlongMetricGroup.gauge(metricName, gauge);
+ }
+
}
diff --git
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java
index e62d2c6a63..ec64f53b39 100644
---
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java
+++
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceMetricData.java
@@ -20,6 +20,7 @@ package org.apache.inlong.sort.base.metric;
import org.apache.inlong.audit.AuditOperator;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.SimpleCounter;
@@ -29,6 +30,8 @@ import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
+import static
org.apache.inlong.sort.base.Constants.CURRENT_EMIT_EVENT_TIME_LAG;
+import static
org.apache.inlong.sort.base.Constants.CURRENT_FETCH_EVENT_TIME_LAG;
import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN;
import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN_FOR_METER;
import static org.apache.inlong.sort.base.Constants.NUM_BYTES_IN_PER_SECOND;
@@ -54,6 +57,29 @@ public class SourceMetricData implements MetricData {
private AuditOperator auditOperator;
private List<Integer> auditKeys;
+ /**
+ * currentFetchEventTimeLag = FetchTime - messageTimestamp, where the
FetchTime is the time the
+ * record fetched into the source operator.
+ */
+ private Gauge currentFetchEventTimeLag;
+ /**
+ * currentEmitEventTimeLag = EmitTime - messageTimestamp, where the
EmitTime is the time the record leaves the
+ * source operator.
+ */
+ private Gauge currentEmitEventTimeLag;
+
+ /**
+ * fetchDelay = FetchTime - messageTimestamp, where the FetchTime is the
time the
+ * record fetched into the source operator.
+ */
+ private volatile long fetchDelay = 0L;
+
+ /**
+ * emitDelay = EmitTime - messageTimestamp, where the EmitTime is the time
the record leaves the
+ * source operator.
+ */
+ private volatile long emitDelay = 0L;
+
public SourceMetricData(MetricOption option, MetricGroup metricGroup) {
this.metricGroup = metricGroup;
this.labels = option.getLabels();
@@ -70,6 +96,8 @@ public class SourceMetricData implements MetricData {
registerMetricsForNumRecordsInForMeter(new
ThreadSafeCounter());
registerMetricsForNumBytesInPerSecond();
registerMetricsForNumRecordsInPerSecond();
+ registerMetricsForCurrentFetchEventTimeLag();
+ registerMetricsForCurrentEmitEventTimeLag();
break;
}
@@ -160,6 +188,14 @@ public class SourceMetricData implements MetricData {
numBytesInPerSecond = registerMeter(NUM_BYTES_IN_PER_SECOND,
this.numBytesInForMeter);
}
+ public void registerMetricsForCurrentFetchEventTimeLag() {
+ currentFetchEventTimeLag = registerGauge(CURRENT_FETCH_EVENT_TIME_LAG,
(Gauge<Long>) this::getFetchDelay);
+ }
+
+ public void registerMetricsForCurrentEmitEventTimeLag() {
+ currentEmitEventTimeLag = registerGauge(CURRENT_EMIT_EVENT_TIME_LAG,
(Gauge<Long>) this::getEmitDelay);
+ }
+
public Counter getNumRecordsIn() {
return numRecordsIn;
}
@@ -184,6 +220,14 @@ public class SourceMetricData implements MetricData {
return numBytesInForMeter;
}
+ public long getFetchDelay() {
+ return fetchDelay;
+ }
+
+ public long getEmitDelay() {
+ return emitDelay;
+ }
+
@Override
public MetricGroup getMetricGroup() {
return metricGroup;
@@ -198,6 +242,12 @@ public class SourceMetricData implements MetricData {
outputMetrics(1, getDataSize(data));
}
+ public void outputMetricsWithEstimate(Object data, long fetchDelay, long
emitDelay) {
+ outputMetrics(1, getDataSize(data));
+ this.fetchDelay = fetchDelay;
+ this.emitDelay = emitDelay;
+ }
+
public void outputMetricsWithEstimate(Object data, long dataTime) {
outputMetrics(1, getDataSize(data), dataTime);
}
@@ -219,6 +269,23 @@ public class SourceMetricData implements MetricData {
}
}
+ public void outputMetrics(long rowCountSize, long rowDataSize, long
fetchDelay, long emitDelay) {
+ outputDefaultMetrics(rowCountSize, rowDataSize, fetchDelay, emitDelay);
+
+ if (auditOperator != null) {
+ for (Integer key : auditKeys) {
+ auditOperator.add(
+ key,
+ getGroupId(),
+ getStreamId(),
+ System.currentTimeMillis(),
+ rowCountSize,
+ rowDataSize);
+ }
+
+ }
+ }
+
public void outputMetrics(long rowCountSize, long rowDataSize, long
dataTime) {
outputDefaultMetrics(rowCountSize, rowDataSize);
@@ -253,6 +320,12 @@ public class SourceMetricData implements MetricData {
}
}
+ private void outputDefaultMetrics(long rowCountSize, long rowDataSize,
long fetchDelay, long emitDelay) {
+ outputDefaultMetrics(rowCountSize, rowDataSize);
+ this.fetchDelay = fetchDelay;
+ this.emitDelay = emitDelay;
+ }
+
@Override
public String toString() {
return "SourceMetricData{"
@@ -264,6 +337,8 @@ public class SourceMetricData implements MetricData {
+ ", numBytesInForMeter=" + numBytesInForMeter.getCount()
+ ", numRecordsInPerSecond=" + numRecordsInPerSecond.getRate()
+ ", numBytesInPerSecond=" + numBytesInPerSecond.getRate()
+ + ", currentFetchEventTimeLag=" +
currentFetchEventTimeLag.getValue()
+ + ", currentEmitEventTimeLag=" +
currentEmitEventTimeLag.getValue()
+ ", auditOperator=" + auditOperator
+ '}';
}
diff --git
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java
index 80072b95ff..65aefd91d4 100644
---
a/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java
+++
b/inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java
@@ -24,6 +24,7 @@ import
org.apache.inlong.sort.base.metric.MetricOption.RegisteredMetric;
import org.apache.inlong.sort.base.metric.MetricState;
import org.apache.inlong.sort.base.metric.SourceMetricData;
import org.apache.inlong.sort.base.metric.phase.ReadPhaseMetricData;
+import org.apache.inlong.sort.base.util.CalculateObjectSizeUtils;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
@@ -161,6 +162,28 @@ public class SourceTableMetricData extends
SourceMetricData implements SourceSub
outputMetricsWithEstimate(new String[]{database, table},
isSnapshotRecord, data);
}
+ /**
+ * output metrics with estimate
+ *
+ * @param database the database name of record
+ * @param table the table name of record
+ * @param isSnapshotRecord is it snapshot record
+ * @param data the data of record
+ * @param fetchDelay fetchDelay = FetchTime - messageTimestamp, where the
FetchTime is the time the record fetched into the source operator.
+ * @param emitDelay emitDelay = EmitTime - messageTimestamp, where the
EmitTime is the time the record leaves the source operator.
+ */
+ public void outputMetricsWithEstimate(String database, String table,
boolean isSnapshotRecord, Object data,
+ long fetchDelay, long emitDelay) {
+ if (StringUtils.isBlank(database) || StringUtils.isBlank(table)) {
+ outputMetricsWithEstimate(data, fetchDelay, emitDelay);
+ // output read phase metric
+ outputReadPhaseMetrics((isSnapshotRecord) ?
ReadPhase.SNAPSHOT_PHASE : ReadPhase.INCREASE_PHASE);
+ return;
+ }
+ // output sub source metric
+ outputMetricsWithEstimate(new String[]{database, table},
isSnapshotRecord, data, fetchDelay, emitDelay);
+ }
+
/**
* output metrics with estimate
*
@@ -180,6 +203,27 @@ public class SourceTableMetricData extends
SourceMetricData implements SourceSub
outputMetricsWithEstimate(new String[]{database, schema, table},
isSnapshotRecord, data);
}
+ /**
+ * output metrics with estimate
+ *
+ * @param database the database name of record
+ * @param schema the schema name of record
+ * @param table the table name of record
+ * @param isSnapshotRecord is it snapshot record
+ * @param data the data of record
+ * @param fetchDelay fetchDelay = FetchTime - messageTimestamp, where the
FetchTime is the time the record fetched into the source operator.
+ * @param emitDelay emitDelay = EmitTime - messageTimestamp, where the
EmitTime is the time the record leaves the source operator.
+ */
+ public void outputMetricsWithEstimate(String database, String schema,
String table,
+ boolean isSnapshotRecord, Object data, long fetchDelay, long
emitDelay) {
+ if (StringUtils.isBlank(database) || StringUtils.isBlank(schema) ||
StringUtils.isBlank(table)) {
+ outputMetricsWithEstimate(data, fetchDelay, emitDelay);
+ return;
+ }
+ // output sub source metric
+ outputMetricsWithEstimate(new String[]{database, schema, table},
isSnapshotRecord, data);
+ }
+
/**
* output metrics with estimate
*
@@ -210,6 +254,39 @@ public class SourceTableMetricData extends
SourceMetricData implements SourceSub
outputReadPhaseMetrics((isSnapshotRecord) ? ReadPhase.SNAPSHOT_PHASE :
ReadPhase.INCREASE_PHASE);
}
+ /**
+ * output metrics with estimate
+ *
+ * @param recordSchemaInfoArray the schema info of record
+ * @param isSnapshotRecord is it snapshot record
+ * @param data the data of record
+ * @param fetchDelay fetchDelay = FetchTime - messageTimestamp, where the
FetchTime is the time the record fetched into the source operator.
+ * @param emitDelay emitDelay = EmitTime - messageTimestamp, where the
EmitTime is the time the record leaves the source operator.
+ */
+ public void outputMetricsWithEstimate(String[] recordSchemaInfoArray,
boolean isSnapshotRecord, Object data,
+ long fetchDelay, long emitDelay) {
+ if (recordSchemaInfoArray == null) {
+ outputMetricsWithEstimate(data, fetchDelay, emitDelay);
+ return;
+ }
+ String identify = String.join(Constants.SEMICOLON,
recordSchemaInfoArray);
+ SourceMetricData subSourceMetricData;
+ if (subSourceMetricMap.containsKey(identify)) {
+ subSourceMetricData = subSourceMetricMap.get(identify);
+ } else {
+ subSourceMetricData =
buildSubSourceMetricData(recordSchemaInfoArray, this);
+ subSourceMetricMap.put(identify, subSourceMetricData);
+ }
+ // source metric and sub source metric output metrics
+ long rowCountSize = 1L;
+ long rowDataSize = CalculateObjectSizeUtils.getDataSize(data);
+ this.outputMetrics(rowCountSize, rowDataSize, fetchDelay, emitDelay);
+ subSourceMetricData.outputMetrics(rowCountSize, rowDataSize,
fetchDelay, emitDelay);
+
+ // output read phase metric
+ outputReadPhaseMetrics((isSnapshotRecord) ? ReadPhase.SNAPSHOT_PHASE :
ReadPhase.INCREASE_PHASE);
+ }
+
/**
* output read phase metric
*
@@ -259,6 +336,12 @@ public class SourceTableMetricData extends
SourceMetricData implements SourceSub
return "SourceTableMetricData{"
+ "numRecordsIn=" + getNumRecordsIn().getCount()
+ ", numBytesIn=" + getNumBytesIn().getCount()
+ + ", numRecordsInForMeter=" +
getNumRecordsInForMeter().getCount()
+ + ", numBytesInForMeter=" + getNumBytesInForMeter().getCount()
+ + ", numRecordsInPerSecond=" +
getNumRecordsInPerSecond().getRate()
+ + ", numBytesInPerSecond=" + getNumBytesInPerSecond().getRate()
+ + ", currentFetchEventTimeLag=" + getFetchDelay()
+ + ", currentEmitEventTimeLag=" + getEmitDelay()
+ ", readPhaseMetricDataMap=" + readPhaseMetricDataMap
+ ", subSourceMetricMap=" + subSourceMetricMap
+ '}';
diff --git
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/metrics/MySqlSourceReaderMetrics.java
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/metrics/MySqlSourceReaderMetrics.java
index 41b4d85057..361c79c33a 100644
---
a/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/metrics/MySqlSourceReaderMetrics.java
+++
b/inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/metrics/MySqlSourceReaderMetrics.java
@@ -108,7 +108,8 @@ public class MySqlSourceReaderMetrics {
public void outputMetrics(String database, String table, boolean
isSnapshotRecord, Object data) {
if (sourceTableMetricData != null) {
- sourceTableMetricData.outputMetricsWithEstimate(database, table,
isSnapshotRecord, data);
+ sourceTableMetricData.outputMetricsWithEstimate(database, table,
isSnapshotRecord, data, fetchDelay,
+ emitDelay);
}
}
