This repository has been archived by the owner on Aug 25, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
18 changed files
with
282 additions
and
100 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| import datetime | ||
| import json | ||
| import logging | ||
| import uuid | ||
|
|
||
| from apiclient.errors import Error | ||
| import httplib2 | ||
| import googleapiclient.discovery | ||
| from oauth2client.client import GoogleCredentials | ||
|
|
||
| from src.commons.big_query.big_query_table import BigQueryTable | ||
| from src.commons.decorators.retry import retry | ||
| from src.commons.error_reporting import ErrorReporting | ||
|
|
||
|
|
||
| class DataStreamer(object): | ||
|
|
||
| @staticmethod | ||
| def _create_http(): | ||
| return httplib2.Http(timeout=60) | ||
|
|
||
| def __init__(self, project_id, dataset_id, table_id): | ||
| self.big_query_table = BigQueryTable(project_id, dataset_id, table_id) | ||
| self.service = googleapiclient.discovery.build( | ||
| 'bigquery', | ||
| 'v2', | ||
| credentials=GoogleCredentials.get_application_default(), | ||
| http=self._create_http() | ||
| ) | ||
|
|
||
| def stream_stats(self, rows, insert_id=None): | ||
| insert_id = insert_id or uuid.uuid4() | ||
| insert_all_data = { | ||
| 'rows': [{ | ||
| 'json': data, | ||
| 'insertId': str(insert_id) | ||
| } for data in rows] | ||
| } | ||
| logging.info("Streaming data to table %s (insertId:%s)", | ||
| self.big_query_table, insert_id) | ||
| insert_all_response = self._stream_metadata(insert_all_data) | ||
| if 'insertErrors' in insert_all_response: | ||
| logging.debug("Sent json: \n%s", json.dumps(insert_all_data)) | ||
| error_message = "Error during streaming metadata to BigQuery: \n{}"\ | ||
| .format(json.dumps(insert_all_response['insertErrors'])) | ||
| logging.error(error_message) | ||
| ErrorReporting().report(error_message) | ||
| else: | ||
| logging.debug("Stats have been sent successfully to %s table", | ||
| self.big_query_table) | ||
|
|
||
| @retry(Error, tries=2, delay=2, backoff=2) | ||
| def _stream_metadata(self, insert_all_data): | ||
| partition = datetime.datetime.now().strftime("%Y%m%d") | ||
| return self.service.tabledata().insertAll( | ||
| projectId=self.big_query_table.get_project_id(), | ||
| datasetId=self.big_query_table.get_dataset_id(), | ||
| tableId='{}${}'.format(self.big_query_table.get_table_id(), | ||
| partition), | ||
| body=insert_all_data).execute(num_retries=3) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,12 +1,22 @@ | ||
| import logging | ||
|
|
||
| from src.commons.big_query.streaming.data_streamer import DataStreamer | ||
| from src.commons.config.configuration import configuration | ||
|
|
||
| class SLIResultsStreamer(object): | ||
|
|
||
| def __init__(self, big_query): | ||
| self.big_query = big_query | ||
| class SLIResultsStreamer(object): | ||
| def __init__(self, | ||
| project_id=configuration.backup_project_id, | ||
| dataset_id="SLI_history", | ||
| table_id="SLI_backup_creation_latency" | ||
| ): | ||
| self.data_streamer = DataStreamer(project_id=project_id, dataset_id=dataset_id, table_id=table_id) | ||
|
|
||
| def stream(self, sli_results): | ||
| logging.info("TODO here results should be streamed into BQ") | ||
| logging.info("SLI results: %s", sli_results) | ||
| pass | ||
| if len(sli_results) == 0: | ||
| logging.info("Nothing to stream") | ||
| return | ||
|
|
||
| logging.info("Streaming SLI results: %s", sli_results) | ||
| self.data_streamer.stream_stats(sli_results) | ||
| logging.info("SLI results streamed") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| resource "google_bigquery_table" "SLI_filtered_table" { | ||
| project = "${var.SLI_views_destination_project}" | ||
| dataset_id = "${var.SLI_history_dataset}" | ||
| table_id = "SLI_backup_creation_latency" | ||
|
|
||
| time_partitioning { | ||
| type = "DAY" | ||
| } | ||
|
|
||
| schema= "${file("SLI_filtered_table_schema.json")}" | ||
|
|
||
| depends_on = ["google_bigquery_dataset.SLI_history_dataset"] | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| [ | ||
| { | ||
| "name": "snapshotTime", | ||
| "type": "TIMESTAMP", | ||
| "mode": "REQUIRED", | ||
| "description": "Where the data was filtered and put into the table" | ||
| }, | ||
| { | ||
| "name": "projectId", | ||
| "type": "STRING", | ||
| "mode": "REQUIRED", | ||
| "description": "ProjectId of table that could not be backed up" | ||
| }, | ||
| { | ||
| "name": "datasetId", | ||
| "type": "STRING", | ||
| "mode": "REQUIRED", | ||
| "description": "DatasetId of table that could not be backed up" | ||
| }, | ||
| { | ||
| "name": "tableId", | ||
| "type": "STRING", | ||
| "mode": "REQUIRED", | ||
| "description": "TableId of table that could not be backed up" | ||
| }, | ||
| { | ||
| "name": "partitionId", | ||
| "type": "STRING", | ||
| "mode": "NULLABLE", | ||
| "description": "PartitionId of table that could not be backed up" | ||
| }, | ||
| { | ||
| "name": "creationTime", | ||
| "type": "TIMESTAMP", | ||
| "mode": "REQUIRED", | ||
| "description": "CreationTime of table that could not be backed up" | ||
| }, | ||
| { | ||
| "name": "lastModifiedTime", | ||
| "type": "TIMESTAMP", | ||
| "mode": "REQUIRED", | ||
| "description": "LastModifiedTime of table that could not be backed up" | ||
| }, | ||
| { | ||
| "name": "backupCreated", | ||
| "type": "TIMESTAMP", | ||
| "mode": "REQUIRED", | ||
| "description": "date of backup table creation (copyJob end)" | ||
| }, | ||
| { | ||
| "name": "backupLastModified", | ||
| "type": "TIMESTAMP", | ||
| "mode": "REQUIRED", | ||
| "description": "date of last modification that will be included in the Backup (copyJob start time - as it is atomic operation and every change before that point is included in copy)" | ||
| }, | ||
| { | ||
| "name": "xDays", | ||
| "type": "STRING", | ||
| "mode": "REQUIRED", | ||
| "description": "the number of days, the table was not backed up" | ||
| } | ||
| ] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| resource "google_bigquery_dataset" "SLI_history_dataset" { | ||
| project = "${var.SLI_views_destination_project}" | ||
| dataset_id = "${var.SLI_history_dataset}" | ||
| location = "${var.SLI_views_location}" | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| resource "google_bigquery_dataset" "SLI_views_for_datastudio_legacy_dataset" { | ||
| dataset_id = "SLI_views_for_datastudio" | ||
| project = "${var.SLI_views_destination_project}" | ||
| location = "${var.SLI_views_location}" | ||
| } | ||
|
|
||
| resource "google_bigquery_table" "SLI_X_days_view" { | ||
| project = "${var.SLI_views_destination_project}" | ||
| dataset_id = "SLI_views_for_datastudio" | ||
| table_id = "SLI_X_days" | ||
|
|
||
| view { | ||
| query = <<EOF | ||
| #legacySQL | ||
| SELECT * FROM | ||
| (select *, "3" as days from [${var.SLI_views_destination_project}.${var.SLI_views_legacy_dataset}.SLI_3_days]), | ||
| (select *, "4" as days from [${var.SLI_views_destination_project}.${var.SLI_views_legacy_dataset}.SLI_4_days]), | ||
| (select *, "5" as days from [${var.SLI_views_destination_project}.${var.SLI_views_legacy_dataset}.SLI_5_days]), | ||
| (select *, "7" as days from [${var.SLI_views_destination_project}.${var.SLI_views_legacy_dataset}.SLI_7_days]) | ||
| EOF | ||
| use_legacy_sql = true | ||
| } | ||
|
|
||
| depends_on = ["google_bigquery_table.SLI_3_days_view", "google_bigquery_table.SLI_4_days_view", "google_bigquery_table.SLI_5_days_view", "google_bigquery_table.SLI_7_days_view"] | ||
| } | ||
|
|
||
| resource "google_bigquery_table" "SLI_days_by_count_view" { | ||
| project = "${var.SLI_views_destination_project}" | ||
| dataset_id = "SLI_views_for_datastudio" | ||
| table_id = "SLI_days_by_count" | ||
|
|
||
| view { | ||
| query = <<EOF | ||
| #legacySQL | ||
| SELECT days, count(*) as count FROM [SLI_views_for_datastudio.SLI_X_days] GROUP BY days | ||
| EOF | ||
| use_legacy_sql = true | ||
| } | ||
|
|
||
| depends_on = ["google_bigquery_table.SLI_X_days_view", "google_bigquery_dataset.SLI_views_for_datastudio_legacy_dataset"] | ||
| } |
Oops, something went wrong.