[gcp][feat] Add metrics collection

fixlib/fixlib/baseresources.py

@@ -116,8 +116,7 @@ def get(self) -> Dict[str, Any]:
         return changes
 
 
-# todo: replace to StrEnum once resoto is on 3.11
-class MetricName(str, Enum):
+class MetricName(StrEnum):
     def __str__(self) -> str:
         return self.value
 
@@ -195,6 +194,8 @@ def __str__(self) -> str:
     DiskQueueDepth = "disk_queue_depth"
     NetworkReceiveThroughput = "network_receive_throughput"
     NetworkTransmitThroughput = "network_transmit_throughput"
+    NetworkBytesSent = "network_bytes_sent"
+    NetworkBytesReceived = "network_bytes_received"
 
     # serverless
     Invocations = "invocations"

plugins/aws/fix_plugin_aws/collector.py

@@ -334,8 +334,9 @@ def collect_usage_metrics(self, builder: GraphBuilder) -> None:
                 continue
             # region can be overridden in the query: s3 is global, but need to be queried per region
             if region := cast(AwsRegion, resource.region()):
-                lookup_map[resource.id] = resource
                 resource_queries: List[cloudwatch.AwsCloudwatchQuery] = resource.collect_usage_metrics(builder)
+                if resource_queries:
+                    lookup_map[resource.id] = resource
                 for query in resource_queries:
                     query_region = query.region or region
                     start = query.start_delta or builder.metrics_delta

plugins/azure/fix_plugin_azure/resource/metrics.py

@@ -1,3 +1,4 @@
+from copy import deepcopy
 from datetime import datetime, timedelta
 from concurrent.futures import as_completed
 import logging
@@ -271,12 +272,13 @@ def _query_for_single(
         interval: str,
     ) -> "Tuple[Optional[AzureMetricData], Optional[str]]":
         try:
+            local_api_spec = deepcopy(api_spec)
             # Set the path for the API call based on the instance ID of the query
-            api_spec.path = f"{query.instance_id}/providers/Microsoft.Insights/metrics"
+            local_api_spec.path = f"{query.instance_id}/providers/Microsoft.Insights/metrics"
             # Retrieve metric data from the API
             aggregation = ",".join(query.aggregation)
             part = builder.client.list(
-                api_spec,
+                local_api_spec,
                 metricnames=query.metric_name,
                 metricNamespace=query.metric_namespace,
                 timespan=timespan,

plugins/gcp/fix_plugin_gcp/__init__.py

@@ -10,6 +10,7 @@
 from fixlib.core.actions import CoreFeedback
 from fixlib.graph import Graph, MaxNodesExceeded
 from fixlib.logger import log, setup_logger
+from fixlib.types import Json
 from .collector import GcpProjectCollector
 from .config import GcpConfig
 from .resources.base import GcpProject
@@ -77,10 +78,11 @@ def collect(self) -> None:
                     project_id,
                     feedback,
                     cloud,
+                    self.task_data or {},
                     max_resources_per_account=self.max_resources_per_account,
                     **collect_args,
                 )
-                for project_id in credentials.keys()
+                for project_id in credentials
             ]
             for future in futures.as_completed(wait_for):
                 project_graph = future.result()
