columnSelectors = Set.of();
-
- /**
- * Descriptive name of the contents. This will be used as baseline for the target archive name and its contents.
- * If not specified then this will be deduced from the source dataset name.
- * Should not include file suffixes such as .csv or .json.
- */
- private String targetContentName;
-
- /**
- * The content type of the resulting file.
- * Defaults to application/json.
- */
- @Schema(implementation = String.class, allowableValues = {
- MediaType.APPLICATION_JSON, MoreMediaTypes.TEXT_CSV})
- private MediaType targetContentType = MediaType.APPLICATION_JSON_TYPE;
-
- /** The password for the resulting archive */
- @NotBlank
- @Schema(implementation = String.class)
- @Min(9)
- private char[] targetPassword;
-
- /** Whether or not to depseudonymize dataset during export */
- private Boolean depseudonymize = false;
-
- /**
- * Pseudonymization rules to be used to depseudonymize the dataset. This is only
- * relevant if depseudonymize=true.
- *
- * If not specified then pseudonymization rules are read from the
- * dataset's dataset-meta.json file.
- */
- private List pseudoRules = new ArrayList<>();
- }
-}
diff --git a/src/main/java/no/ssb/dlp/pseudo/service/export/ExportService.java b/src/main/java/no/ssb/dlp/pseudo/service/export/ExportService.java
deleted file mode 100644
index b41c2a4..0000000
--- a/src/main/java/no/ssb/dlp/pseudo/service/export/ExportService.java
+++ /dev/null
@@ -1,311 +0,0 @@
-package no.ssb.dlp.pseudo.service.export;
-
-import com.google.common.base.CharMatcher;
-import com.google.common.base.Stopwatch;
-import io.micronaut.context.event.ApplicationEventPublisher;
-import io.micronaut.core.annotation.Introspected;
-import io.micronaut.http.MediaType;
-import io.micronaut.runtime.event.annotation.EventListener;
-import io.micronaut.scheduling.annotation.Async;
-import io.reactivex.Flowable;
-import lombok.*;
-import lombok.experimental.Accessors;
-import lombok.extern.slf4j.Slf4j;
-import no.ssb.dapla.dataset.api.DatasetMeta;
-import no.ssb.dapla.parquet.FieldInterceptor;
-import no.ssb.dapla.storage.client.DatasetStorage;
-import no.ssb.dapla.storage.client.backend.gcs.GoogleCloudStorageBackend;
-import no.ssb.dlp.pseudo.core.FieldPseudoInterceptor;
-import no.ssb.dlp.pseudo.core.func.PseudoFuncRule;
-import no.ssb.dlp.pseudo.core.func.PseudoFuncs;
-import no.ssb.dlp.pseudo.core.PseudoOperation;
-import no.ssb.dlp.pseudo.core.file.Compression;
-import no.ssb.dlp.pseudo.core.file.MoreMediaTypes;
-import no.ssb.dlp.pseudo.core.map.RecordMapSerializerFactory;
-import no.ssb.dlp.pseudo.core.util.Json;
-import no.ssb.dlp.pseudo.core.util.PathJoiner;
-import no.ssb.dlp.pseudo.core.util.Zips;
-import no.ssb.dlp.pseudo.service.datasetmeta.DatasetMetaService;
-import no.ssb.dlp.pseudo.service.pseudo.PseudoConfig;
-import no.ssb.dlp.pseudo.service.pseudo.PseudoSecrets;
-
-import javax.inject.Singleton;
-import javax.validation.constraints.NotNull;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.time.Instant;
-import java.time.ZoneOffset;
-import java.time.format.DateTimeFormatter;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static com.google.common.base.CharMatcher.anyOf;
-import static com.google.common.base.CharMatcher.inRange;
-import static no.ssb.dlp.pseudo.core.util.Zips.ZipOptions.zipOpts;
-
-@RequiredArgsConstructor
-@Singleton
-@Slf4j
-public class ExportService {
-
- private static final CharMatcher ALLOWED_FILENAME_CHARACTERS = inRange('a', 'z')
- .or(inRange('A', 'Z'))
- .or(inRange('0', '9'))
- .or(anyOf("_-"));
-
- private final ExportConfig exportConfig;
- private final DatasetStorage datasetClient;
- private final GoogleCloudStorageBackend storageBackend;
- private final PseudoSecrets pseudoSecrets;
- private final DatasetMetaService datasetMetaService;
- private final ApplicationEventPublisher eventPublisher;
-
- public DatasetExportResult export(DatasetExport e) {
- final DatasetExportReport report = new DatasetExportReport(e);
- final DatasetMeta sourceDatasetMeta = datasetMetaService.readDatasetMeta(e.getSourceDataset().legacyDatasetUri()).orElse(null);
-
- // Initialize target names
- if (e.getTargetContentName() == null || e.getTargetContentName().isBlank()) {
- // if not specified then deduce content name from dataset name
- e.setTargetContentName(e.getSourceDataset().getPath().replaceFirst(".*/([^/?]+).*", "$1"));
- }
- String targetRootLocation = targetRootLocationOf(e.getSourceDataset());
- String targetArchiveUri = targetFileLocationOf(targetRootLocation, archiveFilenameOf(e.getTargetContentName(), e.getCompression().getType()));
-
- eventPublisher.publishEvent(ExportEvent.builder()
- .targetRootLocation(targetRootLocation)
- .targetArchiveUri(targetArchiveUri)
- .datasetExport(e)
- .report(report)
- .sourceDataset(e.getSourceDataset())
- .sourceDatasetMeta(sourceDatasetMeta)
- .build());
-
- return DatasetExportResult.builder()
- .targetUri(targetArchiveUri)
- .build();
- }
-
- @EventListener
- @Async
- public void onExportEvent(ExportEvent e) {
- final Stopwatch stopwatch = Stopwatch.createStarted();
- final AtomicInteger counter = new AtomicInteger();
-
- // Initialize depseudo mechanism if needed - else use a noOp interceptor
- FieldInterceptor fieldPseudoInterceptor = initPseudoInterceptor(e.datasetExport(), e.sourceDatasetMeta(), e.report());
-
- // Initiate record stream
- log.debug("Read parquet records");
- Flowable> records = datasetClient.readParquetRecords(e.sourceDataset().legacyDatasetUri(), e.datasetExport().getColumnSelectors(), fieldPseudoInterceptor)
- .doOnNext(s -> {
- if (counter.incrementAndGet() % 100000 == 0) {
- log.debug(String.format("Processed %,d records", counter.get()));
- }
- });
-
- // Serialize records
- MediaType targetContentType = MoreMediaTypes.validContentType(e.datasetExport().getTargetContentType());
- log.debug("Serialize records as %s".formatted(targetContentType));
- Flowable serializedRecords = RecordMapSerializerFactory.newFromMediaType(targetContentType).serialize(records);
-
- // Encrypt and compress stream contents:
- Flowable compressedRecords = encryptAndCompress(e.datasetExport(), serializedRecords);
- // Upload stream contents
-
- storageBackend
- .write(e.targetArchiveUri(), compressedRecords)
- .timeout(30, TimeUnit.SECONDS)
- .doOnError(throwable -> log.error("Upload failed: %s".formatted(e.targetArchiveUri()), throwable))
- .subscribe(() -> {
- e.report().setElapsedMillis(stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
- log.info("Successful upload: %s".formatted(e.targetArchiveUri()));
- uploadExportReport(e.report(), e.targetRootLocation());
- });
- }
-
- void uploadExportReport(DatasetExportReport report, String targetRootLocation) {
- String targetUri = targetFileLocationOf(targetRootLocation, ".export-meta.json");
- try {
- storageBackend.write(targetUri, Json.prettyFrom(report).getBytes(StandardCharsets.UTF_8));
- } catch (IOException e) {
- log.error("Error uploading export report to " + targetUri, e);
- }
- }
-
- /**
- * Encrypt and compress serialized records to temporary file storage
- */
- Flowable encryptAndCompress(DatasetExport e, Flowable serializedRecords) {
- Stopwatch stopwatch = Stopwatch.createStarted();
- Zips.ZipOptions zipOptions = zipOpts()
- .password(e.getCompression().getPassword())
- .encryptionMethod(e.getCompression().getEncryption())
- .build();
- log.debug("Compress and encrypt serialized stream to temporary file. Encryption type: %s, Content name: %s. This can take some time...".formatted(e.getCompression().getEncryption(), e.getTargetContentName()));
- Flowable compressedRecords = Zips.zip(serializedRecords, archiveFilenameOf(e.getTargetContentName(), e.getTargetContentType()), zipOptions);
- log.debug("Compression/encryption done in %s".formatted(stopwatch.stop().elapsed()));
- return compressedRecords;
- }
-
- String targetRootLocationOf(DatasetUri datasetPath) {
- return PathJoiner.joinWithoutLeadingOrTrailingSlash(
- exportConfig.getDefaultTargetRoot(),
- datasetPath.getPath(),
- "" + System.currentTimeMillis()
- );
- }
-
- String targetFileLocationOf(String rootLocation, String filename) {
- return PathJoiner.joinWithoutLeadingOrTrailingSlash(
- rootLocation,
- filename
- );
- }
-
- String archiveFilenameOf(String contentName, MediaType contentType) {
- String timestamp = DateTimeFormatter.ofPattern("yyyyMMdd")
- .withZone(ZoneOffset.UTC)
- .format(Instant.now());
- contentName = ALLOWED_FILENAME_CHARACTERS.retainFrom(contentName);
-
- return "%s-%s.%s".formatted(timestamp, contentName, contentType.getExtension().toLowerCase());
- }
-
- FieldInterceptor initPseudoInterceptor(DatasetExport e, DatasetMeta datasetMeta, DatasetExportReport report) {
- if (!e.getDepseudonymize()) {
- return FieldInterceptor.noOp();
- }
-
- final List pseudoRules;
-
- // Use any explicitly specified pseudo rules
- if (e.getPseudoRules() != null && !e.getPseudoRules().isEmpty()) {
- log.debug("Pseudo rules were explicitly specified");
- pseudoRules = e.getPseudoRules();
- }
-
- // Or try to retrieve pseudo rules from another dataset's metadata file if this has been specified
- else if (e.getPseudoRulesDataset() != null) {
- log.debug("Retrieve pseudo rules from explicitly specified dataset path {}", e.getPseudoRulesDataset());
- PseudoConfig pseudoConfig = datasetMetaService.readDatasetPseudoConfig(e.getPseudoRulesDataset().legacyDatasetUri());
- pseudoRules = pseudoConfig.getRules();
- }
- // Or try to retrieve the pseudo rules from the source dataset's metadata (if present)
- else if (e.getSourceDataset() != null) {
- log.debug("Retrieve pseudo rules from source dataset at {}", e.getSourceDataset());
- PseudoConfig pseudoConfig = datasetMetaService.pseudoConfigOf(datasetMeta);
- pseudoRules = pseudoConfig.getRules();
- }
-
- // Else proceed without any pseudo rules
- else {
- pseudoRules = List.of();
- }
-
- if (e.getDepseudonymize() && pseudoRules.isEmpty()) {
- throw new NoPseudoRulesFoundException("no pseudonymization rules found - unable to depseudonymize dataset %s".formatted(e.getSourceDataset()));
- } else {
- log.info("Pseudo rules: {}", pseudoRules);
- }
-
- report.setAppliedPseudoRules(pseudoRules);
- PseudoFuncs pseudoFuncs = new PseudoFuncs(pseudoRules, pseudoSecrets.resolve(), List.of());
- return new FieldPseudoInterceptor(pseudoFuncs, PseudoOperation.DEPSEUDONYMIZE);
- }
-
- @Data
- @NoArgsConstructor
- @AllArgsConstructor
- public static class DatasetUri {
- @NonNull
- private String root;
- @NonNull
- private String path;
- @NonNull
- private String version;
-
- no.ssb.dapla.dataset.uri.DatasetUri legacyDatasetUri() {
- return no.ssb.dapla.dataset.uri.DatasetUri.of(root, path, version);
- }
- }
-
- @Data
- @Builder
- @Introspected
- static class DatasetExport {
- private String userId;
-
- @NotNull
- private DatasetUri sourceDataset;
- private Set columnSelectors;
-
- @NotNull
- private Compression compression;
-
- private Boolean depseudonymize;
-
- private List pseudoRules;
- private DatasetUri pseudoRulesDataset;
-
- private String targetContentName;
- private MediaType targetContentType;
- }
-
- @Value
- @Builder
- @Accessors(fluent = true)
- public static class ExportEvent {
- @NonNull DatasetUri sourceDataset;
- DatasetMeta sourceDatasetMeta;
- @NonNull DatasetExport datasetExport;
- @NonNull DatasetExportReport report;
- @NonNull String targetRootLocation;
- @NonNull String targetArchiveUri;
- }
-
- @Data
- @Builder
- static class DatasetExportResult {
- private String targetUri;
- }
-
- @Data
- static class DatasetExportReport {
- public DatasetExportReport(DatasetExport e) {
- this.userId = e.getUserId();
- this.exportTimestamp = Instant.now().toString();
- this.depseudonymize = e.getDepseudonymize();
- this.sourceDataset = e.getSourceDataset();
- }
-
- private final DatasetUri sourceDataset;
- private final String userId;
- private final String exportTimestamp;
- private Long elapsedMillis;
- private boolean depseudonymize;
- private List appliedPseudoRules;
- }
-
- public static class ExportServiceException extends RuntimeException {
- public ExportServiceException(String message) {
- super(message);
- }
- }
-
- public static class DatasetNotFoundException extends ExportServiceException {
- public DatasetNotFoundException(String message) {
- super(message);
- }
- }
-
- public static class NoPseudoRulesFoundException extends ExportServiceException {
- public NoPseudoRulesFoundException(String message) {
- super(message);
- }
- }
-
-}
diff --git a/src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoController.java b/src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoController.java
index 216f31a..b02d890 100644
--- a/src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoController.java
+++ b/src/main/java/no/ssb/dlp/pseudo/service/pseudo/PseudoController.java
@@ -22,7 +22,6 @@
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import no.ssb.dapla.dlp.pseudo.func.PseudoFuncFactory;
-import no.ssb.dapla.storage.client.backend.gcs.GoogleCloudStorageBackend;
import no.ssb.dlp.pseudo.core.PseudoOperation;
import no.ssb.dlp.pseudo.core.StreamProcessor;
import no.ssb.dlp.pseudo.core.exception.NoSuchPseudoKeyException;
@@ -64,7 +63,6 @@ public class PseudoController {
private final StreamProcessorFactory streamProcessorFactory;
private final RecordMapProcessorFactory recordProcessorFactory;
- private final GoogleCloudStorageBackend storageBackend;
private final PseudoConfigSplitter pseudoConfigSplitter;
/**
@@ -133,16 +131,6 @@ public HttpResponse pseudonymizeFile(@Schema(implementation = PseudoRe
RecordMapProcessor recordProcessor = recordProcessorFactory.newPseudonymizeRecordProcessor(pseudoConfigs);
ProcessFileResult res = processFile(data, PseudoOperation.PSEUDONYMIZE, recordProcessor, req.getTargetContentType(), req.getCompression());
Flowable file = res.getFlowable();
- if (req.getTargetUri() != null) {
- URI targetUri = req.getTargetUri();
- Completable fileUpload = storageBackend
- .write(targetUri.toString(), file)
- .timeout(30, TimeUnit.SECONDS)
- .doOnError(throwable -> log.error("Upload failed: %s".formatted(targetUri), throwable))
- .doOnComplete(() -> log.info("Successful upload: %s".formatted(targetUri)));
- return HttpResponse.ok(fileUpload.toFlowable());
- }
-
return HttpResponse.ok(file).contentType(res.getTargetContentType());
}
catch (RuntimeException e) {
@@ -192,16 +180,6 @@ public HttpResponse depseudonymizeFile(@Schema(implementation = Pseudo
RecordMapProcessor recordProcessor = recordProcessorFactory.newDepseudonymizeRecordProcessor(pseudoConfigs);
ProcessFileResult res = processFile(data, PseudoOperation.DEPSEUDONYMIZE, recordProcessor, req.getTargetContentType(), req.getCompression());
Flowable file = res.getFlowable();
- if (req.getTargetUri() != null) {
- URI targetUri = req.getTargetUri();
- Completable fileUpload = storageBackend
- .write(targetUri.toString(), file)
- .timeout(30, TimeUnit.SECONDS)
- .doOnError(throwable -> log.error("Upload failed: %s".formatted(targetUri), throwable))
- .doOnComplete(() -> log.info("Successful upload: %s".formatted(targetUri)));
- return HttpResponse.ok(fileUpload.toFlowable());
- }
-
return HttpResponse.ok(file).contentType(res.getTargetContentType());
} catch (Exception e) {
log.error(String.format("Failed to depseudonymize:%nrequest:%n%s", request), e);
@@ -249,16 +227,6 @@ public HttpResponse repseudonymizeFile(@Schema(implementation = Repseu
RecordMapProcessor recordProcessor = recordProcessorFactory.newRepseudonymizeRecordProcessor(req.getSourcePseudoConfig(), req.getTargetPseudoConfig());
ProcessFileResult res = processFile(data, PseudoOperation.REPSEUDONYMIZE, recordProcessor, req.getTargetContentType(), req.getCompression());
Flowable file = res.getFlowable();
- if (req.getTargetUri() != null) {
- URI targetUri = req.getTargetUri();
- Completable fileUpload = storageBackend
- .write(targetUri.toString(), file)
- .timeout(30, TimeUnit.SECONDS)
- .doOnError(throwable -> log.error("Upload failed: %s".formatted(targetUri), throwable))
- .doOnComplete(() -> log.info("Successful upload: %s".formatted(targetUri)));
- return HttpResponse.ok(fileUpload.toFlowable());
- }
-
return HttpResponse.ok(file).contentType(res.getTargetContentType());
} catch (Exception e) {
log.error(String.format("Failed to repseudonymize:%nrequest:%n%s", request), e);
@@ -368,14 +336,6 @@ public static class PseudoRequest {
*/
private PseudoConfig pseudoConfig;
- /**
- * Specify this if you want to stream the result to a specific location such as a GCS bucket. Note that the pseudo
- * service needs to have access to the bucket. Leave this unspecified in order to just stream the result back to
- * the client.
- */
- @Schema(implementation = String.class)
- private URI targetUri;
-
/**
* The content type of the resulting file.
*/
@@ -414,14 +374,6 @@ public static class RepseudoRequest {
*/
private PseudoConfig targetPseudoConfig;
- /**
- * Specify this if you want to stream the result to a specific location such as a GCS bucket. Note that the pseudo
- * service needs to have access to the bucket. Leave this unspecified in order to just stream the result back to
- * the client.
- */
- @Schema(implementation = String.class)
- private URI targetUri;
-
/**
* The content type of the resulting file.
*/
diff --git a/src/main/java/no/ssb/dlp/pseudo/service/storage/DatasetStorageFactory.java b/src/main/java/no/ssb/dlp/pseudo/service/storage/DatasetStorageFactory.java
deleted file mode 100644
index 939bdcc..0000000
--- a/src/main/java/no/ssb/dlp/pseudo/service/storage/DatasetStorageFactory.java
+++ /dev/null
@@ -1,22 +0,0 @@
-package no.ssb.dlp.pseudo.service.storage;
-
-import io.micronaut.context.annotation.Factory;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import no.ssb.dapla.storage.client.DatasetStorage;
-import no.ssb.dapla.storage.client.backend.gcs.GoogleCloudStorageBackend;
-
-import javax.inject.Singleton;
-
-@Factory
-@RequiredArgsConstructor
-@Slf4j
-public class DatasetStorageFactory {
-
- private final GoogleCloudStorageBackend googleCloudStorageBackend;
-
- @Singleton
- public DatasetStorage datasetStorage() {
- return DatasetStorage.builder().withBinaryBackend(googleCloudStorageBackend).build();
- }
-}
diff --git a/src/main/java/no/ssb/dlp/pseudo/service/storage/GoogleCloudStorageBackendFactory.java b/src/main/java/no/ssb/dlp/pseudo/service/storage/GoogleCloudStorageBackendFactory.java
deleted file mode 100644
index a363639..0000000
--- a/src/main/java/no/ssb/dlp/pseudo/service/storage/GoogleCloudStorageBackendFactory.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package no.ssb.dlp.pseudo.service.storage;
-
-import io.micronaut.context.annotation.Factory;
-import io.micronaut.context.annotation.Property;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import no.ssb.dapla.storage.client.backend.gcs.GoogleCloudStorageBackend;
-import no.ssb.dapla.storage.client.backend.gcs.GoogleCloudStorageBackend.Configuration;
-
-import javax.annotation.Nullable;
-import javax.inject.Singleton;
-import java.nio.file.Files;
-import java.nio.file.Path;
-
-@Factory
-@RequiredArgsConstructor
-@Slf4j
-public class GoogleCloudStorageBackendFactory {
-
- @Nullable
- @Property(name = "storage.gcs-service-account-file")
- private Path serviceAccountCredentials;
-
- @Singleton
- public GoogleCloudStorageBackend googleCloudStorageBackend() {
- Configuration configuration = new Configuration();
- if (serviceAccountCredentials != null) {
- if (Files.notExists(serviceAccountCredentials)) {
- throw new StorageBackendException("Could not find service account credentials: " + serviceAccountCredentials);
- }
- configuration.setServiceAccountCredentials(serviceAccountCredentials);
- }
- return new GoogleCloudStorageBackend(configuration);
- }
-
- public static class StorageBackendException extends RuntimeException {
- public StorageBackendException(String message) {
- super(message);
- }
- }
-
-}