Skip to content

Commit 711c711

Browse files
intr3p1dintr3p1d
committed
[#10882] Add server map module based on redis-timeseries
1 parent c8a7016 commit 711c711

27 files changed

+1267
-43
lines changed

‎collector/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -222,6 +222,12 @@
222222
<artifactId>jakarta.annotation-api</artifactId>
223223
</dependency>
224224

225+
226+
<dependency>
227+
<groupId>com.navercorp.pinpoint</groupId>
228+
<artifactId>pinpoint-redis-timeseries</artifactId>
229+
<version>3.1.0-SNAPSHOT</version>
230+
</dependency>
225231
</dependencies>
226232

227233
<build>

‎collector/src/main/java/com/navercorp/pinpoint/collector/PinpointCollectorModule.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package com.navercorp.pinpoint.collector;
22

33

4+
import com.navercorp.pinpoint.collector.applicationmap.ApplicationMapModule;
45
import com.navercorp.pinpoint.collector.config.ClusterModule;
56
import com.navercorp.pinpoint.collector.config.CollectorCommonConfiguration;
67
import com.navercorp.pinpoint.collector.config.CollectorConfiguration;
@@ -35,6 +36,8 @@
3536
GrpcSslModule.class,
3637

3738
RealtimeCollectorModule.class,
39+
40+
ApplicationMapModule.class,
3841
})
3942
@ComponentScan(basePackages = {
4043
"com.navercorp.pinpoint.collector.handler",

‎collector/src/main/java/com/navercorp/pinpoint/collector/applicationmap/ApplicationMapModule.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Copyright 2024 NAVER Corp.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.navercorp.pinpoint.collector.applicationmap;
17+
18+
import org.springframework.context.annotation.ComponentScan;
19+
import org.springframework.context.annotation.Configuration;
20+
21+
/**
22+
* @author intr3p1d
23+
*/
24+
@Configuration(proxyBeanMethods = false)
25+
@ComponentScan(basePackages = {
26+
"com.navercorp.pinpoint.collector.applicationmap",
27+
"com.navercorp.pinpoint.collector.applicationmap.dao",
28+
"com.navercorp.pinpoint.collector.applicationmap.redis",
29+
"com.navercorp.pinpoint.collector.applicationmap.service",
30+
})
31+
public class ApplicationMapModule {
32+
}

‎collector/src/main/java/com/navercorp/pinpoint/collector/applicationmap/dao/InboundDao.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Copyright 2024 NAVER Corp.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.navercorp.pinpoint.collector.applicationmap.dao;
17+
18+
import com.navercorp.pinpoint.collector.dao.CachedStatisticsDao;
19+
import com.navercorp.pinpoint.common.trace.ServiceType;
20+
21+
/**
22+
* @author intr3p1d
23+
*/
24+
public interface InboundDao extends CachedStatisticsDao {
25+
// src -> dest
26+
// inbound (rowKey dest <- columnName src)
27+
// outbound (rowKey src -> columnName dest)
28+
void update(
29+
String srcServiceName, String srcApplicationName, ServiceType srcApplicationType,
30+
String destServiceName, String destApplicationName, ServiceType destApplicationType,
31+
String srcHost, int elapsed, boolean isError
32+
);
33+
}

‎collector/src/main/java/com/navercorp/pinpoint/collector/applicationmap/dao/OutboundDao.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Copyright 2024 NAVER Corp.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.navercorp.pinpoint.collector.applicationmap.dao;
17+
18+
import com.navercorp.pinpoint.collector.dao.CachedStatisticsDao;
19+
import com.navercorp.pinpoint.common.trace.ServiceType;
20+
21+
/**
22+
* @author intr3p1d
23+
*/
24+
public interface OutboundDao extends CachedStatisticsDao {
25+
// src -> dest
26+
// inbound (rowKey dest <- columnName src)
27+
// outbound (rowKey src -> columnName dest)
28+
void update(
29+
String srcServiceName, String srcApplicationName, ServiceType srcApplicationType,
30+
String destServiceName, String destApplicationName, ServiceType destApplicationType,
31+
String srcHost, int elapsed, boolean isError
32+
);
33+
}

‎collector/src/main/java/com/navercorp/pinpoint/collector/applicationmap/dao/SelfDao.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
/*
2+
* Copyright 2024 NAVER Corp.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.navercorp.pinpoint.collector.applicationmap.dao;
17+
18+
19+
import com.navercorp.pinpoint.collector.dao.CachedStatisticsDao;
20+
import com.navercorp.pinpoint.common.trace.ServiceType;
21+
22+
/**
23+
* @author intr3p1d
24+
*/
25+
public interface SelfDao extends CachedStatisticsDao {
26+
void received(String serviceName, String applicationName, ServiceType applicationType, int elapsed, boolean isError);
27+
void updatePing(String serviceName, String applicationName, ServiceType applicationType, int elapsed, boolean isError);
28+
}

‎collector/src/main/java/com/navercorp/pinpoint/collector/applicationmap/redis/RedisInboundDao.java‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
/*
2+
* Copyright 2024 NAVER Corp.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.navercorp.pinpoint.collector.applicationmap.redis;
17+
18+
import com.navercorp.pinpoint.collector.applicationmap.dao.InboundDao;
19+
import com.navercorp.pinpoint.collector.applicationmap.redis.schema.ApplicationMapTable;
20+
import com.navercorp.pinpoint.collector.applicationmap.redis.schema.TimeSeriesKey;
21+
import com.navercorp.pinpoint.collector.applicationmap.redis.schema.TimeSeriesValue;
22+
import com.navercorp.pinpoint.collector.dao.hbase.IgnoreStatFilter;
23+
import com.navercorp.pinpoint.collector.dao.hbase.statistics.MapLinkConfiguration;
24+
import com.navercorp.pinpoint.collector.applicationmap.redis.statistics.RedisBulkWriter;
25+
import com.navercorp.pinpoint.common.server.util.AcceptedTimeService;
26+
import com.navercorp.pinpoint.common.server.util.ApplicationMapStatisticsUtils;
27+
import com.navercorp.pinpoint.common.server.util.TimeSlot;
28+
import com.navercorp.pinpoint.common.trace.HistogramSchema;
29+
import com.navercorp.pinpoint.common.trace.ServiceType;
30+
import org.apache.logging.log4j.LogManager;
31+
import org.apache.logging.log4j.Logger;
32+
import org.springframework.beans.factory.annotation.Qualifier;
33+
import org.springframework.stereotype.Repository;
34+
35+
import java.util.Objects;
36+
37+
/**
38+
* @author intr3p1d
39+
*/
40+
@Repository
41+
public class RedisInboundDao implements InboundDao {
42+
43+
private final Logger logger = LogManager.getLogger(this.getClass());
44+
45+
private final AcceptedTimeService acceptedTimeService;
46+
private final IgnoreStatFilter ignoreStatFilter;
47+
private final RedisBulkWriter bulkWriter;
48+
private final MapLinkConfiguration mapLinkConfiguration;
49+
50+
public RedisInboundDao(
51+
MapLinkConfiguration mapLinkConfiguration,
52+
AcceptedTimeService acceptedTimeService,
53+
IgnoreStatFilter ignoreStatFilter,
54+
@Qualifier("inboundBulkWriter") RedisBulkWriter bulkWriter
55+
) {
56+
this.mapLinkConfiguration = Objects.requireNonNull(mapLinkConfiguration, "mapLinkConfiguration");
57+
this.acceptedTimeService = Objects.requireNonNull(acceptedTimeService, "acceptedTimeService");
58+
this.ignoreStatFilter = Objects.requireNonNull(ignoreStatFilter, "ignoreStatFilter");
59+
this.bulkWriter = Objects.requireNonNull(bulkWriter, "inboundBulkWriter");
60+
}
61+
62+
63+
@Override
64+
public void update(
65+
String srcServiceName, String srcApplicationName, ServiceType srcApplicationType,
66+
String destServiceName, String destApplicationName, ServiceType destApplicationType,
67+
String srcHost, int elapsed, boolean isError
68+
) {
69+
Objects.requireNonNull(srcServiceName, "srcServiceName");
70+
Objects.requireNonNull(destServiceName, "destServiceName");
71+
Objects.requireNonNull(srcApplicationName, "srcApplicationName");
72+
Objects.requireNonNull(destServiceName, "destApplicationName");
73+
74+
if (logger.isDebugEnabled()) {
75+
logger.debug("[Inbound] {} {}({}) <- {} {}({})[{}]",
76+
destServiceName, destApplicationName, destApplicationType,
77+
srcServiceName, srcApplicationName, srcApplicationType, srcHost
78+
);
79+
}
80+
81+
if (ignoreStatFilter.filter(srcApplicationType, srcHost)) {
82+
logger.debug("[Ignore-Inbound] {} {}({}) <- {} {}({})[{}]",
83+
destServiceName, destApplicationName, destApplicationType,
84+
srcServiceName, srcApplicationName, srcApplicationType, srcHost
85+
);
86+
return;
87+
}
88+
89+
final short srcSlotNumber = ApplicationMapStatisticsUtils.getSlotNumber(srcApplicationType, elapsed, isError);
90+
HistogramSchema histogramSchema = srcApplicationType.getHistogramSchema();
91+
final long acceptedTime = acceptedTimeService.getAcceptedTime();
92+
93+
// for inbound, main is destination
94+
// and sub is source
95+
final TimeSeriesKey applicationTypeKey = new TimeSeriesKey(
96+
ApplicationMapTable.Inbound, "tenantId",
97+
destServiceName, destApplicationName,
98+
srcServiceName, srcApplicationName, srcSlotNumber
99+
);
100+
TimeSeriesValue addOne = new TimeSeriesValue(acceptedTime);
101+
this.bulkWriter.increment(applicationTypeKey, addOne);
102+
103+
if (mapLinkConfiguration.isEnableAvg()) {
104+
final TimeSeriesKey sumStatKey = new TimeSeriesKey(
105+
ApplicationMapTable.Inbound, "tenantId",
106+
destServiceName, destApplicationName,
107+
srcServiceName, srcApplicationName,
108+
histogramSchema.getSumStatSlot().getSlotTime()
109+
);
110+
final TimeSeriesValue sumValue = new TimeSeriesValue(acceptedTime);
111+
this.bulkWriter.increment(sumStatKey, sumValue, elapsed);
112+
}
113+
if (mapLinkConfiguration.isEnableMax()) {
114+
final TimeSeriesKey maxStatKey = new TimeSeriesKey(
115+
ApplicationMapTable.Inbound, "tenantId",
116+
destServiceName, destApplicationName,
117+
srcServiceName, srcApplicationName,
118+
histogramSchema.getMaxStatSlot().getSlotTime()
119+
);
120+
final TimeSeriesValue maxValue = new TimeSeriesValue(acceptedTime);
121+
this.bulkWriter.updateMax(maxStatKey, maxValue, elapsed);
122+
}
123+
124+
}
125+
126+
@Override
127+
public void flushLink() {
128+
this.bulkWriter.flushLink();
129+
}
130+
131+
@Override
132+
public void flushAvgMax() {
133+
this.bulkWriter.flushAvgMax();
134+
}
135+
136+
}

0 commit comments

Comments
 (0)