AWS Step Functions - 서버리스 워크플로우 오케스트레이션

AWS Step Functions란?

AWS Step Functions는 분산 애플리케이션의 워크플로우를 시각적으로 조정할 수 있는 서버리스 오케스트레이션 서비스임.

여러 AWS 서비스를 연결하여 복잡한 비즈니스 프로세스를 구현하고, 각 단계의 실행, 재시도, 오류 처리를 자동으로 관리함.

---

왜 Step Functions를 사용해야 하는가?

Lambda 체이닝의 문제점

Lambda 함수를 직접 연결하는 방식:

// Lambda 1: 주문 검증
export const validateOrder = async (event) => {
  const order = JSON.parse(event.body);

  if (isValid(order)) {
    // Lambda 2 호출
    await lambdaClient.send(new InvokeCommand({
      FunctionName: 'processPayment',
      Payload: JSON.stringify(order)
    }));
  }
};

// Lambda 2: 결제 처리
export const processPayment = async (event) => {
  const payment = await processPayment(event);

  // Lambda 3 호출
  await lambdaClient.send(new InvokeCommand({
    FunctionName: 'sendNotification',
    Payload: JSON.stringify(payment)
  }));
};

문제점:

1. 코드 복잡도: 각 Lambda가 다음 단계를 알아야 함
2. 오류 처리 어려움: 재시도 로직을 직접 구현해야 함
3. 모니터링 부족: 전체 워크플로우 진행 상황 파악 어려움
4. 병렬 처리 복잡: 여러 작업을 동시에 실행하려면 복잡한 코드 필요
5. 15분 제한: 단일 Lambda는 최대 15분만 실행 가능

Step Functions의 해결 방법

상태 기계(State Machine)로 워크플로우 정의:

{
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:validateOrder",
      "Next": "ProcessPayment"
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:processPayment",
      "Next": "SendNotification",
      "Retry": [{
        "ErrorEquals": ["PaymentError"],
        "MaxAttempts": 3
      }]
    },
    "SendNotification": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:sendNotification",
      "End": true
    }
  }
}

장점:

1. 선언적 정의: JSON으로 워크플로우 명시
2. 자동 재시도: 오류 시 자동 재시도 정책
3. 시각적 모니터링: AWS 콘솔에서 실행 상태 확인
4. 병렬/조건 처리: Parallel, Choice 상태로 복잡한 로직 구현
5. 무제한 실행 시간: 최대 1년까지 실행 가능

---

Step Functions 핵심 개념

1. 상태 기계 (State Machine)

워크플로우를 정의하는 JSON 문서.

기본 구조:

{
  "Comment": "CSV 처리 파이프라인",
  "StartAt": "ValidateCSV",
  "States": {
    "ValidateCSV": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "csv-validator",
        "Payload.$": "$"
      },
      "ResultPath": "$.validationResult",
      "Next": "CheckValidation"
    },
    "CheckValidation": {
      "Type": "Choice",
      "Choices": [{
        "Variable": "$.validationResult.status",
        "StringEquals": "PASSED",
        "Next": "TransformData"
      }],
      "Default": "ValidationFailed"
    },
    "TransformData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:transformer",
      "Next": "LoadData"
    },
    "LoadData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:loader",
      "End": true
    },
    "ValidationFailed": {
      "Type": "Fail",
      "Error": "ValidationError",
      "Cause": "CSV validation failed"
    }
  }
}

2. 상태 (State) 유형

Task State (작업):

{
  "ProcessOrder": {
    "Type": "Task",
    "Resource": "arn:aws:lambda:...:function:processOrder",
    "TimeoutSeconds": 300,
    "HeartbeatSeconds": 60,
    "Next": "SendNotification"
  }
}

Choice State (조건 분기):

{
  "CheckOrderAmount": {
    "Type": "Choice",
    "Choices": [
      {
        "Variable": "$.totalAmount",
        "NumericGreaterThan": 1000000,
        "Next": "ManagerApproval"
      },
      {
        "Variable": "$.totalAmount",
        "NumericLessThanEquals": 1000000,
        "Next": "AutoApprove"
      }
    ],
    "Default": "DefaultHandler"
  }
}

Parallel State (병렬 실행):

