[ghinfra] Set up ingestion from s3 -> clickhouse for cloudwatch (#6898)

Path: cloudwatch metrics -> firehose -> s3 (new bucket
fbossci-cloudwatch-metrics) -> clickhouse

This is the s3 -> clickhouse part
I think clickhouse has some in built ingestions for kinesis but I'm
lazy...

Requires meta-pytorch/pytorch-gha-infra#751

Testing: ran the python code via
`python tools/rockset_migration/s32ch.py --clickhouse-table
"infra_metrics.cloudwatch_metrics" --stored-data t.json --s3-bucket
fbossci-cloudwatch-metrics --s3-prefix ghci-related`

Author: clee2000

aws/lambda/clickhouse-replicator-s3/lambda_function.py

@@ -544,6 +544,25 @@ def disabled_tests_historical_adapter(table, bucket, key):
     general_adapter(table, bucket, key, schema, ["none"], "JSONEachRow")
 
 
+def cloudwatch_metrics_adapter(table, bucket, key):
+    schema = """
+    `metric_stream_name` LowCardinality(String),
+    `account_id` LowCardinality(String),
+    `region` LowCardinality(String),
+    `namespace` LowCardinality(String),
+    `metric_name` LowCardinality(String),
+    `dimensions` Map(String, String),
+    `timestamp` DateTime,
+    `value` Tuple(
+        max Float32,
+        min Float32,
+        sum Float32,
+        count Float32),
+    `unit` LowCardinality(String)
+    """
+    general_adapter(table, bucket, key, schema, ["none"], "JSONEachRow")
+
+
 SUPPORTED_PATHS = {
     "merges": "default.merges",
     "queue_times_historical": "default.queue_times_historical",
@@ -564,6 +583,8 @@ def disabled_tests_historical_adapter(table, bucket, key):
     "util_metadata": "misc.oss_ci_utilization_metadata",
     "util_timeseries": "misc.oss_ci_time_series",
     "disabled_tests_historical": "misc.disabled_tests_historical",
+    # fbossci-cloudwatch-metrics bucket
+    "ghci-related": "infra_metrics.cloudwatch_metrics",
 }
 
 OBJECT_CONVERTER = {
@@ -586,6 +607,7 @@ def disabled_tests_historical_adapter(table, bucket, key):
     "misc.oss_ci_utilization_metadata": oss_ci_util_metadata_adapter,
     "misc.oss_ci_time_series": oss_ci_util_time_series_adapter,
     "misc.disabled_tests_historical": disabled_tests_historical_adapter,
+    "infra_metrics.cloudwatch_metrics": cloudwatch_metrics_adapter,
 }
 
 

clickhouse_db_schema/infra_metrics.cloudwatch_metrics/schema.sql

@@ -0,0 +1,24 @@
+CREATE TABLE infra_metrics.cloudwatch_metrics
+(
+    `metric_stream_name` LowCardinality(String),
+    `account_id` LowCardinality(String),
+    `region` LowCardinality(String),
+    `namespace` LowCardinality(String),
+    `metric_name` LowCardinality(String),
+    `dimensions` Map(String, String),
+    `timestamp` DateTime,
+    `value` Tuple(
+        max Float32,
+        min Float32,
+        sum Float32,
+        count Float32),
+    `unit` LowCardinality(String),
+    `_meta` Tuple(
+        bucket String,
+        key String)
+)
+ENGINE = SharedMergeTree('/clickhouse/tables/{uuid}/{shard}', '{replica}')
+PARTITION BY toYYYYMM(timestamp)
+ORDER BY (namespace, metric_name, timestamp, dimensions)
+TTL timestamp + toIntervalMonth(12)
+SETTINGS index_granularity = 8192