Skip to content

Commit 1fda68f

Browse files
renvinsrenvins
committed
Add LineProtocolPoint class for Point representation
1 parent 7cdf29c commit 1fda68f

File tree

4 files changed

+105
-32
lines changed

4 files changed

+105
-32
lines changed

‎api/src/main/java/it/renvins/serverpulse/api/data/LineProtocolPoint.java

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
package it.renvins.serverpulse.api.data;
2+
3+
import java.util.HashMap;
4+
import java.util.Map;
5+
6+
/**
7+
* Represents a single point in the InfluxDB line protocol format.
8+
* This class is used to construct a line protocol point with measurement, tags, fields, and timestamp.
9+
*/
10+
public class LineProtocolPoint {
11+
12+
private final String measurement;
13+
private final Map<String, String> tags;
14+
private final Map<String, Object> fields;
15+
private long timestamp;
16+
17+
public LineProtocolPoint(String measurement) {
18+
this.measurement = measurement;
19+
this.tags = new HashMap<>();
20+
this.fields = new HashMap<>();
21+
}
22+
23+
public LineProtocolPoint addTag(String key, String value) {
24+
tags.put(key, value);
25+
return this;
26+
}
27+
28+
public LineProtocolPoint addField(String key, Object value) {
29+
fields.put(key, value);
30+
return this;
31+
}
32+
33+
public LineProtocolPoint setTimestamp(long timestamp) {
34+
this.timestamp = timestamp;
35+
return this;
36+
}
37+
38+
public String toLineProtocol() {
39+
StringBuilder sb = new StringBuilder();
40+
sb.append(measurement);
41+
42+
if (!tags.isEmpty()) {
43+
sb.append(",");
44+
tags.forEach((key, value) -> sb.append(key).append("=").append(value).append(","));
45+
sb.setLength(sb.length() - 1); // Remove the last comma
46+
}
47+
48+
sb.append(" ");
49+
50+
if (!fields.isEmpty()) {
51+
fields.forEach((key, value) -> {
52+
sb.append(key).append("=");
53+
if (value instanceof String) {
54+
sb.append("\"").append(value).append("\"");
55+
} else if (value instanceof Integer || value instanceof Long) {
56+
sb.append(value).append("i");
57+
} else {
58+
sb.append(value);
59+
}
60+
sb.append(",");
61+
});
62+
sb.setLength(sb.length() - 1); // Remove the last comma
63+
}
64+
65+
sb.append(" ").append(timestamp);
66+
67+
return sb.toString();
68+
}
69+
}