{
  "ProcessInParallel": {
    "Type": "Parallel",
    "Branches": [
      {
        "StartAt": "GenerateThumbnail",
        "States": {
          "GenerateThumbnail": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:function:thumbnail",
            "End": true
          }
        }
      },
      {
        "StartAt": "ExtractMetadata",
        "States": {
          "ExtractMetadata": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:function:metadata",
            "End": true
          }
        }
      }
    ],
    "Next": "CombineResults"
  }
}

Wait State (대기):

{
  "WaitForApproval": {
    "Type": "Wait",
    "Seconds": 300,
    "Next": "CheckApprovalStatus"
  }
}

// 또는 특정 시간까지 대기
{
  "WaitUntilMidnight": {
    "Type": "Wait",
    "Timestamp": "2025-01-10T00:00:00Z",
    "Next": "DailyReport"
  }
}

Map State (반복):

{
  "ProcessEachItem": {
    "Type": "Map",
    "ItemsPath": "$.items",
    "MaxConcurrency": 5,
    "Iterator": {
      "StartAt": "ProcessItem",
      "States": {
        "ProcessItem": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:...:function:itemProcessor",
          "End": true
        }
      }
    },
    "Next": "GenerateSummary"
  }
}

---

실전 프로젝트 활용 사례

사례 1: CSV 처리 파이프라인

워크플로우:

S3 Upload → Step Functions 실행
  ├─ ValidateCSV (검증)
  │   ├─ PASSED → TransformData
  │   └─ FAILED → HandleError
  ├─ TransformData (변환)
  ├─ LoadData (DynamoDB 저장)
  └─ SendSuccessNotification

상태 기계 정의 (Serverless Framework):

stepFunctions:
  stateMachines:
    csvProcessing:
      name: CsvProcessingWorkflow
      definition:
        Comment: CSV 파일 처리 워크플로우
        StartAt: ValidateCSV
        States:
          ValidateCSV:
            Type: Task
            Resource: arn:aws:states:::lambda:invoke
            Parameters:
              FunctionName: ${self:service}-${self:provider.stage}-validator
              Payload.$: $
            ResultPath: $.validationResult
            Retry:
              - ErrorEquals:
                  - States.Timeout
                  - Lambda.ServiceException
                IntervalSeconds: 2
                MaxAttempts: 2
                BackoffRate: 2.0
            Catch:
              - ErrorEquals:
                  - States.ALL
                ResultPath: $.error
                Next: HandleError
            Next: CheckValidation

          CheckValidation:
            Type: Choice
            Choices:
              - Variable: $.validationResult.Payload.validationStatus
                StringEquals: PASSED
                Next: TransformData
            Default: HandleError

          TransformData:
            Type: Task
            Resource: arn:aws:states:::lambda:invoke
            Parameters:
              FunctionName: ${self:service}-${self:provider.stage}-transformer
              Payload:
                validRecords.$: $.validationResult.Payload.validRecords
                fileId.$: $.validationResult.Payload.fileId
            ResultPath: $.transformResult
            Timeout: 300
            Retry:
              - ErrorEquals:
                  - States.Timeout
                IntervalSeconds: 2
                MaxAttempts: 2
            Catch:
              - ErrorEquals:
                  - States.ALL
                ResultPath: $.error
                Next: HandleError
            Next: LoadData

          LoadData:
            Type: Task
            Resource: arn:aws:states:::lambda:invoke
            Parameters:
              FunctionName: ${self:service}-${self:provider.stage}-loader
              Payload:
                records.$: $.transformResult.Payload.transformedRecords
                statistics.$: $.transformResult.Payload.statistics
                fileId.$: $.transformResult.Payload.fileId
            ResultPath: $.loadResult
            Timeout: 600
            Retry:
              - ErrorEquals:
                  - DynamoDB.ProvisionedThroughputExceededException
                IntervalSeconds: 5
                MaxAttempts: 3
                BackoffRate: 2.0
            Catch:
              - ErrorEquals:
                  - States.ALL
                ResultPath: $.error
                Next: HandleError
            Next: SendSuccessNotification

          SendSuccessNotification:
            Type: Task
            Resource: arn:aws:states:::sns:publish
            Parameters:
              TopicArn: !Ref SuccessTopic
              Subject: CSV 처리 완료
              Message.$: States.Format('파일 {}의 처리가 완료되었습니다', $.fileId)
            End: true

          HandleError:
            Type: Task
            Resource: arn:aws:states:::lambda:invoke
            Parameters:
              FunctionName: ${self:service}-${self:provider.stage}-errorHandler
              Payload.$: $
            End: true

