refactor: access to log by nacos config

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/dao/MilogLogTailDao.java

@@ -383,4 +383,14 @@ public List<MilogLogTailDo> queryByNames(Long storeId, List<String> nameList) {
         Cnd cnd = Cnd.where("store_id", EQUAL_OPERATE, storeId).and("tail", "in", nameList);
         return dao.query(MilogLogTailDo.class, cnd);
     }
+
+    public List<MilogLogTailDo> queryByCondition(Long spaceId, Long storeId, String tailName, Long heraAppId, Long envId,
+                                                 String logPath) {
+        return dao.query(MilogLogTailDo.class, Cnd.where("space_id", EQUAL_OPERATE, spaceId)
+                .and("store_id", EQUAL_OPERATE, storeId)
+                .and("tail", EQUAL_OPERATE, tailName)
+                .and("milog_app_id", EQUAL_OPERATE, heraAppId)
+                .and("env_id", EQUAL_OPERATE, envId)
+                .and("log_path", EQUAL_OPERATE, logPath));
+    }
 }

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/domain/Tpc.java

@@ -15,6 +15,7 @@
  */
 package com.xiaomi.mone.log.manager.domain;
 
+import com.google.common.collect.Lists;
 import com.xiaomi.mone.log.manager.common.context.MoneUserContext;
 import com.xiaomi.mone.log.manager.common.exception.MilogManageException;
 import com.xiaomi.mone.log.manager.dao.MilogSpaceDao;
@@ -38,9 +39,11 @@
 import com.xiaomi.youpin.docean.plugin.dubbo.anno.Reference;
 import com.xiaomi.youpin.infra.rpc.Result;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
 
 import javax.annotation.Resource;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 import static com.xiaomi.mone.log.common.Constant.GSON;
@@ -110,6 +113,50 @@ public Result<PageDataVo<NodeVo>> getUserPermSpace(String spaceName, Integer pag
         return tpcService.orgNodelist(param);
     }
 
+    public List<NodeVo> queryUserListNode(String spaceName, String userName) {
+        handleRemoteTpcId(tpcNodeCode, userName, UserTypeEnum.CAS_TYPE.getCode());
+
+        boolean isAdmin = isAdmin(userName, UserTypeEnum.CAS_TYPE.getCode());
+
+        List<NodeVo> nodeVoList = Lists.newArrayList();
+        int page = 1;
+        Integer pageSize = 100;
+        while (true) {
+            NodeQryParam param = createNodeQryParam(spaceName, userName, page, isAdmin, pageSize);
+            Result<PageDataVo<NodeVo>> pageDataVoResult = tpcService.orgNodelist(param);
+
+            Optional.ofNullable(pageDataVoResult)
+                    .map(Result::getData)
+                    .map(PageDataVo::getList)
+                    .ifPresentOrElse(nodeVoList::addAll, () -> log.info("No more nodes found. Breaking out of the loop."));
+
+            if (pageDataVoResult == null || pageDataVoResult.getData() == null || CollectionUtils.isEmpty(pageDataVoResult.getData().getList())) {
+                break;
+            }
+
+            page++;
+        }
+
+        return nodeVoList;
+    }
+
+    private NodeQryParam createNodeQryParam(String spaceName, String userName, int page, boolean isAdmin, Integer pageSize) {
+        NodeQryParam param = new NodeQryParam();
+        param.setPager(true);
+        param.setPage(page);
+        param.setPageSize(pageSize);
+        param.setParentId(tpcPId);
+        param.setAccount(userName);
+        param.setType(NodeTypeEnum.PRO_SUB_GROUP.getCode());
+        param.setUserType(UserTypeEnum.CAS_TYPE.getCode());
+        param.setStatus(NodeStatusEnum.ENABLE.getCode());
+        param.setMyNode(!isAdmin);
+        if (StringUtils.isNotEmpty(spaceName)) {
+            param.setNodeName(spaceName);
+        }
+        return param;
+    }
+
     /**
      * set tpc id from tpc server
      */

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/model/bo/AutoAccessLogParam.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2020 Xiaomi
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package com.xiaomi.mone.log.manager.model.bo;
+
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ *
+ * @description
+ * @version 1.0
+ * @author wtt
+ * @date 2024/7/10 10:54
+ *
+ */
+@Data
+public class AutoAccessLogParam {
+
+    private Long appId;
+    private String appName;
+    private Long envId;
+    private String envName;
+    private List<String> ips;
+    private String logPath;
+
+    private String tailName;
+
+    private Long spaceId;
+    private Long storeId;
+}

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/service/SpaceAuthService.java

@@ -21,6 +21,8 @@
 import com.xiaomi.mone.tpc.common.vo.PageDataVo;
 import com.xiaomi.youpin.infra.rpc.Result;
 
+import java.util.List;
+
 /**
  * @author wtt
  * @version 1.0
@@ -76,4 +78,13 @@ public interface SpaceAuthService {
      * @param memberCode
      */
     void addSpaceMember(Long spaceId, String userAccount, Integer userType, Integer memberCode);
