You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2022/01/23 02:23:32 UTC
[incubator-inlong] branch master updated: [INLONG-2266][Feature]Support reporting metrics by audit-sdk in sort (#2267)
This is an automated email from the ASF dual-hosted git repository.
healchow 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 2557dcb [INLONG-2266][Feature]Support reporting metrics by audit-sdk in sort (#2267)
2557dcb is described below
commit 2557dcb56e4060ceda307fcfc66ad15a47ad115b
Author: TianqiWan <52...@users.noreply.github.com>
AuthorDate: Sun Jan 23 10:23:24 2022 +0800
[INLONG-2266][Feature]Support reporting metrics by audit-sdk in sort (#2267)
---
.../inlong/sort/configuration/Constants.java | 11 +++++
inlong-sort/sort-core/pom.xml | 6 +++
.../deserialization/DeserializationSchema.java | 51 +++++++++++++++++++++-
.../sort/flink/hive/HiveMultiTenantWriter.java | 41 +++++++++++++++++
.../org/apache/inlong/sort/util/CommonUtils.java | 35 +++++++++++++--
5 files changed, 139 insertions(+), 5 deletions(-)
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
index 4a01685..ac21a6f 100644
--- a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
@@ -40,6 +40,12 @@ public class Constants {
public static final String METRIC_DATA_OUTPUT_TAG_ID = "metric_data_side_output";
+ public static final int METRIC_AUDIT_ID_FOR_INPUT = 7;
+
+ public static final int METRIC_AUDIT_ID_FOR_OUTPUT = 8;
+
+ public static final String INLONG_GROUP_ID = "inlong.group.id";
+
// ------------------------------------------------------------------------
// Operator uid
// ------------------------------------------------------------------------
@@ -294,6 +300,11 @@ public class Constants {
.defaultValue(5)
.withDescription("minutes");
+ public static final ConfigOption<String> METRICS_AUDIT_SDK_HOSTS =
+ key("metrics.audit.sdk.hosts")
+ .noDefaultValue()
+ .withDescription("Host address for reporting audit metrics. e.g. 0.0.0.0:54041,0.0.0.1:54041");
+
// ------------------------------------------------------------------------
// Single tenant related
// ------------------------------------------------------------------------
diff --git a/inlong-sort/sort-core/pom.xml b/inlong-sort/sort-core/pom.xml
index 5a50c9e..d5f4b38 100644
--- a/inlong-sort/sort-core/pom.xml
+++ b/inlong-sort/sort-core/pom.xml
@@ -170,6 +170,12 @@
</dependency>
<dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>audit-sdk</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils_${flink.scala.binary.version}</artifactId>
<version>${flink.version}</version>
diff --git a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/deserialization/DeserializationSchema.java b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/deserialization/DeserializationSchema.java
index 5827ef2..5aa92f1 100644
--- a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/deserialization/DeserializationSchema.java
+++ b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/deserialization/DeserializationSchema.java
@@ -18,8 +18,14 @@
package org.apache.inlong.sort.flink.deserialization;
import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
+import org.apache.inlong.audit.AuditImp;
import org.apache.inlong.sort.configuration.Configuration;
import org.apache.inlong.sort.configuration.Constants;
import org.apache.inlong.sort.flink.Record;
@@ -34,6 +40,7 @@ import org.apache.inlong.sort.meta.MetaManager;
import org.apache.inlong.sort.meta.MetaManager.DataFlowInfoListener;
import org.apache.inlong.sort.protocol.DataFlowInfo;
import org.apache.flink.util.OutputTag;
+import org.apache.inlong.sort.util.CommonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,6 +69,11 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
private transient Boolean enableOutputMetrics;
+ // dataflow id -> Pair<group id, stream id>
+ private transient Map<Long, Pair<String, String>> inLongGroupIdAndStreamIdMap;
+
+ private transient AuditImp auditImp;
+
public DeserializationSchema(Configuration config) {
this.config = Preconditions.checkNotNull(config);
}
@@ -76,6 +88,13 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
metaManager = MetaManager.getInstance(config);
metaManager.registerDataFlowInfoListener(new DataFlowInfoListenerImpl());
enableOutputMetrics = config.getBoolean(Constants.METRICS_ENABLE_OUTPUT);
+
+ inLongGroupIdAndStreamIdMap = new HashMap<>();
+ String auditHostAndPorts = config.getString(Constants.METRICS_AUDIT_SDK_HOSTS);
+ if (auditHostAndPorts != null) {
+ AuditImp.getInstance().setAuditProxy(new HashSet<>(Arrays.asList(auditHostAndPorts.split(","))));
+ auditImp = AuditImp.getInstance();
+ }
}
@Override
@@ -84,6 +103,10 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
metaManager.close();
metaManager = null;
}
+
+ if (auditImp != null) {
+ auditImp.sendReport();
+ }
}
@Override
@@ -122,7 +145,23 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
context.output(METRIC_DATA_OUTPUT_TAG, metricData);
}
- collector.collect(recordTransformer.toSerializedRecord(sinkRecord));
+ SerializedRecord serializedSinkRecord = recordTransformer.toSerializedRecord(sinkRecord);
+
+ if (auditImp != null) {
+ Pair<String, String> groupIdAndStreamId = inLongGroupIdAndStreamIdMap.getOrDefault(
+ serializedRecord.getDataFlowId(),
+ Pair.of("", ""));
+
+ auditImp.add(
+ Constants.METRIC_AUDIT_ID_FOR_INPUT,
+ groupIdAndStreamId.getLeft(),
+ groupIdAndStreamId.getRight(),
+ sinkRecord.getTimestampMillis(),
+ 1,
+ serializedSinkRecord.getData().length);
+ }
+
+ collector.collect(serializedSinkRecord);
});
if (serializedRecord instanceof TDMsgMixedSerializedRecord) {
@@ -161,6 +200,10 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
multiTenancyDeserializer.addDataFlow(dataFlowInfo);
fieldMappingTransformer.addDataFlow(dataFlowInfo);
recordTransformer.addDataFlow(dataFlowInfo);
+
+ inLongGroupIdAndStreamIdMap.put(
+ dataFlowInfo.getId(),
+ CommonUtils.getInLongGroupIdAndStreamId(dataFlowInfo));
}
}
@@ -171,6 +214,10 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
multiTenancyDeserializer.updateDataFlow(dataFlowInfo);
fieldMappingTransformer.updateDataFlow(dataFlowInfo);
recordTransformer.updateDataFlow(dataFlowInfo);
+
+ inLongGroupIdAndStreamIdMap.put(
+ dataFlowInfo.getId(),
+ CommonUtils.getInLongGroupIdAndStreamId(dataFlowInfo));
}
}
@@ -181,6 +228,8 @@ public class DeserializationSchema extends ProcessFunction<SerializedRecord, Ser
multiTenancyDeserializer.removeDataFlow(dataFlowInfo);
fieldMappingTransformer.removeDataFlow(dataFlowInfo);
recordTransformer.removeDataFlow(dataFlowInfo);
+
+ inLongGroupIdAndStreamIdMap.remove(dataFlowInfo.getId());
}
}
}
diff --git a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/hive/HiveMultiTenantWriter.java b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/hive/HiveMultiTenantWriter.java
index 3d2a39f..5286432 100644
--- a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/hive/HiveMultiTenantWriter.java
+++ b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/hive/HiveMultiTenantWriter.java
@@ -18,8 +18,11 @@
package org.apache.inlong.sort.flink.hive;
import com.google.common.base.Preconditions;
+import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.runtime.state.CheckpointListener;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
@@ -29,6 +32,7 @@ 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.OutputTag;
+import org.apache.inlong.audit.AuditImp;
import org.apache.inlong.sort.configuration.Configuration;
import org.apache.inlong.sort.configuration.Constants;
import org.apache.inlong.sort.flink.SerializedRecord;
@@ -40,6 +44,7 @@ import org.apache.inlong.sort.meta.MetaManager.DataFlowInfoListener;
import org.apache.inlong.sort.protocol.DataFlowInfo;
import org.apache.inlong.sort.protocol.sink.HiveSinkInfo;
import org.apache.inlong.sort.protocol.sink.SinkInfo;
+import org.apache.inlong.sort.util.CommonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,6 +66,11 @@ public class HiveMultiTenantWriter extends ProcessFunction<SerializedRecord, Par
private transient ProcessFunctionContext proxyContext;
+ // dataflow id -> Pair<group id, stream id>
+ private transient Map<Long, Pair<String, String>> inLongGroupIdAndStreamIdMap;
+
+ private transient AuditImp auditImp;
+
public HiveMultiTenantWriter(Configuration configuration) {
this.configuration = Preconditions.checkNotNull(configuration);
}
@@ -73,6 +83,13 @@ public class HiveMultiTenantWriter extends ProcessFunction<SerializedRecord, Par
MetaManager metaManager = MetaManager.getInstance(configuration);
metaManager.registerDataFlowInfoListener(new DataFlowInfoListenerImpl());
proxyContext = new ProcessFunctionContext();
+
+ inLongGroupIdAndStreamIdMap = new HashMap<>();
+ String auditHostAndPorts = configuration.getString(Constants.METRICS_AUDIT_SDK_HOSTS);
+ if (auditHostAndPorts != null) {
+ AuditImp.getInstance().setAuditProxy(new HashSet<>(Arrays.asList(auditHostAndPorts.split(","))));
+ auditImp = AuditImp.getInstance();
+ }
}
@Override
@@ -87,6 +104,10 @@ public class HiveMultiTenantWriter extends ProcessFunction<SerializedRecord, Par
}
hiveWriters.clear();
}
+
+ if (auditImp != null) {
+ auditImp.sendReport();
+ }
}
@Override
@@ -126,6 +147,20 @@ public class HiveMultiTenantWriter extends ProcessFunction<SerializedRecord, Par
hiveWriter.processElement(recordTransformer.toRecord(serializedRecord).getRow(),
proxyContext.setContext(context), collector);
+
+ if (auditImp != null) {
+ Pair<String, String> groupIdAndStreamId = inLongGroupIdAndStreamIdMap.getOrDefault(
+ serializedRecord.getDataFlowId(),
+ Pair.of("", ""));
+
+ auditImp.add(
+ Constants.METRIC_AUDIT_ID_FOR_OUTPUT,
+ groupIdAndStreamId.getLeft(),
+ groupIdAndStreamId.getRight(),
+ serializedRecord.getTimestampMillis(),
+ 1,
+ serializedRecord.getData().length);
+ }
}
}
@@ -176,6 +211,10 @@ public class HiveMultiTenantWriter extends ProcessFunction<SerializedRecord, Par
hiveWriter.open(new org.apache.flink.configuration.Configuration());
hiveWriters.put(dataFlowId, hiveWriter);
recordTransformer.addDataFlow(dataFlowInfo);
+
+ inLongGroupIdAndStreamIdMap.put(
+ dataFlowInfo.getId(),
+ CommonUtils.getInLongGroupIdAndStreamId(dataFlowInfo));
}
}
@@ -195,6 +234,8 @@ public class HiveMultiTenantWriter extends ProcessFunction<SerializedRecord, Par
existingHiveWriter.close();
}
recordTransformer.removeDataFlow(dataFlowInfo);
+
+ inLongGroupIdAndStreamIdMap.remove(dataFlowInfo.getId());
}
}
diff --git a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/util/CommonUtils.java b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/util/CommonUtils.java
index 303c147..2edb554 100644
--- a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/util/CommonUtils.java
+++ b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/util/CommonUtils.java
@@ -17,17 +17,22 @@
package org.apache.inlong.sort.util;
-import org.apache.inlong.sort.formats.base.TableFormatUtils;
-import org.apache.inlong.sort.formats.common.FormatInfo;
-import org.apache.inlong.sort.formats.common.RowFormatInfo;
-import org.apache.inlong.sort.protocol.FieldInfo;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.api.java.typeutils.runtime.RowSerializer;
+import org.apache.inlong.sort.configuration.Constants;
+import org.apache.inlong.sort.formats.base.TableFormatUtils;
+import org.apache.inlong.sort.formats.common.FormatInfo;
+import org.apache.inlong.sort.formats.common.RowFormatInfo;
+import org.apache.inlong.sort.protocol.DataFlowInfo;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.deserialization.DeserializationInfo;
+import org.apache.inlong.sort.protocol.deserialization.TDMsgDeserializationInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -83,4 +88,26 @@ public class CommonUtils {
return (RowSerializer) new RowTypeInfo(typeInformations).createSerializer(new ExecutionConfig());
}
+
+ 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 TDMsgDeserializationInfo) {
+ streamId = ((TDMsgDeserializationInfo) deserializationInfo).getTid();
+ }
+
+ }
+
+ return Pair.of(groupId, streamId);
+ }
}