Merge pull request #492 from dynatrace-oss/PCLOUDS-3555-filter_metric…

…s_and_dimensions-V2

PCLOUDS-3555: Filter metrics and dimensions - V2

k8s/helm-chart/dynatrace-gcp-monitor/templates/function-config.yaml

@@ -29,6 +29,7 @@ data:
   SCOPING_PROJECT_SUPPORT_ENABLED: {{ .Values.scopingProjectSupportEnabled | quote }}
   EXCLUDED_PROJECTS: {{ .Values.excludedProjects | quote }}
   EXCLUDED_PROJECTS_BY_PREFIX: {{ .Values.excludedProjectsByPrefix | quote }}
+  EXCLUDED_METRICS_AND_DIMENSIONS: {{ .Values.excludedMetricsAndDimensions | quote }}
   KEEP_REFRESHING_EXTENSIONS_CONFIG: {{ .Values.keepRefreshingExtensionsConfig | quote }}
   {{- if or (eq .Values.deploymentType "metrics") (eq .Values.deploymentType "all") }}
   PRINT_METRIC_INGEST_INPUT: {{ .Values.printMetricIngestInput | quote }}

k8s/helm-chart/dynatrace-gcp-monitor/templates/function-deployment.yml

@@ -59,6 +59,8 @@ spec:
           - key: AUTODISCOVERY_METRIC_BLOCK_LIST_YAML
             path: "autodiscovery-block-list.yaml"
           {{- end }}
+          - key: EXCLUDED_METRICS_AND_DIMENSIONS
+            path: "metrics-filter-out.yaml"
       {{- end }}
       {{- if (.Values.imagePullSecrets) }}      
       imagePullSecrets:
@@ -167,7 +169,6 @@ spec:
             configMapKeyRef:
               name: dynatrace-gcp-monitor-config
               key: EXCLUDED_PROJECTS_BY_PREFIX
-
         - name:   KEEP_REFRESHING_EXTENSIONS_CONFIG
           valueFrom:
             configMapKeyRef:

k8s/helm-chart/dynatrace-gcp-monitor/values.yaml

@@ -132,7 +132,27 @@ scopingProjectSupportEnabled: "false"
 excludedProjects: ""
 # excludedProjectsByPrefix: comma separated list of projects substring that will be excluded from monitoring (e.g. "project-a,proj,pro").
 excludedProjectsByPrefix: ""
-
+# excludedMetricsAndDimensions: map structure of metrics and dimensions that will be excluded from monitoring.
+# If you want to exclude a metric, add a "metric" entry. Any global metric matching its prefix will not be ingested.
+# If you want to exclude dimensions from a specific metric, add the "metric" entry and specify the dimensions. Selected dimensions will be cut off and the metric will be ingested without them.
+# Example:
+# this WILL NOT ingest any metric starting with "bigquery.googleapis.com/query/"
+# and WILL ingest "compute.googleapis.com/instance/disk/max_read_bytes_count" but WITHOUT the "storage_type" and "device_type" dimensions
+#
+# excludedMetricsAndDimensions: |
+#   filter_out:
+#   - metric: bigquery.googleapis.com/query/
+#   - metric: compute.googleapis.com/instance/disk/max_read_bytes_count
+#     dimensions:
+#       - storage_type
+#       - device_type
+#
+excludedMetricsAndDimensions: |
+  filter_out:
+#   - metric: metric_1
+#   - metric: metric_2
+#     dimensions:
+#       - dimension_A
 metricResources:
   requests:
     memory: "1536Mi"

src/lib/metric_ingest.py

@@ -161,8 +161,25 @@ async def fetch_metric(
         context: MetricsContext,
         project_id: str,
         service: GCPService,
-        metric: Metric
+        metric: Metric,
+        excluded_metrics_and_dimensions: list
 ) -> List[IngestLine]:
+    def should_exclude_dimension(dimension: Dimension):
+        found_excluded_metric = None
+
+        for excluded_metric in excluded_metrics_and_dimensions:
+            if metric.google_metric.startswith(excluded_metric.get("metric")):
+                found_excluded_metric = excluded_metric
+                break
+
+        if not found_excluded_metric:
+            return False
+
+        dimension_key = dimension.key_for_fetch_metric
+        has_dimension_key = dimension_key[dimension_key.rfind(".") + 1:] in found_excluded_metric.get("dimensions", [])
+
+        return has_dimension_key
+
     end_time = (context.execution_time - metric.ingest_delay)
     start_time = (end_time - context.execution_interval)
 
