Skip to content

Commit 4b7e9fa

Browse files
ldecheverz-splitldecheverz-split
authored
Merge pull request #228 from splitio/telemetry
Telemetry
2 parents 99cf461 + c43f1f9 commit 4b7e9fa

File tree

122 files changed

+4162
-1782
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

122 files changed

+4162
-1782
lines changed

‎client/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
<parent>
66
<groupId>io.split.client</groupId>
77
<artifactId>java-client-parent</artifactId>
8-
<version>4.1.7</version>
8+
<version>4.2.0</version>
99
</parent>
1010
<artifactId>java-client</artifactId>
1111
<packaging>jar</packaging>
@@ -121,7 +121,7 @@
121121
<dependency>
122122
<groupId>com.google.guava</groupId>
123123
<artifactId>guava</artifactId>
124-
<version>29.0-jre</version>
124+
<version>30.0-jre</version>
125125
</dependency>
126126
<dependency>
127127
<groupId>org.slf4j</groupId>

‎client/src/main/java/io/split/cache/SegmentCache.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
11
package io.split.cache;
22

3+
import io.split.engine.segments.SegmentImp;
4+
35
import java.util.List;
6+
import java.util.Set;
47

58
/**
69
* Memory for segments
@@ -42,4 +45,16 @@ public interface SegmentCache {
4245
* clear all segments
4346
*/
4447
void clear();
48+
49+
/**
50+
* return every segment
51+
* @return
52+
*/
53+
List<SegmentImp> getAll();
54+
55+
/**
56+
* return key count
57+
* @return
58+
*/
59+
long getKeyCount();
4560
}

‎client/src/main/java/io/split/cache/SegmentCacheInMemoryImpl.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,9 @@
66
import org.slf4j.LoggerFactory;
77

88
import java.util.List;
9+
import java.util.Set;
910
import java.util.concurrent.ConcurrentMap;
11+
import java.util.stream.Collectors;
1012