+
+    /**
+     * query access to user node list
+     * @param spaceName
+     * @param userName
+     * @return
+     */
+    Result<List<NodeVo>> queryUserListNode(String spaceName, String userName);
+
 }

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/service/impl/TpcSpaceAuthService.java

@@ -25,6 +25,7 @@
 import com.xiaomi.youpin.infra.rpc.Result;
 
 import javax.annotation.Resource;
+import java.util.List;
 
 /**
  * @author wtt
@@ -67,4 +68,9 @@ public void addSpaceMember(Long spaceId, String userAccount, Integer userType, I
         tpc.addSpaceMember(spaceId, userAccount, userType, memberCode);
     }
 
+    @Override
+    public Result<List<NodeVo>> queryUserListNode(String spaceName, String userName) {
+        return Result.success(tpc.queryUserListNode(spaceName, userName));
+    }
+
 }

ozhera-log/log-manager/src/main/java/com/xiaomi/mone/log/manager/service/listener/AutoLogAccessListener.java

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2020 Xiaomi
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package com.xiaomi.mone.log.manager.service.listener;
+
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.listener.Listener;
+import com.alibaba.nacos.api.exception.NacosException;
+import com.google.gson.reflect.TypeToken;
+import com.xiaomi.mone.app.api.response.AppBaseInfo;
+import com.xiaomi.mone.log.api.enums.OperateEnum;
+import com.xiaomi.mone.log.api.enums.ProjectTypeEnum;
+import com.xiaomi.mone.log.manager.dao.MilogLogTailDao;
+import com.xiaomi.mone.log.manager.dao.MilogLogstoreDao;
+import com.xiaomi.mone.log.manager.dao.MilogSpaceDao;
+import com.xiaomi.mone.log.manager.model.bo.AutoAccessLogParam;
+import com.xiaomi.mone.log.manager.model.bo.LogTailParam;
+import com.xiaomi.mone.log.manager.model.pojo.MilogLogStoreDO;
+import com.xiaomi.mone.log.manager.model.pojo.MilogLogTailDo;
+import com.xiaomi.mone.log.manager.service.HeraAppService;
+import com.xiaomi.mone.log.manager.service.extension.tail.TailExtensionService;
+import com.xiaomi.mone.log.manager.service.extension.tail.TailExtensionServiceFactory;
+import com.xiaomi.mone.log.manager.service.impl.LogTailServiceImpl;
+import com.xiaomi.mone.log.parse.LogParserFactory;
+import com.xiaomi.youpin.docean.anno.Component;
+import com.xiaomi.youpin.docean.plugin.config.anno.Value;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.jetbrains.annotations.NotNull;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static com.xiaomi.mone.log.common.Constant.*;
+
+/**
+ *
+ * @description
+ * @version 1.0
+ * @author wtt
+ * @date 2024/7/10 10:45
+ *
+ */
+@Component
+@Slf4j
+public class AutoLogAccessListener {
+    @Resource
+    private MilogSpaceDao spaceDao;
+
+    @Resource
+    private MilogLogstoreDao logStoreDao;
+
+    @Resource
+    private MilogLogTailDao logTailDao;
+
+    @Resource
+    private LogTailServiceImpl logTailService;
+
+    private TailExtensionService tailExtensionService;
+
+    @Resource
+    private HeraAppService heraAppService;
+
+    @Value("$defaultNacosAddres")
+    private String nacosAddress;
+
+    private final String ACCESS_LOG_DATA_ID_KEY = "auto_access_log_key";
+
+    private ConfigService configService;
+
+    public void init() throws NacosException {
+        freshAccessLogList();
+    }
+
+    private void freshAccessLogList() throws NacosException {
+        if (null == configService) {
+            tailExtensionService = TailExtensionServiceFactory.getTailExtensionService();
+            this.configService = initConfigService();
+            startListening();
+        }
+    }
+
+    private ConfigService initConfigService() throws NacosException {
+        Properties properties = new Properties();
+        properties.setProperty("serverAddr", nacosAddress);
+        return NacosFactory.createConfigService(properties);
+    }
+
+    private void startListening() throws NacosException {
+        configService.addListener(ACCESS_LOG_DATA_ID_KEY, DEFAULT_GROUP_ID, new Listener() {
+            @Override
+            public void receiveConfigInfo(String configInfo) {
+                autoLogAccess(configInfo);
+            }
+
+            @Override
+            public Executor getExecutor() {
+                return null;
+            }
+        });
+    }
+
+    private void autoLogAccess(String dataContent) {
+        if (StringUtils.isNotEmpty(dataContent)) {
+            List<AutoAccessLogParam> accessLogParamList = GSON.fromJson(dataContent, new TypeToken<List<AutoAccessLogParam>>() {
+            }.getType());
+            if (CollectionUtils.isNotEmpty(accessLogParamList)) {
+                accessLogParamList.forEach(this::processLogParam);
+            }
+        }
+    }
+
+    private void processLogParam(AutoAccessLogParam logParam) {
+        if (!isValidLogParam(logParam)) {
+            log.warn("AutoLogAccess, data param valid error, data:{}", GSON.toJson(logParam));
+            return;
+        }
+
+        if (!isValidSpace(logParam.getSpaceId())) {
+            log.warn("AutoLogAccess, space not exist:{}", logParam.getSpaceId());
+            return;
+        }
+
+        MilogLogStoreDO logStoreDO = logStoreDao.queryById(logParam.getStoreId());
+        if (logStoreDO == null) {
+            log.warn("AutoLogAccess, store not exist:{}", logParam.getStoreId());
+            return;
+        }
+
+        AppBaseInfo appBaseInfo = heraAppService.queryByAppId(logParam.getAppId(), ProjectTypeEnum.MIONE_TYPE.getCode());
+        if (appBaseInfo == null) {
+            log.warn("AppBaseInfo not exist, appId:{}, appName:{}", logParam.getAppId(), logParam.getAppName());
+            appBaseInfo = buildAppBaseInfo(logParam);
+        }
+
+        String tailName = generateTailName(logParam.getAppName(), logParam.getEnvName(), logParam.getEnvId());
+        if (!logTailExists(logParam, Long.valueOf(appBaseInfo.getId()), tailName)) {
+            addLogTail(logParam, logStoreDO, appBaseInfo, tailName);
+        } else {
+            log.info("AutoLogAccess, tail already exist, tailName:{}", tailName);
+        }
+    }
+
+    @NotNull
+    private static AppBaseInfo buildAppBaseInfo(AutoAccessLogParam logParam) {
+        AppBaseInfo appBaseInfo;
+        appBaseInfo = new AppBaseInfo();
+        appBaseInfo.setId((int) (logParam.getAppId() * 100));
+        appBaseInfo.setAppName(logParam.getAppName());
+        appBaseInfo.setBindId(logParam.getAppId().toString());
+        return appBaseInfo;
+    }
+
+    private boolean isValidLogParam(AutoAccessLogParam logParam) {
+        return logParam.getSpaceId() != null && logParam.getAppId() != null &&
+                logParam.getStoreId() != null && logParam.getLogPath() != null &&
+                logParam.getEnvId() != null && CollectionUtils.isNotEmpty(logParam.getIps());
+    }
+
+    private boolean isValidSpace(Long spaceId) {
+        return spaceDao.queryById(spaceId) != null;
+    }
+
+    private boolean logTailExists(AutoAccessLogParam logParam, Long heraAppId, String tailName) {
+        List<MilogLogTailDo> logTailDos = logTailDao.queryByCondition(logParam.getSpaceId(), logParam.getStoreId(), tailName,
+                heraAppId, logParam.getEnvId(), logParam.getLogPath());
+        return CollectionUtils.isNotEmpty(logTailDos);
+    }
+
+    private void addLogTail(AutoAccessLogParam logParam, MilogLogStoreDO logStoreDO, AppBaseInfo appBaseInfo, String tailName) {
+        LogTailParam logTailParam = buildLogTailParam(logParam.getSpaceId(), logStoreDO, logParam, appBaseInfo, tailName, logParam.getIps());
+        MilogLogTailDo logTailDo = logTailService.buildLogTailDo(logTailParam, logStoreDO, appBaseInfo, DEFAULT_OPERATOR);
+        logTailDao.add(logTailDo);
+
+        String topicName = buildTopicName(logParam, logTailDo.getTail());
+        tailExtensionService.defaultBindingAppTailConfigRel(logTailDo.getId(), logTailDo.getMilogAppId(), logStoreDO.getMqResourceId(), topicName, null);
+        logTailService.sengMessageToStream(logTailDo, OperateEnum.ADD_OPERATE.getCode());
+
+        CompletableFuture.runAsync(() -> logTailService.sengMessageToAgent(Long.valueOf(appBaseInfo.getId()), logTailDo));
+    }
+
+    private String buildTopicName(AutoAccessLogParam logParam, String tailName) {
+        return String.format("%s-%s", logParam.getAppId(), tailName);
+    }
+
+    private String generateTailName(String appName, String envName, Long envId) {
+        return String.format("%s-%s-%s", appName, envName, envId);
+    }
+
+    private LogTailParam buildLogTailParam(Long spaceId, MilogLogStoreDO logStoreDO, AutoAccessLogParam logParam,
+                                           AppBaseInfo appBaseInfo, String tailName, List<String> ips) {
+        return LogTailParam.builder()
+                .spaceId(spaceId)
+                .storeId(logStoreDO.getId())
+                .appId(logParam.getAppId())
+                .milogAppId(Long.valueOf(appBaseInfo.getId()))
+                .envId(logParam.getEnvId())
+                .envName(logParam.getEnvName())
+                .tail(tailName)
+                .parseType(LogParserFactory.LogParserEnum.SEPARATOR_PARSE.getCode())
+                .parseScript(DEFAULT_TAIL_SEPARATOR)
+                .logPath(logParam.getLogPath())
+                .ips(ips)
+                .valueList(DEFAULT_VALUE_LIST)
+                .appType(ProjectTypeEnum.MIONE_TYPE.getCode())
+                .deployWay(ProjectTypeEnum.MIONE_TYPE.getCode())
+                .build();
+    }
+}