diff --git a/README.md b/README.md index 27b1f15..aeef92f 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ * [BigQuery Snapshoter](#bigquery-snapshoter) * [GCS Snapshoter](#gcs-snapshoter) * [Tagger](#tagger) - * [Design Notes](#design-notes) + * [Design Notes](#design-notes) * [Assumptions](#assumptions) * [Deployment](#deployment) * [Install Maven](#install-maven) @@ -135,16 +135,16 @@ A cloud scheduler is used to send a BigQuery “Scan Scope” to the dispatcher The solution uses multiple steps, as explained above, to list, backup and tag tables. These steps are designed so that each one of them is responsible for doing one main task and allow for checkpointing to handle retries in a cost-efficient way, and using PubSub to decouple these steps from each other. For example, if a table is backed up in a single run, via the Snpashoter step, -and then the Tagger step fails to update its backup policy (i.e. last_update_at field) +and then the Tagger step fails to update its backup policy (i.e. last_update_at field) due to temporarily GCP API quotas/limits with the underlying service, the request will not be acknowledged to PubSub, and it will be retried again with exponential backoff. In this case only the tagging logic will be retried and not the entire backup operation -that was already successful. +that was already successful. These steps are implemented in separate Cloud Run services, vs one service with multiple endpoints, to allow fine grain control on the concurrency, CPU, Memory and timeout settings for each step. This is especially useful since each step could use different settings depending on its traffic pattern and processing requirements. For example: * The Dispatcher executes once per run while the other services executes once per table; meaning they could use different concurrency settings (i.e. number of container, requests per container). -Another + Another * The Dispatcher uses relatively higher memory since it's listing all tables in the scan scope which could be in thousands * The Dispatcher and GCS Snapshoter needs relatively longer time to finish compared to the BQ Snapshoter. They could use different timeout settings to avoid un-wanted retries by PubSub diff --git a/scripts/prepare_backup_storage_projects.sh b/scripts/prepare_backup_storage_projects.sh index 5ad167d..bdf10ce 100755 --- a/scripts/prepare_backup_storage_projects.sh +++ b/scripts/prepare_backup_storage_projects.sh @@ -35,9 +35,9 @@ do echo "Preparing GCS Snapshoter SA permissions on backup storage project ${project} .." # GCS Snapshoter needs to write to GCS - # permission: storage.objects.create + # permission: storage.objects.create, storage.objects.delete gcloud projects add-iam-policy-binding "${project}" \ --member="serviceAccount:${SA_SNAPSHOTER_GCS_EMAIL}" \ - --role="roles/storage.objectCreator" + --role="roles/storage.objectAdmin" done diff --git a/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/ConfiguratorController.java b/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/ConfiguratorController.java index 1cac659..e63ea60 100644 --- a/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/ConfiguratorController.java +++ b/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/ConfiguratorController.java @@ -19,18 +19,17 @@ import com.google.cloud.Tuple; import com.google.cloud.pso.bq_snapshot_manager.entities.NonRetryableApplicationException; import com.google.cloud.pso.bq_snapshot_manager.entities.PubSubEvent; -import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.FallbackBackupPolicy; import com.google.cloud.pso.bq_snapshot_manager.functions.f02_configurator.Configurator; import com.google.cloud.pso.bq_snapshot_manager.functions.f02_configurator.ConfiguratorRequest; import com.google.cloud.pso.bq_snapshot_manager.functions.f02_configurator.ConfiguratorResponse; -import com.google.cloud.pso.bq_snapshot_manager.functions.f04_tagger.TaggerRequest; -import com.google.cloud.pso.bq_snapshot_manager.functions.f04_tagger.TaggerResponse; import com.google.cloud.pso.bq_snapshot_manager.helpers.ControllerExceptionHelper; import com.google.cloud.pso.bq_snapshot_manager.helpers.LoggingHelper; import com.google.cloud.pso.bq_snapshot_manager.helpers.TrackingHelper; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyServiceGCSImpl; import com.google.cloud.pso.bq_snapshot_manager.services.bq.BigQueryServiceImpl; -import com.google.cloud.pso.bq_snapshot_manager.services.catalog.DataCatalogServiceImpl; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyService; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyServiceFireStoreImpl; import com.google.cloud.pso.bq_snapshot_manager.services.pubsub.PubSubServiceImpl; import com.google.cloud.pso.bq_snapshot_manager.services.set.GCSPersistentSetImpl; import com.google.gson.Gson; @@ -63,7 +62,8 @@ public ConfiguratorController() throws JsonProcessingException { logger = new LoggingHelper( ConfiguratorController.class.getSimpleName(), functionNumber, - environment.getProjectId() + environment.getProjectId(), + environment.getApplicationName() ); logger.logInfoWithTracker( @@ -88,7 +88,7 @@ public ConfiguratorController() throws JsonProcessingException { public ResponseEntity receiveMessage(@RequestBody PubSubEvent requestBody) { - DataCatalogServiceImpl dataCatalogService = null; + BackupPolicyService backupPolicyService = null; // These values will be updated based on the execution flow and logged at the end ResponseEntity responseEntity; @@ -119,12 +119,12 @@ public ResponseEntity receiveMessage(@RequestBody PubSubEvent requestBody) { logger.logInfoWithTracker(configuratorRequest.isDryRun(), trackingId, configuratorRequest.getTargetTable(), String.format("Parsed Request: %s", configuratorRequest.toString())); - dataCatalogService = new DataCatalogServiceImpl(); + backupPolicyService = new BackupPolicyServiceGCSImpl(environment.getGcsBackupPoliciesBucket()); Configurator configurator = new Configurator( environment.toConfig(), new BigQueryServiceImpl(configuratorRequest.getTargetTable().getProject()), - dataCatalogService, + backupPolicyService, new PubSubServiceImpl(), new GCSPersistentSetImpl(environment.getGcsFlagsBucket()), fallbackBackupPolicy, @@ -152,8 +152,8 @@ public ResponseEntity receiveMessage(@RequestBody PubSubEvent requestBody) { } finally { - if (dataCatalogService != null) { - dataCatalogService.shutdown(); + if (backupPolicyService != null) { + backupPolicyService.shutdown(); } } diff --git a/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/Environment.java b/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/Environment.java index e419764..c488560 100644 --- a/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/Environment.java +++ b/services/configurator-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/configurator/Environment.java @@ -29,7 +29,8 @@ public ConfiguratorConfig toConfig (){ getProjectId(), getBqSnapshoterOutputTopic(), getGCSSnapshoterOutputTopic(), - getBackupTagTemplateId() + getBackupTagTemplateId(), + getApplicationName() ); } @@ -58,4 +59,13 @@ public String getBackupPolicyJson(){ public String getGcsFlagsBucket(){ return Utils.getConfigFromEnv("GCS_FLAGS_BUCKET", true); } + + public String getApplicationName(){ + return Utils.getConfigFromEnv("APPLICATION_NAME", true); + } + + public String getGcsBackupPoliciesBucket(){ + return Utils.getConfigFromEnv("GCS_BACKUP_POLICIES_BUCKET", true); + } + } diff --git a/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/DispatcherController.java b/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/DispatcherController.java index 98306f5..2466bc8 100644 --- a/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/DispatcherController.java +++ b/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/DispatcherController.java @@ -56,7 +56,9 @@ public DispatcherController() { logger = new LoggingHelper( DispatcherController.class.getSimpleName(), functionNumber, - environment.getProjectId()); + environment.getProjectId(), + environment.getApplicationName() + ); } @RequestMapping(value = "/", method = RequestMethod.POST) diff --git a/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/Environment.java b/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/Environment.java index f524c14..075f6a6 100644 --- a/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/Environment.java +++ b/services/dispatcher-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/dispatcher/Environment.java @@ -26,7 +26,8 @@ public DispatcherConfig toConfig(){ getProjectId(), getComputeRegionId(), getDataRegionId(), - getOutputTopic() + getOutputTopic(), + getApplicationName() ); } @@ -47,4 +48,8 @@ public String getDataRegionId(){ public String getGcsFlagsBucket(){ return Utils.getConfigFromEnv("GCS_FLAGS_BUCKET", true); } + + public String getApplicationName(){ + return Utils.getConfigFromEnv("APPLICATION_NAME", true); + } } diff --git a/services/library/pom.xml b/services/library/pom.xml index fa23963..fa22c14 100644 --- a/services/library/pom.xml +++ b/services/library/pom.xml @@ -117,6 +117,11 @@ google-cloud-datacatalog + + com.google.cloud + google-cloud-datastore + + com.google.apis google-api-services-cloudresourcemanager diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/Globals.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/Globals.java deleted file mode 100644 index 4c50492..0000000 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/Globals.java +++ /dev/null @@ -1,6 +0,0 @@ -package com.google.cloud.pso.bq_snapshot_manager.entities; - -public class Globals { - public final static String APPLICATION_NAME = "bq_backup_manager"; - -} diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/TableSpec.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/TableSpec.java index 156803c..1e71426 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/TableSpec.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/TableSpec.java @@ -109,4 +109,12 @@ public String toResourceUrl(){ getTable() ); } + + public String toHivePartitionPostfix(){ + return String.format("project=%s/dataset=%s/table=%s", + project, + dataset, + table + ); + } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicy.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicy.java index a778fb3..dbc812c 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicy.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicy.java @@ -3,42 +3,64 @@ import com.google.cloud.Timestamp; import com.google.cloud.datacatalog.v1.Tag; import com.google.cloud.datacatalog.v1.TagField; +import com.google.cloud.datastore.Entity; +import com.google.cloud.datastore.FullEntity; +import com.google.cloud.datastore.IncompleteKey; +import com.google.cloud.datastore.Key; +import com.google.cloud.pso.bq_snapshot_manager.functions.f02_configurator.ConfiguratorRequest; import com.google.cloud.pso.bq_snapshot_manager.helpers.Utils; import com.google.common.base.Objects; import com.google.gson.Gson; +import com.google.gson.annotations.SerializedName; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; public class BackupPolicy { + @SerializedName("backup_cron") private final String cron; + @SerializedName("backup_method") private final BackupMethod method; + @SerializedName("backup_time_travel_offset_days") private final TimeTravelOffsetDays timeTravelOffsetDays; - + @SerializedName("bq_snapshot_expiration_days") private final Double bigQuerySnapshotExpirationDays; + @SerializedName("backup_storage_project") private final String backupStorageProject; + @SerializedName("backup_operation_project") private final String backupOperationProject; + @SerializedName("bq_snapshot_storage_dataset") private final String bigQuerySnapshotStorageDataset; + @SerializedName("gcs_snapshot_storage_location") private final String gcsSnapshotStorageLocation; + @SerializedName("gcs_snapshot_format") private final GCSSnapshotFormat gcsExportFormat; + @SerializedName("gcs_csv_delimiter") private final String gcsCsvDelimiter; + @SerializedName("gcs_csv_export_header") private final Boolean gcsCsvExportHeader; + @SerializedName("gcs_avro_use_logical_types") private final Boolean gcsUseAvroLogicalTypes; + @SerializedName("config_source") private final BackupConfigSource configSource; + @SerializedName("last_backup_at") private Timestamp lastBackupAt; + @SerializedName("last_gcs_snapshot_storage_uri") private String lastBqSnapshotStorageUri; + @SerializedName("last_bq_snapshot_storage_uri") private String lastGcsSnapshotStorageUri; public BackupPolicy(BackupPolicyBuilder builder) throws IllegalArgumentException { - List missingFields = validate(builder); + List missingFields = validate(builder); if (!missingFields.isEmpty()) { throw new IllegalArgumentException( @@ -67,65 +89,65 @@ public BackupPolicy(BackupPolicyBuilder builder) throws IllegalArgumentException this.lastGcsSnapshotStorageUri = builder.lastGcsSnapshotStorageUri; } - public static List validate(BackupPolicyBuilder builder){ + public static List validate(BackupPolicyBuilder builder) { // validate that all required fields are provided depending on the backup method - List missingRequired = new ArrayList<>(); - List missingOptional = new ArrayList<>(); + List missingRequired = new ArrayList<>(); + List missingOptional = new ArrayList<>(); if (builder.cron == null) { - missingRequired.add(DataCatalogBackupPolicyTagFields.backup_cron); + missingRequired.add(BackupPolicyFields.backup_cron); } if (builder.method == null) { - missingRequired.add(DataCatalogBackupPolicyTagFields.backup_method); + missingRequired.add(BackupPolicyFields.backup_method); } if (builder.timeTravelOffsetDays == null) { - missingRequired.add(DataCatalogBackupPolicyTagFields.backup_time_travel_offset_days); + missingRequired.add(BackupPolicyFields.backup_time_travel_offset_days); } if (builder.configSource == null) { - missingRequired.add(DataCatalogBackupPolicyTagFields.config_source); + missingRequired.add(BackupPolicyFields.config_source); } if (builder.backupStorageProject == null) { - missingRequired.add(DataCatalogBackupPolicyTagFields.backup_storage_project); + missingRequired.add(BackupPolicyFields.backup_storage_project); } // if required params are missing return and don't continue with other checks - if(!missingRequired.isEmpty()){ + if (!missingRequired.isEmpty()) { return missingRequired; } if (builder.method.equals(BackupMethod.BIGQUERY_SNAPSHOT) || builder.method.equals(BackupMethod.BOTH)) { if (builder.bigQuerySnapshotStorageDataset == null) { - missingOptional.add(DataCatalogBackupPolicyTagFields.bq_snapshot_storage_dataset); + missingOptional.add(BackupPolicyFields.bq_snapshot_storage_dataset); } if (builder.bigQuerySnapshotExpirationDays == null) { - missingOptional.add(DataCatalogBackupPolicyTagFields.bq_snapshot_expiration_days); + missingOptional.add(BackupPolicyFields.bq_snapshot_expiration_days); } } // check for GCS params - if (builder.method.equals(BackupMethod.GCS_SNAPSHOT) ||builder. method.equals(BackupMethod.BOTH)) { + if (builder.method.equals(BackupMethod.GCS_SNAPSHOT) || builder.method.equals(BackupMethod.BOTH)) { String format = builder.gcsExportFormat == null ? null : GCSSnapshotFormat.getFormatAndCompression(builder.gcsExportFormat).x(); if (builder.gcsExportFormat == null) { - missingOptional.add(DataCatalogBackupPolicyTagFields.gcs_snapshot_format); + missingOptional.add(BackupPolicyFields.gcs_snapshot_format); } if (builder.gcsSnapshotStorageLocation == null) { - missingOptional.add(DataCatalogBackupPolicyTagFields.gcs_snapshot_storage_location); + missingOptional.add(BackupPolicyFields.gcs_snapshot_storage_location); } // check required fields for CSV exports - if(format != null && format.equals("CSV")){ - if(builder.gcsCsvDelimiter == null){ - missingOptional.add(DataCatalogBackupPolicyTagFields.gcs_csv_delimiter); + if (format != null && format.equals("CSV")) { + if (builder.gcsCsvDelimiter == null) { + missingOptional.add(BackupPolicyFields.gcs_csv_delimiter); } - if(builder.gcsCsvExportHeader == null){ - missingOptional.add(DataCatalogBackupPolicyTagFields.gcs_csv_export_header); + if (builder.gcsCsvExportHeader == null) { + missingOptional.add(BackupPolicyFields.gcs_csv_export_header); } } // check for required fields for avro export - if(format != null && format.equals("AVRO")){ - if(builder.gcsUseAvroLogicalTypes == null){ - missingOptional.add(DataCatalogBackupPolicyTagFields.gcs_avro_use_logical_types); + if (format != null && format.equals("AVRO")) { + if (builder.gcsUseAvroLogicalTypes == null) { + missingOptional.add(BackupPolicyFields.gcs_avro_use_logical_types); } } } @@ -206,6 +228,9 @@ public static BackupPolicy fromJson(String jsonStr) { return fromMap(jsonMap); } + public String toJson(){ + return new Gson().toJson(toMap()); + } @Override public String toString() { return "BackupPolicy{" + @@ -255,20 +280,20 @@ public Tag toDataCatalogTag(String tagTemplateId, String tagName) { .setTemplate(tagTemplateId); // required: cron - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.backup_cron.toString(), + tagBuilder.putFields(BackupPolicyFields.backup_cron.toString(), TagField.newBuilder().setStringValue(cron).build() - ); + ); // required: backup method - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.backup_method.toString(), + tagBuilder.putFields(BackupPolicyFields.backup_method.toString(), TagField.newBuilder().setEnumValue( - TagField.EnumValue.newBuilder().setDisplayName( - method.getText()) - .build()) - .build()); + TagField.EnumValue.newBuilder().setDisplayName( + method.getText()) + .build()) + .build()); // required: time travel - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.backup_time_travel_offset_days.toString(), + tagBuilder.putFields(BackupPolicyFields.backup_time_travel_offset_days.toString(), TagField.newBuilder().setEnumValue( TagField.EnumValue.newBuilder().setDisplayName( timeTravelOffsetDays.getText()) @@ -277,36 +302,36 @@ public Tag toDataCatalogTag(String tagTemplateId, String tagName) { //required: backup storage project TagField backupStorageProjectField = TagField.newBuilder().setStringValue(backupStorageProject).build(); - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.backup_storage_project.toString(), backupStorageProjectField); + tagBuilder.putFields(BackupPolicyFields.backup_storage_project.toString(), backupStorageProjectField); //optional: backup operation project - if(backupOperationProject != null){ - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.backup_operation_project.toString(), + if (backupOperationProject != null) { + tagBuilder.putFields(BackupPolicyFields.backup_operation_project.toString(), TagField.newBuilder().setStringValue(backupOperationProject).build()); } // optional: bq snapshot expiration if (bigQuerySnapshotExpirationDays != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.bq_snapshot_expiration_days.toString(), + tagBuilder.putFields(BackupPolicyFields.bq_snapshot_expiration_days.toString(), TagField.newBuilder().setDoubleValue(bigQuerySnapshotExpirationDays).build() - ); + ); } // optional: bq snapshot dataset if (bigQuerySnapshotStorageDataset != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.bq_snapshot_storage_dataset.toString(), + tagBuilder.putFields(BackupPolicyFields.bq_snapshot_storage_dataset.toString(), TagField.newBuilder().setStringValue(bigQuerySnapshotStorageDataset).build()); } // optional: gcs snapshot storage location if (gcsSnapshotStorageLocation != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.gcs_snapshot_storage_location.toString(), + tagBuilder.putFields(BackupPolicyFields.gcs_snapshot_storage_location.toString(), TagField.newBuilder().setStringValue(gcsSnapshotStorageLocation).build()); } // optional: gcs export format if (gcsExportFormat != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.gcs_snapshot_format.toString(), + tagBuilder.putFields(BackupPolicyFields.gcs_snapshot_format.toString(), TagField.newBuilder().setEnumValue( TagField.EnumValue.newBuilder().setDisplayName( gcsExportFormat.toString()) @@ -317,36 +342,36 @@ public Tag toDataCatalogTag(String tagTemplateId, String tagName) { // optional: gcs csv field delimiter if (gcsCsvDelimiter != null) { tagBuilder.putFields( - DataCatalogBackupPolicyTagFields.gcs_csv_delimiter.toString(), + BackupPolicyFields.gcs_csv_delimiter.toString(), TagField.newBuilder().setStringValue(gcsCsvDelimiter).build()); } // optional: gcs_csv_export_header if (gcsCsvExportHeader != null) { tagBuilder.putFields( - DataCatalogBackupPolicyTagFields.gcs_csv_export_header.toString(), + BackupPolicyFields.gcs_csv_export_header.toString(), TagField.newBuilder().setBoolValue(gcsCsvExportHeader).build()); } // optional: gcs csv field delimiter if (gcsUseAvroLogicalTypes != null) { tagBuilder.putFields( - DataCatalogBackupPolicyTagFields.gcs_avro_use_logical_types.toString(), + BackupPolicyFields.gcs_avro_use_logical_types.toString(), TagField.newBuilder().setBoolValue(gcsUseAvroLogicalTypes).build()); } // required: config source - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.config_source.toString(), + tagBuilder.putFields(BackupPolicyFields.config_source.toString(), TagField.newBuilder().setEnumValue( - TagField.EnumValue.newBuilder().setDisplayName( - configSource.toString()) - .build()) - .build()); + TagField.EnumValue.newBuilder().setDisplayName( + configSource.toString()) + .build()) + .build()); // optional: last backup at if (lastBackupAt != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.last_backup_at.toString(), + tagBuilder.putFields(BackupPolicyFields.last_backup_at.toString(), TagField.newBuilder().setTimestampValue( com.google.protobuf.Timestamp.newBuilder() .setSeconds(lastBackupAt.getSeconds()) @@ -358,13 +383,13 @@ public Tag toDataCatalogTag(String tagTemplateId, String tagName) { // optional: last bq snapshot uri if (lastBqSnapshotStorageUri != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.last_bq_snapshot_storage_uri.toString(), + tagBuilder.putFields(BackupPolicyFields.last_bq_snapshot_storage_uri.toString(), TagField.newBuilder().setStringValue(lastBqSnapshotStorageUri).build()); } // optional: last gcs snapshot uri if (lastGcsSnapshotStorageUri != null) { - tagBuilder.putFields(DataCatalogBackupPolicyTagFields.last_gcs_snapshot_storage_uri.toString(), + tagBuilder.putFields(BackupPolicyFields.last_gcs_snapshot_storage_uri.toString(), TagField.newBuilder().setStringValue(lastGcsSnapshotStorageUri).build()); } @@ -375,27 +400,142 @@ public Tag toDataCatalogTag(String tagTemplateId, String tagName) { return tagBuilder.build(); } + public Entity toFireStoreEntity(Key entityKey) { + + // build required fields + Entity.Builder builder = Entity.newBuilder(entityKey) + .set(BackupPolicyFields.backup_cron.toString(), this.getCron()) + .set(BackupPolicyFields.backup_method.toString(), this.getMethod().getText()) + .set(BackupPolicyFields.backup_time_travel_offset_days.toString(), this.getTimeTravelOffsetDays().getText()) + .set(BackupPolicyFields.config_source.toString(), this.getConfigSource().getText()) + .set(BackupPolicyFields.backup_storage_project.toString(), this.getBackupStorageProject()); + + if(this.getBackupOperationProject() != null){ + builder.set(BackupPolicyFields.backup_operation_project.toString(), this.getBackupOperationProject()); + } + + if(this.getBigQuerySnapshotExpirationDays() != null){ + builder.set(BackupPolicyFields.bq_snapshot_expiration_days.toString(), this.getBigQuerySnapshotExpirationDays()); + } + + if(this.getBigQuerySnapshotStorageDataset() != null){ + builder.set(BackupPolicyFields.bq_snapshot_storage_dataset.toString(), this.getBigQuerySnapshotStorageDataset()); + } + + if(this.getGcsSnapshotStorageLocation() != null){ + builder.set(BackupPolicyFields.gcs_snapshot_storage_location.toString(), this.getGcsSnapshotStorageLocation()); + } + + if(this.getGcsExportFormat() != null){ + builder.set(BackupPolicyFields.gcs_snapshot_format.toString(), this.getGcsExportFormat().toString()); + } + + if(this.getGcsCsvDelimiter() != null){ + builder.set(BackupPolicyFields.gcs_csv_delimiter.toString(), this.getGcsCsvDelimiter()); + } + + if(this.getGcsCsvExportHeader() != null){ + builder.set(BackupPolicyFields.gcs_csv_export_header.toString(), this.getGcsCsvExportHeader()); + } + + if(this.getGcsUseAvroLogicalTypes() != null){ + builder.set(BackupPolicyFields.gcs_avro_use_logical_types.toString(), this.getGcsUseAvroLogicalTypes()); + } + + if(this.getLastBackupAt() != null){ + builder.set(BackupPolicyFields.last_backup_at.toString(), this.getLastBackupAt()); + } + + if(this.getLastBqSnapshotStorageUri() != null){ + builder.set(BackupPolicyFields.last_bq_snapshot_storage_uri.toString(), this.getLastBqSnapshotStorageUri()); + } + + if(this.getLastGcsSnapshotStorageUri() != null){ + builder.set(BackupPolicyFields.last_gcs_snapshot_storage_uri.toString(), this.getLastGcsSnapshotStorageUri()); + } + + return builder.build(); + } + + public Map toMap(){ + Map fields = new HashMap<>(); + // required fields + fields.put(BackupPolicyFields.backup_cron.toString(), cron); + fields.put(BackupPolicyFields.backup_method.toString(), method.getText()); + fields.put(BackupPolicyFields.backup_time_travel_offset_days.toString(), timeTravelOffsetDays.getText()); + fields.put(BackupPolicyFields.backup_storage_project.toString(), backupStorageProject); + fields.put(BackupPolicyFields.config_source.toString(), configSource.getText()); + fields.put(BackupPolicyFields.backup_storage_project.toString(), backupStorageProject); + fields.put(BackupPolicyFields.config_source.toString(), configSource.getText()); + + // optional fields + if(this.getBackupOperationProject() != null){ + fields.put(BackupPolicyFields.backup_operation_project.toString(), this.getBackupOperationProject()); + } + + if(this.getBigQuerySnapshotExpirationDays() != null){ + fields.put(BackupPolicyFields.bq_snapshot_expiration_days.toString(), this.getBigQuerySnapshotExpirationDays().toString()); + } + + if(this.getBigQuerySnapshotStorageDataset() != null){ + fields.put(BackupPolicyFields.bq_snapshot_storage_dataset.toString(), this.getBigQuerySnapshotStorageDataset()); + } + + if(this.getGcsSnapshotStorageLocation() != null){ + fields.put(BackupPolicyFields.gcs_snapshot_storage_location.toString(), this.getGcsSnapshotStorageLocation()); + } + + if(this.getGcsExportFormat() != null){ + fields.put(BackupPolicyFields.gcs_snapshot_format.toString(), this.getGcsExportFormat().toString()); + } + + if(this.getGcsCsvDelimiter() != null){ + fields.put(BackupPolicyFields.gcs_csv_delimiter.toString(), this.getGcsCsvDelimiter()); + } + + if(this.getGcsCsvExportHeader() != null){ + fields.put(BackupPolicyFields.gcs_csv_export_header.toString(), this.getGcsCsvExportHeader().toString()); + } + + if(this.getGcsUseAvroLogicalTypes() != null){ + fields.put(BackupPolicyFields.gcs_avro_use_logical_types.toString(), this.getGcsUseAvroLogicalTypes().toString()); + } + + if(this.getLastBackupAt() != null){ + fields.put(BackupPolicyFields.last_backup_at.toString(), this.getLastBackupAt().toString()); + } + + if(this.getLastBqSnapshotStorageUri() != null){ + fields.put(BackupPolicyFields.last_bq_snapshot_storage_uri.toString(), this.getLastBqSnapshotStorageUri()); + } + + if(this.getLastGcsSnapshotStorageUri() != null){ + fields.put(BackupPolicyFields.last_gcs_snapshot_storage_uri.toString(), this.getLastGcsSnapshotStorageUri()); + } + + return fields; + } // used to parse from json map (fallback policies) or data catalog tags (backup policies) - public static BackupPolicy fromMap(Map tagTemplate) throws IllegalArgumentException { + public static BackupPolicy fromMap(Map fieldsMap) throws IllegalArgumentException { // parse required fields - String cron = Utils.getOrFail(tagTemplate, DataCatalogBackupPolicyTagFields.backup_cron.toString()); + String cron = Utils.getOrFail(fieldsMap, BackupPolicyFields.backup_cron.toString()); BackupMethod method = BackupMethod.fromString( - Utils.getOrFail(tagTemplate, DataCatalogBackupPolicyTagFields.backup_method.toString()) + Utils.getOrFail(fieldsMap, BackupPolicyFields.backup_method.toString()) ); TimeTravelOffsetDays timeTravelOffsetDays = TimeTravelOffsetDays.fromString( - Utils.getOrFail(tagTemplate, DataCatalogBackupPolicyTagFields.backup_time_travel_offset_days.toString()) + Utils.getOrFail(fieldsMap, BackupPolicyFields.backup_time_travel_offset_days.toString()) ); - String backupStorageProject = Utils.getOrFail(tagTemplate, DataCatalogBackupPolicyTagFields.backup_storage_project.toString()); + String backupStorageProject = Utils.getOrFail(fieldsMap, BackupPolicyFields.backup_storage_project.toString()); // config source is not required in the fallback policies. It defaults to SYSTEM if not present - String configSourceStr = tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.config_source.toString(), + String configSourceStr = fieldsMap.getOrDefault( + BackupPolicyFields.config_source.toString(), null); BackupConfigSource configSource = configSourceStr == null ? BackupConfigSource.SYSTEM : BackupConfigSource.fromString(configSourceStr); @@ -406,83 +546,83 @@ public static BackupPolicy fromMap(Map tagTemplate) throws Illeg // these fields might not exist in the attached tag template if not filled. Same for fallback policies backupPolicyBuilder.setBackupOperationProject( - tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.backup_operation_project.toString(), + fieldsMap.getOrDefault( + BackupPolicyFields.backup_operation_project.toString(), null ) ); backupPolicyBuilder.setBigQuerySnapshotStorageDataset( - tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.bq_snapshot_storage_dataset.toString(), + fieldsMap.getOrDefault( + BackupPolicyFields.bq_snapshot_storage_dataset.toString(), null) ); - String bqSnapshotExpirationDaysStr = tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.bq_snapshot_expiration_days.toString(), + String bqSnapshotExpirationDaysStr = fieldsMap.getOrDefault( + BackupPolicyFields.bq_snapshot_expiration_days.toString(), null); backupPolicyBuilder.setBigQuerySnapshotExpirationDays( - bqSnapshotExpirationDaysStr == null? null: Double.parseDouble(bqSnapshotExpirationDaysStr) + bqSnapshotExpirationDaysStr == null ? null : Double.parseDouble(bqSnapshotExpirationDaysStr) ); // parse optional GCS snapshot settings backupPolicyBuilder.setGcsSnapshotStorageLocation( - tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.gcs_snapshot_storage_location.toString(), + fieldsMap.getOrDefault( + BackupPolicyFields.gcs_snapshot_storage_location.toString(), null) ); - String gcsSnapshotFormatStr = tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.gcs_snapshot_format.toString(), + String gcsSnapshotFormatStr = fieldsMap.getOrDefault( + BackupPolicyFields.gcs_snapshot_format.toString(), null); backupPolicyBuilder.setGcsExportFormat( - gcsSnapshotFormatStr == null? null: GCSSnapshotFormat.valueOf(gcsSnapshotFormatStr) + gcsSnapshotFormatStr == null ? null : GCSSnapshotFormat.valueOf(gcsSnapshotFormatStr) ); backupPolicyBuilder.setGcsCsvDelimiter( - tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.gcs_csv_delimiter.toString(), + fieldsMap.getOrDefault( + BackupPolicyFields.gcs_csv_delimiter.toString(), null) ); // if optional boolean values are not provided, set them to null and not false - String gcsCsvExportHeaderStr = tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.gcs_csv_export_header.toString(), + String gcsCsvExportHeaderStr = fieldsMap.getOrDefault( + BackupPolicyFields.gcs_csv_export_header.toString(), null); backupPolicyBuilder.setGcsCsvExportHeader( - gcsCsvExportHeaderStr == null? null: Boolean.valueOf(gcsCsvExportHeaderStr) + gcsCsvExportHeaderStr == null ? null : Boolean.valueOf(gcsCsvExportHeaderStr) ); // if optional boolean values are not provided, set them to null and not false - String gcsAvroUseLogicalAvroTypeStr = tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.gcs_avro_use_logical_types.toString(), + String gcsAvroUseLogicalAvroTypeStr = fieldsMap.getOrDefault( + BackupPolicyFields.gcs_avro_use_logical_types.toString(), null); backupPolicyBuilder.setGcsUseAvroLogicalTypes( - gcsAvroUseLogicalAvroTypeStr == null? null: Boolean.valueOf(gcsAvroUseLogicalAvroTypeStr) + gcsAvroUseLogicalAvroTypeStr == null ? null : Boolean.valueOf(gcsAvroUseLogicalAvroTypeStr) ); - String lastBackupAtStr = tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.last_backup_at.toString(), + String lastBackupAtStr = fieldsMap.getOrDefault( + BackupPolicyFields.last_backup_at.toString(), null); backupPolicyBuilder.setLastBackupAt( lastBackupAtStr == null ? null : Timestamp.parseTimestamp(lastBackupAtStr) ); backupPolicyBuilder.setLastBqSnapshotStorageUri( - tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.last_bq_snapshot_storage_uri.toString(), + fieldsMap.getOrDefault( + BackupPolicyFields.last_bq_snapshot_storage_uri.toString(), null) ); - backupPolicyBuilder.setLastGcsSnapshotStorageUri(tagTemplate.getOrDefault( - DataCatalogBackupPolicyTagFields.last_gcs_snapshot_storage_uri.toString(), + backupPolicyBuilder.setLastGcsSnapshotStorageUri(fieldsMap.getOrDefault( + BackupPolicyFields.last_gcs_snapshot_storage_uri.toString(), null)); return backupPolicyBuilder.build(); } - public static class BackupPolicyBuilder{ + public static class BackupPolicyBuilder { // required private String cron; @@ -508,7 +648,7 @@ public static class BackupPolicyBuilder{ private String lastBqSnapshotStorageUri; private String lastGcsSnapshotStorageUri; - public static BackupPolicyBuilder from(BackupPolicy backupPolicy){ + public static BackupPolicyBuilder from(BackupPolicy backupPolicy) { return new BackupPolicyBuilder( backupPolicy.cron, backupPolicy.method, @@ -528,7 +668,7 @@ public static BackupPolicyBuilder from(BackupPolicy backupPolicy){ .setLastGcsSnapshotStorageUri(backupPolicy.lastGcsSnapshotStorageUri); } - public BackupPolicyBuilder(String cron, BackupMethod method, TimeTravelOffsetDays timeTravelOffsetDays, BackupConfigSource configSource, String backupStorageProject){ + public BackupPolicyBuilder(String cron, BackupMethod method, TimeTravelOffsetDays timeTravelOffsetDays, BackupConfigSource configSource, String backupStorageProject) { this.cron = cron; this.method = method; this.timeTravelOffsetDays = timeTravelOffsetDays; @@ -551,7 +691,7 @@ public BackupPolicyBuilder setLastGcsSnapshotStorageUri(String lastGcsSnapshotSt return this; } - public BackupPolicyBuilder setBackupOperationProject(String backupOperationProject) { + public BackupPolicyBuilder setBackupOperationProject(String backupOperationProject) { this.backupOperationProject = backupOperationProject; return this; } @@ -616,7 +756,7 @@ public BackupPolicyBuilder setGcsUseAvroLogicalTypes(Boolean gcsUseAvroLogicalTy return this; } - public BackupPolicy build(){ + public BackupPolicy build() { return new BackupPolicy(this); } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/DataCatalogBackupPolicyTagFields.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicyFields.java similarity index 82% rename from services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/DataCatalogBackupPolicyTagFields.java rename to services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicyFields.java index 2cdb1df..2441781 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/DataCatalogBackupPolicyTagFields.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/entities/backup_policy/BackupPolicyFields.java @@ -1,6 +1,7 @@ package com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy; -public enum DataCatalogBackupPolicyTagFields { +// Field names used to represent a backup policy on external storage layer(s) +public enum BackupPolicyFields { backup_cron, backup_method, backup_time_travel_offset_days, diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/Dispatcher.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/Dispatcher.java index 5239e83..07ef336 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/Dispatcher.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/Dispatcher.java @@ -66,7 +66,8 @@ public Dispatcher(DispatcherConfig config, logger = new LoggingHelper( Dispatcher.class.getSimpleName(), functionNumber, - config.getProjectId() + config.getProjectId(), + config.getApplicationName() ); } @@ -96,7 +97,8 @@ public PubSubPublishResults execute(DispatcherRequest dispatcherRequest, String new LoggingHelper( BigQueryScopeLister.class.getSimpleName(), functionNumber, - config.getProjectId() + config.getProjectId(), + config.getApplicationName() ), runId ); diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherConfig.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherConfig.java index b709a01..0c66af1 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherConfig.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherConfig.java @@ -23,15 +23,19 @@ public class DispatcherConfig { private String dataRegionId; private String outputTopic; + private String applicationName; + public DispatcherConfig(String projectId, String computeRegionId, String dataRegionId, - String outputTopic + String outputTopic, + String applicationName ) { this.projectId = projectId; this.computeRegionId = computeRegionId; this.dataRegionId = dataRegionId; this.outputTopic = outputTopic; + this.applicationName = applicationName; } public String getDataRegionId() { @@ -50,6 +54,9 @@ public String getOutputTopic() { return outputTopic; } + public String getApplicationName() { + return applicationName; + } @Override public String toString() { @@ -58,6 +65,7 @@ public String toString() { ", computeRegionId='" + computeRegionId + '\'' + ", dataRegionId='" + dataRegionId + '\'' + ", outputTopic='" + outputTopic + '\'' + + ", applicationName='" + applicationName + '\'' + '}'; } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/Configurator.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/Configurator.java index 4f1b984..4b31f96 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/Configurator.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/Configurator.java @@ -11,10 +11,9 @@ import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.FallbackBackupPolicy; import com.google.cloud.pso.bq_snapshot_manager.functions.f03_snapshoter.SnapshoterRequest; import com.google.cloud.pso.bq_snapshot_manager.helpers.LoggingHelper; -import com.google.cloud.pso.bq_snapshot_manager.helpers.TrackingHelper; import com.google.cloud.pso.bq_snapshot_manager.helpers.Utils; import com.google.cloud.pso.bq_snapshot_manager.services.bq.BigQueryService; -import com.google.cloud.pso.bq_snapshot_manager.services.catalog.DataCatalogService; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyService; import com.google.cloud.pso.bq_snapshot_manager.services.pubsub.PubSubPublishResults; import com.google.cloud.pso.bq_snapshot_manager.services.pubsub.PubSubService; import com.google.cloud.pso.bq_snapshot_manager.services.set.PersistentSet; @@ -34,7 +33,7 @@ public class Configurator { private final BigQueryService bqService; - private final DataCatalogService dataCatalogService; + private final BackupPolicyService backupPolicyService; private final PubSubService pubSubService; private final PersistentSet persistentSet; private final FallbackBackupPolicy fallbackBackupPolicy; @@ -43,7 +42,7 @@ public class Configurator { public Configurator(ConfiguratorConfig config, BigQueryService bqService, - DataCatalogService dataCatalogService, + BackupPolicyService backupPolicyService, PubSubService pubSubService, PersistentSet persistentSet, FallbackBackupPolicy fallbackBackupPolicy, @@ -51,7 +50,7 @@ public Configurator(ConfiguratorConfig config, Integer functionNumber) { this.config = config; this.bqService = bqService; - this.dataCatalogService = dataCatalogService; + this.backupPolicyService = backupPolicyService; this.pubSubService = pubSubService; this.persistentSet = persistentSet; this.fallbackBackupPolicy = fallbackBackupPolicy; @@ -61,7 +60,8 @@ public Configurator(ConfiguratorConfig config, logger = new LoggingHelper( Configurator.class.getSimpleName(), functionNumber, - config.getProjectId() + config.getProjectId(), + config.getApplicationName() ); } @@ -218,16 +218,28 @@ public ConfiguratorResponse execute(ConfiguratorRequest request, String pubSubMe public BackupPolicy getBackupPolicy(ConfiguratorRequest request) throws IOException { // Check if the table has a back policy attached to it in data catalog - BackupPolicy attachedBackupPolicy = dataCatalogService.getBackupPolicyTag( - request.getTargetTable(), - config.getBackupTagTemplateId() + BackupPolicy attachedBackupPolicy = backupPolicyService.getBackupPolicyForTable( + request.getTargetTable() ); // if there is manually attached backup policy (e.g. by the table designer) then use it. if (attachedBackupPolicy != null && attachedBackupPolicy.getConfigSource().equals(BackupConfigSource.MANUAL)) { + + logger.logInfoWithTracker(request.isDryRun(), + request.getTrackingId(), + request.getTargetTable(), + String.format("Attached backup policy found for table %s", request.getTargetTable()) + ); + return attachedBackupPolicy; }else{ + logger.logInfoWithTracker(request.isDryRun(), + request.getTrackingId(), + request.getTargetTable(), + String.format("No 'config_source=MANUAL' backup policy found for table %s. Will search for a fallback policy.", request.getTargetTable()) + ); + // find the most granular fallback policy table > dataset > project BackupPolicy fallbackPolicy = findFallbackBackupPolicy(fallbackBackupPolicy, request.getTargetTable()).y(); diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorConfig.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorConfig.java index fcd3e65..a460a2b 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorConfig.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorConfig.java @@ -7,14 +7,19 @@ public class ConfiguratorConfig { private final String gcsSnapshoterTopic; private final String backupTagTemplateId; + private final String applicationName; + public ConfiguratorConfig(String projectId, String bigQuerySnapshoterTopic, String gcsSnapshoterTopic, - String backupTagTemplateId) { + String backupTagTemplateId, + String applicationName + ) { this.projectId = projectId; this.bigQuerySnapshoterTopic = bigQuerySnapshoterTopic; this.gcsSnapshoterTopic = gcsSnapshoterTopic; this.backupTagTemplateId = backupTagTemplateId; + this.applicationName = applicationName; } public String getProjectId() { @@ -33,6 +38,10 @@ public String getBackupTagTemplateId() { return backupTagTemplateId; } + public String getApplicationName() { + return applicationName; + } + @Override public String toString() { return "ConfiguratorConfig{" + @@ -40,6 +49,7 @@ public String toString() { ", bigQuerySnapshoterTopic='" + bigQuerySnapshoterTopic + '\'' + ", gcsSnapshoterTopic='" + gcsSnapshoterTopic + '\'' + ", backupTagTemplateId='" + backupTagTemplateId + '\'' + + ", applicationName='" + applicationName + '\'' + '}'; } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoter.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoter.java index c9db13b..009a920 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoter.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoter.java @@ -63,7 +63,8 @@ public BigQuerySnapshoter(SnapshoterConfig config, logger = new LoggingHelper( BigQuerySnapshoter.class.getSimpleName(), functionNumber, - config.getProjectId() + config.getProjectId(), + config.getApplicationName() ); } @@ -139,7 +140,7 @@ public BigQuerySnapshoterResponse execute(SnapshoterRequest request, Timestamp o if(!request.isDryRun()){ - String jobId = TrackingHelper.generateBQSnapshotJobId(request.getTrackingId()); + String jobId = TrackingHelper.generateBQSnapshotJobId(request.getTrackingId(), config.getApplicationName()); // API Call bqService.createSnapshot( diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoter.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoter.java index 95b1c5b..caf0c5f 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoter.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoter.java @@ -2,7 +2,6 @@ import com.google.cloud.Timestamp; import com.google.cloud.Tuple; -import com.google.cloud.pso.bq_snapshot_manager.entities.Globals; import com.google.cloud.pso.bq_snapshot_manager.entities.NonRetryableApplicationException; import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupMethod; @@ -53,7 +52,8 @@ public GCSSnapshoter(SnapshoterConfig config, logger = new LoggingHelper( GCSSnapshoter.class.getSimpleName(), functionNumber, - config.getProjectId() + config.getProjectId(), + config.getApplicationName() ); } @@ -130,7 +130,7 @@ public GCSSnapshoterResponse execute(SnapshoterRequest request, Timestamp operat if(!request.isDryRun()){ // create an async bq export job - String jobId = TrackingHelper.generateBQExportJobId(request.getTrackingId()); + String jobId = TrackingHelper.generateBQExportJobId(request.getTrackingId(), config.getApplicationName()); // We create the tagging request and added it to a persistent storage // The Tagger service will receive notifications of export job completion via log sinks and pick up the tagger request from the persistent storage @@ -152,7 +152,7 @@ public GCSSnapshoterResponse execute(SnapshoterRequest request, Timestamp operat Map jobLabels = new HashMap<>(); // labels has to be max 63 chars, contain only lowercase letters, numeric characters, underscores, and dashes. All characters must use UTF-8 encoding, and international characters are allowed. - jobLabels.put("app", Globals.APPLICATION_NAME); + jobLabels.put("app", config.getApplicationName()); // API Call bqService.exportToGCS( diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/SnapshoterConfig.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/SnapshoterConfig.java index 66be11f..885fe22 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/SnapshoterConfig.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/SnapshoterConfig.java @@ -21,9 +21,12 @@ public class SnapshoterConfig { private String projectId; private String outputTopic; - public SnapshoterConfig(String projectId, String outputTopic) { + private String applicationName; + + public SnapshoterConfig(String projectId, String outputTopic, String applicationName) { this.projectId = projectId; this.outputTopic = outputTopic; + this.applicationName = applicationName; } public String getProjectId() { @@ -34,11 +37,16 @@ public String getOutputTopic() { return outputTopic; } + public String getApplicationName() { + return applicationName; + } + @Override public String toString() { return "SnapshoterConfig{" + "projectId='" + projectId + '\'' + ", outputTopic='" + outputTopic + '\'' + + ", applicationName='" + applicationName + '\'' + '}'; } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/Tagger.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/Tagger.java index bd55a28..5b37be9 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/Tagger.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/Tagger.java @@ -20,26 +20,38 @@ import com.google.cloud.pso.bq_snapshot_manager.entities.RetryableApplicationException; import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupMethod; import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupPolicy; +import com.google.cloud.pso.bq_snapshot_manager.functions.f03_snapshoter.GCSSnapshoter; import com.google.cloud.pso.bq_snapshot_manager.helpers.LoggingHelper; import com.google.cloud.pso.bq_snapshot_manager.helpers.Utils; -import com.google.cloud.pso.bq_snapshot_manager.services.catalog.DataCatalogService; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyService; import com.google.cloud.pso.bq_snapshot_manager.services.set.PersistentSet; +import java.io.IOException; + public class Tagger { private final LoggingHelper logger; private final TaggerConfig config; - private final DataCatalogService dataCatalogService; + private final BackupPolicyService backupPolicyService; private final PersistentSet persistentSet; private final String persistentSetObjectPrefix; - public Tagger(LoggingHelper logger, TaggerConfig config, DataCatalogService dataCatalogService, PersistentSet persistentSet, String persistentSetObjectPrefix) { - this.logger = logger; + private final Integer functionNumber; + + public Tagger(TaggerConfig config, BackupPolicyService backupPolicyService, PersistentSet persistentSet, String persistentSetObjectPrefix, Integer functionNumber) { this.config = config; - this.dataCatalogService = dataCatalogService; + this.backupPolicyService = backupPolicyService; this.persistentSet = persistentSet; this.persistentSetObjectPrefix = persistentSetObjectPrefix; + this.functionNumber = functionNumber; + + logger = new LoggingHelper( + Tagger.class.getSimpleName(), + functionNumber, + config.getProjectId(), + config.getApplicationName() + ); } /** @@ -52,7 +64,7 @@ public Tagger(LoggingHelper logger, TaggerConfig config, DataCatalogService data public TaggerResponse execute( TaggerRequest request, String pubSubMessageId - ) throws NonRetryableApplicationException, RetryableApplicationException { + ) throws NonRetryableApplicationException, RetryableApplicationException, IOException { // run common service start logging and checks Utils.runServiceStartRoutines( @@ -99,10 +111,9 @@ public TaggerResponse execute( if(!request.isDryRun()){ // update the tag // API Calls - dataCatalogService.createOrUpdateBackupPolicyTag( + backupPolicyService.createOrUpdateBackupPolicyForTable( request.getTargetTable(), - upDatedBackupPolicy, - config.getTagTemplateId() + upDatedBackupPolicy ); } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/TaggerConfig.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/TaggerConfig.java index 488120d..0490697 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/TaggerConfig.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/functions/f04_tagger/TaggerConfig.java @@ -18,13 +18,18 @@ public class TaggerConfig { - private String projectId; - private String tagTemplateId; + private final String projectId; + private final String tagTemplateId; + + private final String applicationName; public TaggerConfig(String projectId, - String tagTemplateId) { + String tagTemplateId, + String applicationName + ) { this.projectId = projectId; this.tagTemplateId = tagTemplateId; + this.applicationName = applicationName; } public String getProjectId() { @@ -35,11 +40,16 @@ public String getTagTemplateId() { return tagTemplateId; } + public String getApplicationName() { + return applicationName; + } + @Override public String toString() { return "TaggerConfig{" + "projectId='" + projectId + '\'' + ", tagTemplateId='" + tagTemplateId + '\'' + + ", applicationName='" + applicationName + '\'' + '}'; } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/LoggingHelper.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/LoggingHelper.java index 19dcb20..02e984f 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/LoggingHelper.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/LoggingHelper.java @@ -37,10 +37,13 @@ public class LoggingHelper { // Used to create a trace private final String projectId; - public LoggingHelper(String loggerName, Integer functionNumber, String projectId) { + private final String applicationName; + + public LoggingHelper(String loggerName, Integer functionNumber, String projectId, String applicationName) { this.loggerName = loggerName; this.functionNumber = functionNumber; this.projectId = projectId; + this.applicationName = applicationName; logger = LoggerFactory.getLogger(loggerName); } @@ -257,7 +260,7 @@ private void logWithTracker(ApplicationLog log, Boolean isDryRun, String tracker // Enable JSON logging with Logback and SLF4J by enabling the Logstash JSON Encoder in your logback.xml configuration. String payload = String.format("%s | %s | %s | %s | %s | %s", - Globals.APPLICATION_NAME, + applicationName, log, loggerName, isDryRun!=null? (isDryRun?"Dry-Run":"Wet-Run") : null, @@ -274,7 +277,7 @@ private void logWithTracker(ApplicationLog log, Boolean isDryRun, String tracker } Object [] globalAttributes = new Object[]{ - kv("global_app", Globals.APPLICATION_NAME), + kv("global_app", applicationName), kv("global_logger_name", this.loggerName), kv("global_app_log", log), kv("global_tracker", tracker), diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelper.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelper.java index 955dc25..0b0b744 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelper.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelper.java @@ -17,7 +17,6 @@ package com.google.cloud.pso.bq_snapshot_manager.helpers; import com.google.cloud.Timestamp; -import com.google.cloud.pso.bq_snapshot_manager.entities.Globals; import java.util.UUID; @@ -70,16 +69,16 @@ public static String generateTrackingId (String runId){ return String.format("%s-%s", runId, UUID.randomUUID().toString()); } - public static String generateBQExportJobId(String trackingId){ - return String.format("%s_%s_%s",trackingId, "export", Globals.APPLICATION_NAME, trackingId); + public static String generateBQExportJobId(String trackingId, String applicationName){ + return String.format("%s_%s_%s",trackingId, "export", applicationName, trackingId); } public static String parseTrackingIdFromBQExportJobId(String bqExportJobId){ return Utils.tokenize(bqExportJobId,"_", true).get(0); } - public static String generateBQSnapshotJobId(String trackingId){ - return String.format("%s_%s_%s",trackingId, "snapshot", Globals.APPLICATION_NAME, trackingId); + public static String generateBQSnapshotJobId(String trackingId, String applicationName){ + return String.format("%s_%s_%s",trackingId, "snapshot", applicationName, trackingId); } } diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/Utils.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/Utils.java index f01d9e7..b4f1154 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/Utils.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/helpers/Utils.java @@ -142,8 +142,8 @@ public static Tuple getTableSpecWithTimeTravel(TableSpec table, // always use time travel for consistency and traceability timeTravelMs = refPointMs; }else{ - // use a buffer (milliseconds) to count for the operation time - Long bufferMs = timeTravelOffsetDays.equals(TimeTravelOffsetDays.DAYS_7) ? 60000L : 0L; + // use a buffer (milliseconds) to count for the operation time. 1 MIN = 60000 MILLISECONDS + Long bufferMs = timeTravelOffsetDays.equals(TimeTravelOffsetDays.DAYS_7) ? 60 * 60000L : 0L; // milli seconds per day * number of days Long timeTravelOffsetMs = (86400000L * Long.parseLong(timeTravelOffsetDays.getText())); timeTravelMs = (refPointMs - timeTravelOffsetMs) + bufferMs; diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyService.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyService.java new file mode 100644 index 0000000..901ae02 --- /dev/null +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyService.java @@ -0,0 +1,16 @@ +package com.google.cloud.pso.bq_snapshot_manager.services.backup_policy; + +import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupPolicy; +import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; + +import javax.annotation.Nullable; +import java.io.IOException; + +public interface BackupPolicyService { + + void createOrUpdateBackupPolicyForTable(TableSpec tableSpec, BackupPolicy backupPolicy) throws IOException; + + @Nullable BackupPolicy getBackupPolicyForTable(TableSpec tableSpec) throws IOException, IllegalArgumentException; + + void shutdown(); +} diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/catalog/DataCatalogServiceImpl.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceDataCatalogImpl.java similarity index 67% rename from services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/catalog/DataCatalogServiceImpl.java rename to services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceDataCatalogImpl.java index bab4879..4dcfad0 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/catalog/DataCatalogServiceImpl.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceDataCatalogImpl.java @@ -1,4 +1,4 @@ -package com.google.cloud.pso.bq_snapshot_manager.services.catalog; +package com.google.cloud.pso.bq_snapshot_manager.services.backup_policy; import com.google.cloud.Timestamp; @@ -14,19 +14,21 @@ import java.util.Map; import java.util.stream.Collectors; -public class DataCatalogServiceImpl implements DataCatalogService { +public class BackupPolicyServiceDataCatalogImpl implements BackupPolicyService { private final DataCatalogClient dataCatalogClient; + private String backupPolicyTagTemplateId; - public DataCatalogServiceImpl() throws IOException { + public BackupPolicyServiceDataCatalogImpl(String backupPolicyTagTemplateId) throws IOException { dataCatalogClient = DataCatalogClient.create(); + this.backupPolicyTagTemplateId = backupPolicyTagTemplateId; } - public void shutdown(){ + public void shutdown() { dataCatalogClient.shutdown(); } - public Tag createOrUpdateBackupPolicyTag(TableSpec tableSpec, BackupPolicy backupPolicy, String backupPolicyTagTemplateId){ + public void createOrUpdateBackupPolicyForTable(TableSpec tableSpec, BackupPolicy backupPolicy) { // API Call String parent = getBigQueryEntryName(tableSpec); @@ -35,7 +37,7 @@ public Tag createOrUpdateBackupPolicyTag(TableSpec tableSpec, BackupPolicy backu DataCatalogClient.ListTagsPagedResponse response = dataCatalogClient.listTags(parent); List allTags = new ArrayList<>(); - for (DataCatalogClient.ListTagsPage l: response.iteratePages()){ + for (DataCatalogClient.ListTagsPage l : response.iteratePages()) { allTags.addAll(l.getResponse().getTagsList()); } @@ -44,12 +46,12 @@ public Tag createOrUpdateBackupPolicyTag(TableSpec tableSpec, BackupPolicy backu backupPolicyTagTemplateId ); - if(tag == null){ + if (tag == null) { // create a new tag - return dataCatalogClient.createTag(parent, backupPolicy.toDataCatalogTag(backupPolicyTagTemplateId, null)); - }else{ + dataCatalogClient.createTag(parent, backupPolicy.toDataCatalogTag(backupPolicyTagTemplateId, null)); + } else { // update existing tag referencing the existing tag.name - return dataCatalogClient.updateTag( + dataCatalogClient.updateTag( backupPolicy.toDataCatalogTag( backupPolicyTagTemplateId, tag.getName() @@ -58,13 +60,13 @@ public Tag createOrUpdateBackupPolicyTag(TableSpec tableSpec, BackupPolicy backu } } - public Tag findTag(List tags, String tagTemplateName){ + public Tag findTag(List tags, String tagTemplateName) { List foundTags = tags.stream().filter(t -> t.getTemplate().equals(tagTemplateName)) .collect(Collectors.toList()); // if more than one tag is found use the first one - return foundTags.size() >= 1? foundTags.get(0): null; + return foundTags.size() >= 1 ? foundTags.get(0) : null; } @@ -72,23 +74,22 @@ public Tag findTag(List tags, String tagTemplateName){ * Return the attached backup policy tag template or null if no template is attached * * @param tableSpec - * @param backupPolicyTagTemplateId * @return * @throws IllegalArgumentException */ - public @Nullable BackupPolicy getBackupPolicyTag(TableSpec tableSpec, String backupPolicyTagTemplateId) throws IllegalArgumentException { + public @Nullable BackupPolicy getBackupPolicyForTable(TableSpec tableSpec) throws IllegalArgumentException { Map tagTemplate = getTagFieldsMap(tableSpec, backupPolicyTagTemplateId); - if(tagTemplate == null){ + if (tagTemplate == null) { // no backup tag template is attached to this table return null; - }else{ + } else { return BackupPolicy.fromMap(convertTagFieldMapToStrMap(tagTemplate)); } } - public Tag getTag(TableSpec tableSpec, String templateId){ + public Tag getTag(TableSpec tableSpec, String templateId) { // API Call String parent = getBigQueryEntryName(tableSpec); // API CALL @@ -97,8 +98,8 @@ public Tag getTag(TableSpec tableSpec, String templateId){ // TODO: handle multiple pages List tags = response.getPage().getResponse().getTagsList(); - for (Tag tagTemplate: tags){ - if (tagTemplate.getTemplate().equals(templateId)){ + for (Tag tagTemplate : tags) { + if (tagTemplate.getTemplate().equals(templateId)) { return tagTemplate; } } @@ -108,10 +109,10 @@ public Tag getTag(TableSpec tableSpec, String templateId){ public Map getTagFieldsMap(TableSpec tableSpec, String templateId) { Tag tag = getTag(tableSpec, templateId); - return tag == null? null: tag.getFieldsMap(); + return tag == null ? null : tag.getFieldsMap(); } - public String getBigQueryEntryName(TableSpec tableSpec){ + public String getBigQueryEntryName(TableSpec tableSpec) { LookupEntryRequest lookupEntryRequest = LookupEntryRequest.newBuilder() .setLinkedResource(tableSpec.toDataCatalogLinkedResource()).build(); @@ -120,30 +121,30 @@ public String getBigQueryEntryName(TableSpec tableSpec){ return dataCatalogClient.lookupEntry(lookupEntryRequest).getName(); } - public static Map convertTagFieldMapToStrMap(Map tagFieldMap){ + public static Map convertTagFieldMapToStrMap(Map tagFieldMap) { Map strMap = new HashMap<>(tagFieldMap.size()); - for(Map.Entry entry: tagFieldMap.entrySet()){ + for (Map.Entry entry : tagFieldMap.entrySet()) { String strValue = ""; - if(entry.getValue().hasBoolValue()){ + if (entry.getValue().hasBoolValue()) { strValue = String.valueOf(entry.getValue().getBoolValue()); } - if(entry.getValue().hasStringValue()){ + if (entry.getValue().hasStringValue()) { strValue = entry.getValue().getStringValue(); } - if(entry.getValue().hasDoubleValue()){ + if (entry.getValue().hasDoubleValue()) { strValue = String.valueOf(entry.getValue().getDoubleValue()); } - if(entry.getValue().hasEnumValue()){ + if (entry.getValue().hasEnumValue()) { strValue = entry.getValue().getEnumValue().getDisplayName(); } - if(entry.getValue().hasTimestampValue()){ + if (entry.getValue().hasTimestampValue()) { strValue = Timestamp.ofTimeSecondsAndNanos( entry.getValue().getTimestampValue().getSeconds(), entry.getValue().getTimestampValue().getNanos() ).toString(); } - if(entry.getValue().hasRichtextValue()){ + if (entry.getValue().hasRichtextValue()) { strValue = entry.getValue().getRichtextValue(); } strMap.put(entry.getKey(), strValue); diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceFireStoreImpl.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceFireStoreImpl.java new file mode 100644 index 0000000..28ca117 --- /dev/null +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceFireStoreImpl.java @@ -0,0 +1,59 @@ +package com.google.cloud.pso.bq_snapshot_manager.services.backup_policy; + + + +import com.google.cloud.datastore.*; +import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; +import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.*; + +import javax.annotation.Nullable; +import java.util.HashMap; +import java.util.Map; + +public class BackupPolicyServiceFireStoreImpl implements BackupPolicyService { + + + private static final String KIND = "BigQueryBackupPolicy"; + private Datastore datastore; + + public BackupPolicyServiceFireStoreImpl() { + datastore = DatastoreOptions.getDefaultInstance().getService(); + } + + + public void createOrUpdateBackupPolicyForTable(TableSpec tableSpec, BackupPolicy backupPolicy) { + + Key backupPolicyKey = datastore.newKeyFactory().setKind(KIND).newKey(tableSpec.toSqlString()); + Entity backupPolicyEntity = backupPolicy.toFireStoreEntity(backupPolicyKey); + datastore.put(backupPolicyEntity); + } + + + public @Nullable BackupPolicy getBackupPolicyForTable(TableSpec tableSpec) { + Key backupPolicyKey = datastore.newKeyFactory().setKind(KIND).newKey(tableSpec.toSqlString()); + Entity backupPolicyEntity = datastore.get(backupPolicyKey); + if(backupPolicyEntity == null){ + // table doesn't have a backup policy stored + return null; + }else{ + Map propertiesMap = entityToStrMap(backupPolicyEntity); + return BackupPolicy.fromMap(propertiesMap); + } + } + + @Override + public void shutdown() { + // do nothing + } + + public Map entityToStrMap (Entity entity){ + Map strMap = new HashMap<>(entity.getProperties().size()); + + for (String key: entity.getProperties().keySet()) { + strMap.put(key, entity.getValue(key).get().toString()); + } + return strMap; + } + + +} diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceGCSImpl.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceGCSImpl.java new file mode 100644 index 0000000..6bde510 --- /dev/null +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/backup_policy/BackupPolicyServiceGCSImpl.java @@ -0,0 +1,59 @@ +package com.google.cloud.pso.bq_snapshot_manager.services.backup_policy; + +import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; +import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupPolicy; +import com.google.cloud.storage.*; + +import javax.annotation.Nullable; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; + +public class BackupPolicyServiceGCSImpl implements BackupPolicyService { + + + private Storage storage; + private String bucketName; + + public BackupPolicyServiceGCSImpl(String bucketName) { + // Instantiates a client + this.storage = StorageOptions.getDefaultInstance().getService(); + this.bucketName = bucketName; + } + + + public void createOrUpdateBackupPolicyForTable(TableSpec tableSpec, BackupPolicy backupPolicy) throws IOException { + + String filePath = tableToGcsKey(tableSpec); + BlobId blobId = BlobId.of(bucketName, filePath); + BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build(); + + byte[] content = backupPolicy.toJson().getBytes(StandardCharsets.UTF_8); + storage.createFrom(blobInfo, new ByteArrayInputStream(content)); + } + + + public @Nullable BackupPolicy getBackupPolicyForTable(TableSpec tableSpec) { + String filePath = tableToGcsKey(tableSpec); + BlobId blobId = BlobId.of(bucketName, filePath); + Blob blob = storage.get(blobId); + if (blob == null) { + return null; + } else { + byte[] contentBytes = storage.readAllBytes(blobId); + String contentStr = new String(contentBytes, StandardCharsets.UTF_8); + return BackupPolicy.fromJson(contentStr); + } + } + + @Override + public void shutdown() { + // do nothing + } + + public String tableToGcsKey(TableSpec tableSpec) { + return String.format("%s/%s" , tableSpec.toHivePartitionPostfix(), "backup_policy.json"); + } + + +} diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/bq/BigQueryServiceImpl.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/bq/BigQueryServiceImpl.java index 17db6f0..7e9e0e5 100644 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/bq/BigQueryServiceImpl.java +++ b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/bq/BigQueryServiceImpl.java @@ -19,13 +19,11 @@ import com.google.cloud.Timestamp; import com.google.cloud.Tuple; import com.google.cloud.bigquery.*; -import com.google.cloud.pso.bq_snapshot_manager.entities.Globals; import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.GCSSnapshotFormat; import javax.annotation.Nullable; import java.io.IOException; -import java.util.HashMap; import java.util.Map; public class BigQueryServiceImpl implements BigQueryService { diff --git a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/catalog/DataCatalogService.java b/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/catalog/DataCatalogService.java deleted file mode 100644 index e03e878..0000000 --- a/services/library/src/main/java/com/google/cloud/pso/bq_snapshot_manager/services/catalog/DataCatalogService.java +++ /dev/null @@ -1,17 +0,0 @@ -package com.google.cloud.pso.bq_snapshot_manager.services.catalog; - -import com.google.cloud.datacatalog.v1.Entry; -import com.google.cloud.datacatalog.v1.LookupEntryRequest; -import com.google.cloud.datacatalog.v1.Tag; -import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupPolicy; -import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; - -import javax.annotation.Nullable; -import java.io.IOException; - -public interface DataCatalogService { - - Tag createOrUpdateBackupPolicyTag(TableSpec tableSpec, BackupPolicy backupPolicy, String backupPolicyTagTemplateId); - - @Nullable BackupPolicy getBackupPolicyTag(TableSpec tableSpec, String backupPolicyTagTemplateId) throws IOException, IllegalArgumentException; -} diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/entities/BackupPolicyTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/entities/BackupPolicyTest.java index fd9f606..50f322c 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/entities/BackupPolicyTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/entities/BackupPolicyTest.java @@ -311,13 +311,13 @@ public void testMissingValues() { public void testValidate_Required(){ BackupPolicy.BackupPolicyBuilder builder = new BackupPolicy.BackupPolicyBuilder(null,null,null,null,null); - List actual = BackupPolicy.validate(builder); - List expected = Arrays.asList( - DataCatalogBackupPolicyTagFields.backup_cron, - DataCatalogBackupPolicyTagFields.backup_method, - DataCatalogBackupPolicyTagFields.backup_time_travel_offset_days, - DataCatalogBackupPolicyTagFields.config_source, - DataCatalogBackupPolicyTagFields.backup_storage_project + List actual = BackupPolicy.validate(builder); + List expected = Arrays.asList( + BackupPolicyFields.backup_cron, + BackupPolicyFields.backup_method, + BackupPolicyFields.backup_time_travel_offset_days, + BackupPolicyFields.config_source, + BackupPolicyFields.backup_storage_project ); assertEquals(expected, actual); @@ -327,10 +327,10 @@ public void testValidate_Required(){ public void testValidate_BQSnapshot(){ BackupPolicy.BackupPolicyBuilder builder = new BackupPolicy.BackupPolicyBuilder("",BackupMethod.BIGQUERY_SNAPSHOT,TimeTravelOffsetDays.DAYS_0,BackupConfigSource.SYSTEM,""); - List actual = BackupPolicy.validate(builder); - List expected = Arrays.asList( - DataCatalogBackupPolicyTagFields.bq_snapshot_storage_dataset, - DataCatalogBackupPolicyTagFields.bq_snapshot_expiration_days + List actual = BackupPolicy.validate(builder); + List expected = Arrays.asList( + BackupPolicyFields.bq_snapshot_storage_dataset, + BackupPolicyFields.bq_snapshot_expiration_days ); assertEquals(expected, actual); @@ -341,10 +341,10 @@ public void testValidate_GCSSnapshot(){ BackupPolicy.BackupPolicyBuilder builder = new BackupPolicy.BackupPolicyBuilder("",BackupMethod.GCS_SNAPSHOT,TimeTravelOffsetDays.DAYS_0,BackupConfigSource.SYSTEM,""); - List actual = BackupPolicy.validate(builder); - List expected = Arrays.asList( - DataCatalogBackupPolicyTagFields.gcs_snapshot_format, - DataCatalogBackupPolicyTagFields.gcs_snapshot_storage_location + List actual = BackupPolicy.validate(builder); + List expected = Arrays.asList( + BackupPolicyFields.gcs_snapshot_format, + BackupPolicyFields.gcs_snapshot_storage_location ); assertEquals(expected, actual); @@ -356,11 +356,11 @@ public void testValidate_GCSSnapshotCSV(){ BackupPolicy.BackupPolicyBuilder builder = new BackupPolicy.BackupPolicyBuilder("",BackupMethod.GCS_SNAPSHOT,TimeTravelOffsetDays.DAYS_0,BackupConfigSource.SYSTEM,""); builder.setGcsExportFormat(GCSSnapshotFormat.CSV); - List actual = BackupPolicy.validate(builder); - List expected = Arrays.asList( - DataCatalogBackupPolicyTagFields.gcs_snapshot_storage_location, - DataCatalogBackupPolicyTagFields.gcs_csv_delimiter, - DataCatalogBackupPolicyTagFields.gcs_csv_export_header + List actual = BackupPolicy.validate(builder); + List expected = Arrays.asList( + BackupPolicyFields.gcs_snapshot_storage_location, + BackupPolicyFields.gcs_csv_delimiter, + BackupPolicyFields.gcs_csv_export_header ); assertEquals(expected, actual); @@ -372,10 +372,10 @@ public void testValidate_GCSSnapshotAvro(){ BackupPolicy.BackupPolicyBuilder builder = new BackupPolicy.BackupPolicyBuilder("",BackupMethod.GCS_SNAPSHOT,TimeTravelOffsetDays.DAYS_0,BackupConfigSource.SYSTEM,""); builder.setGcsExportFormat(GCSSnapshotFormat.AVRO_SNAPPY); - List actual = BackupPolicy.validate(builder); - List expected = Arrays.asList( - DataCatalogBackupPolicyTagFields.gcs_snapshot_storage_location, - DataCatalogBackupPolicyTagFields.gcs_avro_use_logical_types + List actual = BackupPolicy.validate(builder); + List expected = Arrays.asList( + BackupPolicyFields.gcs_snapshot_storage_location, + BackupPolicyFields.gcs_avro_use_logical_types ); assertEquals(expected, actual); diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/BigQueryScopeListerTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/BigQueryScopeListerTest.java index 970912d..07b763b 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/BigQueryScopeListerTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/BigQueryScopeListerTest.java @@ -20,7 +20,7 @@ public class BigQueryScopeListerTest { BigQueryScopeLister lister = new BigQueryScopeLister( new ResourceScannerTestImpl(), - new LoggingHelper("test", 1, "test-project"), + new LoggingHelper("test", 1, "test-project", "bq_backup_manager"), "R-testxxxxxxx" ); diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherTest.java index e14e43f..c0fd571 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f01_dispatcher/DispatcherTest.java @@ -54,7 +54,8 @@ public class DispatcherTest { "testProjectId", "testComputeRegionId", "testDataRegionId", - "testTaggerTopic" + "testTaggerTopic", + "bq_backup_manager" ); String runId = "1679574252412-R"; diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorTest.java index 1f98532..4dce9ce 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f02_configurator/ConfiguratorTest.java @@ -2,7 +2,6 @@ import com.google.cloud.Timestamp; import com.google.cloud.Tuple; -import com.google.cloud.datacatalog.v1.Tag; import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.GCSSnapshotFormat; import com.google.cloud.pso.bq_snapshot_manager.entities.NonRetryableApplicationException; import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; @@ -14,13 +13,12 @@ import com.google.cloud.pso.bq_snapshot_manager.services.PersistentSetTestImpl; import com.google.cloud.pso.bq_snapshot_manager.services.PubSubServiceTestImpl; import com.google.cloud.pso.bq_snapshot_manager.services.bq.BigQueryService; -import com.google.cloud.pso.bq_snapshot_manager.services.catalog.DataCatalogService; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyService; import com.google.cloud.pso.bq_snapshot_manager.services.pubsub.PubSubPublishResults; import org.junit.Test; import javax.annotation.Nullable; import java.io.IOException; -import java.sql.Time; import java.time.LocalDateTime; import java.util.AbstractMap; import java.util.Map; @@ -35,7 +33,8 @@ public class ConfiguratorTest { LoggingHelper testLogger = new LoggingHelper( ConfiguratorTest.class.getSimpleName(), 2, - "testProject" + "testProject", + "bq_backup_manager" ); String jsonPolicyStr = "{\n" + @@ -334,7 +333,8 @@ private ConfiguratorResponse executeConfigurator( "test-project", "test-bqSnapshoterTopic", "test-gcsSnapshoterTopic", - "test-templateId" + "test-templateId", + "bq_backup_manager" ); Configurator configurator = new Configurator( @@ -355,18 +355,22 @@ public Long getTableCreationTime(TableSpec table) { return Utils.timestampToUnixTimeMillis(tableCreationTS); } }, - new DataCatalogService() { + new BackupPolicyService() { @Override - public Tag createOrUpdateBackupPolicyTag(TableSpec tableSpec, BackupPolicy backupPolicy, String backupPolicyTagTemplateId) { + public void createOrUpdateBackupPolicyForTable(TableSpec tableSpec, BackupPolicy backupPolicy) { - return null; } @Override - public @Nullable BackupPolicy getBackupPolicyTag(TableSpec tableSpec, String backupPolicyTagTemplateId) throws IOException, IllegalArgumentException { + public @Nullable BackupPolicy getBackupPolicyForTable(TableSpec tableSpec) throws IOException, IllegalArgumentException { return testBackupPolicy; } + + @Override + public void shutdown() { + + } }, new PubSubServiceTestImpl(), new PersistentSetTestImpl(), diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoterTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoterTest.java index 7419a20..d5b2141 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoterTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/BigQuerySnapshoterTest.java @@ -41,7 +41,7 @@ public void testGetSnapshotTableSpec() { public void testExecute() throws NonRetryableApplicationException, IOException, InterruptedException { BigQuerySnapshoter snapshoter = new BigQuerySnapshoter( - new SnapshoterConfig("host-project", "data-region"), + new SnapshoterConfig("host-project", "data-region", "bq_backup_manager"), new BigQueryService() { @Override public void createSnapshot(String jobId, TableSpec sourceTable, TableSpec destinationId, Timestamp snapshotExpirationTs, String trackingId) throws InterruptedException { diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoterTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoterTest.java index b58dbaa..4260026 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoterTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/functions/f03_snapshoter/GCSSnapshoterTest.java @@ -42,7 +42,7 @@ public void testPrepareGcsUriForMultiFileExport() { public void testExecute() throws NonRetryableApplicationException, IOException, InterruptedException { GCSSnapshoter gcsSnapshoter = new GCSSnapshoter( - new SnapshoterConfig("host-project", "data-region"), + new SnapshoterConfig("host-project", "data-region", "bq_backup_manager"), new BigQueryService() { @Override public void createSnapshot(String jobId, TableSpec sourceTable, TableSpec destinationId, Timestamp snapshotExpirationTs, String trackingId) throws InterruptedException { diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/ControllerExceptionHelperTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/ControllerExceptionHelperTest.java index 12d16e9..ecaffe2 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/ControllerExceptionHelperTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/ControllerExceptionHelperTest.java @@ -33,7 +33,7 @@ public class ControllerExceptionHelperTest { @Test public void testRetryableApiException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test", "bq_backup_manager"); try { @@ -64,7 +64,7 @@ public Object getTransportCode() { @Test public void testNonRetryableApiException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test","bq_backup_manager"); try { @@ -95,7 +95,7 @@ public Object getTransportCode() { @Test public void testRetryableIOException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test", "bq_backup_manager"); try { @@ -115,7 +115,7 @@ public void testRetryableIOException() { @Test public void testRetryableRuntimeStatusResourceExhaustedException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test", "bq_backup_manager"); try { @@ -135,7 +135,7 @@ public void testRetryableRuntimeStatusResourceExhaustedException() { @Test public void testRetryableResourceExhaustedException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test", "bq_backup_manager"); try { @@ -169,7 +169,7 @@ public Object getTransportCode() { @Test public void testNestedRetryableException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test", "bq_backup_manager"); try { Throwable cause = new java.net.SocketException ("test socket exception"); @@ -196,7 +196,7 @@ public void testNestedRetryableException() { @Test public void testNonRetryableException() { - LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test"); + LoggingHelper logger = new LoggingHelper(ControllerExceptionHelperTest.class.getSimpleName(), 0, "test", "bq_backup_manager"); try { diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelperTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelperTest.java index 803848d..ecf3160 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelperTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/TrackingHelperTest.java @@ -21,7 +21,7 @@ public void parseAsTimestamp(){ public void parseTrackingIdFromJobId(){ String trackingId = TrackingHelper.generateTrackingId(TrackingHelper.MIN_RUN_ID); String parsedTrackingId = TrackingHelper.parseTrackingIdFromBQExportJobId( - TrackingHelper.generateBQExportJobId(trackingId) + TrackingHelper.generateBQExportJobId(trackingId, "bq_backup_manager") ); assertEquals(trackingId, parsedTrackingId); diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/UtilsTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/UtilsTest.java index 6431ce3..b3d794f 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/UtilsTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/helpers/UtilsTest.java @@ -59,7 +59,7 @@ public void testGetTableSpecWithTimeTravel(){ refPoint ); - Long expectedMs7Days = (1665665921000L - (7 * 86400000)) + 60000; + Long expectedMs7Days = (1665665921000L - (7 * 86400000)) + 60* 60000; assertEquals(TableSpec.fromSqlString("p.d.t@"+expectedMs7Days.toString()), actualWith7.x()); assertEquals(expectedMs7Days, actualWith7.y()); diff --git a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/services/DataCatalogServiceImplTest.java b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/services/BackupPolicyServiceDataCatalogImplTest.java similarity index 75% rename from services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/services/DataCatalogServiceImplTest.java rename to services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/services/BackupPolicyServiceDataCatalogImplTest.java index 43f32b6..57d2c9e 100644 --- a/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/services/DataCatalogServiceImplTest.java +++ b/services/library/src/test/java/com/google/cloud/pso/bq_snapshot_manager/services/BackupPolicyServiceDataCatalogImplTest.java @@ -1,25 +1,15 @@ package com.google.cloud.pso.bq_snapshot_manager.services; -import com.google.cloud.Timestamp; import com.google.cloud.datacatalog.v1.TagField; -import com.google.cloud.pso.bq_snapshot_manager.entities.*; -import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupConfigSource; -import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupMethod; -import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.BackupPolicy; -import com.google.cloud.pso.bq_snapshot_manager.entities.backup_policy.TimeTravelOffsetDays; -import com.google.cloud.pso.bq_snapshot_manager.helpers.Utils; -import com.google.cloud.pso.bq_snapshot_manager.services.catalog.DataCatalogServiceImpl; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyServiceDataCatalogImpl; import org.junit.Test; -import org.springframework.test.context.TestExecutionListeners; -import java.io.IOException; import java.util.HashMap; import java.util.Map; -import java.util.stream.Collectors; import static org.junit.Assert.assertEquals; -public class DataCatalogServiceImplTest { +public class BackupPolicyServiceDataCatalogImplTest { @Test public void testConvertTagFieldMapToStrMap() { @@ -83,7 +73,7 @@ public void testConvertTagFieldMapToStrMap() { expected.put("bq_snapshot_expiration_days","0.0"); expected.put("gcs_snapshot_storage_location","test-bucket"); - Map actual = DataCatalogServiceImpl.convertTagFieldMapToStrMap(tagMap); + Map actual = BackupPolicyServiceDataCatalogImpl.convertTagFieldMapToStrMap(tagMap); assertEquals(expected, actual); } diff --git a/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/BigQuerySnapshoterController.java b/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/BigQuerySnapshoterController.java index f47c5a5..86f91da 100644 --- a/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/BigQuerySnapshoterController.java +++ b/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/BigQuerySnapshoterController.java @@ -56,7 +56,8 @@ public BigQuerySnapshoterController() { logger = new LoggingHelper( BigQuerySnapshoterController.class.getSimpleName(), functionNumber, - environment.getProjectId() + environment.getProjectId(), + environment.getApplicationName() ); } diff --git a/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/Environment.java b/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/Environment.java index b1efd8e..87916d5 100644 --- a/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/Environment.java +++ b/services/snapshoter-bq-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_bq/Environment.java @@ -26,7 +26,8 @@ public SnapshoterConfig toConfig (){ return new SnapshoterConfig( getProjectId(), - getOutputTopic() + getOutputTopic(), + getApplicationName() ); } @@ -40,4 +41,7 @@ public String getProjectId(){ public String getGcsFlagsBucket(){ return Utils.getConfigFromEnv("GCS_FLAGS_BUCKET", true); } + public String getApplicationName(){ + return Utils.getConfigFromEnv("APPLICATION_NAME", true); + } } diff --git a/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/Environment.java b/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/Environment.java index 827211a..47be1f5 100644 --- a/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/Environment.java +++ b/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/Environment.java @@ -26,7 +26,8 @@ public SnapshoterConfig toConfig (){ return new SnapshoterConfig( getProjectId(), - getOutputTopic() + getOutputTopic(), + getApplicationName() ); } @@ -40,4 +41,8 @@ public String getProjectId(){ public String getGcsFlagsBucket(){ return Utils.getConfigFromEnv("GCS_FLAGS_BUCKET", true); } + public String getApplicationName(){ + return Utils.getConfigFromEnv("APPLICATION_NAME", true); + } + } diff --git a/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/GCSSnapshoterController.java b/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/GCSSnapshoterController.java index 3137185..2877c6d 100644 --- a/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/GCSSnapshoterController.java +++ b/services/snapshoter-gcs-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/snapshoter_gcs/GCSSnapshoterController.java @@ -58,7 +58,8 @@ public GCSSnapshoterController() { logger = new LoggingHelper( GCSSnapshoterController.class.getSimpleName(), functionNumber, - environment.getProjectId() + environment.getProjectId(), + environment.getApplicationName() ); } diff --git a/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/Environment.java b/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/Environment.java index f260cde..6aa0b3a 100644 --- a/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/Environment.java +++ b/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/Environment.java @@ -23,7 +23,8 @@ public class Environment { public TaggerConfig toConfig (){ return new TaggerConfig( getProjectId(), - getTagTemplateId() + getTagTemplateId(), + getApplicationName() ); } @@ -38,5 +39,11 @@ public String getTagTemplateId(){ public String getGcsFlagsBucket(){ return Utils.getConfigFromEnv("GCS_FLAGS_BUCKET", true); } + public String getApplicationName(){ + return Utils.getConfigFromEnv("APPLICATION_NAME", true); + } + public String getGcsBackupPoliciesBucket(){ + return Utils.getConfigFromEnv("GCS_BACKUP_POLICIES_BUCKET", true); + } } diff --git a/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/TaggerController.java b/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/TaggerController.java index 214c44b..69a3b1e 100644 --- a/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/TaggerController.java +++ b/services/tagger-app/src/main/java/com/google/cloud/pso/bq_snapshot_manager/tagger/TaggerController.java @@ -18,21 +18,19 @@ import com.google.cloud.Tuple; import com.google.cloud.pso.bq_snapshot_manager.entities.NonRetryableApplicationException; import com.google.cloud.pso.bq_snapshot_manager.entities.PubSubEvent; -import com.google.cloud.pso.bq_snapshot_manager.entities.TableSpec; -import com.google.cloud.pso.bq_snapshot_manager.functions.f03_snapshoter.BigQuerySnapshoter; -import com.google.cloud.pso.bq_snapshot_manager.functions.f03_snapshoter.GCSSnapshoter; import com.google.cloud.pso.bq_snapshot_manager.functions.f04_tagger.Tagger; import com.google.cloud.pso.bq_snapshot_manager.functions.f04_tagger.TaggerResponse; import com.google.cloud.pso.bq_snapshot_manager.functions.f04_tagger.TaggerRequest; import com.google.cloud.pso.bq_snapshot_manager.helpers.ControllerExceptionHelper; import com.google.cloud.pso.bq_snapshot_manager.helpers.LoggingHelper; import com.google.cloud.pso.bq_snapshot_manager.helpers.TrackingHelper; -import com.google.cloud.pso.bq_snapshot_manager.services.catalog.DataCatalogServiceImpl; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyService; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyServiceFireStoreImpl; +import com.google.cloud.pso.bq_snapshot_manager.services.backup_policy.BackupPolicyServiceGCSImpl; import com.google.cloud.pso.bq_snapshot_manager.services.map.GcsPersistentMapImpl; import com.google.cloud.pso.bq_snapshot_manager.services.map.PersistentMap; import com.google.cloud.pso.bq_snapshot_manager.services.set.GCSPersistentSetImpl; import com.google.gson.Gson; -import com.google.gson.JsonElement; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import org.springframework.boot.SpringApplication; @@ -60,7 +58,8 @@ public TaggerController() { logger = new LoggingHelper( TaggerController.class.getSimpleName(), functionNumber, - environment.getProjectId() + environment.getProjectId(), + environment.getApplicationName() ); } @@ -68,7 +67,7 @@ public TaggerController() { public ResponseEntity receiveMessage(@RequestBody PubSubEvent requestBody) { String trackingId = TrackingHelper.MIN_RUN_ID; - DataCatalogServiceImpl dataCatalogService = null; + BackupPolicyService backupPolicyService = null; // These values will be updated based on the execution flow and logged at the end ResponseEntity responseEntity; @@ -128,13 +127,13 @@ public ResponseEntity receiveMessage(@RequestBody PubSubEvent requestBody) { logger.logInfoWithTracker(taggerRequest.isDryRun(), trackingId, taggerRequest.getTargetTable(), String.format("Parsed Request: %s", taggerRequest.toString())); - dataCatalogService = new DataCatalogServiceImpl(); + backupPolicyService = new BackupPolicyServiceGCSImpl(environment.getGcsBackupPoliciesBucket()); Tagger tagger = new Tagger( - new LoggingHelper(Tagger.class.getSimpleName(), functionNumber, environment.getProjectId()), environment.toConfig(), - dataCatalogService, + backupPolicyService, new GCSPersistentSetImpl(environment.getGcsFlagsBucket()), - "tagger-flags" + "tagger-flags", + functionNumber ); taggerResponse = tagger.execute( @@ -157,8 +156,8 @@ public ResponseEntity receiveMessage(@RequestBody PubSubEvent requestBody) { error = e; }finally { - if(dataCatalogService != null){ - dataCatalogService.shutdown(); + if(backupPolicyService != null){ + backupPolicyService.shutdown(); } } diff --git a/terraform/main.tf b/terraform/main.tf index c074146..90b4de1 100644 --- a/terraform/main.tf +++ b/terraform/main.tf @@ -70,6 +70,10 @@ locals { name = "GCS_FLAGS_BUCKET", value = module.gcs.create_gcs_flags_bucket_name }, + { + name = "GCS_BACKUP_POLICIES_BUCKET", + value = module.gcs.create_gcs_backup_policies_bucket_name + }, { name = "APPLICATION_NAME", value = var.application_name @@ -111,7 +115,6 @@ module "gcs" { source = "./modules/gcs" gcs_flags_bucket_name = "${var.project}-${var.gcs_flags_bucket_name}" project = var.project - region = var.compute_region # because it's used by the cloud run services # both dispatchers should be admins. Add the inspection-dispatcher-sa only if it's being deployed gcs_flags_bucket_admins = [ @@ -122,6 +125,13 @@ module "gcs" { "serviceAccount:${module.iam.sa_tagger_email}", ] common_labels = local.common_labels + gcs_backup_policies_bucket_admins = [ + "serviceAccount:${module.iam.sa_configurator_email}", + "serviceAccount:${module.iam.sa_tagger_email}", + ] + gcs_backup_policies_bucket_name = "${var.project}-${var.gcs_backup_policies_bucket_name}" + compute_region = var.compute_region + data_region = var.data_region } module "bigquery" { @@ -131,6 +141,7 @@ module "bigquery" { dataset = var.bigquery_dataset_name logging_sink_sa = module.cloud_logging.service_account common_labels = local.common_labels + gcs_backup_policies_bucket_name = module.gcs.create_gcs_backup_policies_bucket_name } module "cloud_logging" { diff --git a/terraform/modules/bigquery/main.tf b/terraform/modules/bigquery/main.tf index e508ba8..037c31a 100644 --- a/terraform/modules/bigquery/main.tf +++ b/terraform/modules/bigquery/main.tf @@ -8,7 +8,6 @@ resource "google_bigquery_dataset" "results_dataset" { project = var.project location = var.region dataset_id = var.dataset - description = "To store DLP results from BQ Security Classifier app" labels = var.common_labels } @@ -246,6 +245,29 @@ resource "google_bigquery_table" "view_run_duration" { } +########## External tables ##################################### + +resource "google_bigquery_table" "external_gcs_backup_policies" { + dataset_id = google_bigquery_dataset.results_dataset.dataset_id + table_id = "ext_backup_policies" + + external_data_configuration { + source_format = "NEWLINE_DELIMITED_JSON" + hive_partitioning_options { + mode = "CUSTOM" # Custom means you must encode the partition key schema within the source_uri_prefix + source_uri_prefix = "gs://${var.gcs_backup_policies_bucket_name}/{project:STRING}/{dataset:STRING}/{table:STRING}" + + } + source_uris = [ + "gs://${var.gcs_backup_policies_bucket_name}/*.json", + ] + autodetect = false # Let BigQuery try to autodetect the schema and format of the table. + schema = file("modules/bigquery/schema/ext_backup_policies.json") + } + + deletion_protection = false +} + diff --git a/terraform/modules/bigquery/schema/ext_backup_policies.json b/terraform/modules/bigquery/schema/ext_backup_policies.json new file mode 100644 index 0000000..3fadd01 --- /dev/null +++ b/terraform/modules/bigquery/schema/ext_backup_policies.json @@ -0,0 +1,88 @@ +[ + { + "mode": "NULLABLE", + "name": "config_source", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "backup_cron", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "backup_method", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "backup_time_travel_offset_days", + "type": "INT64" + }, + { + "mode": "NULLABLE", + "name": "backup_storage_project", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "backup_operation_project", + "type": "STRING" + }, + + + + + + { + "mode": "NULLABLE", + "name": "bq_snapshot_storage_dataset", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "bq_snapshot_expiration_days", + "type": "FLOAT" + }, + { + "mode": "NULLABLE", + "name": "gcs_snapshot_storage_location", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "gcs_snapshot_format", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "gcs_csv_delimiter", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "gcs_csv_export_header", + "type": "BOOLEAN" + }, + { + "mode": "NULLABLE", + "name": "gcs_avro_use_logical_types", + "type": "BOOLEAN" + }, + { + "mode": "NULLABLE", + "name": "last_backup_at", + "type": "TIMESTAMP" + }, + { + "mode": "NULLABLE", + "name": "last_bq_snapshot_storage_uri", + "type": "STRING" + }, + { + "mode": "NULLABLE", + "name": "last_gcs_snapshot_storage_uri", + "type": "STRING" + } +] + diff --git a/terraform/modules/bigquery/variables.tf b/terraform/modules/bigquery/variables.tf index dd774cb..57432a0 100644 --- a/terraform/modules/bigquery/variables.tf +++ b/terraform/modules/bigquery/variables.tf @@ -18,6 +18,10 @@ variable "common_labels" { type = map(string) } +variable "gcs_backup_policies_bucket_name" { + type = string +} + diff --git a/terraform/modules/bigquery/views/v_audit_log_by_table.tpl b/terraform/modules/bigquery/views/v_audit_log_by_table.tpl index 4c039ef..5b738cd 100644 --- a/terraform/modules/bigquery/views/v_audit_log_by_table.tpl +++ b/terraform/modules/bigquery/views/v_audit_log_by_table.tpl @@ -21,7 +21,7 @@ CAST(JSON_VALUE(jsonPayload.unified_input_json, '$.isForceRun') AS BOOL) AS is_f CAST(JSON_VALUE(jsonPayload.unified_output_json, '$.isBackupCronTime') AS BOOL) AS is_backup_cron_time, CAST(JSON_VALUE(jsonPayload.unified_output_json, '$.isTableCreatedBeforeTimeTravel') AS BOOL) AS is_table_created_before_time_travel, CAST(JSON_VALUE(jsonPayload.unified_output_json, '$.isBackupTime') AS BOOL) AS is_backup_time, -JSON_VALUE(jsonPayload.unified_output_json, '$.backupPolicy.method') AS backup_method, +JSON_VALUE(jsonPayload.unified_output_json, '$.backupPolicy.backup_method') AS backup_method, CAST(JSON_VALUE(jsonPayload.unified_input_json, '$.isDryRun') AS BOOL) AS is_dry_run, timestamp AS configurator_log_ts FROM `${project}.${dataset}.${logging_table}` diff --git a/terraform/modules/bigquery/views/v_service_calls.tpl b/terraform/modules/bigquery/views/v_service_calls.tpl index 832aaee..55f3515 100644 --- a/terraform/modules/bigquery/views/v_service_calls.tpl +++ b/terraform/modules/bigquery/views/v_service_calls.tpl @@ -12,5 +12,5 @@ SUM(CASE WHEN step = 'START' AND function_number = 4 THEN 1 ELSE 0 END) AS tagg SUM(CASE WHEN step = 'END' AND function_number = 4 THEN 1 ELSE 0 END) AS tagger_ends FROM `${project}.${dataset}.${logging_view_steps}` -WHERE function_number > 1 +WHERE function_number <> 1 GROUP BY 1,2,3 \ No newline at end of file diff --git a/terraform/modules/gcs/main.tf b/terraform/modules/gcs/main.tf index a0e558c..4baf248 100644 --- a/terraform/modules/gcs/main.tf +++ b/terraform/modules/gcs/main.tf @@ -4,7 +4,7 @@ resource "google_storage_bucket" "gcs_flags_bucket" { project = var.project name = var.gcs_flags_bucket_name # This bucket is used by the services so let's create in the same compute region - location = var.region + location = var.compute_region # force_destroy = true @@ -28,3 +28,25 @@ resource "google_storage_bucket_iam_binding" "gcs_flags_bucket_iam_bindings" { role = "roles/storage.objectAdmin" members = var.gcs_flags_bucket_admins } + + +// Backup Polices + +resource "google_storage_bucket" "gcs_backup_policies_bucket" { + project = var.project + name = var.gcs_backup_policies_bucket_name + # This bucket must be created in the same region that BigQuery dataset is created + location = var.data_region + + force_destroy = false + + uniform_bucket_level_access = true + + labels = var.common_labels +} + +resource "google_storage_bucket_iam_binding" "gcs_backup_policies_iam_bindings" { + bucket = google_storage_bucket.gcs_backup_policies_bucket.name + role = "roles/storage.objectAdmin" + members = var.gcs_backup_policies_bucket_admins +} \ No newline at end of file diff --git a/terraform/modules/gcs/output.tf b/terraform/modules/gcs/output.tf index 66cef1d..41fa88d 100644 --- a/terraform/modules/gcs/output.tf +++ b/terraform/modules/gcs/output.tf @@ -1,3 +1,7 @@ output "create_gcs_flags_bucket_name" { value = google_storage_bucket.gcs_flags_bucket.name +} + +output "create_gcs_backup_policies_bucket_name" { + value = google_storage_bucket.gcs_backup_policies_bucket.name } \ No newline at end of file diff --git a/terraform/modules/gcs/variables.tf b/terraform/modules/gcs/variables.tf index 295b224..7fd67c8 100644 --- a/terraform/modules/gcs/variables.tf +++ b/terraform/modules/gcs/variables.tf @@ -1,6 +1,8 @@ variable "project" {type = string} -variable "region" {type = string} +variable "compute_region" {type = string} + +variable "data_region" {type = string} variable "gcs_flags_bucket_name" {type = string} @@ -8,6 +10,12 @@ variable "gcs_flags_bucket_admins" { type = list(string) } +variable "gcs_backup_policies_bucket_name" {type = string} + +variable "gcs_backup_policies_bucket_admins" { + type = list(string) +} + variable "common_labels" { type = map(string) } \ No newline at end of file diff --git a/terraform/modules/iam/main.tf b/terraform/modules/iam/main.tf index 884e749..fef0131 100644 --- a/terraform/modules/iam/main.tf +++ b/terraform/modules/iam/main.tf @@ -83,6 +83,14 @@ resource "google_service_account_iam_member" "sa_dispatcher_account_user_sa_disp #### Configurator SA Permissions ### +// read backup policies when using datastore as backend +resource "google_project_iam_member" "sa_configurator_datastore_viewer" { +project = var.project +role = "roles/datastore.viewer" +member = "serviceAccount:${google_service_account.sa_configurator.email}" +} + +// read backup policies when using data catalog as backend resource "google_project_iam_member" "sa_configurator_datacatalog_viewer" { project = var.project role = "roles/datacatalog.viewer" @@ -116,6 +124,14 @@ resource "google_service_account_iam_member" "sa_snapshoter_gcs_account_user_sa_ } #### Tagger SA Permissions ### + +// read / write for backup policies when using datastore as backend +resource "google_project_iam_member" "sa_tagger_datastore_user" { +project = var.project +role = "roles/datastore.user" +member = "serviceAccount:${google_service_account.sa_tagger.email}" +} + resource "google_project_iam_member" "sa_tagger_datacatalog_viewer" { project = var.project role = "roles/datacatalog.viewer" diff --git a/terraform/variables.tf b/terraform/variables.tf index 339d28d..1724554 100644 --- a/terraform/variables.tf +++ b/terraform/variables.tf @@ -171,7 +171,12 @@ variable "tagger_pubsub_sub" { variable "gcs_flags_bucket_name" { type = string - default = "bq-snapshot-mgr-flags" + default = "bq-backup-manager-flags" +} + +variable "gcs_backup_policies_bucket_name" { + type = string + default = "bq-backup-manager-policies" } # Images