functions:
  validator:
    handler: lambdas/validator/index.handler
    timeout: 300
    memorySize: 512

  transformer:
    handler: lambdas/transformer/index.handler
    timeout: 300

  loader:
    handler: lambdas/loader/index.handler
    timeout: 600
    memorySize: 1024

  errorHandler:
    handler: lambdas/error-handler/index.handler

Step Functions 실행:

import { SFNClient, StartExecutionCommand } from '@aws-sdk/client-sfn';

const sfnClient = new SFNClient({});

// S3 이벤트 트리거 시 Step Functions 실행
export const handler = async (event) => {
  const s3Event = event.Records[0].s3;
  const bucket = s3Event.bucket.name;
  const key = decodeURIComponent(s3Event.object.key.replace(/\+/g, ' '));

  const executionName = `csv-processing-${Date.now()}`;

  await sfnClient.send(new StartExecutionCommand({
    stateMachineArn: process.env.STATE_MACHINE_ARN,
    name: executionName,
    input: JSON.stringify({
      bucket,
      key,
      fileId: key.split('/').pop().replace('.csv', ''),
      timestamp: new Date().toISOString()
    })
  }));

  return {
    statusCode: 200,
    body: JSON.stringify({ executionName })
  };
};

---

사례 2: 승인 워크플로우 (사람 개입)

워크플로우:

주문 생성
  ↓
금액 확인 (Choice)
  ├─ > 100만원 → 관리자 승인 대기 (Callback)
  │               ↓ (승인 링크 클릭)
  │              자동 진행
  └─ ≤ 100만원 → 자동 승인

상태 기계:

{
  "StartAt": "CheckAmount",
  "States": {
    "CheckAmount": {
      "Type": "Choice",
      "Choices": [{
        "Variable": "$.totalAmount",
        "NumericGreaterThan": 1000000,
        "Next": "RequestApproval"
      }],
      "Default": "ProcessOrder"
    },
    "RequestApproval": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Parameters": {
        "FunctionName": "sendApprovalEmail",
        "Payload": {
          "order.$": "$",
          "taskToken.$": "$$.Task.Token"
        }
      },
      "TimeoutSeconds": 86400,
      "Next": "ProcessOrder",
      "Catch": [{
        "ErrorEquals": ["States.Timeout"],
        "Next": "ApprovalTimeout"
      }]
    },
    "ProcessOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:processOrder",
      "End": true
    },
    "ApprovalTimeout": {
      "Type": "Fail",
      "Error": "ApprovalTimeout",
      "Cause": "승인 대기 시간 초과"
    }
  }
}

승인 이메일 Lambda:

import { SESClient, SendEmailCommand } from '@aws-sdk/client-ses';

const sesClient = new SESClient({});

export const handler = async (event) => {
  const order = event.order;
  const taskToken = event.taskToken;

  // 승인/거부 링크 생성
  const approveUrl = `https://api.example.com/approve?token=${encodeURIComponent(taskToken)}`;
  const rejectUrl = `https://api.example.com/reject?token=${encodeURIComponent(taskToken)}`;

  await sesClient.send(new SendEmailCommand({
    Source: 'approval@example.com',
    Destination: { ToAddresses: ['manager@example.com'] },
    Message: {
      Subject: { Data: `주문 승인 요청 - ${order.orderId}` },
      Body: {
        Html: {
          Data: `
            <h2>주문 승인 요청</h2>
            <p>주문 번호: ${order.orderId}</p>
            <p>금액: ${order.totalAmount.toLocaleString()}원</p>
            <p>
              <a href="${approveUrl}">승인</a> |
              <a href="${rejectUrl}">거부</a>
            </p>
          `
        }
      }
    }
  }));

  // Task Token은 저장하지 않고 URL에 포함
  // 사용자가 링크 클릭 시 API Gateway → Lambda → SendTaskSuccess 호출
};

승인/거부 처리 API:

import { SFNClient, SendTaskSuccessCommand, SendTaskFailureCommand } from '@aws-sdk/client-sfn';

const sfnClient = new SFNClient({});

