def _create_operational_health_layout()

in Gems/AWSMetrics/cdk/aws_metrics/dashboard.py [0:0]

• 73 lines of code
• 3 McCabe index (conditional complexity)

    def _create_operational_health_layout(
            self,
            input_stream_name: str,
            delivery_stream_name: str,
            analytics_processing_lambda_name: str,
            events_processing_lambda_name: str) -> LayoutWidget:
        """
        This layout contains operational health metrics during events ingestion and analytics processing.

        @param input_stream_name Name of the input Kinesis data stream.
        @param delivery_stream_name Name of the Kinesis Firehose delivery stream.
        @param analytics_processing_lambda_name Name of the analytics processing Lambda function.
        @param events_processing_lambda_name Name of the events processing Lambda function.
        @return Operational health layout widget.
        """
        operational_health_graph_widgets = list()

        event_ingestion_left_widgets = [
            cloudwatch.Metric(
                metric_name="IncomingRecords",
                label="Kinesis Incoming Records",
                namespace="AWS/Kinesis",
                period=Duration.minutes(aws_metrics_constants.DASHBOARD_METRICS_TIME_PERIOD),
                statistic="Sum",
                dimensions_map={
                    "StreamName": input_stream_name
                }
            )
        ]
        if delivery_stream_name:
            event_ingestion_left_widgets.append(
                cloudwatch.Metric(
                    metric_name="DeliveryToS3.Records",
                    label="Firehose Delivery To S3 Records",
                    namespace="AWS/Firehose",
                    period=Duration.minutes(aws_metrics_constants.DASHBOARD_METRICS_TIME_PERIOD),
                    statistic="Sum",
                    dimensions_map={
                        "DeliveryStreamName": delivery_stream_name
                    }
                )
            )

        operational_health_graph_widgets.append(
            cloudwatch.GraphWidget(
                title="Events Ingestion",
                left=event_ingestion_left_widgets,
                live_data=True
            )
        )

        analytics_processing_lambda_errors_metrics, analytics_processing_lambda_error_rate_metrics = \
            self._get_lambda_operational_health_metrics(
                analytics_processing_lambda_name,
                "Analytics Processing Lambda"
            )

        lambda_processing_left_widgets = [analytics_processing_lambda_errors_metrics]
        lambda_processing_right_widgets = [analytics_processing_lambda_error_rate_metrics]

        if events_processing_lambda_name:
            events_processing_lambda_errors_metrics, events_processing_lambda_error_rate_metrics = \
                self._get_lambda_operational_health_metrics(
                    events_processing_lambda_name,
                    "Events Processing Lambda"
                )

            lambda_processing_left_widgets.append(events_processing_lambda_errors_metrics)
            lambda_processing_right_widgets.append(events_processing_lambda_error_rate_metrics)

        operational_health_graph_widgets.append(
            cloudwatch.GraphWidget(
                title="Lambda Processing",
                left=lambda_processing_left_widgets,
                right=lambda_processing_right_widgets,
                right_y_axis=cloudwatch.YAxisProps(
                    show_units=False,
                    min=0,
                    max=100
                ),
                live_data=True,
                view=cloudwatch.GraphWidgetView.TIME_SERIES
            )
        )

        operational_health_layout = LayoutWidget(
            layout_description=aws_metrics_constants.DASHBOARD_OPERATIONAL_HEALTH_DESCRIPTION,
            widgets=operational_health_graph_widgets,
            max_width=aws_metrics_constants.DASHBOARD_MAX_WIDGET_WIDTH // 2)

        return operational_health_layout