[AWS] RDS Aurora: Lambda로 CDC 걸기

CDC(Change Data Capture:변경 데이터 캡쳐)는 DB 상에서 데이터가 변경될때마다 그걸 가져다 뭔가 액션을 취하는 기능 요소를 말한다.
변경을 기반으로 다른 데이터소스에 데이터를 동기화하거나, 웹훅을 쏜다거나 할때 자주 사용되는 패턴이다.

이걸 구현할 때는 보통 Kafka 같은걸로 데이터 폴링하면서 쏘기도 하고 그러는데, RDS AuroraDB를 사용할 경우에는 이런 목표를 더욱 간단하게 달성할 수 있다. MySQL, PostgreSQL 둘 다 된다.

RDS AuroraDB는 자체적으로 위의 그림과 같이 DB에서 AWS Lambda를 바로 호출할 수 있는 기능을 제공한다. 이걸 응용해서 테이블에 update 트리거를 걸고, 트리거 프로시저에서 Lambda를 호출하면 Lambda가 그걸 받아서 추가적인 처리를 하도록 할 수 있다.

Lambda 확장 설치

전용 확장을 설치해줘야 한다.
아무데나 cli 들어가서 아래 커맨드를 입력해준다,

CREATE EXTENSION IF NOT EXISTS aws_lambda CASCADE;

Lambda 함수 생성

함수를 만들때 VPC를 켜고, RDS와 동일한 VPC, 보안그룹, AZ에 놓이도록 생성한다.

런타임같은건 상관없다.

그리고 적당히 들어오는 값만 찍어주도록 수정했다.

IAM 권한 설정

먼저 RDS가 Lambda에 접근할 수 있게 해줄 권한과 정책을 설정해줘야 한다.

먼저 역할을 하나 만든다.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "rds.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

그리고 위의 신뢰 정책을 복붙해 넣어준다.

이름은 적당히 짓고

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "lambda:InvokeFunction",
            "Resource": "*"
        }
    ]
}

역할에 권한을 추가해준다.
나는 인라인 정책으로 넣었다.

RDS 추가설정

방금 만든 IAM 역할을 RDS 페이지에서 Lambda 타겟으로 등록해준다.

그리고 연결된 컴퓨팅 리소스에 Lambda를 연결해준다.

프록시는 필요하다면 달면 된다. 당연히 추가비용이 나온다.

그래서 이렇게 뜨면 된다.

RDS에서 Lambda 호출해보기

함수의 형태로 호출이 가능하다.
아래는 방금 만든 Lambda 함수를 RDS에서 호출하는 간단한 예제다.

SELECT * from aws_lambda.invoke(aws_commons.create_lambda_function_arn('cdc_function', 'ap-northeast-2'), '{"body": "Hello from Postgres!"}'::json );

그럼 저 페이로드로 요청을 날려서

응답을 받아올 것이다.

Lambda 쪽 로그에도 잘 찍혔다.

트리거로 CDC 구성하기

이번에는 트리거를 활용해서 Lambda를 통한 CDC를 구성해보겠다.

먼저 간단한 테이블을 하나 생성했다.

별건 없다.

그리고 아래와 같은 형태로 함수와 트리거를 생성해준다.

CREATE OR REPLACE FUNCTION public.trigger_test_table_function()
 RETURNS trigger
 LANGUAGE plpgsql
AS $function$
    begin
        IF (TG_OP = 'DELETE') THEN
            perform aws_lambda.invoke(
           		aws_commons.create_lambda_function_arn('cdc_function', 'ap-northeast-2'), 
           		json_build_object(
           			'type', 'delete',
           			'value', row_to_json(old)
           		)
           	);
        ELSIF (TG_OP = 'UPDATE') THEN
            perform aws_lambda.invoke(
           		aws_commons.create_lambda_function_arn('cdc_function', 'ap-northeast-2'), 
           		json_build_object(
           			'type', 'update',
           			'value', row_to_json(new)
           		)
           	);
        ELSIF (TG_OP = 'INSERT') THEN
            perform aws_lambda.invoke(
           		aws_commons.create_lambda_function_arn('cdc_function', 'ap-northeast-2'), 
           		json_build_object(
           			'type', 'insert',
           			'value', row_to_json(new)
           		)
           	);
        END IF;
        RETURN NULL; 
    END;
$function$;

CREATE TRIGGER trigger_test_table
AFTER INSERT OR update or delete
ON test
FOR EACH ROW
EXECUTE Function trigger_test_table_function();

어떤 operation으로 날라온건지, 그리고 테이블 row의 전체 값을 그대로 쏴주도록 했다.

그리고 insert를 날려보면

날린 대로 들어올 것이고

delete를 날려보면

그것도 그것대로 잘 날라올 것이다.

이런식으로 써먹으면 된다.

참조
https://docs.aws.amazon.com/ko_kr/AmazonRDS/latest/AuroraUserGuide/PostgreSQL-Lambda.html
https://stackoverflow.com/questions/73474076/invoking-aws-lambda-with-postgresql
https://aws.amazon.com/ko/blogs/database/capturing-data-changes-in-amazon-aurora-using-aws-lambda/