export const handler = async (event) => {
  const taskToken = event.queryStringParameters.token;
  const action = event.path; // /approve 또는 /reject

  if (action === '/approve') {
    await sfnClient.send(new SendTaskSuccessCommand({
      taskToken,
      output: JSON.stringify({ approved: true })
    }));

    return {
      statusCode: 200,
      body: '주문이 승인되었습니다'
    };
  } else {
    await sfnClient.send(new SendTaskFailureCommand({
      taskToken,
      error: 'ApprovalRejected',
      cause: '관리자가 주문을 거부했습니다'
    }));

    return {
      statusCode: 200,
      body: '주문이 거부되었습니다'
    };
  }
};

---

고급 기능

1. 입력/출력 처리

InputPath, ResultPath, OutputPath:

{
  "ProcessOrder": {
    "Type": "Task",
    "Resource": "arn:aws:lambda:...:function:processOrder",
    "InputPath": "$.order",        // 입력 필터링: order 속성만 전달
    "ResultPath": "$.orderResult", // 결과 저장 경로
    "OutputPath": "$",             // 출력: 전체 상태 반환
    "Next": "SendNotification"
  }
}

데이터 흐름 예시:

// 초기 입력
{
  "orderId": "order-123",
  "customerId": "user-456",
  "order": {
    "items": [...],
    "totalAmount": 50000
  }
}

// InputPath: "$.order" → Lambda에 전달되는 데이터
{
  "items": [...],
  "totalAmount": 50000
}

// Lambda 반환값
{
  "paymentId": "pay-789",
  "status": "success"
}

// ResultPath: "$.orderResult" → 결과 병합
{
  "orderId": "order-123",
  "customerId": "user-456",
  "order": {
    "items": [...],
    "totalAmount": 50000
  },
  "orderResult": {
    "paymentId": "pay-789",
    "status": "success"
  }
}

2. Parameters (동적 데이터 전달)

{
  "InvokeLambda": {
    "Type": "Task",
    "Resource": "arn:aws:states:::lambda:invoke",
    "Parameters": {
      "FunctionName": "my-function",
      "Payload": {
        "orderId.$": "$.orderId",
        "timestamp.$": "$$.Execution.StartTime",
        "executionArn.$": "$$.Execution.Id",
        "staticValue": "constant"
      }
    }
  }
}

컨텍스트 객체 ($$):

• $$.Execution.Id: 실행 ID
• $$.Execution.StartTime: 시작 시간
• $$.State.Name: 현재 상태 이름
• $$.StateMachine.Id: 상태 기계 ARN

3. 조건 연산자

{
  "CheckOrderStatus": {
    "Type": "Choice",
    "Choices": [
      {
        "And": [
          {
            "Variable": "$.status",
            "StringEquals": "pending"
          },
          {
            "Variable": "$.amount",
            "NumericGreaterThan": 100000
          }
        ],
        "Next": "HighValuePendingOrder"
      },
      {
        "Or": [
          {
            "Variable": "$.status",
            "StringEquals": "cancelled"
          },
          {
            "Variable": "$.status",
            "StringEquals": "refunded"
          }
        ],
        "Next": "ClosedOrder"
      },
      {
        "Not": {
          "Variable": "$.paymentVerified",
          "BooleanEquals": true
        },
        "Next": "VerifyPayment"
      }
    ],
    "Default": "StandardProcessing"
  }
}

4. 에러 처리

Retry 정책:

{
  "ProcessPayment": {
    "Type": "Task",
    "Resource": "arn:aws:lambda:...:function:payment",
    "Retry": [
      {
        "ErrorEquals": ["TimeoutError"],
        "IntervalSeconds": 2,
        "MaxAttempts": 3,
        "BackoffRate": 2.0
      },
      {
        "ErrorEquals": ["NetworkError"],
        "IntervalSeconds": 1,
        "MaxAttempts": 5,
        "BackoffRate": 1.5
      }
    ],
    "Next": "Success"
  }
}

Catch 핸들러:

{
  "ProcessOrder": {
    "Type": "Task",
    "Resource": "arn:aws:lambda:...:function:processOrder",
    "Catch": [
      {
        "ErrorEquals": ["PaymentError"],
        "ResultPath": "$.error",
        "Next": "RefundCustomer"
      },
      {
        "ErrorEquals": ["InventoryError"],
        "ResultPath": "$.error",
        "Next": "NotifyOutOfStock"
      },
      {
        "ErrorEquals": ["States.ALL"],
        "ResultPath": "$.error",
        "Next": "GenericErrorHandler"
      }
    ]
  }
}

---

Standard vs Express 워크플로우