@@ -190,6 +207,11 @@ async def fetch_metric(
     all_dimensions = (service_dimensions + metric.dimensions)
     dt_dimensions_mapping = DtDimensionsMap()
     for dimension in all_dimensions:
+        if should_exclude_dimension(dimension):
+            context.log(
+                f"Skiping fetching dimension {dimension.key_for_create_entity_id} for metric {metric.google_metric}")
+            continue
+
         if dimension.key_for_send_to_dynatrace:
             dt_dimensions_mapping.add_label_mapping(dimension.key_for_fetch_metric, dimension.key_for_send_to_dynatrace)
 

src/lib/utilities.py

@@ -27,6 +27,7 @@ def chunks(full_list: List, chunk_size: int) -> List[List]:
     chunk_size = max(1, chunk_size)
     return [full_list[i:i + chunk_size] for i in range(0, len(full_list), chunk_size)]
 
+
 def safe_read_yaml(filepath: str, alternative_environ_name: str):
     try:
         with open(filepath, encoding="utf-8") as activation_file:
@@ -37,18 +38,33 @@ def safe_read_yaml(filepath: str, alternative_environ_name: str):
         yaml_dict = {}
     return yaml_dict
 
+
 def read_activation_yaml():
     return safe_read_yaml('/code/config/activation/gcp_services.yaml', "ACTIVATION_CONFIG" )
 
+
 def read_autodiscovery_config_yaml():
     return safe_read_yaml('/code/config/activation/autodiscovery-config.yaml', "AUTODISCOVERY_RESOURCES_YAML" )
 
+
+def read_filter_out_list_yaml() -> list:
+    loaded_yaml = safe_read_yaml("/code/config/activation/metrics-filter-out.yaml", "EXCLUDED_METRICS_AND_DIMENSIONS") or {}
+    excluded_metrics = loaded_yaml.get("filter_out", [])
+
+    for metric in excluded_metrics:
+        metric["dimensions"] = set(metric.get("dimensions", []))
+
+    return excluded_metrics
+
+
 def read_autodiscovery_block_list_yaml():
     return safe_read_yaml('/code/config/activation/autodiscovery-block-list.yaml', "AUTODISCOVERY_BLOCK_LIST_YAML" )
 
+
 def read_autodiscovery_resources_mapping():
     return safe_read_yaml('./lib/autodiscovery/config/autodiscovery-mapping.yaml', "AUTODISCOVERY_RESOURCES_MAPPING")
 
+
 def get_activation_config_per_service(activation_yaml):
     return {service_activation.get('service'): service_activation for service_activation in
             activation_yaml['services']} if activation_yaml and activation_yaml['services'] else {}

src/main.py

@@ -34,6 +34,7 @@
 from lib.sfm.for_metrics.metrics_definitions import SfmKeys
 from lib.topology.topology import fetch_topology, build_entity_id_map
 from lib.sfm.api_call_latency import ApiCallLatency
+from lib.utilities import read_filter_out_list_yaml
 
 
 async def async_dynatrace_gcp_extension(services: Optional[List[GCPService]] = None):
@@ -144,8 +145,11 @@ async def query_metrics(execution_id: Optional[str], services: Optional[List[GCP
 
         context.start_processing_timestamp = time.time()
 
+        excluded_metrics_and_dimensions = read_filter_out_list_yaml()
+
         process_project_metrics_tasks = [
-            process_project_metrics(context, project_id, services, disabled_apis_by_project_id.get(project_id, set()))
+            process_project_metrics(context, project_id, services, disabled_apis_by_project_id.get(project_id, set()),
+                                    excluded_metrics_and_dimensions)
             for project_id
             in projects_ids
         ]
@@ -167,10 +171,11 @@ async def query_metrics(execution_id: Optional[str], services: Optional[List[GCP
 
 
 async def process_project_metrics(context: MetricsContext, project_id: str, services: List[GCPService],
-                                  disabled_apis: Set[str]):
+                                  disabled_apis: Set[str], excluded_metrics_and_dimensions: list):
     try:
         context.log(project_id, f"Starting processing...")
-        ingest_lines = await fetch_ingest_lines_task(context, project_id, services, disabled_apis)
+        ingest_lines = await fetch_ingest_lines_task(context, project_id, services, disabled_apis,
+                                                     excluded_metrics_and_dimensions)
         fetch_data_time = time.time() - context.start_processing_timestamp
         context.sfm[SfmKeys.fetch_gcp_data_execution_time].update(project_id, fetch_data_time)
         context.log(project_id, f"Finished fetching data in {fetch_data_time}")
@@ -180,7 +185,17 @@ async def process_project_metrics(context: MetricsContext, project_id: str, serv
 
 
 async def fetch_ingest_lines_task(context: MetricsContext, project_id: str, services: List[GCPService],
-                                  disabled_apis: Set[str]) -> List[IngestLine]:
+                                  disabled_apis: Set[str], excluded_metrics_and_dimensions: list) -> List[IngestLine]:
+    def should_exclude_metric(metric: Metric):
+        found_excluded_metric = None
+
+        for excluded_metric in excluded_metrics_and_dimensions:
+            if metric.google_metric.startswith(excluded_metric.get("metric")):
+                found_excluded_metric = excluded_metric
+                break
+
+        return found_excluded_metric and not found_excluded_metric.get("dimensions")
+
     fetch_metric_coros = []
     metrics_metadata = []
     topology: Dict[GCPService, Iterable[Entity]] = {}
@@ -195,6 +210,7 @@ async def fetch_ingest_lines_task(context: MetricsContext, project_id: str, serv
     # and metrics from all projects are being collected
     skipped_services_with_no_instances = []
     skipped_disabled_apis = set()
+    skipped_excluded_metrics = []
 
     for service in services:
         if not service.is_enabled:
@@ -203,6 +219,10 @@ async def fetch_ingest_lines_task(context: MetricsContext, project_id: str, serv
             skipped_services_with_no_instances.append(f"{service.name}/{service.feature_set}")
             continue  # skip fetching the metrics because there are no instances
         for metric in service.metrics:
+            if should_exclude_metric(metric):
+                context.log(f"Skiping fetching all the data for the metric {metric.google_metric}")
+                continue
+
             # Fetch metric only if it's metric from extensions or is autodiscovered in project_id
             if not metric.autodiscovered_metric or project_id in metric.project_ids:
                 gcp_api_last_index = metric.google_metric.find("/")
@@ -211,7 +231,8 @@ async def fetch_ingest_lines_task(context: MetricsContext, project_id: str, serv
                     skipped_disabled_apis.add(api)
                     continue  # skip fetching the metrics because service API is disabled
                 fetch_metric_coro = run_fetch_metric(
-                    context=context, project_id=project_id, service=service, metric=metric
+                    context=context, project_id=project_id, service=service, metric=metric,
+                    excluded_metrics_and_dimensions=excluded_metrics_and_dimensions
                 )
                 fetch_metric_coros.append(fetch_metric_coro)
 
@@ -224,6 +245,9 @@ async def fetch_ingest_lines_task(context: MetricsContext, project_id: str, serv
         skipped_disabled_apis_string = ", ".join(skipped_disabled_apis)
         context.log(project_id, f"Skipped fetching metrics for disabled APIs: {skipped_disabled_apis_string}")
 
+    if skipped_excluded_metrics:
+        context.log(project_id, f"Skipped fetching for excluded metrics: {', '.join(skipped_excluded_metrics)}")
+
     fetch_metric_results = await asyncio.gather(*fetch_metric_coros, return_exceptions=True)
     entity_id_map = build_entity_id_map(list(topology.values()))
     flat_metric_results = flatten_and_enrich_metric_results(context, fetch_metric_results, entity_id_map)
@@ -236,10 +260,11 @@ async def run_fetch_metric(
         context: MetricsContext,
         project_id: str,
         service: GCPService,
-        metric: Metric
+        metric: Metric,
+        excluded_metrics_and_dimensions: list
 ):
     try:
-        return await fetch_metric(context, project_id, service, metric)
+        return await fetch_metric(context, project_id, service, metric, excluded_metrics_and_dimensions)
     except Exception as e:
         context.log(project_id, f"Failed to finish task for [{metric.google_metric}], reason is {type(e).__name__} {e}")
         return []

tests/e2e/lib-tests.sh

@@ -90,6 +90,9 @@ activeGate:
   dynatracePaasToken: "${DYNATRACE_PAAS_TOKEN}"
 serviceAccount: "${IAM_SERVICE_ACCOUNT}"
 scopingProjectSupportEnabled: "true"
+excludedMetricsAndDimensions: |
+  filter_out:
+    - metric: dummy_metric_name
 gcpServicesYaml: |
   services:
     - service: api