1113
/**
1214
* InMemoryCache Implementation
@@ -59,4 +61,14 @@ public long getChangeNumber(String segmentName) {
5961
public void clear() {
6062
_segments.clear();
6163
}
64+
65+
@Override
66+
public List<SegmentImp> getAll() {
67+
return _segments.values().stream().collect(Collectors.toList());
68+
}
69+
70+
@Override
71+
public long getKeyCount() {
72+
return _segments.values().stream().mapToLong(SegmentImp::getKeysSize).sum();
73+
}
6274
}

‎client/src/main/java/io/split/client/ApiKeyCounter.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@
66
import org.slf4j.Logger;
77
import org.slf4j.LoggerFactory;
88

9+
import java.util.HashMap;
10+
import java.util.Map;
11+
912
public class ApiKeyCounter {
1013

1114
private static final Logger _log = LoggerFactory.getLogger(ApiKeyCounter.class);
@@ -63,4 +66,18 @@ boolean isApiKeyPresent(String apiKey) {
6366
int getCount(String apiKey) {
6467
return USED_API_KEYS.count(apiKey);
6568
}
69+
70+
public Map<String, Long> getFactoryInstances() {
71+
Map<String, Long> factoryInstances = new HashMap<>();
72+
for (String factory :USED_API_KEYS) {
73+
factoryInstances.putIfAbsent(factory, new Long(getCount(factory)));
74+
}
75+
76+
return factoryInstances;
77+
}
78+
79+
@VisibleForTesting
80+
void clearApiKeys() {
81+
USED_API_KEYS.clear();
82+
}
6683
}

‎client/src/main/java/io/split/client/EventClientImpl.java‎

Lines changed: 30 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,11 @@
44
import io.split.client.dtos.Event;
55
import io.split.client.utils.GenericClientUtil;
66
import io.split.client.utils.Utils;
7+
import io.split.telemetry.domain.enums.EventsDataRecordsEnum;
8+
import io.split.telemetry.domain.enums.HTTPLatenciesEnum;
9+
import io.split.telemetry.domain.enums.LastSynchronizationRecordsEnum;
10+
import io.split.telemetry.storage.TelemetryEvaluationProducer;
11+
import io.split.telemetry.storage.TelemetryRuntimeProducer;
712
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
813
import org.slf4j.Logger;
914
import org.slf4j.LoggerFactory;
@@ -23,6 +28,7 @@
2328
import java.util.concurrent.TimeUnit;
2429

2530
import static java.lang.Thread.MIN_PRIORITY;
31+
import static com.google.common.base.Preconditions.checkNotNull;
2632

2733
/**
2834
* Responsible for sending events added via .track() to Split collection services
@@ -45,34 +51,28 @@ public class EventClientImpl implements EventClient {
4551
private final CloseableHttpClient _httpclient;
4652
private final URI _target;
4753
private final int _waitBeforeShutdown;
54+
private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
4855

4956
ThreadFactory eventClientThreadFactory(final String name) {
50-
return new ThreadFactory() {
51-
@Override
52-
public Thread newThread(final Runnable r) {
53-
return new Thread(new Runnable() {
54-
@Override
55-
public void run() {
56-
Thread.currentThread().setPriority(MIN_PRIORITY);
57-
r.run();
58-
}
59-
}, name);
60-
}
61-
};
57+
return r -> new Thread(() -> {
58+
Thread.currentThread().setPriority(MIN_PRIORITY);
59+
r.run();
60+
}, name);
6261
}
6362

6463

65-
public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown) throws URISyntaxException {
66-
return new EventClientImpl(new LinkedBlockingQueue<WrappedEvent>(),
64+
public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
65+
return new EventClientImpl(new LinkedBlockingQueue<>(maxQueueSize),
6766
httpclient,
6867
Utils.appendPath(eventsRootTarget, "api/events/bulk"),
6968
maxQueueSize,
7069
flushIntervalMillis,
71-
waitBeforeShutdown);
70+
waitBeforeShutdown,
71+
telemetryRuntimeProducer);
7272
}
7373

7474
EventClientImpl(BlockingQueue<WrappedEvent> eventQueue, CloseableHttpClient httpclient, URI target, int maxQueueSize,
75-
long flushIntervalMillis, int waitBeforeShutdown) throws URISyntaxException {
75+
long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
7676

7777
_httpclient = httpclient;
7878

@@ -83,6 +83,7 @@ public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsR
8383

8484
_maxQueueSize = maxQueueSize;
8585
_flushIntervalMillis = flushIntervalMillis;
86+
_telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
8687

8788
_senderExecutor = new ThreadPoolExecutor(
8889
1,
@@ -122,9 +123,16 @@ public boolean track(Event event, int eventSize) {
122123
if (event == null) {
123124
return false;
124125
}
125-
_eventQueue.put(new WrappedEvent(event, eventSize));
126+
if(_eventQueue.offer(new WrappedEvent(event, eventSize))) {
127+
_telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
128+
}
129+
else {
130+
_log.warn("Event dropped.");
131+
_telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
132+
}
126133

127-
} catch (InterruptedException e) {
134+
} catch (ClassCastException | NullPointerException | IllegalArgumentException e) {
135+
_telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
128136
_log.warn("Interruption when adding event withed while adding message %s.", event);
129137
return false;
130138
}
@@ -153,7 +161,7 @@ public void run() {
153161
List<Event> events = new ArrayList<>();
154162
long accumulated = 0;
155163
try {
156-
while (true) {
164+
while (!Thread.currentThread().isInterrupted()) {
157165
WrappedEvent data = _eventQueue.take();
158166
Event event = data.event();
159167
Long size = data.size();
@@ -169,7 +177,7 @@ public void run() {
169177

170178
continue;
171179
}
172-
180+
long initTime = System.currentTimeMillis();
173181
if (events.size() >= _maxQueueSize || accumulated >= MAX_SIZE_BYTES || event == CENTINEL) {
174182

175183
// Send over the network
@@ -183,6 +191,8 @@ public void run() {
183191
// Clear the queue of events for the next batch.
184192
events = new ArrayList<>();
185193
accumulated = 0;
194+
_telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.EVENTS, System.currentTimeMillis()-initTime);
195+
_telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.EVENTS, System.currentTimeMillis());
186196
}
187197
}
188198
} catch (InterruptedException e) {

‎client/src/main/java/io/split/client/HttpSegmentChangeFetcher.java‎

Lines changed: 15 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,14 @@
77
import io.split.engine.common.FetchOptions;
88
import io.split.engine.metrics.Metrics;
99
import io.split.engine.segments.SegmentChangeFetcher;
10+
import io.split.telemetry.domain.enums.HTTPLatenciesEnum;
11+
import io.split.telemetry.domain.enums.LastSynchronizationRecordsEnum;
12+
import io.split.telemetry.domain.enums.ResourceEnum;
13+
import io.split.telemetry.storage.TelemetryRuntimeProducer;
1014
import org.apache.hc.client5.http.classic.methods.HttpGet;
1115
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
1216
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
17+
import org.apache.hc.core5.http.HttpStatus;
1318
import org.apache.hc.core5.http.Header;
1419
import org.apache.hc.core5.http.io.entity.EntityUtils;
1520
import org.apache.hc.core5.net.URIBuilder;
@@ -41,21 +46,17 @@ public final class HttpSegmentChangeFetcher implements SegmentChangeFetcher {
4146

4247
private final CloseableHttpClient _client;
4348
private final URI _target;
44-
private final Metrics _metrics;
49+
private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
4550

46-
public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root) throws URISyntaxException {
47-
return create(client, root, new Metrics.NoopMetrics());
51+
public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
52+
return new HttpSegmentChangeFetcher(client, Utils.appendPath(root, "api/segmentChanges"), telemetryRuntimeProducer);
4853
}
4954

50-
public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root, Metrics metrics) throws URISyntaxException {
51-
return new HttpSegmentChangeFetcher(client, Utils.appendPath(root, "api/segmentChanges"), metrics);
52-
}
53-
54-
private HttpSegmentChangeFetcher(CloseableHttpClient client, URI uri, Metrics metrics) {
55+
private HttpSegmentChangeFetcher(CloseableHttpClient client, URI uri, TelemetryRuntimeProducer telemetryRuntimeProducer) {
5556
_client = client;
5657
_target = uri;
57-
_metrics = metrics;
5858
checkNotNull(_target);
59+
_telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
5960
}
6061

6162
@Override
@@ -90,17 +91,17 @@ public SegmentChange fetch(String segmentName, long since, FetchOptions options)
9091

9192
int statusCode = response.getCode();
9293

93-
if (statusCode < 200 || statusCode >= 300) {
94+
if (statusCode < HttpStatus.SC_OK || statusCode >= HttpStatus.SC_MULTIPLE_CHOICES) {
95+
_telemetryRuntimeProducer.recordSyncError(ResourceEnum.SEGMENT_SYNC, statusCode);
9496
_log.error("Response status was: " + statusCode);
95-
if (statusCode == 403) {
97+
if (statusCode == HttpStatus.SC_FORBIDDEN) {
9698
_log.error("factory instantiation: you passed a browser type api_key, " +
9799
"please grab an api key from the Split console that is of type sdk");
98100
}
99-
_metrics.count(PREFIX + ".status." + statusCode, 1);
100101
throw new IllegalStateException("Could not retrieve segment changes for " + segmentName + "; http return code " + statusCode);
101102
}
102103

103-
104+
_telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.SEGMENTS, System.currentTimeMillis());
104105

105106
String json = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
106107
if (_log.isDebugEnabled()) {
@@ -109,11 +110,10 @@ public SegmentChange fetch(String segmentName, long since, FetchOptions options)
109110

110111
return Json.fromJson(json, SegmentChange.class);
111112
} catch (Throwable t) {
112-
_metrics.count(PREFIX + ".exception", 1);
113113
throw new IllegalStateException("Problem fetching segmentChanges: " + t.getMessage(), t);
114114
} finally {
115+
_telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SEGMENTS, System.currentTimeMillis()-start);
115116
Utils.forceClose(response);
116-
_metrics.time(PREFIX + ".time", System.currentTimeMillis() - start);
117117
}
118118

119119

‎client/src/main/java/io/split/client/HttpSplitChangeFetcher.java‎

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,14 @@
77
import io.split.engine.common.FetchOptions;
88
import io.split.engine.experiments.SplitChangeFetcher;
99
import io.split.engine.metrics.Metrics;
10+
import io.split.telemetry.domain.enums.HTTPLatenciesEnum;
11+
import io.split.telemetry.domain.enums.LastSynchronizationRecordsEnum;
12+
import io.split.telemetry.domain.enums.ResourceEnum;
13+
import io.split.telemetry.storage.TelemetryRuntimeProducer;
1014
import org.apache.hc.client5.http.classic.methods.HttpGet;
1115
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
1216
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
17+
import org.apache.hc.core5.http.HttpStatus;
1318
import org.apache.hc.core5.http.Header;
1419
import org.apache.hc.core5.http.io.entity.EntityUtils;
1520
import org.apache.hc.core5.net.URIBuilder;
@@ -42,21 +47,17 @@ public final class HttpSplitChangeFetcher implements SplitChangeFetcher {
4247

4348
private final CloseableHttpClient _client;
4449
private final URI _target;
45-
private final Metrics _metrics;
50+
private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
4651

47-
public static HttpSplitChangeFetcher create(CloseableHttpClient client, URI root) throws URISyntaxException {
48-
return create(client, root, new Metrics.NoopMetrics());
52+
public static HttpSplitChangeFetcher create(CloseableHttpClient client, URI root, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
53+
return new HttpSplitChangeFetcher(client, Utils.appendPath(root, "api/splitChanges"), telemetryRuntimeProducer);
4954
}
5055

51-
public static HttpSplitChangeFetcher create(CloseableHttpClient client, URI root, Metrics metrics) throws URISyntaxException {
52-
return new HttpSplitChangeFetcher(client, Utils.appendPath(root, "api/splitChanges"), metrics);
53-
}
54-
55-
private HttpSplitChangeFetcher(CloseableHttpClient client, URI uri, Metrics metrics) {
56+
private HttpSplitChangeFetcher(CloseableHttpClient client, URI uri, TelemetryRuntimeProducer telemetryRuntimeProducer) {
5657
_client = client;
5758
_target = uri;
58-
_metrics = metrics;
5959
checkNotNull(_target);
60+
_telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
6061
}
6162

6263
long makeRandomTill() {
@@ -93,8 +94,8 @@ public SplitChange fetch(long since, FetchOptions options) {
9394

9495
int statusCode = response.getCode();
9596

96-
if (statusCode < 200 || statusCode >= 300) {
97-
_metrics.count(PREFIX + ".status." + statusCode, 1);
97+
if (statusCode < HttpStatus.SC_OK || statusCode >= HttpStatus.SC_MULTIPLE_CHOICES) {
98+
_telemetryRuntimeProducer.recordSyncError(ResourceEnum.SPLIT_SYNC, statusCode);
9899
throw new IllegalStateException("Could not retrieve splitChanges; http return code " + statusCode);
99100
}
100101

@@ -106,11 +107,10 @@ public SplitChange fetch(long since, FetchOptions options) {
106107

107108
return Json.fromJson(json, SplitChange.class);
108109
} catch (Throwable t) {
109-
_metrics.count(PREFIX + ".exception", 1);
110110
throw new IllegalStateException("Problem fetching splitChanges: " + t.getMessage(), t);
111111
} finally {
112+
_telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SPLITS, System.currentTimeMillis()-start);
112113
Utils.forceClose(response);
113-
_metrics.time(PREFIX + ".time", System.currentTimeMillis() - start);
114114
}
115115
}
116116

‎client/src/main/java/io/split/client/LocalhostSplitFactory.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@
66
import io.split.engine.SDKReadinessGates;
77
import io.split.engine.evaluator.EvaluatorImp;
88
import io.split.engine.metrics.Metrics;
9+
import io.split.telemetry.storage.InMemoryTelemetryStorage;
10+
import io.split.telemetry.storage.NoopTelemetryStorage;
911
import org.slf4j.Logger;
1012
import org.slf4j.LoggerFactory;
1113

@@ -54,12 +56,12 @@ public LocalhostSplitFactory(String directory, String file) throws IOException {
5456
SplitCache splitCache = new InMemoryCacheImp();
5557
SDKReadinessGates sdkReadinessGates = new SDKReadinessGates();
5658

57-
sdkReadinessGates.splitsAreReady();
5859
_cacheUpdaterService = new CacheUpdaterService(splitCache);
5960
_cacheUpdaterService.updateCache(splitAndKeyToTreatment);
61+
sdkReadinessGates.sdkInternalReady();
6062
_client = new SplitClientImpl(this, splitCache,
61-
new ImpressionsManager.NoOpImpressionsManager(), new Metrics.NoopMetrics(), new NoopEventClient(),
62-
SplitClientConfig.builder().setBlockUntilReadyTimeout(1).build(), sdkReadinessGates, new EvaluatorImp(splitCache));
63+
new ImpressionsManager.NoOpImpressionsManager(), new NoopEventClient(),
64+
SplitClientConfig.builder().setBlockUntilReadyTimeout(1).build(), sdkReadinessGates, new EvaluatorImp(splitCache), new NoopTelemetryStorage(), new NoopTelemetryStorage());
6365
_manager = LocalhostSplitManager.of(splitAndKeyToTreatment);
6466

6567
_splitFile.registerWatcher();

0 commit comments

Comments
 (0)