AWS 클라우드의 이벤트를 감지하여 에러 발생시 AI를 이용하여 해결책을 이메일로 전달받는 시스템 개발 방법

6분 분량

콘텐츠 수준: 중급

해당 기사에서는 Cloudtrail의 접속없이 에러 발생시 이를 감지하여 AWS의 Bedrock을 이용하여 답변을 보내주는 시스템 개발 방법에 대해서 다룹니다.

사용 목적

Cloudtrail에서는 클라우드 사용시에 발생하는 다양한 클라우드 이벤트들을 확인할 수 있습니다. 이때 에러 발생시에도 Cloudtrail에서 에러에 대한 정보를 확인을 할수있습니다.
Cloudtrail에 접속하지 않고 아래와 같이 설정을 하는 경우 에러 발생시 이를 감지하여 AI를 이용하여 답변을 받을 수 있어 에러와 관련된 문제는 빠르게 처리가 가능합니다.

아키텍처 개요

아래의 아키텍처에서는 Cloudtrail로 받은 에러 관련된 로그를 Bedrock 을 이용하여 분석한뒤에 해결방안에 대해 이메일로 보내는 서비스를 개발 하는 방법에 대한 가이드입니다.
구체적으로 Cloudtrail의 로그를 Cloudwatch logs로 전달합니다. 전달받은 Cloudwatch logs 구독 필터를 이용하여 Error 가 발생한 로그를 Lambda에 전달합니다. 이후 전달된 에러와 관련된 다양한 정보들을 2가지 방식에 의해서 AI 모델을 이용합니다. 첫번째는 Anthropic모델을 이용하여 해당 에러를 분석하고 해결책을 전달받습니다. 두번째는 s3에 저장된 다양한 문서들에서 해당 에러와 관련된 정보들을 검색하고 그에 따른 답변을 추출합니다. 이렇게 2가지 방식으로 전달 받은 답변은 번역되어 SES 를 이용하여 사용자의 이메일로 전달됩니다.

Enter image description here

에러 발생시 실시간으로 에러 내용과 해결책을 이메일로 받을 수 있습니다.
전달 받은 내용에는 에러와 관련된 정보를 포함하여 해결책에 대한 내용들을 확인할 수 있습니다.

Enter image description here

사용되는 서비스들

Cloudwatch
CloudTrail
Amazon Transcribe
S3
Bedrock
Simple Email Service(SES)

사전 준비사항

AWS CLI에서 자격증명 필요합니다. (해당 가이드에서는 Administartor권한 사용, Bedrock을 사용하기 위해 us-east-1에 자격증명이 필요합니다.)
발신자와 송신자의 SES 자격증명 인증 필요 [1] SES를 이용하기 위해서는 발신자 및 송신자의 자격증명이 필요합니다.
Lambda에서 사용할 IAM 역할 생성 필요합니다. (테스트를 위해 Administartor권한을 부여합니다.)
Bedrock 지식 기반 저장소에 사용될 S3 버킷이 us-east-1리전에 존재해야합니다.

단계1. Cloudtrail 추적 로그 생성

기본적인 설정을 유지합니다.
CloudWatch Logs를 활성화하며 로그 그룹이름을 지정합니다.(예시. cloudtrail-log)
[로그 이벤트 선택], [관리 이벤트],[Insights 이벤트]을 활성화 하여 추적 로그를 생성합니다.

단계2 : Lambda 함수 생성

[함수 이름] 원하는 함수이름 을 입력합니다.
[런타임],[Python 3.12]로 최신버전의 파이썬으로 지정합니다.
이전 단계에서 사전 준비사항에서 생성한 IAM 권한을 연결합니다.
나머지 사항은 그대로 두고 [함수 생성]을 클릭하여 함수를 생성합니다.

단계3: Lambda Layer 생성 및 추가[2]

lib 폴더 생성

LIB_DIR=boto3-mylayer/python
mkdir -p $LIB_DIR

LIB_DIR에 라이브러리를 설치합니다.

pip install boto3 --upgrade --target $LIB_DIR

모든 종속성을 /tmp/boto3-mylayer.zip으로 압축합니다.

cd boto3-mylayer
zip -r /tmp/boto3-mylayer.zip .

계층 게시

aws lambda publish-layer-version --layer-name boto3-mylayer --zip-file fileb:///tmp/boto3-mylayer.zip

이후 이전 단계에서 만든 Lambda 함수에 해당 레이어를 추가합니다.

단계4. Cloudwatch log에서 구독 필터 생성

단계1 에서 만든 Cloudwatch log를 클릭하고 [구독 필터]를 클릭합니다.
[구독 필터],[생성] 에서 [Lambda 구독 필터 생성]을 클릭합니다.
아래와 같이 이전 단계에서 만든 Lambda함수를 입력하고 [구독 필터 패턴]에 error라고 입력합니다.

Enter image description here

단계5. Bedrock 지식기반 생성

[Amazon Bedrock], [지식 기반],[지식 기반 생성]을 클릭하여 지식 기반 생성합니다.
단계1은 기본적인 구성으로 설정합니다.
단계2에서 지식 기반으로 사용될 S3 버킷을 지정합니다.
[임베딩 모델]에는 Titan Embeddings G1 - Text v1.2을 사용하며 기본 설정을 유지합니다.
검토 및 생성을 통해 생성합니다.
지식 기반이 생성되면 해당 지식 기반을 클릭하여 S3에 데이터를 넣은뒤에 [동기화]를 합니다.
지식 기반이 생성되면서 발생한 지식 기반 ID 는 따로 적어 이후의 코드에 입력해야합니다.

