This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new cc65dcc58 [INLONG-3546][Sort] Support reporting metrics by audit-sdk
in sort-single-tenant (#3549)
cc65dcc58 is described below
commit cc65dcc5854747bc51d7fffc90ef3a4e67c09abf
Author: TianqiWan <[email protected]>
AuthorDate: Wed Apr 6 15:28:43 2022 +0800
[INLONG-3546][Sort] Support reporting metrics by audit-sdk in
sort-single-tenant (#3549)
Co-authored-by: tianqiwan <[email protected]>
---
inlong-sort/sort-single-tenant/pom.xml | 6 ++
.../inlong/sort/singletenant/flink/Entrance.java | 13 +++-
.../deserialization/DeserializationFunction.java | 70 +++++++++++++++++++---
.../sort/singletenant/flink/utils/CommonUtils.java | 28 +++++++++
.../singletenant/flink/DebeziumToCanalITCase.java | 2 +-
.../singletenant/flink/WholeDBMigrationITCase.java | 2 +-
.../DeserializationFunctionTest.java | 12 +++-
7 files changed, 117 insertions(+), 16 deletions(-)
diff --git a/inlong-sort/sort-single-tenant/pom.xml
b/inlong-sort/sort-single-tenant/pom.xml
index 4fdf859b6..d4c5c50f1 100644
--- a/inlong-sort/sort-single-tenant/pom.xml
+++ b/inlong-sort/sort-single-tenant/pom.xml
@@ -148,6 +148,12 @@
<version>2.12.6.1</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>audit-sdk</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<!-- test -->
<dependency>
<groupId>org.apache.kafka</groupId>
diff --git
a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/Entrance.java
b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/Entrance.java
index a114ac901..216164aba 100644
---
a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/Entrance.java
+++
b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/Entrance.java
@@ -27,6 +27,7 @@ import java.io.File;
import java.io.IOException;
import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
@@ -85,7 +86,7 @@ public class Entrance {
);
DataStream<Row> deserializedStream =
- buildDeserializationStream(sourceStream,
dataFlowInfo.getSourceInfo(), config);
+ buildDeserializationStream(sourceStream, dataFlowInfo, config);
DataStream<Row> transformationStream =
buildTransformationStream(deserializedStream, dataFlowInfo,
config);
@@ -147,22 +148,28 @@ public class Entrance {
private static DataStream<Row> buildDeserializationStream(
DataStream<SerializedRecord> sourceStream,
- SourceInfo sourceInfo,
+ DataFlowInfo dataFlowInfo,
Configuration config
) throws IOException, ClassNotFoundException {
final boolean orderlyOutput =
config.getBoolean(Constants.JOB_ORDERLY_OUTPUT);
+ SourceInfo sourceInfo = dataFlowInfo.getSourceInfo();
FieldInfo[] sourceFields = sourceInfo.getFields();
DeserializationSchema<Row> schema = DeserializationSchemaFactory.build(
sourceFields, sourceInfo.getDeserializationInfo());
FieldMappingTransformer fieldMappingTransformer = new
FieldMappingTransformer(config, sourceFields);
+ Pair<String, String> inLongGroupIdAndStreamId =
CommonUtils.getInLongGroupIdAndStreamId(dataFlowInfo);
+
// Currently, canal and debezium deserialization schema will put a map
at the first position
// of the deserialized row. So the `appendAttributes` flag should be
set false.
DeserializationFunction function = new DeserializationFunction(
schema,
fieldMappingTransformer,
!(sourceInfo.getDeserializationInfo() instanceof
DebeziumDeserializationInfo)
- && !(sourceInfo.getDeserializationInfo() instanceof
CanalDeserializationInfo));
+ && !(sourceInfo.getDeserializationInfo() instanceof
CanalDeserializationInfo),
+ config,
+ inLongGroupIdAndStreamId.getLeft(),
+ inLongGroupIdAndStreamId.getRight());
DataStream<Row> deserializedStream = sourceStream.process(function)
.uid(Constants.DESERIALIZATION_SCHEMA_UID)
diff --git
a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunction.java
b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunction.java
index e9bec47a7..40a13c3d9 100644
---
a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunction.java
+++
b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunction.java
@@ -17,12 +17,18 @@
package org.apache.inlong.sort.singletenant.flink.deserialization;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.Preconditions;
+import org.apache.inlong.audit.AuditImp;
import org.apache.inlong.common.msg.InLongMsg;
+import org.apache.inlong.sort.configuration.Configuration;
+import org.apache.inlong.sort.configuration.Constants;
import org.apache.inlong.sort.singletenant.flink.SerializedRecord;
import java.util.Iterator;
@@ -37,21 +43,36 @@ public class DeserializationFunction extends
ProcessFunction<SerializedRecord, R
private final boolean appendAttributes;
+ private final Configuration config;
+
+ private final String inLongGroupId;
+
+ private final String inLongStreamId;
+
+ private transient AuditImp auditImp;
+
public DeserializationFunction(
DeserializationSchema<Row> deserializationSchema,
FieldMappingTransformer fieldMappingTransformer,
- boolean appendAttributes
- ) {
+ boolean appendAttributes,
+ Configuration config,
+ String inLongGroupId,
+ String inLongStreamId) {
this.deserializationSchema = deserializationSchema;
this.fieldMappingTransformer = fieldMappingTransformer;
this.appendAttributes = appendAttributes;
+ this.config = Preconditions.checkNotNull(config);
+ this.inLongGroupId = inLongGroupId;
+ this.inLongStreamId = inLongStreamId;
}
- public DeserializationFunction(
- DeserializationSchema<Row> deserializationSchema,
- FieldMappingTransformer fieldMappingTransformer
- ) {
- this(deserializationSchema, fieldMappingTransformer, true);
+ @Override
+ public void open(org.apache.flink.configuration.Configuration parameters)
throws Exception {
+ String auditHostAndPorts =
config.getString(Constants.METRICS_AUDIT_PROXY_HOSTS);
+ if (auditHostAndPorts != null) {
+ AuditImp.getInstance().setAuditProxy(new
HashSet<>(Arrays.asList(auditHostAndPorts.split(","))));
+ auditImp = AuditImp.getInstance();
+ }
}
@Override
@@ -71,7 +92,21 @@ public class DeserializationFunction extends
ProcessFunction<SerializedRecord, R
while (iterator.hasNext()) {
byte[] bodyBytes = iterator.next();
- if (bodyBytes == null || bodyBytes.length == 0) {
+ long bodyLength = bodyBytes == null ? 0 : bodyBytes.length;
+
+ // Currently, we can not get the number of records in source
because they are packed in InLongMsg.
+ // So we output metrics for input here.
+ if (auditImp != null) {
+ auditImp.add(
+ Constants.METRIC_AUDIT_ID_FOR_INPUT,
+ inLongGroupId,
+ inLongStreamId,
+ value.getTimestampMillis(),
+ 1,
+ bodyLength);
+ }
+
+ if (bodyLength == 0) {
continue;
}
@@ -79,10 +114,29 @@ public class DeserializationFunction extends
ProcessFunction<SerializedRecord, R
if (appendAttributes) {
inputRow = Row.join(Row.of(new HashMap<>()), inputRow);
}
+
+ // Currently, the transformer operator and the sink
operator do not discard data.
+ // So we simply output metrics for output here.
+ if (auditImp != null) {
+ auditImp.add(
+ Constants.METRIC_AUDIT_ID_FOR_OUTPUT,
+ inLongGroupId,
+ inLongStreamId,
+ value.getTimestampMillis(),
+ 1,
+ bodyLength);
+ }
+
out.collect(fieldMappingTransformer.transform(inputRow,
value.getTimestampMillis()));
}));
}
}
}
+ @Override
+ public void close() {
+ if (auditImp != null) {
+ auditImp.sendReport();
+ }
+ }
}
diff --git
a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/utils/CommonUtils.java
b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/utils/CommonUtils.java
index eac8154a0..3094c6a67 100644
---
a/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/utils/CommonUtils.java
+++
b/inlong-sort/sort-single-tenant/src/main/java/org/apache/inlong/sort/singletenant/flink/utils/CommonUtils.java
@@ -21,7 +21,9 @@ package org.apache.inlong.sort.singletenant.flink.utils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import org.apache.avro.Schema;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.formats.common.TimestampFormat;
import org.apache.flink.table.api.TableSchema;
@@ -30,6 +32,7 @@ import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.inlong.sort.configuration.Constants;
import org.apache.inlong.sort.formats.base.TableFormatUtils;
import org.apache.inlong.sort.formats.common.DateFormatInfo;
import org.apache.inlong.sort.formats.common.FormatInfo;
@@ -40,6 +43,7 @@ import org.apache.inlong.sort.formats.common.TimeFormatInfo;
import org.apache.inlong.sort.formats.common.TimestampFormatInfo;
import org.apache.inlong.sort.formats.common.TypeInfo;
import org.apache.inlong.sort.protocol.BuiltInFieldInfo;
+import org.apache.inlong.sort.protocol.DataFlowInfo;
import org.apache.inlong.sort.protocol.FieldInfo;
import java.io.ByteArrayInputStream;
@@ -48,6 +52,8 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
+import org.apache.inlong.sort.protocol.deserialization.DeserializationInfo;
+import
org.apache.inlong.sort.protocol.deserialization.InLongMsgDeserializationInfo;
import static
org.apache.flink.formats.avro.typeutils.AvroSchemaConverter.convertToSchema;
import static
org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter.toDataType;
@@ -226,4 +232,26 @@ public class CommonUtils {
return results.toArray(new FieldInfo[0]);
}
+ public static Pair<String, String>
getInLongGroupIdAndStreamId(DataFlowInfo dataFlowInfo) {
+ String groupId = "";
+ String streamId = "";
+
+ if (dataFlowInfo != null) {
+ // Get group id
+ Map<String, Object> properties = dataFlowInfo.getProperties();
+ if (properties != null) {
+ groupId = properties.getOrDefault(Constants.INLONG_GROUP_ID,
"").toString();
+ }
+
+ // Get stream id
+ final DeserializationInfo deserializationInfo =
dataFlowInfo.getSourceInfo().getDeserializationInfo();
+ if (deserializationInfo instanceof InLongMsgDeserializationInfo) {
+ streamId = ((InLongMsgDeserializationInfo)
deserializationInfo).getTid();
+ }
+
+ }
+
+ return Pair.of(groupId, streamId);
+ }
+
}
diff --git
a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/DebeziumToCanalITCase.java
b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/DebeziumToCanalITCase.java
index c81e093d9..470e99b00 100644
---
a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/DebeziumToCanalITCase.java
+++
b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/DebeziumToCanalITCase.java
@@ -106,7 +106,7 @@ public class DebeziumToCanalITCase {
FieldMappingTransformer fieldMappingTransformer = new
FieldMappingTransformer(
new Configuration(), sourceFieldInfos);
DeserializationFunction function = new DeserializationFunction(
- deserializationSchema, fieldMappingTransformer, false);
+ deserializationSchema, fieldMappingTransformer, false,
new Configuration(), "", "");
DataStream<Row> deserializedStream =
sourceStream.process(function);
// Transform
diff --git
a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/WholeDBMigrationITCase.java
b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/WholeDBMigrationITCase.java
index 9d5ff79ac..e6ad945c5 100644
---
a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/WholeDBMigrationITCase.java
+++
b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/WholeDBMigrationITCase.java
@@ -107,7 +107,7 @@ public class WholeDBMigrationITCase {
FieldMappingTransformer fieldMappingTransformer = new
FieldMappingTransformer(
new Configuration(), fieldInfos);
DeserializationFunction function = new DeserializationFunction(
- deserializationSchema, fieldMappingTransformer, false);
+ deserializationSchema, fieldMappingTransformer, false,
new Configuration(), "", "");
DataStream<Row> deserializedStream =
sourceStream.process(function);
// Serialize and output
diff --git
a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunctionTest.java
b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunctionTest.java
index dbe9d3e93..55c2481de 100644
---
a/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunctionTest.java
+++
b/inlong-sort/sort-single-tenant/src/test/java/org/apache/inlong/sort/singletenant/flink/deserialization/DeserializationFunctionTest.java
@@ -48,8 +48,11 @@ public class DeserializationFunctionTest {
FieldInfo[] fieldInfos = {new FieldInfo("content",
StringFormatInfo.INSTANCE)};
DeserializationFunction function = new DeserializationFunction(
DeserializationSchemaFactory.build(fieldInfos, null),
- new FieldMappingTransformer(new Configuration(), fieldInfos)
- );
+ new FieldMappingTransformer(new Configuration(), fieldInfos),
+ true,
+ new Configuration(),
+ "",
+ "");
ListCollector<Row> collector = new ListCollector<>();
function.processElement(serializedRecord,null, collector);
@@ -132,7 +135,10 @@ public class DeserializationFunctionTest {
fieldInfos,
new CanalDeserializationInfo(null, null, false,
"ISO_8601", false)),
new FieldMappingTransformer(new Configuration(), fieldInfos),
- false);
+ false,
+ new Configuration(),
+ "",
+ "");
ListCollector<Row> collector = new ListCollector<>();
function.processElement(serializedRecord,null, collector);
