JavaScript 개발자를 위한 Rust: AWS Lambda로 SQS 배치 오류 처리

이 게시물에서 부분 배치 응답을 사용하여 AWS SQS 비교를 향상시킬 것입니다.
The partial batch response은 2021년 11월에 다시 발표되었으며 개발자가 부분적인 오류를 처리하는 데 도움이 됩니다.

미니 시리즈Rust for JavaScript developers의 일부로 다른 부분을 확인할 수 있습니다.

Part 1 AWS SQS에 대한 기본 비교를 표시했습니다.

Part 2 AWS AppConfig에 대한 기본 비교를 표시했습니다.

Part 3 CSV 파일 읽기 처리에 대한 기본 비교를 보여 주었습니다.

왜

지난 몇 년 동안 저는 .NET, JavaScript 및 Rust 간에 여러 번 언어를 전환했으며 한 언어로 습득한 지식을 새로운 언어로 이전할 수 있습니다. 따라서 유사성을 빠르게 선택하려면 정신적으로 매핑해야 합니다.

Serverless Rust가 작동하는 모습을 보고 싶어하는 친구와 이 작업을 하고 있었기 때문에 이에 대한 짧은 글을 썼습니다.

기본

Rust는 예를 들어 다음과 같은 많은 내장 기능을 제공합니다.


JS
녹


npm
뱃짐

npm 초기화
화물 초기화

npm 설치
화물 설치

npm 실행 빌드
화물 빌드

패키지. json
Cargo.toml

패키지 잠금. json
Cargo.lock

웹팩
화물 빌드

보푸라기
카고 클리피

더 예쁘다
화물 fmt

문서 생성
화물 문서

jest와 같은 테스트 라이브러리
화물 테스트


새로운 SAM 기반 서버리스 앱 생성

sam init --location gh:aws-samples/cookiecutter-aws-sam-rust

아마존 SQS

이전 챌린지the partial batch response가 발표되었는데, 오류를 처리하고 있었습니다. 처리하는 동안 SQS 레코드 중 하나가 실패한다고 가정하면 메시지의 전체 배치가 실패한 것으로 간주됩니다. 우리 중 많은 사람들이 성공 및 실패 메시지를 추적하고 이 제한을 피하기 위해 프로그래밍 방식으로 처리된 메시지를 삭제했습니다.

코드가 이것one과 매우 유사하기 때문에 aws sam 템플릿부터 시작하여 필요한 변경 사항만 보고하겠습니다.

AWSTemplateFormatVersion: 2010-09-09
Transform: 'AWS::Serverless-2016-10-31'
Description: Writer Lambda.

##########################################################################
#  Global values that are applied to all resources                       #
##########################################################################
Globals:
  Function:
    MemorySize: 1024
    Architectures: ["arm64"]
    Handler: bootstrap
    Runtime: provided.al2
    Timeout: 29
    Layers:
      - !Sub arn:aws:lambda:${AWS::Region}:580247275435:layer:LambdaInsightsExtension-Arm64:1
    Environment:
      Variables:
        RUST_BACKTRACE: 1
        RUST_LOG: info

Resources:
##########################################################################
#   Lambda Function                                                      #
##########################################################################
  LambdaFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ../build/handler
      Policies:
        - AWSLambdaBasicExecutionRole
      Events:
        MySQSEvent:
          Type: SQS
          Properties:
            Queue: <ARN of the SQS queue>
            BatchSize: 10
            FunctionResponseTypes:
              - ReportBatchItemFailures

코드는 다음과 같습니다.

pub async fn execute(client: &aws_sdk_dynamodb::Client, event: LambdaEvent<SqsEvent>,) -> Result<Value, Error> {
    let failed_message: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let mut tasks = Vec::with_capacity(event.payload.records.len());
    let shared_client = Arc::from(client.clone());

    for record in event.payload.records.into_iter() {
        let shared_client = shared_client.clone();
        let message_id = record.message_id.unwrap();
        let failed_message = failed_message.clone();

        tasks.push(tokio::spawn(async move {
            if let Some(body) = &record.body {
                let request = serde_json::from_str::<MyStruct>(&body);
                if let Ok(request) = request {
                    DoSomething::new()
                      .execute(&shared_client, &request)
                      .map_or_else(|e| {
                        failed_message.lock().unwrap().push(message_id.clone());
                      }, |_| ());
                } 
            }
        }));
    }

    join_all(tasks).await;

    let response = BatchItemFailures {
        batch_item_failures: failed_message.lock().unwrap().clone()
            .into_iter()
            .map(|message_id| {
                return ItemIdentifier {
                    item_identifier: message_id,
                };
            })
            .collect(),
    };

    Ok(serde_json::to_value(response).unwrap())
}

TypeScript 버전과 크게 다르지 않습니다.

export const handler = async (event: SQSEvent): Promise<BatchItemFailures> => {
  const failedMessageIds: string[] = [];
  await Promise.all(
    event.Records.map(async (record: SQSRecord) => {
      try {
        //DoSomething(record)
      } catch (error) {
        failedMessageIds.push(record.messageId);
      }
    }),
  );

  const batchItemFailures: BatchItemFailures = {
    batchItemFailures: failedMessageIds.map(id => {
      return {
        itemIdentifier: id
      }
    })
  };

  return batchItemFailures;
};

Rust는 더 엄격하며 루프 내에서 tokio::spawn을 사용하기 때문에 failed_message 벡터는 단일 차용 규칙을 위반하기 때문에 외부로 이동되고 더 이상 사용할 수 없으므로 다음을 사용해야 합니다.

 Arc<Mutex<Vec<String>>>

실패에 대해 필터링할 수 있는 join_all 반환 내부에서 SQS 식별자를 설정하는 방법을 찾을 수 없습니다.

결론

서버리스 애플리케이션의 Rust 구문은 다른 언어보다 더 어색할 수 있습니다. 여전히 나는 Rust를 매일 사용하지 않고 있으며 더 이상 여분의 main(), unwrap() 또는 Mutex/Arc를 알아채지 못합니다. 그것은 언어의 일부일 뿐이며 지금은 익숙해졌습니다.

내 목표는 다른 프로그래밍 언어를 사용하고, 기술 세트를 재사용하고, 요구 사항에 따라 더 나은 결과를 얻을 수 있음을 입증하는 것입니다(작업에 가장 적합한 도구 선택). 다른 프로그래밍 언어로 동일한 코드를 작성하면 Rust로 빠르게 이동하는 데 도움이 되며 Rust로 쉽게 마이그레이션하는 데 도움이 되기를 바랍니다.