‎api/src/main/java/it/renvins/serverpulse/api/service/IDatabaseService.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,11 @@ public interface IDatabaseService extends Service {
1818
*/
1919
boolean ping();
2020

21+
/**
22+
* Sets the isConnected flag to false.
23+
*/
24+
void disconnect();
25+
2126
/**
2227
* Returns the last known connection status. Does not perform a live check.
2328
* @return true if the service believes it's connected, false otherwise.

‎common/src/main/java/it/renvins/serverpulse/common/DatabaseService.java

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,11 @@ public boolean ping() {
138138
}
139139
}
140140

141+
@Override
142+
public void disconnect() {
143+
this.isConnected = false;
144+
}
145+
141146
@Override
142147
public boolean isConnected() {
143148
// Return the flag, don't ping here!
@@ -223,10 +228,6 @@ private void connect() {
223228
}
224229
}
225230

226-
private void disconnect() {
227-
this.isConnected = false;
228-
}
229-
230231
/** Stops and nullifies the retry task if it's running. */
231232
private synchronized void stopRetryTask() {
232233
if (retryTask != null) {

‎common/src/main/java/it/renvins/serverpulse/common/MetricsService.java

Lines changed: 26 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,14 @@
11
package it.renvins.serverpulse.common;
22

33
import java.time.Instant;
4+
import java.time.temporal.TemporalField;
45
import java.util.ArrayList;
56
import java.util.List;
67
import java.util.Map;
78
import java.util.concurrent.CompletableFuture;
89
import java.util.concurrent.Executor;
9-
import java.util.logging.Level;
10-
import java.util.logging.Logger;
11-
12-
import com.influxdb.client.domain.WritePrecision;
13-
import com.influxdb.client.write.Point;
1410
import it.renvins.serverpulse.api.ServerPulseProvider;
11+
import it.renvins.serverpulse.api.data.LineProtocolPoint;
1512
import it.renvins.serverpulse.api.utils.MemoryUtils;
1613
import it.renvins.serverpulse.api.data.SyncMetricsSnapshot;
1714
import it.renvins.serverpulse.api.data.WorldData;
@@ -49,7 +46,7 @@ public void load() {
4946

5047
@Override
5148
public void collectAndSendMetrics() {
52-
if (!ServerPulseProvider.get().getDatabaseService().isConnected() || ServerPulseProvider.get().getDatabaseService().getWriteApi() == null) {
49+
if (!ServerPulseProvider.get().getDatabaseService().isConnected()) {
5350
return;
5451
}
5552
if (!ServerPulseProvider.get().getDatabaseService().ping()) {
@@ -77,7 +74,8 @@ public void collectAndSendMetrics() {
7774
}, asyncExecutor).thenAcceptAsync(points -> {
7875
if (!points.isEmpty()) {
7976
try {
80-
ServerPulseProvider.get().getDatabaseService().getWriteApi().writePoints(points);
77+
String body = String.join("\n", points);
78+
ServerPulseProvider.get().getDatabaseService().writeLineProtocol(body);
8179
} catch (Exception e) {
8280
logger.error("Error sending metrics to InfluxDB...", e);
8381
}
@@ -131,42 +129,42 @@ private SyncMetricsSnapshot collectSnapshot() {
131129
* @param usableDisk The usable disk space in bytes.
132130
* @return A list of InfluxDB points representing the metrics.
133131
*/
134-
private List<Point> buildPoints(SyncMetricsSnapshot snapshot, long usedHeap, long committedHeap,
132+
private List<String> buildPoints(SyncMetricsSnapshot snapshot, long usedHeap, long committedHeap,
135133
long totalDisk, long usableDisk, int minPing, int maxPing, int avgPing) {
136-
List<Point> points = new ArrayList<>();
134+
List<String> points = new ArrayList<>();
137135

138136
String serverTag = configuration.getServerTag();
139137
String measurement = configuration.getMeasurementTable();
140138

141-
Point generalPoint = Point.measurement(measurement)
142-
.addTag("server", serverTag)
143-
.addField("tps_1m", snapshot.getTps()[0])
144-
.addField("tps_5m", snapshot.getTps()[1])
145-
.addField("tps_15m", snapshot.getTps()[2])
146-
.addField("players_online", snapshot.getPlayerCount())
147-
.addField("used_memory", usedHeap)
148-
.addField("available_memory", committedHeap)
149-
.addField("total_disk_space", totalDisk)
150-
.addField("usable_disk_space", usableDisk)
151-
.addField("min_ping", minPing)
152-
.addField("max_ping", maxPing)
153-
.addField("avg_ping", avgPing)
154-
.time(Instant.now(), WritePrecision.NS);
139+
LineProtocolPoint generalPoint = new LineProtocolPoint(measurement)
140+
.addTag("server", serverTag)
141+
.addField("tps_1m", snapshot.getTps()[0])
142+
.addField("tps_5m", snapshot.getTps()[1])
143+
.addField("tps_15m", snapshot.getTps()[2])
144+
.addField("players_online", snapshot.getPlayerCount())
145+
.addField("used_memory", usedHeap)
146+
.addField("available_memory", committedHeap)
147+
.addField("total_disk_space", totalDisk)
148+
.addField("usable_disk_space", usableDisk)
149+
.addField("min_ping", minPing)
150+
.addField("max_ping", maxPing)
151+
.addField("avg_ping", avgPing)
152+
.setTimestamp(Instant.now().getNano());
155153
addConfigTags(generalPoint);
156-
points.add(generalPoint);
154+
points.add(generalPoint.toLineProtocol());
157155

158156
for (Map.Entry<String, WorldData> entry : snapshot.getWorldData().entrySet()) {
159157
String worldName = entry.getKey();
160158
WorldData worldData = entry.getValue();
161159

162-
Point worldPoint = Point.measurement(measurement)
160+
LineProtocolPoint worldPoint = new LineProtocolPoint(measurement)
163161
.addTag("server", serverTag)
164162
.addTag("world", worldName)
165163
.addField("entities_count", worldData.getEntities())
166164
.addField("loaded_chunks", worldData.getLoadedChunks())
167-
.time(Instant.now(), WritePrecision.NS);
165+
.setTimestamp(Instant.now().getNano());
168166
addConfigTags(worldPoint);
169-
points.add(worldPoint);
167+
points.add(worldPoint.toLineProtocol());
170168
}
171169
return points;
172170
}
@@ -176,7 +174,7 @@ private List<Point> buildPoints(SyncMetricsSnapshot snapshot, long usedHeap, lon
176174
*
177175
* @param point The InfluxDB point to which tags will be added.
178176
*/
179-
private void addConfigTags(Point point) {
177+
private void addConfigTags(LineProtocolPoint point) {
180178
Map<String, String> tags = configuration.getTags();
181179
tags.forEach(point::addTag);
182180
}

0 commit comments

Comments
 (0)