AWSTemplateFormatVersion: '2010-09-09'
Description: >
  OpenSearch Domain Review Remediator. Review-only module for inventorying Amazon
  OpenSearch Service domains and identifying low-activity or potentially oversized
  domains for owner validation. This module intentionally does not delete or modify domains.

Parameters:
  ProjectName:
    Type: String
    Default: opensearch-domain-review
  TargetRegionsCsv:
    Type: String
    Default: all-enabled
  ExcludedRegionsCsv:
    Type: String
    Default: ap-northeast-3
  LookbackDays:
    Type: Number
    Default: 14
  LowActivityRequestsPerDay:
    Type: Number
    Default: 10
  MinimumDomainAgeDays:
    Type: Number
    Default: 30
  MinStorageGiBForCostFlag:
    Type: Number
    Default: 100
  ScheduleEnabled:
    Type: String
    Default: 'false'
    AllowedValues: ['true', 'false']
  ScheduleExpression:
    Type: String
    Default: rate(30 days)

Conditions:
  IsScheduleEnabled: !Equals [!Ref ScheduleEnabled, 'true']

Resources:
  ReviewRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: OpenSearchDomainReviewPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: DiscoverRegionsAndAccount
                Effect: Allow
                Action:
                  - ec2:DescribeRegions
                  - sts:GetCallerIdentity
                Resource: '*'
              - Sid: ReadOpenSearch
                Effect: Allow
                Action:
                  - es:ListDomainNames
                  - es:DescribeElasticsearchDomain
                  - es:ListTags
                  - opensearch:ListDomainNames
                  - opensearch:DescribeDomain
                  - opensearch:ListTags
                Resource: '*'
              - Sid: ReadMetrics
                Effect: Allow
                Action:
                  - cloudwatch:GetMetricStatistics
                Resource: '*'

  ReviewFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Ref ProjectName
      Runtime: python3.12
      Handler: index.handler
      Timeout: 900
      MemorySize: 512
      Role: !GetAtt ReviewRole.Arn
      Environment:
        Variables:
          TARGET_REGIONS_CSV: !Ref TargetRegionsCsv
          EXCLUDED_REGIONS_CSV: !Ref ExcludedRegionsCsv
          LOOKBACK_DAYS: !Ref LookbackDays
          LOW_ACTIVITY_REQUESTS_PER_DAY: !Ref LowActivityRequestsPerDay
          MINIMUM_DOMAIN_AGE_DAYS: !Ref MinimumDomainAgeDays
          MIN_STORAGE_GIB_FOR_COST_FLAG: !Ref MinStorageGiBForCostFlag
      Code:
        ZipFile: |
          import json
          import os
          import traceback
          from datetime import datetime, timedelta, timezone

          import boto3
          from botocore.exceptions import ClientError

          def now():
              return datetime.now(timezone.utc)

          def split_csv(value):
              return [x.strip() for x in str(value or '').split(',') if x.strip()]

          def env_int(name, default):
              try:
                  return int(os.environ.get(name, str(default)))
              except Exception:
                  return default

          def age_days(dt):
              if not dt:
                  return None
              if dt.tzinfo is None:
                  dt = dt.replace(tzinfo=timezone.utc)
              return (now() - dt).days

          def enabled_regions(excluded):
              requested = os.environ.get('TARGET_REGIONS_CSV', 'all-enabled')
              if requested.lower() != 'all-enabled':
                  regions = split_csv(requested)
              else:
                  ec2 = boto3.client('ec2', region_name=os.environ.get('AWS_REGION', 'us-east-1'))
                  regions = sorted([r['RegionName'] for r in ec2.describe_regions(AllRegions=False).get('Regions', [])])
              return [r for r in regions if r not in excluded]

          def account_id():
              try:
                  return boto3.client('sts').get_caller_identity().get('Account')
              except Exception:
                  return None

          def list_domains(region):
              try:
                  c = boto3.client('opensearch', region_name=region)
                  names = [d['DomainName'] for d in c.list_domain_names().get('DomainNames', []) if d.get('DomainName')]
                  return c, 'opensearch', names
              except Exception:
                  c = boto3.client('es', region_name=region)
                  names = [d['DomainName'] for d in c.list_domain_names().get('DomainNames', []) if d.get('DomainName')]
                  return c, 'es', names

          def describe_domain(client, api_type, name):
              if api_type == 'opensearch':
                  return client.describe_domain(DomainName=name).get('DomainStatus', {})
              return client.describe_elasticsearch_domain(DomainName=name).get('DomainStatus', {})

          def tags(client, arn):
              if not arn:
                  return {}, None
              try:
                  tag_list = client.list_tags(ARN=arn).get('TagList', [])
                  return {t.get('Key'): t.get('Value', '') for t in tag_list if t.get('Key')}, None
              except ClientError as e:
                  return {}, e

          def metric(cw, namespace, name, dimensions, stat, start, end):
              try:
                  resp = cw.get_metric_statistics(
                      Namespace=namespace,
                      MetricName=name,
                      Dimensions=dimensions,
                      StartTime=start,
                      EndTime=end,
                      Period=86400,
                      Statistics=[stat]
                  )
                  vals = [float(p.get(stat, 0.0)) for p in resp.get('Datapoints', []) if p.get(stat) is not None]
                  if not vals:
                      return None, 0
                  if stat == 'Sum':
                      return sum(vals), len(vals)
                  if stat == 'Minimum':
                      return min(vals), len(vals)
                  return sum(vals) / len(vals), len(vals)
              except ClientError:
                  return None, 0

          def metrics(region, domain, acct, days):
              cw = boto3.client('cloudwatch', region_name=region)
              start = now() - timedelta(days=days)
              end = now()
              dims = [{'Name': 'DomainName', 'Value': domain}]
              if acct:
                  dims.append({'Name': 'ClientId', 'Value': acct})

              for ns in ['AWS/OpenSearchService', 'AWS/ES']:
                  search, sp = metric(cw, ns, 'SearchRate', dims, 'Sum', start, end)
                  index, ip = metric(cw, ns, 'IndexingRate', dims, 'Sum', start, end)
                  cpu, cp = metric(cw, ns, 'CPUUtilization', dims, 'Average', start, end)
                  jvm, jp = metric(cw, ns, 'JVMMemoryPressure', dims, 'Average', start, end)
                  free, fp = metric(cw, ns, 'FreeStorageSpace', dims, 'Minimum', start, end)
                  points = sp + ip + cp + jp + fp
                  if points:
                      search = search or 0.0
                      index = index or 0.0
                      return {
                          'namespace_used': ns,
                          'lookback_days': days,
                          'search_rate_sum': round(search, 4),
                          'indexing_rate_sum': round(index, 4),
                          'requests_per_day_estimate': round((search + index) / max(days, 1), 4),
                          'cpu_average': round(cpu, 4) if cpu is not None else None,
                          'jvm_memory_pressure_average': round(jvm, 4) if jvm is not None else None,
                          'free_storage_space_minimum_mib': round(free, 4) if free is not None else None,
                          'metric_datapoints': points
                      }
              return {'lookback_days': days, 'metric_datapoints': 0}

          def summarize(status, tag_map, metric_map, settings):
              cluster = status.get('ClusterConfig', {}) or {}
              ebs = status.get('EBSOptions', {}) or {}
              vpc = status.get('VPCOptions', {}) or {}
              endpoint = status.get('DomainEndpointOptions', {}) or {}
              enc = status.get('EncryptionAtRestOptions', {}) or {}
              n2n = status.get('NodeToNodeEncryptionOptions', {}) or {}
              adv = status.get('AdvancedSecurityOptions', {}) or {}

              count = cluster.get('InstanceCount') or 0
              size = ebs.get('VolumeSize') or 0
              total_gib = int(count or 0) * int(size or 0) if ebs.get('EBSEnabled') else 0
              d_age = age_days(status.get('Created'))
              flags = []

              if d_age is not None and d_age >= settings['minimum_domain_age_days']:
                  if metric_map.get('metric_datapoints', 0) and metric_map.get('requests_per_day_estimate', 0) < settings['low_activity_requests_per_day']:
                      flags.append('low_activity')
              if total_gib >= settings['min_storage_gib_for_cost_flag']:
                  flags.append('large_ebs_storage')
              if cluster.get('WarmCount'):
                  flags.append('warm_nodes_present')
              if cluster.get('ColdStorageOptions', {}).get('Enabled'):
                  flags.append('cold_storage_enabled')
              if count and count > 1:
                  flags.append('multi_node_domain')

              return {
                  'domain_name': status.get('DomainName'),
                  'arn': status.get('ARN'),
                  'engine_version': status.get('EngineVersion') or status.get('ElasticsearchVersion'),
                  'domain_age_days': d_age,
                  'processing': status.get('Processing'),
                  'upgrade_processing': status.get('UpgradeProcessing'),
                  'instance_type': cluster.get('InstanceType'),
                  'instance_count': count,
                  'dedicated_master_enabled': cluster.get('DedicatedMasterEnabled'),
                  'dedicated_master_type': cluster.get('DedicatedMasterType'),
                  'dedicated_master_count': cluster.get('DedicatedMasterCount'),
                  'zone_awareness_enabled': cluster.get('ZoneAwarenessEnabled'),
                  'warm_enabled': cluster.get('WarmEnabled'),
                  'warm_type': cluster.get('WarmType'),
                  'warm_count': cluster.get('WarmCount'),
                  'cold_storage_enabled': cluster.get('ColdStorageOptions', {}).get('Enabled'),
                  'ebs_enabled': ebs.get('EBSEnabled'),
                  'ebs_volume_type': ebs.get('VolumeType'),
                  'ebs_volume_size_gib_per_node': size,
                  'total_ebs_gib_estimate': total_gib,
                  'ebs_iops': ebs.get('Iops'),
                  'ebs_throughput': ebs.get('Throughput'),
                  'encryption_at_rest_enabled': enc.get('Enabled'),
                  'node_to_node_encryption_enabled': n2n.get('Enabled'),
                  'advanced_security_enabled': adv.get('Enabled'),
                  'enforce_https': endpoint.get('EnforceHTTPS'),
                  'tls_security_policy': endpoint.get('TLSSecurityPolicy'),
                  'vpc_id': vpc.get('VPCId'),
                  'subnet_ids': vpc.get('SubnetIds', []),
                  'security_group_ids': vpc.get('SecurityGroupIds', []),
                  'metrics': metric_map,
                  'cost_review_flags': flags,
                  'tags': tag_map
              }

          def scan_region(region, settings, acct):
              result = {
                  'region': region,
                  'domains': 0,
                  'domains_with_metric_data': 0,
                  'low_activity_candidates': 0,
                  'large_storage_candidates': 0,
                  'cost_review_candidates': 0,
                  'tag_read_access_denied': 0,
                  'access_denied': 0,
                  'failed': 0,
                  'failures': [],
                  'domain_inventory': []
              }
              try:
                  client, api_type, names = list_domains(region)
              except ClientError as e:
                  code = e.response.get('Error', {}).get('Code', 'Unknown')
                  msg = e.response.get('Error', {}).get('Message', str(e))
                  result['failures'].append({'operation': 'ListDomainNames', 'error_code': code, 'error_message': msg})
                  if code in ('AccessDenied', 'AccessDeniedException') or 'explicit deny' in msg.lower():
                      result['access_denied'] += 1
                  else:
                      result['failed'] += 1
                  return result

              result['domains'] = len(names)
              for name in names:
                  try:
                      status = describe_domain(client, api_type, name)
                      tag_map, tag_err = tags(client, status.get('ARN'))
                      if tag_err:
                          result['tag_read_access_denied'] += 1
                      metric_map = metrics(region, name, acct, settings['lookback_days'])
                      item = summarize(status, tag_map, metric_map, settings)
                      result['domain_inventory'].append(item)
                      if metric_map.get('metric_datapoints', 0):
                          result['domains_with_metric_data'] += 1
                      if 'low_activity' in item['cost_review_flags']:
                          result['low_activity_candidates'] += 1
                      if 'large_ebs_storage' in item['cost_review_flags']:
                          result['large_storage_candidates'] += 1
                      if item['cost_review_flags']:
                          result['cost_review_candidates'] += 1
                  except Exception as e:
                      result['failed'] += 1
                      result['failures'].append({'domain_name': name, 'error_code': type(e).__name__, 'error_message': str(e)})
              return result

          def handler(event, context):
              settings = {
                  'excluded_regions': split_csv(os.environ.get('EXCLUDED_REGIONS_CSV')),
                  'lookback_days': env_int('LOOKBACK_DAYS', 14),
                  'low_activity_requests_per_day': env_int('LOW_ACTIVITY_REQUESTS_PER_DAY', 10),
                  'minimum_domain_age_days': env_int('MINIMUM_DOMAIN_AGE_DAYS', 30),
                  'min_storage_gib_for_cost_flag': env_int('MIN_STORAGE_GIB_FOR_COST_FLAG', 100)
              }

              if isinstance(event, dict):
                  if 'lookback_days' in event:
                      settings['lookback_days'] = int(event['lookback_days'])
                  if 'target_regions' in event and event['target_regions']:
                      os.environ['TARGET_REGIONS_CSV'] = ','.join(event['target_regions'])

              acct = account_id()
              summary = {
                  'started_at': now().isoformat(),
                  'review_only': True,
                  'account_id': acct,
                  'lookback_days': settings['lookback_days'],
                  'low_activity_requests_per_day': settings['low_activity_requests_per_day'],
                  'minimum_domain_age_days': settings['minimum_domain_age_days'],
                  'min_storage_gib_for_cost_flag': settings['min_storage_gib_for_cost_flag'],
                  'excluded_regions': settings['excluded_regions'],
                  'regions': [],
                  'totals': {
                      'domains': 0,
                      'domains_with_metric_data': 0,
                      'low_activity_candidates': 0,
                      'large_storage_candidates': 0,
                      'cost_review_candidates': 0,
                      'tag_read_access_denied': 0,
                      'access_denied': 0,
                      'failed': 0
                  },
                  'region_results': []
              }
              try:
                  regions = enabled_regions(settings['excluded_regions'])
                  summary['regions'] = regions
                  for region in regions:
                      try:
                          rr = scan_region(region, settings, acct)
                      except Exception as e:
                          rr = {'region': region, 'domains': 0, 'domains_with_metric_data': 0, 'low_activity_candidates': 0, 'large_storage_candidates': 0, 'cost_review_candidates': 0, 'tag_read_access_denied': 0, 'access_denied': 0, 'failed': 1, 'failures': [{'error_code': type(e).__name__, 'error_message': str(e), 'trace': traceback.format_exc()}], 'domain_inventory': []}
                      summary['region_results'].append(rr)
                      for key in summary['totals']:
                          summary['totals'][key] += int(rr.get(key, 0) or 0)
              finally:
                  summary['ended_at'] = now().isoformat()
                  print(json.dumps(summary, indent=2, default=str))
              return summary

  ReviewScheduleRule:
    Type: AWS::Events::Rule
    Condition: IsScheduleEnabled
    Properties:
      Name: !Sub '${ProjectName}-schedule'
      ScheduleExpression: !Ref ScheduleExpression
      State: ENABLED
      Targets:
        - Arn: !GetAtt ReviewFunction.Arn
          Id: OpenSearchDomainReviewTarget
          Input: '{"source":"eventbridge-schedule"}'

  ReviewSchedulePermission:
    Type: AWS::Lambda::Permission
    Condition: IsScheduleEnabled
    Properties:
      FunctionName: !Ref ReviewFunction
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt ReviewScheduleRule.Arn

Outputs:
  ReviewFunctionName:
    Description: Lambda function name.
    Value: !Ref ReviewFunction
  ReviewFunctionArn:
    Description: Lambda function ARN.
    Value: !GetAtt ReviewFunction.Arn
  ReviewOnly:
    Description: This module is intentionally review-only.
    Value: 'true'
