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 55e3714 [INLONG-2982][Manager] Add group logs in manager client
(#2983)
55e3714 is described below
commit 55e3714512f709e9621d409f93a01e5fe7bae5da
Author: kipshi <[email protected]>
AuthorDate: Mon Mar 7 20:24:27 2022 +0800
[INLONG-2982][Manager] Add group logs in manager client (#2983)
---
.../manager/client/api/InlongGroupContext.java | 8 +++++++-
.../manager/client/api/impl/InlongGroupImpl.java | 13 +++++++++++--
.../manager/client/api/impl/InlongStreamImpl.java | 21 ++++++++++++---------
.../manager/common/pojo/workflow/EventLogView.java | 3 +++
4 files changed, 33 insertions(+), 12 deletions(-)
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
index 2d2865d..dd788f8 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/InlongGroupContext.java
@@ -45,7 +45,12 @@ public class InlongGroupContext implements Serializable {
private Map<String, String> extensions;
/**
- * Error message for Inlong group, taskName->logs.
+ * Logs for Inlong group, taskName->logs.
+ */
+ private Map<String, List<String>> groupLogs;
+
+ /**
+ * Error message for Inlong group, taskName->exceptionMsg.
*/
private Map<String, List<String>> errMsgs;
@@ -64,6 +69,7 @@ public class InlongGroupContext implements Serializable {
this.groupConf = streamGroupConf;
this.inlongStreamMap = groupContext.getStreamMap();
this.errMsgs = Maps.newHashMap();
+ this.groupLogs = Maps.newHashMap();
this.state = InlongGroupState.parseByBizStatus(groupInfo.getStatus());
this.extensions = Maps.newHashMap();
List<InlongGroupExtInfo> extInfos = groupInfo.getExtList();
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongGroupImpl.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongGroupImpl.java
index f5d5db3..58cf865 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongGroupImpl.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongGroupImpl.java
@@ -200,17 +200,26 @@ public class InlongGroupImpl implements InlongGroup {
}
//Create group context
InlongGroupContext inlongGroupContext = new
InlongGroupContext(groupContext, groupConf);
+ //Fetch group logs
List<EventLogView> logViews =
managerClient.getInlongGroupError(inlongGroupId);
if (CollectionUtils.isNotEmpty(logViews)) {
Map<String, List<String>> errMsgs = Maps.newHashMap();
+ Map<String, List<String>> groupLogs = Maps.newHashMap();
logViews.stream()
- .filter(x -> null != x.getElementName() && null !=
x.getException())
+ .filter(x -> StringUtils.isNotEmpty(x.getElementName()))
.forEach(eventLogView -> {
String taskName = eventLogView.getElementName();
- errMsgs.computeIfAbsent(taskName,
Lists::newArrayList).add(eventLogView.getException());
+ if
(StringUtils.isNotEmpty(eventLogView.getException())) {
+ errMsgs.computeIfAbsent(taskName,
Lists::newArrayList).add(eventLogView.getException());
+ }
+ if (StringUtils.isNotEmpty(eventLogView.getRemark())) {
+ groupLogs.computeIfAbsent(taskName,
Lists::newArrayList).add(eventLogView.getRemark());
+ }
});
inlongGroupContext.setErrMsgs(errMsgs);
+ inlongGroupContext.setGroupLogs(groupLogs);
}
+ //Fetch stream logs
Map<String, InlongStream> streams =
inlongGroupContext.getInlongStreamMap();
streams.keySet().stream().forEach(streamName -> {
String inlongStreamId = "b_" + streamName;
diff --git
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
index 7727e69..ce89b8c 100644
---
a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
+++
b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/impl/InlongStreamImpl.java
@@ -25,6 +25,7 @@ import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.compress.utils.Lists;
import org.apache.inlong.manager.client.api.InlongStream;
import org.apache.inlong.manager.client.api.StreamField;
import org.apache.inlong.manager.client.api.StreamField.FieldType;
@@ -50,20 +51,22 @@ public class InlongStreamImpl extends InlongStream {
private Map<String, StreamSink> streamSinks = Maps.newHashMap();
- private List<StreamField> streamFields;
+ private List<StreamField> streamFields = Lists.newArrayList();
public InlongStreamImpl(FullStreamResponse fullStreamResponse) {
InlongStreamInfo streamInfo = fullStreamResponse.getStreamInfo();
this.name = streamInfo.getName();
List<InlongStreamFieldInfo> streamFieldInfos =
streamInfo.getFieldList();
- this.streamFields = streamFieldInfos.stream().map(streamFieldInfo -> {
- return new StreamField(streamFieldInfo.getId(),
- FieldType.forName(streamFieldInfo.getFieldType()),
- streamFieldInfo.getFieldName(),
- streamFieldInfo.getFieldComment(),
- streamFieldInfo.getFieldValue()
- );
- }).collect(Collectors.toList());
+ if (CollectionUtils.isNotEmpty(streamFieldInfos)) {
+ this.streamFields = streamFieldInfos.stream().map(streamFieldInfo
-> {
+ return new StreamField(streamFieldInfo.getId(),
+ FieldType.forName(streamFieldInfo.getFieldType()),
+ streamFieldInfo.getFieldName(),
+ streamFieldInfo.getFieldComment(),
+ streamFieldInfo.getFieldValue()
+ );
+ }).collect(Collectors.toList());
+ }
List<SinkResponse> sinkList = fullStreamResponse.getSinkInfo();
if (CollectionUtils.isNotEmpty(sinkList)) {
this.streamSinks = sinkList.stream()
diff --git
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/EventLogView.java
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/EventLogView.java
index 7ec0b6d..1e58f50 100644
---
a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/EventLogView.java
+++
b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/EventLogView.java
@@ -84,6 +84,9 @@ public class EventLogView {
@ApiModelProperty("End time")
private Date endTime;
+ @ApiModelProperty("Remark")
+ private String remark;
+
@ApiModelProperty("Exception message")
private String exception;
