Skip to content

Commit 2b8a889

Browse files
skykaninskykanin
authored
implement custom WithSpan annotation (#196)
1 parent b7ed178 commit 2b8a889

File tree

6 files changed

+257
-182
lines changed

6 files changed

+257
-182
lines changed

‎src/main/java/no/ssb/dlp/pseudo/service/accessgroups/CloudIdentityService.java‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
package no.ssb.dlp.pseudo.service.accessgroups;
22

33
import io.micronaut.cache.annotation.Cacheable;
4+
import io.opentelemetry.api.GlobalOpenTelemetry;
5+
import io.opentelemetry.api.trace.Tracer;
46
import io.reactivex.Flowable;
57
import jakarta.inject.Singleton;
68
import lombok.RequiredArgsConstructor;
9+
import no.ssb.dlp.pseudo.service.tracing.WithSpan;
710

811
import java.util.ArrayList;
912
import java.util.List;
@@ -13,8 +16,10 @@
1316
public class CloudIdentityService {
1417
private final CloudIdentityClient cloudIdentityClient;
1518

19+
@WithSpan
1620
@Cacheable(value = "cloud-identity-service-cache", parameters = {"groupEmail"})
1721
public List<Membership> listMembers(String groupEmail) {
22+
1823
return Flowable.fromPublisher(cloudIdentityClient.lookup(groupEmail))
1924
.flatMap(lookupResponse -> fetchMemberships(lookupResponse.getGroupName(), null,
2025
new ArrayList<>()))
@@ -24,13 +29,14 @@ public List<Membership> listMembers(String groupEmail) {
2429
/**
2530
* Paginate through all memberships of a group.
2631
*
27-
* @param groupId the id of the group
28-
* @param nextPageToken a token for pagination (will be null on first call)
32+
* @param groupId the id of the group
33+
* @param nextPageToken a token for pagination (will be null on first call)
2934
* @param allMemberships a list that will be populated with all memberships
3035
* @return the list of all memberships
3136
*/
32-
private Flowable<List<Membership>> fetchMemberships(String groupId, String nextPageToken,
33-
List<Membership> allMemberships) {
37+
@WithSpan
38+
protected Flowable<List<Membership>> fetchMemberships(String groupId, String nextPageToken,
39+
List<Membership> allMemberships) {
3440
if (groupId == null || groupId.isEmpty()) {
3541
return Flowable.just(allMemberships);
3642
}

‎src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoField.java‎

Lines changed: 39 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import no.ssb.dlp.pseudo.core.util.Json;
2121
import no.ssb.dlp.pseudo.service.pseudo.metadata.FieldMetric;
2222
import no.ssb.dlp.pseudo.service.pseudo.metadata.PseudoMetadataProcessor;
23+
import no.ssb.dlp.pseudo.service.tracing.WithSpan;
2324

2425
import java.util.List;
2526
import java.util.Map;
@@ -32,7 +33,6 @@
3233
@Data
3334
@Slf4j
3435
public class PseudoField {
35-
private static final Tracer tracer = GlobalOpenTelemetry.getTracer("pseudo-service");
3636
@Getter(AccessLevel.PROTECTED)
3737
private static final int BUFFER_SIZE = 10000;
3838
@Getter(AccessLevel.PROTECTED)
@@ -87,52 +87,46 @@ public PseudoField(String name, String pattern, String pseudoFunc, EncryptedKeys
8787
* @param values The values to be processed.
8888
* @return A Flowable stream that processes the field values by applying the configured pseudo rules, and returns them as a lists of strings.
8989
*/
90-
@AddingSpanAttributes
91-
public Flowable<String> process(@SpanAttribute("pseudoConfigSplitter") PseudoConfigSplitter pseudoConfigSplitter,
92-
@SpanAttribute("recordProcessorFactory") RecordMapProcessorFactory recordProcessorFactory,
93-
@SpanAttribute("values") List<String> values,
94-
@SpanAttribute("pseudoOperation") PseudoOperation pseudoOperation,
90+
@WithSpan
91+
public Flowable<String> process(PseudoConfigSplitter pseudoConfigSplitter,
92+
RecordMapProcessorFactory recordProcessorFactory,
93+
List<String> values,
94+
PseudoOperation pseudoOperation,
9595
String correlationId) {
96-
final var span = tracer.spanBuilder("PseudoField.process").startSpan();
97-
try (Scope scope = span.makeCurrent()) {
98-
99-
Stopwatch stopwatch = Stopwatch.createStarted();
100-
List<PseudoConfig> pseudoConfigs = pseudoConfigSplitter.splitIfNecessary(this.getPseudoConfig());
101-
102-
RecordMapProcessor<PseudoMetadataProcessor> recordMapProcessor;
103-
switch (pseudoOperation) {
104-
case PSEUDONYMIZE -> recordMapProcessor = recordProcessorFactory.
105-
newPseudonymizeRecordProcessor(pseudoConfigs, correlationId);
106-
case DEPSEUDONYMIZE -> recordMapProcessor = recordProcessorFactory.
107-
newDepseudonymizeRecordProcessor(pseudoConfigs, correlationId);
108-
default -> throw new RuntimeException(
109-
String.format("Pseudo operation \"%s\" not supported for this method", pseudoOperation));
110-
}
111-
Completable preprocessor = getPreprocessor(values, recordMapProcessor);
112-
// Metadata will be processes in parallel with the data, but must be collected separately
113-
final PseudoMetadataProcessor metadataProcessor = recordMapProcessor.getMetadataProcessor();
114-
final Flowable<String> metadata = Flowable.fromPublisher(metadataProcessor.getMetadata());
115-
final Flowable<String> logs = Flowable.fromPublisher(metadataProcessor.getLogs());
116-
final Flowable<String> metrics = Flowable.fromPublisher(metadataProcessor.getMetrics());
117-
118-
Flowable<String> result = preprocessor.andThen(Flowable.fromIterable(values.stream()
119-
.map(v -> mapOptional(v, recordMapProcessor, metadataProcessor)).toList()
120-
))
121-
.map(v -> v.map(Json::from).orElse("null"))
122-
.doOnError(throwable -> {
123-
log.error("Response failed", throwable);
124-
recordMapProcessor.getMetadataProcessor().onErrorAll(throwable);
125-
})
126-
.doOnComplete(() -> {
127-
log.info("{} took {}", pseudoOperation, stopwatch.stop().elapsed());
128-
// Signal the metadataProcessor to stop collecting metadata
129-
recordMapProcessor.getMetadataProcessor().onCompleteAll();
130-
});
131-
132-
return PseudoResponseSerializer.serialize(result, metadata, logs, metrics);
133-
} finally {
134-
span.end();
96+
Stopwatch stopwatch = Stopwatch.createStarted();
97+
List<PseudoConfig> pseudoConfigs = pseudoConfigSplitter.splitIfNecessary(this.getPseudoConfig());
98+
99+
RecordMapProcessor<PseudoMetadataProcessor> recordMapProcessor;
100+
switch (pseudoOperation) {
101+
case PSEUDONYMIZE -> recordMapProcessor = recordProcessorFactory.
102+
newPseudonymizeRecordProcessor(pseudoConfigs, correlationId);
103+
case DEPSEUDONYMIZE -> recordMapProcessor = recordProcessorFactory.
104+
newDepseudonymizeRecordProcessor(pseudoConfigs, correlationId);
105+
default -> throw new RuntimeException(
106+
String.format("Pseudo operation \"%s\" not supported for this method", pseudoOperation));
135107
}
108+
Completable preprocessor = getPreprocessor(values, recordMapProcessor);
109+
// Metadata will be processes in parallel with the data, but must be collected separately
110+
final PseudoMetadataProcessor metadataProcessor = recordMapProcessor.getMetadataProcessor();
111+
final Flowable<String> metadata = Flowable.fromPublisher(metadataProcessor.getMetadata());
112+
final Flowable<String> logs = Flowable.fromPublisher(metadataProcessor.getLogs());
113+
final Flowable<String> metrics = Flowable.fromPublisher(metadataProcessor.getMetrics());
114+
115+
Flowable<String> result = preprocessor.andThen(Flowable.fromIterable(values.stream()
116+
.map(v -> mapOptional(v, recordMapProcessor, metadataProcessor)).toList()
117+
))
118+
.map(v -> v.map(Json::from).orElse("null"))
119+
.doOnError(throwable -> {
120+
log.error("Response failed", throwable);
121+
recordMapProcessor.getMetadataProcessor().onErrorAll(throwable);
122+
})
123+
.doOnComplete(() -> {
124+
log.info("{} took {}", pseudoOperation, stopwatch.stop().elapsed());
125+
// Signal the metadataProcessor to stop collecting metadata
126+
recordMapProcessor.getMetadataProcessor().onCompleteAll();
127+
});
128+
129+
return PseudoResponseSerializer.serialize(result, metadata, logs, metrics);
136130
}
137131

138132
/**

0 commit comments

Comments
 (0)