feat:add deploy_space nacos config

apache · Dec 4, 2024 · 8191e07 · 8191e07
1 parent 66402b2
commit 8191e07
Show file tree

Hide file tree

Showing 8 changed files with 17 additions and 5 deletions.
diff --git a/ozhera-log/log-common/src/main/java/org/apache/ozhera/log/model/LogtailConfig.java b/ozhera-log/log-common/src/main/java/org/apache/ozhera/log/model/LogtailConfig.java
@@ -38,6 +38,8 @@ public class LogtailConfig {
     private String type;
     private Integer appType;
 
+    private String deploySpace;
+
     private Integer parseType;
     private String tail;
     /**

diff --git a/...-manager/src/main/java/org/apache/ozhera/log/manager/service/MilogConfigNacosService.java b/...-manager/src/main/java/org/apache/ozhera/log/manager/service/MilogConfigNacosService.java
@@ -32,7 +32,7 @@ public interface MilogConfigNacosService {
 
     void publishStreamConfig(Long spaceId, Integer type, Integer projectType, String motorRoomEn);
 
-    void publishNameSpaceConfig(String motorRoomEn, Long spaceId, Long storeId, Long tailId, Integer type, String changeType);
+    void publishNameSpaceConfig(String motorRoomEn, Long spaceId, Long storeId, Long tailId, String deploySpace,Integer type, String changeType);
 
     void removeStreamConfig(String motorRoomEn, Long id);
 }
diff --git a/...-manager/src/main/java/org/apache/ozhera/log/manager/service/impl/LogTailServiceImpl.java b/...-manager/src/main/java/org/apache/ozhera/log/manager/service/impl/LogTailServiceImpl.java
@@ -297,7 +297,7 @@ public void handleNacosConfigByMotorRoom(MilogLogTailDo mt, String motorRoomEn,
         if (OperateEnum.ADD_OPERATE.getCode().equals(type) || OperateEnum.UPDATE_OPERATE.getCode().equals(type)) {
             milogConfigNacosServiceImpl.publishStreamConfig(mt.getSpaceId(), type, projectType, motorRoomEn);
         }
-        milogConfigNacosServiceImpl.publishNameSpaceConfig(motorRoomEn, mt.getSpaceId(), mt.getStoreId(), mt.getId(), type, "");
+        milogConfigNacosServiceImpl.publishNameSpaceConfig(motorRoomEn, mt.getSpaceId(), mt.getStoreId(), mt.getId(), mt.getDeploySpace(), type, "");
     }
 
     @Override
@@ -310,11 +310,11 @@ public boolean deleteConfigRemote(Long spaceId, Long id, String motorRoomEn, Log
             return true;
         }
         if (LogStructureEnum.STORE == logStructureEnum) {
-            milogConfigNacosServiceImpl.publishNameSpaceConfig(motorRoomEn, spaceId, id, null, OperateEnum.DELETE_OPERATE.getCode(), logStructureEnum.getCode());
+            milogConfigNacosServiceImpl.publishNameSpaceConfig(motorRoomEn, spaceId, id, null, null, OperateEnum.DELETE_OPERATE.getCode(), logStructureEnum.getCode());
             return true;
         }
         MilogLogTailDo tailDo = milogLogtailDao.queryById(id);
-        milogConfigNacosServiceImpl.publishNameSpaceConfig(motorRoomEn, spaceId, tailDo.getStoreId(), id, OperateEnum.DELETE_OPERATE.getCode(), logStructureEnum.getCode());
+        milogConfigNacosServiceImpl.publishNameSpaceConfig(motorRoomEn, spaceId, tailDo.getStoreId(), id, null, OperateEnum.DELETE_OPERATE.getCode(), logStructureEnum.getCode());
         return true;
     }
 

diff --git a/...src/main/java/org/apache/ozhera/log/manager/service/impl/MilogConfigNacosServiceImpl.java b/...src/main/java/org/apache/ozhera/log/manager/service/impl/MilogConfigNacosServiceImpl.java
@@ -240,7 +240,7 @@ private List<String> ensureDefaultCompatibility(MiLogStreamConfig existConfig, L
     }
 
     @Override
-    public void publishNameSpaceConfig(String motorRoomEn, Long spaceId, Long storeId, Long tailId, Integer type, String changeType) {
+    public void publishNameSpaceConfig(String motorRoomEn, Long spaceId, Long storeId, Long tailId, String deploySpace, Integer type, String changeType) {
         Assert.notNull(spaceId, "logSpaceId not empty");
         Assert.notNull(storeId, "storeId not empty");
         //send msg
@@ -436,6 +436,11 @@ private StorageInfo buildEsInfo(MilogEsClusterDO clusterDO) {
         return new StorageInfo(clusterDO.getId(), clusterDO.getAddr(), clusterDO.getToken(), clusterDO.getDtCatalog(), clusterDO.getDtDatabase());
     }
 
+    public static void main(String[] args) {
+        LogtailConfig logtailConfig = new LogtailConfig();
+        logtailConfig.setDeploySpace("123");
+        System.out.println(logtailConfig.getDeploySpace());
+    }
 
     public LogtailConfig assembleLogTailConfigs(Long tailId) {
         LogtailConfig logtailConfig = new LogtailConfig();
@@ -447,6 +452,7 @@ public LogtailConfig assembleLogTailConfigs(Long tailId) {
             logtailConfig.setParseScript(milogLogTail.getParseScript());
             logtailConfig.setValueList(milogLogTail.getValueList());
             logtailConfig.setAppType(milogLogTail.getAppType());
+            logtailConfig.setDeploySpace(milogLogTail.getDeploySpace());
             // Query MQ information
             handleTailConfig(tailId, milogLogTail.getStoreId(), milogLogTail.getSpaceId(),
                     milogLogTail.getMilogAppId(), logtailConfig, milogLogTail.getAppType());

diff --git a/...-log/log-stream/src/main/java/org/apache/ozhera/log/stream/common/LogStreamConstants.java b/...-log/log-stream/src/main/java/org/apache/ozhera/log/stream/common/LogStreamConstants.java
@@ -38,5 +38,6 @@ public class LogStreamConstants {
     public static final String LOG_STREAM_SPACE_ID = "spaceId";
     public static final String LOG_STREAM_STORE_ID = "storeId";
     public static final String LOG_STREAM_TAIL_ID = "tailId";
+    public static final String DEPLOY_SPACE = "deploy_space";
 
 }
diff --git a/ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/job/JobManager.java b/ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/job/JobManager.java
@@ -186,6 +186,7 @@ private SinkJobConfig buildSinkJobConfig(String type, String ak, String sk, Stri
                 .jobType(SinkJobEnum.NORMAL_JOB.name())
                 .storageType(sinkConfig.getStorageType())
                 .consumerGroup(logtailConfig.getConsumerGroup())
+                .deploySpace(logtailConfig.getDeploySpace())
                 .build();
         sinkJobConfig.setLogTailId(logtailConfig.getLogtailId());
         sinkJobConfig.setLogStoreId(sinkConfig.getLogstoreId());

diff --git a/ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/job/LogDataTransfer.java b/ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/job/LogDataTransfer.java
@@ -143,6 +143,7 @@ private void putCommonData(Map<String, Object> dataMap) {
         dataMap.putIfAbsent(LOG_STREAM_SPACE_ID, sinkJobConfig.getLogSpaceId());
         dataMap.putIfAbsent(LOG_STREAM_STORE_ID, sinkJobConfig.getLogStoreId());
         dataMap.putIfAbsent(LOG_STREAM_TAIL_ID, sinkJobConfig.getLogTailId());
+        dataMap.putIfAbsent(DEPLOY_SPACE, sinkJobConfig.getDeploySpace());
     }
 
     private void sendMessage(Map<String, Object> dataMap) throws Exception {

diff --git a/ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/job/SinkJobConfig.java b/ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/job/SinkJobConfig.java
@@ -61,4 +61,5 @@ public class SinkJobConfig extends LogConfig {
      */
     private String jobType;
     private String consumerGroup;
+    private String deploySpace;
 }