@@ -98,6 +100,7 @@ def collect_project(
         project_id: str,
         core_feedback: CoreFeedback,
         cloud: Cloud,
+        task_data: Json,
         args: Optional[Namespace] = None,
         running_config: Optional[RunningConfig] = None,
         credentials: Optional[Dict[str, Any]] = None,
@@ -130,7 +133,7 @@ def collect_project(
 
         try:
             core_feedback.progress_done(project_id, 0, 1)
-            gpc = GcpProjectCollector(Config.gcp, cloud, project, core_feedback, max_resources_per_account)
+            gpc = GcpProjectCollector(Config.gcp, cloud, project, core_feedback, task_data, max_resources_per_account)
             try:
                 gpc.collect()
             except MaxNodesExceeded as ex:

plugins/gcp/fix_plugin_gcp/collector.py

@@ -1,6 +1,8 @@
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
 import logging
 from concurrent.futures import ThreadPoolExecutor
-from typing import Type, List, Any, Optional
+from typing import Type, List, Any, Optional, cast
 
 from fix_plugin_gcp.config import GcpConfig
 from fix_plugin_gcp.gcp_client import GcpApiSpec
@@ -14,12 +16,15 @@
     firestore,
     filestore,
     cloudfunctions,
+    monitoring,
 )
 from fix_plugin_gcp.resources.base import GcpResource, GcpProject, ExecutorQueue, GraphBuilder, GcpRegion, GcpZone
 from fix_plugin_gcp.utils import Credentials
 from fixlib.baseresources import Cloud
 from fixlib.core.actions import CoreFeedback, ErrorAccumulator
 from fixlib.graph import Graph
+from fixlib.json import value_in_path
+from fixlib.types import Json
 
 log = logging.getLogger("fix.plugins.gcp")
 all_resources: List[Type[GcpResource]] = (
@@ -58,6 +63,7 @@ def __init__(
         cloud: Cloud,
         project: GcpProject,
         core_feedback: CoreFeedback,
+        task_data: Json,
         max_resources_per_account: Optional[int] = None,
     ) -> None:
         self.config = config
@@ -67,6 +73,7 @@ def __init__(
         self.error_accumulator = ErrorAccumulator()
         self.graph = Graph(root=self.project, max_nodes=max_resources_per_account)
         self.credentials = Credentials.get(self.project.id)
+        self.task_data = task_data
 
     def collect(self) -> None:
         with ThreadPoolExecutor(
@@ -77,7 +84,20 @@ def collect(self) -> None:
             # It should only be used in scenarios, where it is safe to do so.
             # This executor is shared between all regions.
             shared_queue = ExecutorQueue(executor, self.project.safe_name)
+
+            def get_last_run() -> Optional[datetime]:
+                td = self.task_data
+                if not td:
+                    return None
+                timestamp = value_in_path(td, ["timing", td.get("step", ""), "started_at"])
+
+                if timestamp is None:
+                    return None
+
+                return datetime.fromtimestamp(timestamp, timezone.utc)
+
             project_global_region = GcpRegion.fallback_global_region(self.project)
+            last_run = get_last_run()
             global_builder = GraphBuilder(
                 self.graph,
                 self.cloud,
@@ -87,6 +107,8 @@ def collect(self) -> None:
                 self.core_feedback,
                 self.error_accumulator,
                 project_global_region,
+                config=self.config,
+                last_run_started_at=last_run,
             )
             global_builder.add_node(project_global_region, {})
 
@@ -113,6 +135,13 @@ def collect(self) -> None:
 
             self.error_accumulator.report_all(global_builder.core_feedback)
 
+            if global_builder.config.collect_usage_metrics:
+                try:
+                    log.info(f"[GCP:{self.project.id}] Collect usage metrics.")
+                    self.collect_usage_metrics(global_builder)
+                except Exception as e:
+                    log.warning(f"Failed to collect usage metrics in project {self.project.id}: {e}")
+            shared_queue.wait_for_submitted_work()
             log.info(f"[GCP:{self.project.id}] Connect resources and create edges.")
             # connect nodes
             for node, data in list(self.graph.nodes(data=True)):
@@ -131,6 +160,42 @@ def collect(self) -> None:
             self.core_feedback.progress_done(self.project.id, 1, 1, context=[self.cloud.id])
             log.info(f"[GCP:{self.project.id}] Collecting resources done.")
 
+    def collect_usage_metrics(self, builder: GraphBuilder) -> None:
+        metrics_queries = defaultdict(list)
+        two_hours = timedelta(hours=2)
+        thirty_minutes = timedelta(minutes=30)
+        lookup_map = {}
+        for resource in builder.graph.nodes:
+            if not isinstance(resource, GcpResource):
+                continue
+            # region can be overridden in the query
+            if region := cast(GcpRegion, resource.region()):
+                resource_queries: List[monitoring.GcpMonitoringQuery] = resource.collect_usage_metrics(builder)
+                if resource_queries:
+                    # set unique GcpMonitoringQuery.ref_id
+                    lookup_map[f"{resource.kind}/{resource.id}/{region.id}"] = resource
+                for query in resource_queries:
+                    query_region = query.region or region
+                    start = builder.metrics_delta
+                    if query.period and query.period < thirty_minutes:
+                        start = min(start, two_hours)
+                    metrics_queries[(query_region, start)].append(query)
+        for (region, start), queries in metrics_queries.items():
+
+            def collect_and_set_metrics(
+                start: timedelta, region: GcpRegion, queries: List[monitoring.GcpMonitoringQuery]
+            ) -> None:
+                start_at = builder.created_at - start
+                try:
+                    result = monitoring.GcpMonitoringMetricData.query_for(
+                        builder.for_region(region), queries, start_at, builder.created_at
+                    )
+                    monitoring.update_resource_metrics(lookup_map, result)
+                except Exception as e:
+                    log.warning(f"Error occurred in region {region}: {e}")
+
+            builder.submit_work(collect_and_set_metrics, start, region, queries)
+
     def remove_unconnected_nodes(self, builder: GraphBuilder) -> None:
         def rm_leaf_nodes(clazz: Any, ignore_kinds: Optional[Type[Any]] = None) -> None:
             remove_nodes = set()

plugins/gcp/fix_plugin_gcp/config.py

@@ -1,6 +1,6 @@
 from fixlib.proc import num_default_threads
 from attrs import define, field
-from typing import List, ClassVar
+from typing import List, ClassVar, Optional
 
 
 @define
@@ -31,6 +31,10 @@ class GcpConfig:
             "If false, the error is logged and the resource is skipped."
         },
     )
+    collect_usage_metrics: Optional[bool] = field(
+        default=True,
+        metadata={"description": "Collect resource usage metrics via GCP Monitoring, enabled by default"},
+    )
 
     def should_collect(self, name: str) -> bool:
         if self.collect:

plugins/gcp/fix_plugin_gcp/resources/aiplatform.py

@@ -50,8 +50,6 @@ class AIPlatformRegionFilter:
     def collect_resources(cls, builder: GraphBuilder, **kwargs: Any) -> List[GcpResource]:
         # Default behavior: in case the class has an ApiSpec, call the api and call collect.
         if issubclass(cls, GcpResource):
-            region_name = "global" if not builder.region else builder.region.safe_name
-            log.info(f"[GCP:{builder.project.id}:{region_name}] Collecting {cls.kind}")
             if spec := cls.api_spec:
                 expected_errors = GcpExpectedErrorCodes | (spec.expected_errors or set()) | {"HttpError:none:none"}
                 with GcpErrorHandler(
@@ -66,7 +64,9 @@ def collect_resources(cls, builder: GraphBuilder, **kwargs: Any) -> List[GcpReso
                     if builder.region:
                         items = builder.client.list(spec, **kwargs)
                         collected_resources = cls.collect(items, builder)
-                        log.info(f"[GCP:{builder.project.id}] finished collecting: {cls.kind}")
+                        log.info(
+                            f"[GCP:{builder.project.id}:{builder.region.safe_name}] finished collecting: {cls.kind}"
+                        )
                         return collected_resources
         return []
 

plugins/gcp/fix_plugin_gcp/resources/base.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from datetime import datetime, timedelta
 import json
 import logging
 from concurrent.futures import Future
@@ -8,6 +9,7 @@
 from typing import Callable, List, ClassVar, Optional, TypeVar, Type, Any, Dict, Set, Tuple
 
 from attr import define, field
+from fix_plugin_gcp.config import GcpConfig
 from google.auth.credentials import Credentials as GoogleAuthCredentials
 from googleapiclient.errors import HttpError
 
@@ -32,6 +34,8 @@
 from fixlib.types import Json
 from fixinventorydata.cloud import regions as cloud_region_data
 
+from fixlib.utils import utc
+
 log = logging.getLogger("fix.plugins.gcp")
 
 
@@ -81,7 +85,9 @@ def __init__(
         core_feedback: CoreFeedback,
         error_accumulator: ErrorAccumulator,
         fallback_global_region: GcpRegion,
+        config: GcpConfig,
         region: Optional[GcpRegion] = None,
+        last_run_started_at: Optional[datetime] = None,
         graph_nodes_access: Optional[Lock] = None,
         graph_edges_access: Optional[Lock] = None,
     ) -> None:
@@ -95,12 +101,39 @@ def __init__(
         self.core_feedback = core_feedback
         self.error_accumulator = error_accumulator
         self.fallback_global_region = fallback_global_region
+        self.config = config
+        self.created_at = utc()
+        self.last_run_started_at = last_run_started_at
         self.region_by_name: Dict[str, GcpRegion] = {}
         self.region_by_zone_name: Dict[str, GcpRegion] = {}
         self.zone_by_name: Dict[str, GcpZone] = {}
         self.graph_nodes_access = graph_nodes_access or Lock()
         self.graph_edges_access = graph_edges_access or Lock()
 
+        if last_run_started_at:
+            now = utc()
+
+            # limit the metrics to the last 2 hours
+            if now - last_run_started_at > timedelta(hours=2):
+                start = now - timedelta(hours=2)
+            else:
+                start = last_run_started_at
+
+            delta = now - start
+
+            min_delta = max(delta, timedelta(seconds=60))
+            # in case the last collection happened too quickly, raise the metrics timedelta to 60s,
+            if min_delta != delta:
+                start = now - min_delta
+                delta = min_delta
+        else:
+            now = utc()
+            delta = timedelta(hours=1)
+            start = now - delta
+
+        self.metrics_start = start
+        self.metrics_delta = delta
+
     def submit_work(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
         """
         Use this method for work that can be done in parallel.
@@ -275,7 +308,9 @@ def for_region(self, region: GcpRegion) -> GraphBuilder:
             self.core_feedback,
             self.error_accumulator,
             self.fallback_global_region,
+            self.config,
             region,
+            self.last_run_started_at,
             self.graph_nodes_access,
             self.graph_edges_access,
         )
@@ -374,13 +409,13 @@ def post_process_instance(self, builder: GraphBuilder, source: Json) -> None:
         """
         pass
 
+    def collect_usage_metrics(self, builder: GraphBuilder) -> List:  # type: ignore
+        # Default behavior: do nothing
+        return []
+
     @classmethod
     def collect_resources(cls: Type[GcpResource], builder: GraphBuilder, **kwargs: Any) -> List[GcpResource]:
         # Default behavior: in case the class has an ApiSpec, call the api and call collect.
-        if kwargs:
-            log.info(f"[GCP:{builder.project.id}] Collecting {cls.kind} with ({kwargs})")
-        else:
-            log.info(f"[GCP:{builder.project.id}] Collecting {cls.kind}")
         if spec := cls.api_spec:
             expected_errors = GcpExpectedErrorCodes | (spec.expected_errors or set())
             with GcpErrorHandler(
@@ -393,7 +428,9 @@ def collect_resources(cls: Type[GcpResource], builder: GraphBuilder, **kwargs: A
             ):
                 items = builder.client.list(spec, **kwargs)
                 resources = cls.collect(items, builder)
-                log.info(f"[GCP:{builder.project.id}] finished collecting: {cls.kind}")
+                log.info(
+                    f"[GCP:{builder.project.id}:{builder.region.safe_name if builder.region else "global"}] finished collecting: {cls.kind}"
+                )
                 return resources
         return []