You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by ve...@apache.org on 2023/07/13 11:33:18 UTC
[inlong] branch branch-1.8 updated (4c94d5a1ee -> c3c448d0bf)
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a change to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
from 4c94d5a1ee [INLONG-8507][Dataproxy] Modify nodeIp in dataproxy message to clientIp (#8508)
new 788a21cb06 [INLONG-8520][Agent] Makeup the code to figure out whether it is kvm、k8s or not (#8521)
new 5189f56377 [INLONG-8511][Manager] Add admin permission of public tenant to all users (#8513)
new 7f924a6c2b [INLONG-8524][Agent] Update the JVM Options for Agent (#8525)
new c5a2d063ce [INLONG-8527][Manager] Fix wrong capitalization of InlongRequestWrapper (#8528)
new e96016a04a [INLONG-8509][Manager] Optimize preProcessTemplateFileTask in AgentServiceImpl (#8510)
new c3c448d0bf [INLONG-8522][Manager] optimize log print and status judgement in preProcessLabelFileTasks (#8523)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../sources/reader/file/FileReaderOperator.java | 6 ++
inlong-agent/bin/agent-env.sh | 9 +--
.../service/core/impl/AgentServiceImpl.java | 65 ++++++++++++----------
.../manager/service/user/TenantRoleService.java | 5 ++
.../service/user/TenantRoleServiceImpl.java | 12 ++++
.../manager/service/user/UserServiceImpl.java | 5 ++
.../inlong/manager/service/ServiceBaseTest.java | 12 +++-
.../service/core/impl/AgentServiceTest.java | 40 +++++++------
inlong-manager/manager-web/sql/changes-1.8.0.sql | 4 ++
.../manager/web/filter/TenantInsertionFilter.java | 1 -
.../manager/web/utils/InlongRequestWrapper.java | 4 +-
.../web/filter/TenantInsertionFilterTest.java | 2 -
12 files changed, 106 insertions(+), 59 deletions(-)
[inlong] 06/06: [INLONG-8522][Manager] optimize log print and status judgement in preProcessLabelFileTasks (#8523)
Posted by ve...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit c3c448d0bf63d95c6df41696fd377abd7a45e4c5
Author: kipshi <48...@users.noreply.github.com>
AuthorDate: Thu Jul 13 17:37:05 2023 +0800
[INLONG-8522][Manager] optimize log print and status judgement in preProcessLabelFileTasks (#8523)
Co-authored-by: kipshi <ki...@tencent.com>
---
.../service/core/impl/AgentServiceImpl.java | 12 ++++----
.../service/core/impl/AgentServiceTest.java | 34 +++++++++++++---------
2 files changed, 27 insertions(+), 19 deletions(-)
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
index 174929741e..b20d76a8c0 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
@@ -202,7 +202,7 @@ public class AgentServiceImpl implements AgentService {
// Update task status, other tasks with status 20x will change to 30x in next request
if (CollectionUtils.isEmpty(request.getCommandInfo())) {
- LOGGER.info("task result was empty in request: {}, just return", request);
+ LOGGER.debug("task result was empty in request: {}, just return", request);
return;
}
for (CommandEntity command : request.getCommandInfo()) {
@@ -339,7 +339,7 @@ public class AgentServiceImpl implements AgentService {
private List<DataConfig> processQueuedTasks(TaskRequest request) {
HashSet<SourceStatus> needAddStatusSet = Sets.newHashSet(SourceStatus.TOBE_ISSUED_SET);
if (PullJobTypeEnum.NEVER == PullJobTypeEnum.getPullJobType(request.getPullJobType())) {
- LOGGER.warn("agent pull job type is [NEVER], just pull to be active tasks");
+ LOGGER.debug("agent pull job type is [NEVER], just pull to be active tasks");
needAddStatusSet.remove(SourceStatus.TO_BE_ISSUED_ADD);
}
@@ -369,7 +369,7 @@ public class AgentServiceImpl implements AgentService {
private void preProcessNonFileTasks(TaskRequest taskRequest) {
List<Integer> needAddStatusList;
if (PullJobTypeEnum.NEVER == PullJobTypeEnum.getPullJobType(taskRequest.getPullJobType())) {
- LOGGER.warn("agent pull job type is [NEVER], just pull to be active tasks");
+ LOGGER.debug("agent pull job type is [NEVER], just pull to be active tasks");
needAddStatusList = Collections.singletonList(SourceStatus.TO_BE_ISSUED_ACTIVE.getCode());
} else {
needAddStatusList = Arrays.asList(SourceStatus.TO_BE_ISSUED_ADD.getCode(),
@@ -482,12 +482,12 @@ public class AgentServiceImpl implements AgentService {
SourceStatus.SOURCE_NORMAL,
SourceStatus.TO_BE_ISSUED_ADD,
SourceStatus.TO_BE_ISSUED_ACTIVE);
- Set<GroupStatus> exceptedMatchedGroupStatus = Sets.newHashSet(GroupStatus.SUSPENDED,
- GroupStatus.SUSPENDING);
+ Set<GroupStatus> matchedGroupStatus = Sets.newHashSet(
+ GroupStatus.CONFIG_SUCCESSFUL, GroupStatus.RESTARTED);
if (matchGroup(sourceEntity, clusterNodeEntity)
&& groupEntity != null
&& !exceptedMatchedSourceStatus.contains(SourceStatus.forCode(sourceEntity.getStatus()))
- && !exceptedMatchedGroupStatus.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
+ && matchedGroupStatus.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
LOGGER.info("Transform task({}) from {} to {} because tag rematch "
+ "for agent({}) in cluster({})", sourceEntity.getAgentIp(),
sourceEntity.getStatus(), SourceStatus.TO_BE_ISSUED_ACTIVE.getCode(),
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
index d27c018653..6f0cbebaf3 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
@@ -48,7 +48,6 @@ import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@@ -161,18 +160,13 @@ class AgentServiceTest extends ServiceBaseTest {
sourceService.logicDeleteAll(groupId, streamId, GLOBAL_OPERATOR);
}
- @BeforeAll
- public static void setUp(
- @Autowired AgentService agentService,
- @Autowired HeartbeatService heartbeatService) {
- agent = new MockAgent(agentService, heartbeatService, 2);
- agent.sendHeartbeat();
- }
-
@BeforeEach
public void setupEach() {
+ agent = new MockAgent(agentService, heartbeatService, 2);
+ agent.sendHeartbeat();
groupStreamCache = new ArrayList<>();
groupCache = new ArrayList<>();
+ super.login();
}
@AfterEach
@@ -251,13 +245,27 @@ class AgentServiceTest extends ServiceBaseTest {
// bind group and rematch
bindGroup(true, "group1");
TaskResult t2 = agent.pullTask();
- Assertions.assertEquals(1, t2.getDataConfigs().size());
- Assertions.assertEquals(1, t2.getDataConfigs().stream()
+ Assertions.assertEquals(0, t2.getDataConfigs().size());
+ Assertions.assertEquals(0, t2.getDataConfigs().stream()
+ .filter(dataConfig -> Integer.valueOf(dataConfig.getOp()) == ManagerOpEnum.ACTIVE.getType())
+ .collect(Collectors.toSet())
+ .size());
+
+ // update group to config success
+ final String groupId = sourceService.listSource(groupStream.getLeft(), groupStream.getRight()).stream()
+ .filter(source -> source.getTemplateId() != null)
+ .findAny()
+ .get()
+ .getInlongGroupId();
+ groupMapper.updateStatus(groupId, GroupStatus.CONFIG_SUCCESSFUL.getCode(), GLOBAL_OPERATOR);
+ TaskResult t3 = agent.pullTask();
+ Assertions.assertEquals(1, t3.getDataConfigs().size());
+ Assertions.assertEquals(1, t3.getDataConfigs().stream()
.filter(dataConfig -> Integer.valueOf(dataConfig.getOp()) == ManagerOpEnum.ACTIVE.getType())
.collect(Collectors.toSet())
.size());
- DataConfig d2 = t2.getDataConfigs().get(0);
- Assertions.assertEquals(sourceId, d2.getTaskId());
+ DataConfig d3 = t3.getDataConfigs().get(0);
+ Assertions.assertEquals(sourceId, d3.getTaskId());
}
/**
[inlong] 02/06: [INLONG-8511][Manager] Add admin permission of public tenant to all users (#8513)
Posted by ve...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 5189f56377876676deb184ec091a6716e190dd59
Author: vernedeng <ve...@apache.org>
AuthorDate: Thu Jul 13 12:10:44 2023 +0800
[INLONG-8511][Manager] Add admin permission of public tenant to all users (#8513)
---
.../inlong/manager/service/user/TenantRoleService.java | 5 +++++
.../inlong/manager/service/user/TenantRoleServiceImpl.java | 12 ++++++++++++
.../apache/inlong/manager/service/user/UserServiceImpl.java | 5 +++++
.../org/apache/inlong/manager/service/ServiceBaseTest.java | 12 +++++++++---
inlong-manager/manager-web/sql/changes-1.8.0.sql | 4 ++++
.../inlong/manager/web/filter/TenantInsertionFilter.java | 1 -
.../inlong/manager/web/filter/TenantInsertionFilterTest.java | 2 --
7 files changed, 35 insertions(+), 6 deletions(-)
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleService.java
index 6add8e0a28..43fa062138 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleService.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleService.java
@@ -66,4 +66,9 @@ public interface TenantRoleService {
* @return true = delete success/ false = delete fail
*/
Boolean delete(Integer id);
+
+ /**
+ * Add user to default tenant
+ */
+ Integer saveDefault(String username, String operator);
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleServiceImpl.java
index a0c038c3de..618d014cec 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/TenantRoleServiceImpl.java
@@ -31,6 +31,7 @@ import org.apache.inlong.manager.pojo.user.LoginUserUtils;
import org.apache.inlong.manager.pojo.user.TenantRoleInfo;
import org.apache.inlong.manager.pojo.user.TenantRolePageRequest;
import org.apache.inlong.manager.pojo.user.TenantRoleRequest;
+import org.apache.inlong.manager.pojo.user.UserRoleCode;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
@@ -40,6 +41,7 @@ import org.springframework.stereotype.Service;
import java.util.List;
+import static org.apache.inlong.common.util.BasicAuth.DEFAULT_TENANT;
import static org.apache.inlong.manager.common.enums.ErrorCodeEnum.TENANT_NOT_EXIST;
/**
@@ -137,4 +139,14 @@ public class TenantRoleServiceImpl implements TenantRoleService {
return true;
}
+ @Override
+ public Integer saveDefault(String username, String operator) {
+ // make default public tenant permission
+ TenantRoleRequest tenantRoleRequest = new TenantRoleRequest();
+ tenantRoleRequest.setTenant(DEFAULT_TENANT);
+ tenantRoleRequest.setRoleCode(UserRoleCode.TENANT_OPERATOR);
+ tenantRoleRequest.setUsername(username);
+ return this.save(tenantRoleRequest, operator);
+ }
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/UserServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/UserServiceImpl.java
index 8f79bcb1c6..450699746f 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/UserServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/user/UserServiceImpl.java
@@ -104,6 +104,8 @@ public class UserServiceImpl implements UserService {
@Autowired
private UserEntityMapper userMapper;
@Autowired
+ private TenantRoleService tenantRoleService;
+ @Autowired
private InlongGroupEntityMapper groupMapper;
@Autowired
private InlongClusterEntityMapper clusterMapper;
@@ -147,6 +149,9 @@ public class UserServiceImpl implements UserService {
}
Preconditions.expectTrue(userMapper.insert(entity) > 0, "Create user failed");
+ Preconditions.expectTrue(tenantRoleService.saveDefault(username, currentUser) > 0,
+ "Add default tenant role failed");
+
LOGGER.debug("success to create user info={}", request);
return entity.getId();
}
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/ServiceBaseTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/ServiceBaseTest.java
index 4aae873949..27bed28ac8 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/ServiceBaseTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/ServiceBaseTest.java
@@ -32,7 +32,8 @@ import org.apache.inlong.manager.service.group.InlongGroupService;
import org.apache.inlong.manager.service.stream.InlongStreamService;
import org.apache.inlong.manager.test.BaseTest;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,14 +63,19 @@ public class ServiceBaseTest extends BaseTest {
@Autowired
protected InlongStreamService streamService;
- @BeforeEach
- public void login() {
+ @BeforeAll
+ public static void login() {
UserInfo userInfo = new UserInfo();
userInfo.setTenant(PUBLIC_TENANT);
userInfo.setName(GLOBAL_OPERATOR);
LoginUserUtils.setUserLoginInfo(userInfo);
}
+ @AfterAll
+ public static void logout() {
+ LoginUserUtils.removeUserLoginInfo();
+ }
+
@Test
public void test() {
LOGGER.info("The test class cannot be empty, otherwise 'No runnable methods exception' will be reported");
diff --git a/inlong-manager/manager-web/sql/changes-1.8.0.sql b/inlong-manager/manager-web/sql/changes-1.8.0.sql
index edd0abb7ac..b1ef5f8a78 100644
--- a/inlong-manager/manager-web/sql/changes-1.8.0.sql
+++ b/inlong-manager/manager-web/sql/changes-1.8.0.sql
@@ -78,6 +78,10 @@ ALTER TABLE tenant_user_role
CREATE INDEX index_tenant
ON tenant_user_role (tenant, is_deleted);
+INSERT INTO tenant_user_role(username, role_code, tenant, creator)
+ SELECT name, 'TENANT_ADMIN', 'public', 'inlong_init'
+ FROM user;
+
-- To avoid the ambiguity, rename "tenant" in PulsarGroup & PulsarCluster to "pulsarTenant"
UPDATE inlong_group SET ext_params = replace(ext_params, '"tenant"', '"pulsarTenant"');
UPDATE inlong_cluster SET ext_params = replace(ext_params, '"tenant"', '"pulsarTenant"');
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/filter/TenantInsertionFilter.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/filter/TenantInsertionFilter.java
index 6cce47a23a..9b9f5ef3b1 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/filter/TenantInsertionFilter.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/filter/TenantInsertionFilter.java
@@ -68,7 +68,6 @@ public class TenantInsertionFilter implements Filter {
}
inlongWrapper.addParameter(BASIC_AUTH_TENANT_HEADER, tenant);
- inlongWrapper.addBodyParam(BASIC_AUTH_TENANT_HEADER, tenant);
filterChain.doFilter(inlongWrapper, servletResponse);
}
diff --git a/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/filter/TenantInsertionFilterTest.java b/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/filter/TenantInsertionFilterTest.java
index 73c3b092a1..db62200a70 100644
--- a/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/filter/TenantInsertionFilterTest.java
+++ b/inlong-manager/manager-web/src/test/java/org/apache/inlong/manager/web/filter/TenantInsertionFilterTest.java
@@ -111,8 +111,6 @@ public class TenantInsertionFilterTest extends WebFilterConfig {
Assertions.assertEquals(targetTenant, tenant);
Map<String, String> paraMap = HttpContextUtils.getParameterMapAll(request);
Assertions.assertTrue(paraMap.containsKey(BasicAuth.BASIC_AUTH_TENANT_HEADER));
- String body = HttpContextUtils.getBodyString(request);
- Assertions.assertTrue(body.contains(BasicAuth.BASIC_AUTH_TENANT_HEADER));
}
}
}
\ No newline at end of file
[inlong] 04/06: [INLONG-8527][Manager] Fix wrong capitalization of InlongRequestWrapper (#8528)
Posted by ve...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit c5a2d063cee201ecb470b1afc11a0249b9e56cf2
Author: vernedeng <ve...@apache.org>
AuthorDate: Thu Jul 13 14:42:30 2023 +0800
[INLONG-8527][Manager] Fix wrong capitalization of InlongRequestWrapper (#8528)
---
.../org/apache/inlong/manager/web/utils/InlongRequestWrapper.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/utils/InlongRequestWrapper.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/utils/InlongRequestWrapper.java
index 08bc0cc729..efbe6846e6 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/utils/InlongRequestWrapper.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/utils/InlongRequestWrapper.java
@@ -132,7 +132,7 @@ public class InlongRequestWrapper extends HttpServletRequestWrapper {
@Override
public String getHeader(String name) {
- return this.headers.get(name);
+ return this.headers.get(name.toLowerCase());
}
@Override
@@ -141,7 +141,7 @@ public class InlongRequestWrapper extends HttpServletRequestWrapper {
}
public void addHeader(String name, String value) {
- headers.put(name, value);
+ headers.put(name.toLowerCase(), value);
}
public void addBodyParam(String key, String value) throws JsonProcessingException {
[inlong] 03/06: [INLONG-8524][Agent] Update the JVM Options for Agent (#8525)
Posted by ve...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 7f924a6c2baab6fcc372480097811f2745be2c3b
Author: Charles Zhang <do...@apache.org>
AuthorDate: Thu Jul 13 12:51:51 2023 +0800
[INLONG-8524][Agent] Update the JVM Options for Agent (#8525)
---
inlong-agent/bin/agent-env.sh | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
diff --git a/inlong-agent/bin/agent-env.sh b/inlong-agent/bin/agent-env.sh
index 98718573c7..555381caf9 100755
--- a/inlong-agent/bin/agent-env.sh
+++ b/inlong-agent/bin/agent-env.sh
@@ -37,15 +37,16 @@ else
fi
if [ -z "$AGENT_JVM_HEAP_OPTS" ]; then
- HEAP_OPTS="-Xms512m -Xmx6656m"
+ HEAP_OPTS="-Xmx512m -Xss512k"
else
HEAP_OPTS="$AGENT_JVM_HEAP_OPTS"
fi
-GC_OPTS="-XX:SurvivorRatio=6 -XX:+UseMembar -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+CMSScavengeBeforeRemark -XX:ParallelCMSThreads=3 -XX:+TieredCompilation -XX:+UseCMSCompactAtFullCollection -verbose:gc -Xloggc:$BASE_DIR/logs/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$BASE_DIR/logs/ -XX:+CMSClassUnloadingEnabled -XX:CMSInitiatingOccupancyFraction=60 -XX:CMSFullGCsBeforeCompaction=1 -Dsun.net.inetaddr.ttl=3 -Dsun. [...]
+GC_OPTS="-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:InitiatingHeapOccupancyPercent=60 -Djava.net.preferIPv4Stack=true -Dfile.encoding=UTF-8"
+LOG_OPTS="-Xloggc:$BASE_DIR/logs/gc.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=20M"
if [ -n "$NEED_TRACK_NATIVE_MEMORY" ] && [ "$NEED_TRACK_NATIVE_MEMORY" = "true" ]; then
GC_OPTS="$GC_OPTS -XX:NativeMemoryTracking"
fi
-AGENT_JVM_ARGS="$HEAP_OPTS $GC_OPTS"
+AGENT_JVM_ARGS="$HEAP_OPTS $GC_OPTS $LOG_OPTS"
# Add Agent Rmi Args when necessary
AGENT_RMI_ARGS="-Dcom.sun.management.jmxremote \
@@ -59,4 +60,4 @@ if [[ $JMX_ENABLED == 1 ]]; then
export AGENT_ARGS="$AGENT_JVM_ARGS $AGENT_RMI_ARGS -cp $CLASSPATH -Dagent.home=$BASE_DIR"
else
export AGENT_ARGS="$AGENT_JVM_ARGS -cp $CLASSPATH -Dagent.home=$BASE_DIR"
-fi
\ No newline at end of file
+fi
[inlong] 05/06: [INLONG-8509][Manager] Optimize preProcessTemplateFileTask in AgentServiceImpl (#8510)
Posted by ve...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit e96016a04a07f482bb34e1b11fdaaaf30aead637
Author: kipshi <48...@users.noreply.github.com>
AuthorDate: Thu Jul 13 15:39:05 2023 +0800
[INLONG-8509][Manager] Optimize preProcessTemplateFileTask in AgentServiceImpl (#8510)
Co-authored-by: kipshi <ki...@tencent.com>
---
.../service/core/impl/AgentServiceImpl.java | 53 ++++++++++++----------
.../service/core/impl/AgentServiceTest.java | 6 +--
2 files changed, 31 insertions(+), 28 deletions(-)
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
index ad594c3514..174929741e 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
@@ -407,30 +407,35 @@ public class AgentServiceImpl implements AgentService {
// find those node whose tag match stream_source tag and agent ip match stream_source agent ip
List<StreamSourceEntity> sourceEntities = sourceMapper.selectTemplateSourceByCluster(needCopiedStatusList,
Lists.newArrayList(SourceType.FILE), agentClusterName);
- Set<GroupStatus> noNeedAddTask = Sets.newHashSet(GroupStatus.SUSPENDED, GroupStatus.SUSPENDING);
- sourceEntities.forEach(sourceEntity -> {
- InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId());
- if (groupEntity != null && noNeedAddTask.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
- return;
- }
- StreamSourceEntity subSource = sourceMapper.selectOneByTemplatedIdAndAgentIp(sourceEntity.getId(),
- agentIp);
- if (subSource == null) {
- // if not, clone a subtask for this Agent.
- // note: a new source name with random suffix is generated to adhere to the unique constraint
- StreamSourceEntity fileEntity =
- CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new);
- fileEntity.setSourceName(fileEntity.getSourceName() + "-"
- + RandomStringUtils.randomAlphanumeric(10).toLowerCase(Locale.ROOT));
- fileEntity.setTemplateId(sourceEntity.getId());
- fileEntity.setAgentIp(agentIp);
- fileEntity.setStatus(SourceStatus.TO_BE_ISSUED_ADD.getCode());
- // create new sub source task
- sourceMapper.insert(fileEntity);
- LOGGER.info("Transform new template task({}) for agent({}) in cluster({}).",
- fileEntity.getId(), taskRequest.getAgentIp(), taskRequest.getClusterName());
- }
- });
+ Set<GroupStatus> noNeedAddTask = Sets.newHashSet(
+ GroupStatus.SUSPENDED, GroupStatus.SUSPENDING, GroupStatus.DELETING, GroupStatus.DELETED);
+ sourceEntities.stream()
+ .forEach(sourceEntity -> {
+ InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId());
+ if (groupEntity != null && noNeedAddTask.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
+ return;
+ }
+ StreamSourceEntity subSource = sourceMapper.selectOneByTemplatedIdAndAgentIp(sourceEntity.getId(),
+ agentIp);
+ if (subSource == null) {
+ InlongClusterNodeEntity clusterNodeEntity = selectByIpAndCluster(agentClusterName, agentIp);
+ // if stream_source match node_group with node, clone a subtask for this Agent.
+ // note: a new source name with random suffix is generated to adhere to the unique constraint
+ if (matchGroup(sourceEntity, clusterNodeEntity)) {
+ StreamSourceEntity fileEntity =
+ CommonBeanUtils.copyProperties(sourceEntity, StreamSourceEntity::new);
+ fileEntity.setSourceName(fileEntity.getSourceName() + "-"
+ + RandomStringUtils.randomAlphanumeric(10).toLowerCase(Locale.ROOT));
+ fileEntity.setTemplateId(sourceEntity.getId());
+ fileEntity.setAgentIp(agentIp);
+ fileEntity.setStatus(SourceStatus.TO_BE_ISSUED_ADD.getCode());
+ // create new sub source task
+ sourceMapper.insert(fileEntity);
+ LOGGER.info("Transform new template task({}) for agent({}) in cluster({}).",
+ fileEntity.getId(), taskRequest.getAgentIp(), taskRequest.getClusterName());
+ }
+ }
+ });
}
/**
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
index 0fdae8b096..d27c018653 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
@@ -71,8 +71,6 @@ class AgentServiceTest extends ServiceBaseTest {
@Autowired
private AgentService agentService;
@Autowired
- private HeartbeatService heartbeatService;
- @Autowired
private InlongGroupEntityMapper groupMapper;
@Autowired
private InlongStreamServiceTest streamServiceTest;
@@ -212,12 +210,12 @@ class AgentServiceTest extends ServiceBaseTest {
TaskResult taskResult = agent.pullTask();
Assertions.assertTrue(taskResult.getCmdConfigs().isEmpty());
- Assertions.assertEquals(4, taskResult.getDataConfigs().size());
+ Assertions.assertEquals(3, taskResult.getDataConfigs().size());
Assertions.assertEquals(3, taskResult.getDataConfigs().stream()
.filter(dataConfig -> Integer.valueOf(dataConfig.getOp()) == ManagerOpEnum.ADD.getType())
.collect(Collectors.toSet())
.size());
- Assertions.assertEquals(1, taskResult.getDataConfigs().stream()
+ Assertions.assertEquals(0, taskResult.getDataConfigs().stream()
.filter(dataConfig -> Integer.valueOf(dataConfig.getOp()) == ManagerOpEnum.FROZEN.getType())
.collect(Collectors.toSet())
.size());
[inlong] 01/06: [INLONG-8520][Agent] Makeup the code to figure out whether it is kvm、k8s or not (#8521)
Posted by ve...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 788a21cb06a632410296035b3cf144b30d544f78
Author: justinwwhuang <hw...@163.com>
AuthorDate: Thu Jul 13 11:50:17 2023 +0800
[INLONG-8520][Agent] Makeup the code to figure out whether it is kvm、k8s or not (#8521)
---
.../inlong/agent/plugin/sources/reader/file/FileReaderOperator.java | 6 ++++++
1 file changed, 6 insertions(+)
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/file/FileReaderOperator.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/file/FileReaderOperator.java
index bcd4b59272..24db072eed 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/file/FileReaderOperator.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/file/FileReaderOperator.java
@@ -113,6 +113,7 @@ public class FileReaderOperator extends AbstractReader {
private final long monitorActiveInterval = 60 * 1000;
private final BlockingQueue<String> queue = new LinkedBlockingQueue<>(CACHE_QUEUE_SIZE);
private final StringBuffer sb = new StringBuffer();
+ private boolean needMetadata = false;
public FileReaderOperator(File file, int position) {
this(file, position, "");
@@ -371,6 +372,9 @@ public class FileReaderOperator extends AbstractReader {
}
public String fillMetaData(String message) {
+ if (!needMetadata) {
+ return message;
+ }
long timestamp = System.currentTimeMillis();
boolean isJson = FileDataUtils.isJSON(message);
Map<String, String> mergeData = new HashMap<>(metadata);
@@ -390,8 +394,10 @@ public class FileReaderOperator extends AbstractReader {
String[] env = jobConf.get(JOB_FILE_META_ENV_LIST).split(COMMA);
Arrays.stream(env).forEach(data -> {
if (data.equalsIgnoreCase(KUBERNETES)) {
+ needMetadata = true;
new KubernetesMetadataProvider(this).getData();
} else if (data.equalsIgnoreCase(ENV_CVM)) {
+ needMetadata = true;
metadata.put(METADATA_HOST_NAME, AgentUtils.getLocalHost());
metadata.put(METADATA_SOURCE_IP, AgentUtils.fetchLocalIp());
metadata.put(METADATA_FILE_NAME, file.getName());