refactor: compile log with new groupId (#472)

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/LogAgentServerBootstrap.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server;
+
+import com.google.common.collect.Lists;
+import com.xiaomi.data.push.rpc.RpcCmd;
+import com.xiaomi.data.push.rpc.RpcServer;
+import com.xiaomi.data.push.rpc.common.Pair;
+import org.apache.ozhera.log.common.Config;
+import org.apache.ozhera.log.common.Constant;
+import org.apache.ozhera.log.server.porcessor.AgentCollectProgressProcessor;
+import org.apache.ozhera.log.server.porcessor.AgentConfigProcessor;
+import org.apache.ozhera.log.server.porcessor.PingProcessor;
+import com.xiaomi.youpin.docean.Ioc;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+
+import static org.apache.ozhera.log.server.common.ServerConstant.SERVER_PORT;
+
+/**
+ * @author wtt
+ * @version 1.0
+ * @description
+ * @date 2022/12/5 11:24
+ */
+@Slf4j
+public class LogAgentServerBootstrap {
+
+    public static void main(String[] args) throws IOException {
+        String nacosAddr = Config.ins().get("nacosAddr", "");
+        String serverName = Config.ins().get("serverName", "");
+        log.info("nacos:{} name:{}", nacosAddr, serverName);
+        RpcServer rpcServer = new RpcServer(nacosAddr, serverName);
+        rpcServer.setListenPort(SERVER_PORT);
+        //Register the processor
+        rpcServer.setProcessorList(Lists.newArrayList(
+                new Pair<>(RpcCmd.pingReq, new PingProcessor()),
+                new Pair<>(Constant.RPCCMD_AGENT_CODE, new AgentCollectProgressProcessor()),
+                new Pair<>(Constant.RPCCMD_AGENT_CONFIG_CODE, new AgentConfigProcessor())
+        ));
+        rpcServer.init();
+        rpcServer.start();
+
+        Ioc.ins().putBean(rpcServer);
+        Ioc.ins().init("com.xiaomi.mone", "com.xiaomi.youpin");
+        log.info("log server start finish");
+    }
+}

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/common/ServerConstant.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server.common;
+
+/**
+ * @author wtt
+ * @version 1.0
+ * @description
+ * @date 2023/10/9 19:34
+ */
+public class ServerConstant {
+    public static final Integer SERVER_PORT = 9899;
+}

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/common/Version.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server.common;
+
+import java.io.Serializable;
+
+/**
+ * @author wangtao
+ */
+public class Version implements Serializable {
+
+
+    @Override
+    public String toString() {
+        return "log-agent-server:2022-12-05:0.0.2";
+    }
+}

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/porcessor/AgentCollectProgressProcessor.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server.porcessor;
+
+import com.xiaomi.data.push.rpc.netty.NettyRequestProcessor;
+import com.xiaomi.data.push.rpc.protocol.RemotingCommand;
+import org.apache.ozhera.log.api.model.vo.UpdateLogProcessCmd;
+import org.apache.ozhera.log.common.Constant;
+import org.apache.ozhera.log.server.common.Version;
+import org.apache.ozhera.log.server.service.DefaultLogProcessCollector;
+import com.xiaomi.youpin.docean.Ioc;
+import com.xiaomi.youpin.docean.anno.Component;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.annotation.Resource;
+import java.nio.charset.StandardCharsets;
+
+import static org.apache.ozhera.log.common.Constant.GSON;
+
+/**
+ * @author wtt
+ * @version 1.0
+ * @description The receiver that communicates with the agent ---- the acquisition progress
+ * @date 2021/8/19 15:32
+ */
+@Slf4j
+@Component
+public class AgentCollectProgressProcessor implements NettyRequestProcessor {
+
+    @Resource
+    DefaultLogProcessCollector processService;
+
+    private static Version version = new Version();
+
+    @Override
+    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
+        log.debug("received a message from the agent");
+        RemotingCommand response = RemotingCommand.createResponseCommand(Constant.RPCCMD_AGENT_CODE);
+        String body = new String(request.getBody(), StandardCharsets.UTF_8);
+        UpdateLogProcessCmd cmd = GSON.fromJson(body, UpdateLogProcessCmd.class);
+        log.debug("a request from the client sent by the agent:{}", cmd.getIp());
+        if (null == processService && Ioc.ins().containsBean(DefaultLogProcessCollector.class.getCanonicalName())) {
+            processService = Ioc.ins().getBean(DefaultLogProcessCollector.class);
+        }
+        if (null != processService) {
+            processService.collectLogProcess(cmd);
+        }
+        response.setBody(version.toString().getBytes());
+        response.setBody(Constant.SUCCESS_MESSAGE.getBytes());
+        return response;
+    }
+
+    @Override
+    public boolean rejectRequest() {
+        return false;
+    }
+}

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/porcessor/AgentConfigProcessor.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server.porcessor;
+
+import com.xiaomi.data.push.rpc.netty.NettyRequestProcessor;
+import com.xiaomi.data.push.rpc.protocol.RemotingCommand;
+import org.apache.ozhera.log.api.model.meta.LogCollectMeta;
+import org.apache.ozhera.log.api.service.AgentConfigService;
+import org.apache.ozhera.log.common.Constant;
+import org.apache.ozhera.log.server.service.DefaultAgentConfigAcquirer;
+import com.xiaomi.youpin.docean.Ioc;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.extern.slf4j.Slf4j;
+
+import static org.apache.ozhera.log.common.Constant.GSON;
+
+/**
+ * @author wtt
+ * @version 1.0
+ * @description The receiver that communicates with the agent ---- the agent starts to get the configuration
+ * @date 2021/8/19 15:32
+ */
+@Slf4j
+public class AgentConfigProcessor implements NettyRequestProcessor {
+
+    @Override
+    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
+        RemotingCommand response = RemotingCommand.createResponseCommand(Constant.RPCCMD_AGENT_CONFIG_CODE);
+        String ip = new String(request.getBody());
+        log.info("agent start get metadata config，agent ip:{}", ip);
+
+        AgentConfigService agentConfigService = Ioc.ins().getBean(DefaultAgentConfigAcquirer.class);
+
+        LogCollectMeta logCollectMeta = agentConfigService.getLogCollectMetaFromManager(ip);
+        String responseInfo = GSON.toJson(logCollectMeta);
+        log.info("agent start get metadata config info:{}", responseInfo);
+        response.setBody(responseInfo.getBytes());
+        return response;
+    }
+
+    @Override
+    public boolean rejectRequest() {
+        return false;
+    }
+}

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/porcessor/PingProcessor.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server.porcessor;
+
+import com.xiaomi.data.push.context.AgentContext;
+import com.xiaomi.data.push.rpc.RpcCmd;
+import com.xiaomi.data.push.rpc.common.RemotingHelper;
+import com.xiaomi.data.push.rpc.netty.AgentChannel;
+import com.xiaomi.data.push.rpc.netty.NettyRequestProcessor;
+import com.xiaomi.data.push.rpc.protocol.RemotingCommand;
+import org.apache.ozhera.log.api.model.meta.AppLogMeta;
+import org.apache.ozhera.log.api.model.vo.PingReq;
+import org.apache.ozhera.log.server.common.Version;
+import org.apache.ozhera.log.utils.NetUtil;
+import io.netty.channel.ChannelHandlerContext;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.ozhera.log.common.Constant.GSON;
+import static org.apache.ozhera.log.server.common.ServerConstant.SERVER_PORT;
+
+/**
+ * @Author goodjava@qq.com
+ * @Date 2021/6/24 11:38
+ */
+@Slf4j
+public class PingProcessor implements NettyRequestProcessor {
+
+    public static Map<String, Long> agentHeartTimeStampMap = new ConcurrentHashMap<>(1024);
+
+    private static Version version = new Version();
+
+    private DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss SSS");
+
+    @Override
+    public RemotingCommand processRequest(ChannelHandlerContext channelHandlerContext, RemotingCommand remotingCommand) {
+        final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(channelHandlerContext.channel());
+        RemotingCommand response = RemotingCommand.createResponseCommand(RpcCmd.pingRes);
+        String body = new String(remotingCommand.getBody());
+        PingReq pr = GSON.fromJson(body, PingReq.class);
+
+        AgentChannel ch = AgentContext.ins().map.get(remoteAddress);
+        if (null != ch) {
+            ch.setIp(pr.getIp());
+        }
+        String requestBody = String.format("%s->%s->%s:%s->%s", version.toString(), dateTimeFormatter.format(LocalDateTime.now()), NetUtil.getLocalIp(), SERVER_PORT, remoteAddress);
+        response.setBody(requestBody.getBytes());
+        if (null != pr && StringUtils.isNotBlank(pr.getIp())) {
+            agentHeartTimeStampMap.put(pr.getIp(), Instant.now().toEpochMilli());
+        }
+
+        if (pr.getMessage().equals("load")) {
+            AppLogMeta meta = new AppLogMeta();
+            meta.setAppName("log-manager");
+            meta.setAppId(ThreadLocalRandom.current().nextLong());
+            response.setBody(GSON.toJson(meta).getBytes());
+        }
+
+        return response;
+    }
+
+    @Override
+    public boolean rejectRequest() {
+        return false;
+    }
+}

ozhera-log/log-agent-server/src/main/java/org/apache/ozhera/log/server/service/ApplicationShutdownHook.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (C) 2020 Xiaomi Corporation
+ *
+ *  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 org.apache.ozhera.log.server.service;
+
+import com.xiaomi.data.push.rpc.RpcServer;
+import com.xiaomi.youpin.docean.anno.Component;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.annotation.Resource;
+
+/**
+ * @author wtt
+ * @version 1.0
+ * @description
+ * @date 2023/10/13 14:11
+ */
+@Component
+@Slf4j
+public class ApplicationShutdownHook {
+
+    @Resource
+    private RpcServer rpcServer;
+
+    public void init() {
+        addRuntimeShutdownHook();
+    }
+
+    /**
+     * addRuntimeShutdownHook server deregisterInstance
+     */
+    private void addRuntimeShutdownHook() {
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> rpcServer.shutdown()));
+    }
+
+}