def _add_fsx_widgets()

in cli/src/pcluster/templates/cw_dashboard_builder.py [0:0]
51 lines of code
7 McCabe index (conditional complexity)

    def _add_fsx_widgets(self, storages_list):
        common_metrics = [
            new_pcluster_metric(title="Data Read/Write Ops", metrics=["DataReadOperations", "DataWriteOperations"]),
            new_pcluster_metric(title="Data Read/Write Bytes", metrics=["DataReadBytes", "DataWriteBytes"]),
        ]
        free_data_storage_capacity_graph_title = "Free Data Storage Capacity"
        lustre_metrics = common_metrics + [
            new_pcluster_metric(title=free_data_storage_capacity_graph_title, metrics=["FreeDataStorageCapacity"])
        ]
        namespace = "AWS/FSx"
        metric_graphs = defaultdict(list)
        for index, storage in enumerate(storages_list):
            if isinstance(storage.config, SharedFsxLustre):
                metrics = lustre_metrics
                dimensions_map = {"FileSystemId": storage.id}
            elif isinstance(storage.config, ExistingFileCache):
                metrics = lustre_metrics
                dimensions_map = {"FileCacheId": storage.config.file_cache_id}
            else:
                metrics = common_metrics
                dimensions_map = {"FileSystemId": storage.config.file_system_id, "VolumeId": storage.id}

                # FSx Ontap/OpenZFS do not provide free capacity metric. The code below generates the metric using math.
                free_capacity_metric = cloudwatch.MathExpression(
                    label=f"{storage.id} FreeDataStorageCapacity",
                    expression=f"capacity{index} - used_capacity{index}",
                    using_metrics={
                        f"capacity{index}": cloudwatch.Metric(
                            namespace=namespace,
                            metric_name="StorageCapacity",
                            dimensions_map=dimensions_map,
                        ),
                        f"used_capacity{index}": cloudwatch.Metric(
                            namespace=namespace,
                            metric_name="UsedStorageCapacity",
                            dimensions_map=dimensions_map,
                        ),
                    },
                )
                metric_graphs[free_data_storage_capacity_graph_title].append(free_capacity_metric)
            for metrics_param in metrics:
                metric_list = metric_graphs[metrics_param.title]
                for metric in metrics_param.metrics:
                    cloudwatch_metric = cloudwatch.Metric(
                        namespace=namespace,
                        metric_name=metric,
                        dimensions_map=dimensions_map,
                    )
                    metric_list.append(cloudwatch_metric)
        widgets_list = []
        for title, metric_list in metric_graphs.items():
            widgets_list.append(self._generate_graph_widget(title, metric_list))
        return widgets_list