특성	Standard	Express
최대 실행 시간	1년	5분
실행 기록	완전한 이력 보관	CloudWatch Logs만
실행 보장	Exactly-once	At-least-once (Sync) / At-most-once (Async)
가격	상태 전환당 과금	실행 횟수 + 실행 시간
사용 사례	장기 워크플로우, 승인 프로세스	고속 처리, IoT 데이터

Express 워크플로우 예시:

stepFunctions:
  stateMachines:
    fastProcessing:
      type: EXPRESS
      definition:
        StartAt: ProcessEvent
        States:
          ProcessEvent:
            Type: Task
            Resource: arn:aws:lambda:...:function:processor
            End: true

---

모니터링 및 디버깅

CloudWatch Logs 통합:

stepFunctions:
  stateMachines:
    myWorkflow:
      loggingConfig:
        level: ALL
        includeExecutionData: true
        destinations:
          - !GetAtt StepFunctionsLogGroup.Arn

X-Ray 추적:

stepFunctions:
  stateMachines:
    myWorkflow:
      tracingConfig:
        enabled: true

실행 조회:

import { SFNClient, DescribeExecutionCommand } from '@aws-sdk/client-sfn';

const execution = await sfnClient.send(new DescribeExecutionCommand({
  executionArn: 'arn:aws:states:...:execution:workflow:exec-123'
}));

console.log(execution.status);  // RUNNING, SUCCEEDED, FAILED, TIMED_OUT
console.log(execution.input);
console.log(execution.output);

---

Best Practices

1. 멱등성 보장

// Lambda 함수에서 중복 실행 체크
export const handler = async (event) => {
  const orderId = event.orderId;
  const executionId = event.executionId;

  // DynamoDB에서 이미 처리했는지 확인
  const existing = await checkProcessed(orderId, executionId);
  if (existing) {
    console.log('Already processed');
    return existing.result;
  }

  // 처리 수행
  const result = await processOrder(event);

  // 처리 결과 저장
  await saveProcessed(orderId, executionId, result);

  return result;
};

2. 적절한 타임아웃 설정

{
  "LongRunningTask": {
    "Type": "Task",
    "Resource": "arn:aws:lambda:...:function:longTask",
    "TimeoutSeconds": 900,
    "HeartbeatSeconds": 300,
    "Next": "NextStep"
  }
}

3. 병렬 처리 활용

{
  "ProcessInParallel": {
    "Type": "Parallel",
    "Branches": [
      {
        "StartAt": "Task1",
        "States": {
          "Task1": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:function:task1",
            "End": true
          }
        }
      },
      {
        "StartAt": "Task2",
        "States": {
          "Task2": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:function:task2",
            "End": true
          }
        }
      }
    ],
    "Next": "CombineResults"
  }
}

---

비용 최적화

Standard 워크플로우:

상태 전환당: $0.000025 (처음 4,000개 무료)

예시: 10개 상태, 100만 실행
= 10,000,000 상태 전환
= (10,000,000 - 4,000) × $0.000025
= $250/월

Express 워크플로우:

요청당: $0.000001
실행 시간: $0.00001667 per GB-second

예시: 1분 실행, 512MB, 100만 실행
= 100만 × $0.000001 (요청)
+ 100만 × 60초 × 0.5GB × $0.00001667
= $1 + $500.1 = $501.1/월

최적화 전략:

1. 불필요한 상태 제거
2. 병렬 처리로 실행 시간 단축
3. Express 워크플로우는 짧은 작업에만 사용

---

마치며

AWS Step Functions는 복잡한 워크플로우를 간단하게 구현할 수 있게 해줌.

Step Functions를 사용해야 할 때:

• 여러 Lambda를 조율해야 할 때
• 조건 분기, 병렬 처리 필요
• 재시도 및 오류 처리 자동화
• 15분 이상 실행되는 프로세스
• 사람의 승인이 필요한 워크플로우

핵심 패턴:

1. 순차 처리: Task → Task → Task
2. 조건 분기: Choice 상태로 if-else
3. 병렬 처리: Parallel로 동시 실행
4. 반복: Map으로 배열 처리
5. 대기: Wait로 지연 처리

실전 프로젝트에서는 Step Functions를 Lambda(처리), DynamoDB(상태 저장), SNS(알림)와 조합하여 엔터프라이즈급 워크플로우를 구축할 수 있음.

도움이 되셨길 바랍니다! 😀