[#10704] set the format of the agent inspector table name.

batch/src/main/java/com/navercorp/pinpoint/batch/alarm/dao/pinot/PinotAlarmDao.java

@@ -22,7 +22,7 @@
 import com.navercorp.pinpoint.batch.alarm.vo.AgentUsage;
 import com.navercorp.pinpoint.batch.alarm.vo.AgentUsageCount;
 import com.navercorp.pinpoint.batch.common.BatchProperties;
-import com.navercorp.pinpoint.common.dao.pinot.AgentStatTopicAndTableNameManager;
+import com.navercorp.pinpoint.common.dao.pinot.AgentStatTableNameManager;
 import com.navercorp.pinpoint.common.model.TagInformation;
 import com.navercorp.pinpoint.common.server.util.time.Range;
 import com.navercorp.pinpoint.metric.common.model.Tag;
@@ -46,12 +46,14 @@ public class PinotAlarmDao implements AlarmDao {
     private final PinotAsyncTemplate asyncTemplate;
     private final SqlSessionTemplate syncTemplate;
     private final int tableCount;
+    private final AgentStatTableNameManager agentStatTableNameManager;
 
     public PinotAlarmDao(@Qualifier("batchPinotAsyncTemplate") PinotAsyncTemplate asyncTemplate, @Qualifier("batchPinotTemplate") SqlSessionTemplate syncTemplate, BatchProperties batchProperties) {
         this.asyncTemplate = Objects.requireNonNull(asyncTemplate, "asyncTemplate");
         this.syncTemplate = Objects.requireNonNull(syncTemplate, "syncTemplate");
         Objects.requireNonNull(batchProperties, "batchProperties");
         this.tableCount = batchProperties.getAlarmAgentInspectorStatTableCount();
+        this.agentStatTableNameManager = new AgentStatTableNameManager(batchProperties.getAgentInspectorStatTablePrefix(), batchProperties.getAgentInspectorStatTablePaddingLength());
     }
 
     @Override
@@ -91,6 +93,6 @@ public CompletableFuture<List<TagInformation>> getTagInfoContainedSpecificTag(St
     }
 
     private String getTableName(String applicationName) {
-        return AgentStatTopicAndTableNameManager.getAgentStatTableName(applicationName, tableCount);
+        return agentStatTableNameManager.getAgentStatTableName(applicationName, tableCount);
     }
 }

batch/src/main/java/com/navercorp/pinpoint/batch/common/BatchProperties.java

@@ -52,6 +52,12 @@ public class BatchProperties {
     @Value("${job.alarm.agent.inspector.stat.table.count}")
     private int alarmAgentInspectorStatTableCount;
 
+    @Value("${job.alarm.agent.inspector.stat.table.prefix}")
+    private String agentInspectorStatTablePrefix;
+
+    @Value("${job.alarm.agent.inspector.stat.table.padding.length}")
+    private int agentInspectorStatTablePaddingLength;
+
     @Value("${job.agent.count.enable:true}")
     private boolean agentCountJobEnable;
 
@@ -184,6 +190,14 @@ public int getAlarmAgentInspectorStatTableCount() {
         return alarmAgentInspectorStatTableCount;
     }
 
+    public String getAgentInspectorStatTablePrefix() {
+        return agentInspectorStatTablePrefix;
+    }
+
+    public int getAgentInspectorStatTablePaddingLength() {
+        return agentInspectorStatTablePaddingLength;
+    }
+
     @Override
     public String toString() {
         return "BatchProperties{" +
@@ -193,6 +207,8 @@ public String toString() {
                 ", alarmJobEnable=" + alarmJobEnable +
                 ", alarmJobCron='" + alarmJobCron + '\'' +
                 ", alarmAgentInspectorStatTableCount=" + alarmAgentInspectorStatTableCount +
+                ", agentInspectorStatTablePrefix='" + agentInspectorStatTablePrefix + '\'' +
+                ", agentInspectorStatTablePaddingLength=" + agentInspectorStatTablePaddingLength +
                 ", agentCountJobEnable=" + agentCountJobEnable +
                 ", agentCountJobCron='" + agentCountJobCron + '\'' +
                 ", flinkCheckJobEnable=" + flinkCheckJobEnable +

batch/src/main/resources/batch-root.properties

@@ -30,6 +30,8 @@ batch.flink.rest.port=8081
 
 job.alarm.cron=0 0/3 * * * *
 job.alarm.agent.inspector.stat.table.count=1
+job.alarm.agent.inspector.stat.table.prefix=inspectorStatAgent
+job.alarm.agent.inspector.stat.table.padding.length=2
 
 job.agent.count.cron=0 0 2 * * *
 

inspector-module/inspector-collector/src/main/java/com/navercorp/pinpoint/inspector/collector/config/InspectorCollectorProperties.java

@@ -27,6 +27,10 @@ public class InspectorCollectorProperties {
 
     @Value("${kafka.inspector.agent.topic.count}")
     private int agentStatTopicCount;
+    @Value("${kafka.inspector.agent.topic.prefix}")
+    private String agentStatTopicPrefix;
+    @Value("${kafka.inspector.agent.topic.padding.length}")
+    private int agentStatTopicPaddingLength;
     @Value("${kafka.inspector.application.topic.name}")
     private String applicationStatTopicName;
 
@@ -37,4 +41,12 @@ public int getAgentStatTopicCount() {
     public String getApplicationStatTopicName() {
         return applicationStatTopicName;
     }
+
+    public String getAgentStatTopicPrefix() {
+        return agentStatTopicPrefix;
+    }
+
+    public int getAgentStatTopicPaddingLength() {
+        return agentStatTopicPaddingLength;
+    }
 }

inspector-module/inspector-collector/src/main/java/com/navercorp/pinpoint/inspector/collector/dao/pinot/DefaultAgentStatDao.java

@@ -16,10 +16,10 @@
 
 package com.navercorp.pinpoint.inspector.collector.dao.pinot;
 
-import com.navercorp.pinpoint.common.dao.pinot.AgentStatTopicAndTableNameManager;
+import com.navercorp.pinpoint.common.dao.pinot.AgentStatTopicNameManager;
 import com.navercorp.pinpoint.common.server.bo.stat.AgentStatBo;
 import com.navercorp.pinpoint.common.server.bo.stat.AgentStatDataPoint;
-import com.navercorp.pinpoint.common.server.util.StringPrecondition;
+import com.navercorp.pinpoint.inspector.collector.config.InspectorCollectorProperties;
 import com.navercorp.pinpoint.inspector.collector.dao.AgentStatDao;
 import com.navercorp.pinpoint.inspector.collector.model.kafka.AgentStat;
 import com.navercorp.pinpoint.inspector.collector.model.kafka.ApplicationStat;
@@ -51,23 +51,25 @@ public class DefaultAgentStatDao <T extends AgentStatDataPoint> implements Agent
     private final int agentStatTopicCount;
     private final String applicationStatTopicName;
     private final TenantProvider tenantProvider;
+    private final AgentStatTopicNameManager agentStatTopicNameManager;
 
     public DefaultAgentStatDao(Function<AgentStatBo,
                                List<T>> dataPointFunction,
                                KafkaTemplate<String, AgentStat> kafkaAgentStatTemplate,
                                KafkaTemplate<String, ApplicationStat> kafkaApplicationStatTemplate,
                                BiFunction<List<T>, String, List<AgentStat>> convertToKafkaAgentStatModelFunction,
                                Function<List<AgentStat>, List<ApplicationStat>> convertToKafkaApplicationStatModelFunction,
-                               int agentStatTopicCount,
-                               String applicationStatTopicName,
+                               InspectorCollectorProperties inspectorCollectorProperties,
                                TenantProvider tenantProvider) {
         this.dataPointFunction = Objects.requireNonNull(dataPointFunction, "dataPointFunction");
         this.kafkaAgentStatTemplate = Objects.requireNonNull(kafkaAgentStatTemplate, "kafkaAgentStatTemplate");
         this.kafkaApplicationStatTemplate = Objects.requireNonNull(kafkaApplicationStatTemplate, "kafkaApplicationStatTemplate");
         this.convertToKafkaAgentStatModelFunction = Objects.requireNonNull(convertToKafkaAgentStatModelFunction, "convertToKafkaAgentStatModelFunction");
         this.convertToKafkaApplicationStatModelFunction = Objects.requireNonNull(convertToKafkaApplicationStatModelFunction, "convertToKafkaApplicationStatModelFunction");
-        this.agentStatTopicCount = agentStatTopicCount;
-        this.applicationStatTopicName = StringPrecondition.requireHasLength(applicationStatTopicName, "applicationStatTopic");
+        Objects.requireNonNull(inspectorCollectorProperties, "inspectorCollectorProperties");
+        this.agentStatTopicCount = inspectorCollectorProperties.getAgentStatTopicCount();
+        this.applicationStatTopicName = inspectorCollectorProperties.getApplicationStatTopicName();
+        this.agentStatTopicNameManager = new AgentStatTopicNameManager(inspectorCollectorProperties.getAgentStatTopicPrefix(), inspectorCollectorProperties.getAgentStatTopicPaddingLength());
         this.tenantProvider = Objects.requireNonNull(tenantProvider, "tenantProvider");
     }
 
@@ -78,7 +80,7 @@ public void insert(String applicationName, String agentId, List<T> agentStatData
         };
 
         List<AgentStat> agentStatList = convertToKafkaAgentStatModel(agentStatData);
-        String topicName = AgentStatTopicAndTableNameManager.getAgentStatTopicName(applicationName, agentStatTopicCount);
+        String topicName = agentStatTopicNameManager.getAgentStatTopicName(applicationName, agentStatTopicCount);
 
         for (AgentStat agentStat : agentStatList) {
             kafkaAgentStatTemplate.send(topicName, agentStat.getSortKey(), agentStat);

inspector-module/inspector-collector/src/main/java/com/navercorp/pinpoint/inspector/collector/dao/pinot/PinotDaoConfiguration.java

@@ -42,21 +42,18 @@ public class PinotDaoConfiguration {
 
     private final KafkaTemplate<String, AgentStat> kafkaAgentStatTemplate;
     private final KafkaTemplate<String, ApplicationStat> kafkaApplicationStatTemplate;
-    private final int agentStatTopicCount;
-    private final String applicationStatTopicName;
+    InspectorCollectorProperties inspectorCollectorProperties;
     private final TenantProvider tenantProvider;
 
     public PinotDaoConfiguration(KafkaTemplate<String, AgentStat> kafkaAgentStatTemplate, KafkaTemplate<String, ApplicationStat> kafkaApplicationStatTemplate, InspectorCollectorProperties inspectorCollectorProperties, TenantProvider tenantProvider) {
         this.kafkaAgentStatTemplate = Objects.requireNonNull(kafkaAgentStatTemplate, "kafkaAgentStatTemplate");
         this.kafkaApplicationStatTemplate = Objects.requireNonNull(kafkaApplicationStatTemplate, "kafkaApplicationStatTemplate");
-        Objects.requireNonNull(inspectorCollectorProperties, "inspectorCollectorProperties");
-        this.agentStatTopicCount = inspectorCollectorProperties.getAgentStatTopicCount();
-        this.applicationStatTopicName = inspectorCollectorProperties.getApplicationStatTopicName();
+        this.inspectorCollectorProperties = Objects.requireNonNull(inspectorCollectorProperties, "inspectorCollectorProperties");
         this.tenantProvider = Objects.requireNonNull(tenantProvider, "tenantProvider");
     }
 
     private <T extends AgentStatDataPoint> AgentStatDao<T> newAgentStatDao(Function<AgentStatBo, List<T>> dataPointFunction, BiFunction<List<T>, String, List<AgentStat>> convertToAgentStat, Function<List<AgentStat>, List<ApplicationStat>> convertToKafkaApplicationStat) {
-        return new DefaultAgentStatDao(dataPointFunction, kafkaAgentStatTemplate, kafkaApplicationStatTemplate, convertToAgentStat, convertToKafkaApplicationStat, agentStatTopicCount, applicationStatTopicName, tenantProvider);
+        return new DefaultAgentStatDao(dataPointFunction, kafkaAgentStatTemplate, kafkaApplicationStatTemplate, convertToAgentStat, convertToKafkaApplicationStat, inspectorCollectorProperties, tenantProvider);
     }
 
     @Bean

inspector-module/inspector-collector/src/main/resources/inspector/collector/profiles/local/inspector-collector.properties

@@ -1 +1,3 @@
-kafka.inspector.agent.topic.count=1
+kafka.inspector.agent.topic.count=1
+kafka.inspector.agent.topic.prefix=inspector-stat-agent-
+kafka.inspector.agent.topic.padding.length=2

inspector-module/inspector-collector/src/main/resources/inspector/collector/profiles/release/inspector-collector.properties

@@ -1 +1,3 @@
-kafka.inspector.agent.topic.count=1
+kafka.inspector.agent.topic.count=1
+kafka.inspector.agent.topic.prefix=inspector-stat-agent-
+kafka.inspector.agent.topic.padding.length=2

inspector-module/inspector-commons/src/main/java/com/navercorp/pinpoint/common/dao/pinot/AgentStatNameManager.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2024 NAVER Corp.
+ *
+ * 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.navercorp.pinpoint.common.dao.pinot;
+
+import org.apache.kafka.common.utils.Utils;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * @author minwoo-jung
+ */
+public class AgentStatNameManager {
+    protected int getHashValue(String applicationName, int agentStatTopicCount) {
+        int hash = Utils.murmur2(applicationName.getBytes(StandardCharsets.UTF_8));
+        return Utils.toPositive(hash) % agentStatTopicCount;
+    }
+}

inspector-module/inspector-commons/src/main/java/com/navercorp/pinpoint/common/dao/pinot/AgentStatTableNameManager.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2024 NAVER Corp.
+ *
+ * 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.navercorp.pinpoint.common.dao.pinot;
+
+/**
+ * @author minwoo-jung
+ */
+public class AgentStatTableNameManager extends AgentStatNameManager {
+
+    private final String tablePrefix;
+    private final String numberFormat;
+
+    public AgentStatTableNameManager(String tablePrefix, int paddingLength) {
+        this.tablePrefix = tablePrefix;
+        this.numberFormat = "%0" + String.valueOf(paddingLength) + "d";
+    }
+
+    public String getAgentStatTableName(String applicationName, int agentStatTopicCount) {
+        int hashValue = getHashValue(applicationName, agentStatTopicCount);
+        String postfix = String.format(numberFormat, hashValue);
+        StringBuilder sb = new StringBuilder();
+        sb.append(tablePrefix).append(postfix);
+        return sb.toString();
+    }
+
+}