def __init__()

in cdk/stacks/grafana/grafana_stack.py [0:0]

• 99 lines of code
• 1 McCabe index (conditional complexity)

    def __init__(self, scope: core.Construct, construct_id: str,
                 database: timestream.CfnDatabase, table: timestream.CfnTable,
                 **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        vpc = ec2.Vpc(self, "GrafanaVpc", max_azs=2)

        vpc.add_interface_endpoint('EFSEndpoint', service=ec2.InterfaceVpcEndpointAwsService.ELASTIC_FILESYSTEM)
        vpc.add_interface_endpoint('SMEndpoint', service=ec2.InterfaceVpcEndpointAwsService.SECRETS_MANAGER)

        cluster = ecs.Cluster(self, "MyCluster", vpc=vpc)

        file_system = efs.FileSystem(
            self, "EfsFileSystem",
            vpc=vpc,
            encrypted=True,
            lifecycle_policy=efs.LifecyclePolicy.AFTER_14_DAYS,
            performance_mode=efs.PerformanceMode.GENERAL_PURPOSE,
            throughput_mode=efs.ThroughputMode.BURSTING
        )

        access_point = efs.AccessPoint(
            self, "EfsAccessPoint",
            file_system=file_system,
            path="/var/lib/grafana",
            posix_user=PosixUser(
                gid="1000",
                uid="1000"
            ),
            create_acl=Acl(
                owner_gid="1000",
                owner_uid="1000",
                permissions="755"
            )
        )

        log_group = logs.LogGroup(self, "taskLogGroup",
                                  retention=logs.RetentionDays.ONE_MONTH
                                  )

        container_log_driver = ecs.LogDrivers.aws_logs(stream_prefix="fargate-grafana", log_group=log_group)

        task_role = iam.Role(self, "taskRole", assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"))

        task_role.add_to_policy(iam.PolicyStatement(
            effect=iam.Effect.ALLOW,
            actions=[
                "cloudwatch:DescribeAlarmsForMetric",
                "cloudwatch:DescribeAlarmHistory",
                "cloudwatch:DescribeAlarms",
                "cloudwatch:ListMetrics",
                "cloudwatch:GetMetricStatistics",
                "cloudwatch:GetMetricData",
                "ec2:DescribeTags",
                "ec2:DescribeInstances",
                "ec2:DescribeRegions",
                "tag:GetResources"
            ],
            resources=["*"]
        ))
        self.grant_timestream_read(task_role, database, table)

        execution_role = iam.Role(self, "executionRole", assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"))
        log_group.grant_write(execution_role)

        volume_name = "efsGrafanaVolume"

        volume_config = ecs.Volume(
            name=volume_name,
            efs_volume_configuration=EfsVolumeConfiguration(
                file_system_id=file_system.file_system_id,
                transit_encryption="ENABLED",
                authorization_config=AuthorizationConfig(access_point_id=access_point.access_point_id)
            ))

        task_definition = ecs.FargateTaskDefinition(
            self, "TaskDef",
            task_role=task_role,
            execution_role=execution_role,
            volumes=[volume_config]
        )

        grafana_admin_password = secretsmanager.Secret(self, "grafanaAdminPassword")
        grafana_admin_password.grant_read(task_role)

        container_web = task_definition.add_container(
            "grafana",
            image=ecs.ContainerImage.from_registry("grafana/grafana"),
            logging=container_log_driver,
            environment={
                "GF_INSTALL_PLUGINS": "grafana-timestream-datasource",
                "GF_AWS_default_REGION": core.Aws.REGION
            },
            secrets={
                "GF_SECURITY_ADMIN_PASSWORD": ecs.Secret.from_secrets_manager(
                    grafana_admin_password)
            })

        container_web.add_port_mappings(PortMapping(container_port=3000))
        container_web.add_mount_points(
            MountPoint(container_path="/var/lib/grafana", read_only=False, source_volume=volume_config.name)
        )

        fargate_service = ecs_patterns.ApplicationLoadBalancedFargateService(
            self, "MyFargateService",
            cluster=cluster,
            cpu=1024,
            desired_count=1,
            task_definition=task_definition,
            memory_limit_mib=2048,
            platform_version=ecs.FargatePlatformVersion.LATEST
        )

        fargate_service.target_group.configure_health_check(path="/api/health")
        file_system.connections.allow_default_port_from(fargate_service.service.connections)

        core.CfnOutput(self, "GrafanaAdminSecret", value=grafana_admin_password.secret_name,
                       export_name="GrafanaAdminSecret")