Skip to content

add more tracing spans #189

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
skykanin merged 1 commit into from
Feb 26, 2026
Merged

add more tracing spans #189

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 19 additions & 38 deletions src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,9 +66,7 @@ public class PseudoController {
@Produces(MediaType.APPLICATION_JSON)
@Post(value = "/pseudonymize/field", consumes = MediaType.APPLICATION_JSON)
@ExecuteOn(TaskExecutors.BLOCKING)
public HttpResponse<Flowable<byte[]>> pseudonymizeField(
@SpanAttribute("pseudonymize column request") @Schema(implementation = PseudoFieldRequest.class) String request
) {
public HttpResponse<Flowable<byte[]>> pseudonymizeField(@SpanAttribute("pseudonymize column request") @Schema(implementation = PseudoFieldRequest.class) String request) {
PseudoFieldRequest req = Json.toObject(PseudoFieldRequest.class, request);
Span currentSpan = Span.current();
if (currentSpan.getSpanContext().isValid() && req != null) {
Expand All @@ -82,16 +80,14 @@ public HttpResponse<Flowable<byte[]>> pseudonymizeField(
final String correlationId = MDC.get("CorrelationID");

return HttpResponse.ok(
pseudoField.process(
pseudoConfigSplitter,
recordProcessorFactory,
req.values,
PseudoOperation.PSEUDONYMIZE,
correlationId
)
.map(o -> o.getBytes(StandardCharsets.UTF_8))
)
.characterEncoding(StandardCharsets.UTF_8);
pseudoField.process(
pseudoConfigSplitter,
recordProcessorFactory,
req.values,
PseudoOperation.PSEUDONYMIZE,
correlationId
).map(o -> o.getBytes(StandardCharsets.UTF_8))
).characterEncoding(StandardCharsets.UTF_8);
} catch (Exception e) {
return HttpResponse.serverError(Flowable.error(e));
}
Expand All @@ -110,9 +106,7 @@ public HttpResponse<Flowable<byte[]>> pseudonymizeField(
@Secured({PseudoServiceRole.ADMIN})
@Post(value = "/depseudonymize/field", consumes = MediaType.APPLICATION_JSON)
@ExecuteOn(TaskExecutors.BLOCKING)
public HttpResponse<Flowable<byte[]>> depseudonymizeField(
@SpanAttribute("depseudonymize column request") @Schema(implementation = DepseudoFieldRequest.class) String request
) {
public HttpResponse<Flowable<byte[]>> depseudonymizeField(@SpanAttribute("depseudonymize column request") @Schema(implementation = DepseudoFieldRequest.class) String request) {
DepseudoFieldRequest req = Json.toObject(DepseudoFieldRequest.class, request);
Span currentSpan = Span.current();
if (currentSpan.getSpanContext().isValid() && req != null) {
Expand All @@ -126,10 +120,7 @@ public HttpResponse<Flowable<byte[]>> depseudonymizeField(

final String correlationId = MDC.get("CorrelationID");

return HttpResponse.ok(pseudoField.process(
pseudoConfigSplitter, recordProcessorFactory,req.values, PseudoOperation.DEPSEUDONYMIZE, correlationId)
.map(o -> o.getBytes(StandardCharsets.UTF_8)))
.characterEncoding(StandardCharsets.UTF_8);
return HttpResponse.ok(pseudoField.process(pseudoConfigSplitter, recordProcessorFactory, req.values, PseudoOperation.DEPSEUDONYMIZE, correlationId).map(o -> o.getBytes(StandardCharsets.UTF_8))).characterEncoding(StandardCharsets.UTF_8);
} catch (Exception e) {
return HttpResponse.serverError(Flowable.error(e));
}
Expand All @@ -146,20 +137,15 @@ public HttpResponse<Flowable<byte[]>> depseudonymizeField(
@Secured({PseudoServiceRole.ADMIN})
@Post(value = "/repseudonymize/field", consumes = MediaType.APPLICATION_JSON)
@ExecuteOn(TaskExecutors.BLOCKING)
public HttpResponse<Flowable<byte[]>> repseudonymizeField(
@Schema(implementation = RepseudoFieldRequest.class) String request
) {
public HttpResponse<Flowable<byte[]>> repseudonymizeField(@Schema(implementation = RepseudoFieldRequest.class) String request) {
RepseudoFieldRequest req = Json.toObject(RepseudoFieldRequest.class, request);
log.info(Strings.padEnd(String.format("*** Repseudonymize field: %s ", req.getName()), 80, '*'));
PseudoField sourcePseudoField = new PseudoField(req.getName(), req.getPattern(), req.getSourcePseudoFunc(), req.getSourceKeyset());
PseudoField targetPseudoField = new PseudoField(req.getName(), req.getPattern(), req.getTargetPseudoFunc(), req.getTargetKeyset());
try {

final String correlationId = MDC.get("CorrelationID");
return HttpResponse.ok(
sourcePseudoField.process(recordProcessorFactory, req.values, targetPseudoField, correlationId)
.map(o -> o.getBytes(StandardCharsets.UTF_8)))
.characterEncoding(StandardCharsets.UTF_8);
return HttpResponse.ok(sourcePseudoField.process(recordProcessorFactory, req.values, targetPseudoField, correlationId).map(o -> o.getBytes(StandardCharsets.UTF_8))).characterEncoding(StandardCharsets.UTF_8);
} catch (Exception e) {
return HttpResponse.serverError(Flowable.error(e));
}
Expand Down Expand Up @@ -209,34 +195,29 @@ public static class RepseudoFieldRequest {

@Error
public HttpResponse<JsonError> unknownPseudoKeyError(HttpRequest request, NoSuchPseudoKeyException e) {
JsonError error = new JsonError(e.getMessage())
.link(Link.SELF, Link.of(request.getUri()));
JsonError error = new JsonError(e.getMessage()).link(Link.SELF, Link.of(request.getUri()));
return HttpResponse.<JsonError>badRequest().body(error);
}

@Error
public HttpResponse<JsonError> sidIndexUnavailable(HttpRequest request, SidIndexUnavailableException e) {
JsonError error = new JsonError(e.getMessage())
.link(Link.SELF, Link.of(request.getUri()));
JsonError error = new JsonError(e.getMessage()).link(Link.SELF, Link.of(request.getUri()));
return HttpResponse.<JsonError>serverError().status(HttpStatus.SERVICE_UNAVAILABLE).body(error);
}

@Error
public HttpResponse<JsonError> illegalArgument(HttpRequest request, IllegalArgumentException e) {
JsonError error = new JsonError(e.getMessage())
.link(Link.SELF, Link.of(request.getUri()));
JsonError error = new JsonError(e.getMessage()).link(Link.SELF, Link.of(request.getUri()));
return HttpResponse.<JsonError>badRequest().body(error);
}

@Error
public HttpResponse<JsonError> sidVersionInvalid(HttpRequest request, PseudoFuncFactory.PseudoFuncInitException e) {
if (e.getCause() instanceof InvocationTargetException && e.getCause().getCause() instanceof InvalidSidSnapshotDateException){
JsonError error = new JsonError(e.getCause().getCause().getMessage())
.link(Link.SELF, Link.of(request.getUri()));
if (e.getCause() instanceof InvocationTargetException && e.getCause().getCause() instanceof InvalidSidSnapshotDateException) {
JsonError error = new JsonError(e.getCause().getCause().getMessage()).link(Link.SELF, Link.of(request.getUri()));
return HttpResponse.<JsonError>badRequest().body(error);
}
JsonError error = new JsonError(e.getMessage())
.link(Link.SELF, Link.of(request.getUri()));
JsonError error = new JsonError(e.getMessage()).link(Link.SELF, Link.of(request.getUri()));
return HttpResponse.<JsonError>serverError().status(HttpStatus.SERVICE_UNAVAILABLE).body(error);
}
}
24 changes: 15 additions & 9 deletions src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
package no.ssb.dlp.pseudo.service.pseudo;

import com.google.common.base.Stopwatch;
import io.opentelemetry.instrumentation.annotations.AddingSpanAttributes;
import io.opentelemetry.instrumentation.annotations.SpanAttribute;
import io.opentelemetry.instrumentation.annotations.WithSpan;
import io.reactivex.Completable;
import io.reactivex.Flowable;
import lombok.AccessLevel;
Expand Down Expand Up @@ -75,21 +78,23 @@ public PseudoField(String name, String pattern, String pseudoFunc, EncryptedKeys
/**
* Creates a Flowable that processes each value of the field, by applying the configured pseudo rules using a recordMapProcessor.
* This variant of the process() method is intended for "pseudonymize" and "depseudonymize" operations.
*
* @param pseudoConfigSplitter The PseudoConfigSplitter instance to use for splitting pseudo configurations.
* @param recordProcessorFactory The RecordMapProcessorFactory instance to use for creating a new PseudonymizeRecordProcessor.
* @param values The values to be processed.
* @return A Flowable stream that processes the field values by applying the configured pseudo rules, and returns them as a lists of strings.
*/
public Flowable<String> process(PseudoConfigSplitter pseudoConfigSplitter,
RecordMapProcessorFactory recordProcessorFactory,
List<String> values,
PseudoOperation pseudoOperation,
String correlationId) {
@WithSpan
public Flowable<String> process(@SpanAttribute("pseudoConfigSplitter") PseudoConfigSplitter pseudoConfigSplitter,
@SpanAttribute("recordProcessorFactory") RecordMapProcessorFactory recordProcessorFactory,
@SpanAttribute("values") List<String> values,
@SpanAttribute("pseudoOperation") PseudoOperation pseudoOperation,
String correlationId) {
Stopwatch stopwatch = Stopwatch.createStarted();
List<PseudoConfig> pseudoConfigs = pseudoConfigSplitter.splitIfNecessary(this.getPseudoConfig());

RecordMapProcessor<PseudoMetadataProcessor> recordMapProcessor;
switch (pseudoOperation){
switch (pseudoOperation) {
case PSEUDONYMIZE -> recordMapProcessor = recordProcessorFactory.
newPseudonymizeRecordProcessor(pseudoConfigs, correlationId);
case DEPSEUDONYMIZE -> recordMapProcessor = recordProcessorFactory.
Expand Down Expand Up @@ -124,14 +129,15 @@ public Flowable<String> process(PseudoConfigSplitter pseudoConfigSplitter,
/**
* Creates a Flowable that processes each value of the field, by applying the configured pseudo rules using a recordMapProcessor.
* This variant of the process() method is intended for the "repseudonymize" operation.
*
* @param recordProcessorFactory The RecordMapProcessorFactory instance to use for creating a new PseudonymizeRecordProcessor.
* @param values The values to be processed.
* @return A Flowable stream that processes the field values by applying the configured pseudo rules, and returns them as a lists of strings.
*/
public Flowable<String> process(RecordMapProcessorFactory recordProcessorFactory,
List<String> values,
PseudoField targetPseudoField,
String correlationId) {
List<String> values,
PseudoField targetPseudoField,
String correlationId) {
Stopwatch stopwatch = Stopwatch.createStarted();
PseudoConfig targetPseudoConfig = targetPseudoField.getPseudoConfig();
RecordMapProcessor<PseudoMetadataProcessor> recordMapProcessor = recordProcessorFactory.
Expand Down
10 changes: 4 additions & 6 deletions src/main/java/no/ssb/dlp/pseudo/service/pseudo/RecordMapProcessorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import com.github.benmanes.caffeine.cache.LoadingCache;
import com.google.crypto.tink.Aead;
import io.opentelemetry.instrumentation.annotations.AddingSpanAttributes;
import io.opentelemetry.instrumentation.annotations.WithSpan;
import jakarta.inject.Singleton;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -17,7 +17,6 @@
import no.ssb.dlp.pseudo.core.PseudoException;
import no.ssb.dlp.pseudo.core.PseudoKeyset;
import no.ssb.dlp.pseudo.core.PseudoOperation;
import no.ssb.dlp.pseudo.core.PseudoSecret;
import no.ssb.dlp.pseudo.core.field.FieldDescriptor;
import no.ssb.dlp.pseudo.core.field.ValueInterceptorChain;
import no.ssb.dlp.pseudo.core.func.PseudoFuncDeclaration;
Expand Down Expand Up @@ -48,13 +47,12 @@ public class RecordMapProcessorFactory {
private final PseudoSecrets pseudoSecrets;
private final LoadingCache<String, Aead> aeadCache;

@AddingSpanAttributes
@WithSpan
public RecordMapProcessor<PseudoMetadataProcessor> newPseudonymizeRecordProcessor(List<PseudoConfig> pseudoConfigs, String correlationId) {
ValueInterceptorChain chain = new ValueInterceptorChain();
PseudoMetadataProcessor metadataProcessor = new PseudoMetadataProcessor(correlationId);

for (PseudoConfig config : pseudoConfigs) {
List<PseudoSecret> secrets = pseudoSecrets.resolve();
for (PseudoKeyset keyset : config.getKeysets()) {
log.info(keyset.getKekUri().toString());
}
Expand All @@ -66,7 +64,7 @@ public RecordMapProcessor<PseudoMetadataProcessor> newPseudonymizeRecordProcesso
return new RecordMapProcessor<>(chain, metadataProcessor);
}

@AddingSpanAttributes
@WithSpan
public RecordMapProcessor<PseudoMetadataProcessor> newDepseudonymizeRecordProcessor(List<PseudoConfig> pseudoConfigs, String correlationId) {
ValueInterceptorChain chain = new ValueInterceptorChain();
PseudoMetadataProcessor metadataProcessor = new PseudoMetadataProcessor(correlationId);
Expand All @@ -81,7 +79,7 @@ public RecordMapProcessor<PseudoMetadataProcessor> newDepseudonymizeRecordProces
return new RecordMapProcessor<>(chain, metadataProcessor);
}

@AddingSpanAttributes
@WithSpan
public RecordMapProcessor<PseudoMetadataProcessor> newRepseudonymizeRecordProcessor(PseudoConfig sourcePseudoConfig,
PseudoConfig targetPseudoConfig, String correlationId) {
final PseudoFuncs fieldDepseudonymizer = newPseudoFuncs(sourcePseudoConfig.getRules(),
Expand Down
4 changes: 4 additions & 0 deletions src/main/java/no/ssb/dlp/pseudo/service/pseudo/metadata/PseudoMetadataProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package no.ssb.dlp.pseudo.service.pseudo.metadata;

import io.opentelemetry.instrumentation.annotations.WithSpan;
import io.reactivex.processors.ReplayProcessor;
import lombok.Value;
import no.ssb.dlp.pseudo.core.util.Json;
Expand Down Expand Up @@ -34,12 +35,15 @@ public void addLog(String log) {
public void addMetric(FieldMetric fieldMetric) {
metrics.onNext(fieldMetric);
}
@WithSpan
public Publisher<String> getMetadata() {
return datadocMetadata.map(FieldMetadata::toDatadocVariable).map(Json::from);
}
@WithSpan
public Publisher<String> getLogs() {
return logs.map(Json::from);
}
@WithSpan
public Publisher<String> getMetrics() {
return metrics.groupBy(FieldMetric::name)
.flatMapSingle(group ->
Expand Down
Loading