Enter image description here

단계6. Lambda 코드 수정 및 설정

이전 단계에서 만든 Lambda 함수를 클릭합니다.
해당 가이드에서는 [구성],[일반 구성]에서 제한 시간을 3분으로 설정하였습니다. 해당 부분은 환경에 맞추어 설정해야합니다.
[코드]에서 아래의 코드를 입력한뒤 [Deployment]를 클릭하여 배포합니다.
아래의 코드를 고객님의 환경에 맞추어서 수정해야합니다.

import base64
import gzip
import json
import boto3
from botocore.exceptions import ClientError

def send_email(sender_email, recipient_email, subject, html_body):
    CHARSET = "UTF-8"
    client = boto3.client('ses',region_name="ap-northeast-2")
    try:
        response = client.send_email(
            Destination={
                'ToAddresses': [
                    recipient_email,
                ],
            },
            Message={
                'Body': {
                    'Html': {
                        'Charset': CHARSET,
                        'Data': html_body,
                    },
                },
                'Subject': {
                    'Charset': CHARSET,
                    'Data': subject,
                },
            },
            Source=sender_email,
        )
    except ClientError as e:
        print(e.response['Error']['Message'])
    else:
        print("Email sent! Message ID:", response['MessageId'])

def get_bedrock(prompt):
    bedrock = boto3.client(service_name='bedrock-runtime',region_name="us-east-1")

    body = json.dumps({
        "prompt": "\n\nHuman:{} \n\nAssistant:".format(prompt),
        "max_tokens_to_sample": 300,
        "temperature": 0.1,
        "top_p": 0.9,
    })

    modelId = 'anthropic.claude-v2'
    accept = 'application/json'
    contentType = 'application/json'

    response = bedrock.invoke_model(body=body, modelId=modelId, accept=accept, contentType=contentType)
    response_body = json.loads(response.get('body').read())

    input_string = response_body.get('completion')
    sentences = input_string.split('\n\n')  
    return sentences

def format_html(event_name,event_source,error_code,error_message,bedrock_result_ko,bedrock_result,kb_result):
    html_frame='''
        <html>
        <head>
            <title>Page Title</title>
        </head>
        <body>
            <h1></h1>
            <h2>문제 상황</h2>
            이벤트 이름: {} <br />
            이벤트 소스:{} <br />
            에러 코드: {} <br />
            에러 메시지: {} <br />
            <h2>AI에서 제시하는 해결 방안</h2>
            {}
            <h2>AI에서 제시하는 해결 방안(원문)</h2>
            {}
            <h2>Knowledge Bases 답변</h2>
            {}
        </body>
    </html>
    '''.format(event_name,event_source,error_code,error_message,bedrock_result_ko,bedrock_result,kb_result)
    return html_frame

def translate_text(text):
    translate = boto3.client('translate')
    response = translate.translate_text(
        Text=text,
        SourceLanguageCode='en',
        TargetLanguageCode='ko'
    )
    translated_text = response['TranslatedText']
    return translated_text

def retrieveAndGenerate(input, kbId):
    bedrock_agent_runtime = boto3.client(
        service_name="bedrock-agent-runtime",region_name="us-east-1"
    )
    return bedrock_agent_runtime.retrieve_and_generate(
        input={
            'text': input
        },
        retrieveAndGenerateConfiguration={
            'type': 'KNOWLEDGE_BASE',
            'knowledgeBaseConfiguration': {
                'knowledgeBaseId': kbId,
                'modelArn': 'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-v2'
                }
            }
        )

def lambda_handler(event, context):
    encoded_gzip_data=event['awslogs']['data']
    decoded_data = base64.b64decode(encoded_gzip_data)
    uncompressed_data = gzip.decompress(decoded_data)

    data_dict = json.loads(uncompressed_data)
    error_lst=[]

    for log_event in data_dict['logEvents']:
        event_message = json.loads(log_event['message'])
        event_name = event_message['eventName']
        event_source=event_message.get('eventSource').split(".")[0]
        error_code = event_message.get('errorCode', 'NULL')
        error_message = event_message.get('errorMessage', 'NULL')

        if error_code == "NULL" or error_message == "NULL":
            continue
        error_lst.append([event_name,event_source,error_code,error_message])

    for error in error_lst:
        event_name, event_source, error_code, error_message=error[0],error[1],error[2],error[3]
        prompt=''' The following problems are occurring.

        Event Name: {}
        Event Source: {}
        Error Code: {}
        Error Message: {}
        ====================
        Please make the solution below.
        Solution:
        '''.format(event_name, event_source, error_code, error_message)
        print(prompt)
        try:
            result_lst=get_bedrock(prompt)
            bedrock_result=" ".join(result_lst)
        except Exception as e:    
            print('예외가 발생했습니다.', e)
        
        kb_prompt="How can i resolve {} {}".format(error_code,error_message)
        kb_text = retrieveAndGenerate(kb_prompt, "<지식 기반 ID>")["output"]["text"]

        bedrock_result_ko=translate_text(bedrock_result)
        kb_result=translate_text(kb_text)

        sender_email = "<SES 자격증명된 발신자 이메일>"
        recipient_email = "<SES 자격증명된 수신자 이메일>"
        subject = "AI Automatic Detection: Error Occurred"
        html_body =format_html(event_name,event_source,error_code,error_message,bedrock_result_ko,bedrock_result,kb_result)
        send_email(sender_email, recipient_email, subject, html_body)