📦 S3 + Lambda: 이벤트 기반 처리

S3(Simple Storage Service)는 파일 저장소고, Lambda는 서버 없이 코드를 실행하는 서비스야. 이 둘을 연동하면 파일이 업로드되면 자동으로 처리하는 시스템을 만들 수 있어!

실전 시나리오: 사용자가 이미지를 업로드하면 자동으로 썸네일을 생성하는 시스템 🖼️

구현 단계:
1️⃣ S3 버킷 생성
2️⃣ Lambda 함수 작성 (이미지 리사이징 코드)
3️⃣ S3 이벤트 트리거 설정
4️⃣ Lambda에 S3 접근 권한 부여

Lambda 함수 예제 (Python):

import boto3
from PIL import Image
import io

s3 = boto3.client('s3')

def lambda_handler(event, context):
    # S3 이벤트에서 버킷과 파일명 추출
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    # 원본 이미지 다운로드
    response = s3.get_object(Bucket=bucket, Key=key)
    image_data = response['Body'].read()
    
    # 이미지 리사이징
    image = Image.open(io.BytesIO(image_data))
    image.thumbnail((200, 200))
    
    # 썸네일 저장
    buffer = io.BytesIO()
    image.save(buffer, 'JPEG')
    buffer.seek(0)
    
    # S3에 업로드
    thumbnail_key = f"thumbnails/{key}"
    s3.put_object(Bucket=bucket, Key=thumbnail_key, Body=buffer)
    
    return {
        'statusCode': 200,
        'body': f'썸네일 생성 완료: {thumbnail_key}'
    }

이렇게 하면 이미지가 업로드될 때마다 자동으로 Lambda가 실행되어 썸네일을 만들어줘! 서버 관리 걱정 없이 말이야. 😎

🗄️ API Gateway + Lambda + DynamoDB: 서버리스 API

이 조합은 완전한 서버리스 백엔드를 만들 수 있어. 서버 없이 REST API를 구축하는 거지!

아키텍처 흐름:
클라이언트 → API Gateway → Lambda → DynamoDB

장점:
• 서버 관리 불필요
• 자동 스케일링
• 사용한 만큼만 비용 지불
• 높은 가용성

실전 예제: 간단한 TODO 앱 API 만들기 📝

import boto3
import json
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('TodoItems')

def lambda_handler(event, context):
    http_method = event['httpMethod']
    
    # POST: 새 TODO 생성
    if http_method == 'POST':
        body = json.loads(event['body'])
        item = {
            'id': str(datetime.now().timestamp()),
            'title': body['title'],
            'completed': False,
            'created_at': datetime.now().isoformat()
        }
        table.put_item(Item=item)
        return {
            'statusCode': 201,
            'body': json.dumps(item),
            'headers': {'Content-Type': 'application/json'}
        }
    
    # GET: TODO 목록 조회
    elif http_method == 'GET':
        response = table.scan()
        return {
            'statusCode': 200,
            'body': json.dumps(response['Items']),
            'headers': {'Content-Type': 'application/json'}
        }
    
    return {
        'statusCode': 400,
        'body': json.dumps({'error': 'Unsupported method'})
    }

API Gateway에서 이 Lambda를 연결하면, 바로 사용 가능한 REST API가 완성돼! 정말 간단하지? 🎯

📨 SQS + Lambda: 비동기 메시지 처리

SQS(Simple Queue Service)는 메시지 큐 서비스야. Lambda와 연동하면 대량의 작업을 안정적으로 처리할 수 있어!

사용 사례:
• 이메일 발송 대기열
• 주문 처리 시스템
• 데이터 처리 파이프라인
• 마이크로서비스 간 통신

왜 큐를 사용할까?
만약 1000명의 사용자가 동시에 주문을 하면? 서버가 터질 수 있어. 하지만 큐를 사용하면 주문을 차례대로 안전하게 처리할 수 있지! 🛡️

import boto3
import json

sqs = boto3.client('sqs')
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789/my-queue'

# 메시지 전송
def send_message(order_data):
    response = sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps(order_data),
        MessageAttributes={
            'OrderType': {
                'StringValue': 'Premium',
                'DataType': 'String'
            }
        }
    )
    return response['MessageId']

# Lambda에서 메시지 처리
def lambda_handler(event, context):
    for record in event['Records']:
        # 메시지 내용 추출
        body = json.loads(record['body'])
        
        # 주문 처리 로직
        process_order(body)
        
        print(f"주문 처리 완료: {body['order_id']}")
    
    return {'statusCode': 200}

def process_order(order):
    # 실제 주문 처리 로직
    print(f"주문 처리 중: {order}")
    # 재고 확인, 결제 처리, 배송 준비 등...

⚡ Azure Functions + Blob Storage

Azure Functions는 AWS Lambda와 비슷한 서버리스 컴퓨팅 서비스야. Blob Storage와 연동하면 파일 처리 자동화가 가능해!

실전 시나리오: CSV 파일이 업로드되면 자동으로 데이터베이스에 저장 📊

Azure Functions 예제 (C#):

using System.IO;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Azure.Storage.Blobs;

public static class BlobTriggerFunction
{
    [FunctionName("ProcessCsvFile")]
    public static void Run(
        [BlobTrigger("uploads/{name}", Connection = "AzureWebJobsStorage")] 
        Stream myBlob,
        string name,
        ILogger log)
    {
        log.LogInformation($"C# Blob trigger function Processed blob\n Name:{name} \n Size: {myBlob.Length} Bytes");
        
        // CSV 파일 읽기
        using (var reader = new StreamReader(myBlob))
        {
            string line;
            while ((line = reader.ReadLine()) != null)
            {
                // 각 라인 처리
                ProcessCsvLine(line);
            }
        }
        
        log.LogInformation("CSV 처리 완료! ✅");
    }
    
    private static void ProcessCsvLine(string line)
    {
        // 데이터베이스에 저장하는 로직
        // ...
    }
}

Python으로도 가능해:

import logging
import azure.functions as func
import csv
from io import StringIO

def main(myblob: func.InputStream):
    logging.info(f"Python blob trigger function processed blob \n"
                 f"Name: {myblob.name}\n"
                 f"Blob Size: {myblob.length} bytes")
    
    # Blob 내용 읽기
    blob_content = myblob.read().decode('utf-8')
    csv_reader = csv.DictReader(StringIO(blob_content))
    
    # 각 행 처리
    for row in csv_reader:
        process_row(row)
    
    logging.info("CSV 처리 완료! 🎉")

def process_row(row):
    # 데이터 처리 로직
    print(f"Processing: {row}")

🗃️ Cosmos DB + Azure Functions: 글로벌 데이터베이스

Cosmos DB는 전 세계에 분산된 NoSQL 데이터베이스야. 99.999%의 가용성을 보장하고, 밀리초 단위의 응답 속도를 제공해!

Change Feed 기능:
Cosmos DB의 데이터가 변경되면 자동으로 Azure Functions를 트리거할 수 있어. 실시간 데이터 처리에 완벽하지! ⚡

import logging
import azure.functions as func

def main(documents: func.DocumentList) -> str:
    if documents:
        logging.info(f'Document id: {documents[0]["id"]}')
        
        for doc in documents:
            # 변경된 문서 처리
            process_document(doc)
            
            # 예: 사용자 활동 로그 분석
            if doc.get('event_type') == 'purchase':
                send_recommendation(doc['user_id'])
    
    return 'OK'

def process_document(doc):
    logging.info(f"Processing document: {doc}")
    # 비즈니스 로직 실행

def send_recommendation(user_id):
    # 추천 시스템 트리거
    logging.info(f"Sending recommendations to user: {user_id}")

이렇게 하면 사용자가 구매를 하는 즉시 추천 시스템이 작동하는 실시간 서비스를 만들 수 있어! 🚀

🔐 Azure Key Vault: 비밀 관리

애플리케이션에서 사용하는 비밀번호, API 키, 인증서를 안전하게 관리하는 서비스야. 절대 코드에 하드코딩하지 말고 Key Vault를 사용하자! 🔒

Python에서 Key Vault 사용하기:

from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

# Key Vault 클라이언트 생성
credential = DefaultAzureCredential()
vault_url = "https://my-key-vault.vault.azure.net/"
client = SecretClient(vault_url=vault_url, credential=credential)

# 비밀 값 가져오기
database_password = client.get_secret("DatabasePassword").value
api_key = client.get_secret("ThirdPartyApiKey").value

# 이제 안전하게 사용!
connect_to_database(password=database_password)
call_external_api(api_key=api_key)

Managed Identity를 사용하면 코드에 인증 정보를 전혀 넣지 않아도 돼. Azure가 자동으로 인증을 처리해주거든! 👍

⚡ Cloud Functions + Cloud Storage

GCP의 Cloud Functions도 서버리스 컴퓨팅 서비스야. Cloud Storage와 연동하면 파일 처리를 자동화할 수 있어!

실전 예제: 이미지 업로드 시 Vision API로 자동 태깅 🏷️

from google.cloud import vision
from google.cloud import storage
import json

def process_image(event, context):
    """Cloud Storage 트리거 함수"""
    
    file_name = event['name']
    bucket_name = event['bucket']
    
    print(f'Processing file: {file_name}')
    
    # Vision API 클라이언트 생성
    vision_client = vision.ImageAnnotatorClient()
    storage_client = storage.Client()
    
    # 이미지 분석
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    
    image = vision.Image()
    image.source.image_uri = f'gs://{bucket_name}/{file_name}'
    
    # 라벨 감지
    response = vision_client.label_detection(image=image)
    labels = response.label_annotations
    
    # 결과 저장
    tags = [label.description for label in labels]
    result = {
        'file': file_name,
        'tags': tags,
        'confidence': [label.score for label in labels]
    }
    
    # 메타데이터로 저장
    result_blob = bucket.blob(f'results/{file_name}.json')
    result_blob.upload_from_string(json.dumps(result))
    
    print(f'Tags detected: {tags}')
    return 'OK'

이렇게 하면 사용자가 이미지를 업로드하는 즉시 AI가 자동으로 태그를 붙여줘! 정말 똑똑하지? 🤖

📊 BigQuery + Cloud Functions: 실시간 데이터 분석

BigQuery는 페타바이트급 데이터를 초고속으로 분석하는 데이터 웨어하우스야. Cloud Functions와 연동하면 실시간 분석 파이프라인을 만들 수 있어!

사용 사례:
• 실시간 로그 분석
• 사용자 행동 분석
• 매출 대시보드 자동 업데이트
• 이상 거래 탐지

실전 예제: Pub/Sub 메시지를 받아서 BigQuery에 저장

from google.cloud import bigquery
import base64
import json

def process_pubsub(event, context):
    """Pub/Sub 트리거 함수"""
    
    # Pub/Sub 메시지 디코딩
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    message_data = json.loads(pubsub_message)
    
    # BigQuery 클라이언트 생성
    client = bigquery.Client()
    table_id = "my-project.my_dataset.events"
    
    # 데이터 삽입
    rows_to_insert = [
        {
            "event_id": message_data['id'],
            "user_id": message_data['user_id'],
            "event_type": message_data['type'],
            "timestamp": message_data['timestamp'],
            "properties": json.dumps(message_data.get('properties', {}))
        }
    ]
    
    errors = client.insert_rows_json(table_id, rows_to_insert)
    
    if errors:
        print(f"Errors occurred: {errors}")
    else:
        print(f"Event {message_data['id']} inserted successfully!")
    
    return 'OK'

이제 BigQuery에서 SQL로 실시간 분석을 할 수 있어:

-- 최근 1시간 동안의 이벤트 통계
SELECT 
    event_type,
    COUNT(*) as event_count,
    COUNT(DISTINCT user_id) as unique_users
FROM `my-project.my_dataset.events`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
GROUP BY event_type
ORDER BY event_count DESC;

BigQuery는 수십억 건의 데이터도 몇 초 만에 분석해! 정말 빠르지? ⚡

🤖 Cloud Run + AI Platform: 머신러닝 서비스 배포

Cloud Run은 컨테이너를 서버리스로 실행하는 서비스야. 머신러닝 모델을 API로 배포하기에 완벽해!

실전 시나리오: 텍스트 감정 분석 API 만들기 😊😢😡

from flask import Flask, request, jsonify
from google.cloud import language_v1
import os

app = Flask(__name__)

# Natural Language API 클라이언트
client = language_v1.LanguageServiceClient()

@app.route('/analyze', methods=['POST'])
def analyze_sentiment():
    """텍스트 감정 분석 엔드포인트"""
    
    data = request.get_json()
    text = data.get('text', '')
    
    if not text:
        return jsonify({'error': 'No text provided'}), 400
    
    # 감정 분석 수행
    document = language_v1.Document(
        content=text,
        type_=language_v1.Document.Type.PLAIN_TEXT
    )
    
    sentiment = client.analyze_sentiment(
        request={'document': document}
    ).document_sentiment
    
    # 결과 반환
    result = {
        'text': text,
        'score': sentiment.score,  # -1.0 (부정) ~ 1.0 (긍정)
        'magnitude': sentiment.magnitude,
        'sentiment': get_sentiment_label(sentiment.score)
    }
    
    return jsonify(result)

def get_sentiment_label(score):
    """감정 점수를 라벨로 변환"""
    if score > 0.25:
        return 'positive 😊'
    elif score < -0.25:
        return 'negative 😢'
    else:
        return 'neutral 😐'

if __name__ == '__main__':
    port = int(os.environ.get('PORT', 8080))
    app.run(host='0.0.0.0', port=port)

Dockerfile:

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 app:app

배포 명령어:

# 컨테이너 빌드 및 배포
gcloud run deploy sentiment-api \
  --source . \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated

# 배포 완료! URL이 자동으로 생성돼
# https://sentiment-api-xxxxx-uc.a.run.app

이제 전 세계 어디서든 이 API를 호출할 수 있어! 자동 스케일링도 되고, 사용한 만큼만 비용을 내면 돼. 완벽하지? 🎯

🔄 Pub/Sub: 이벤트 기반 아키텍처

Pub/Sub은 메시지 큐 서비스로, 마이크로서비스 간 통신에 최적화되어 있어. AWS의 SQS/SNS와 비슷하지만 더 강력해!

Pub/Sub의 특징:
• 초당 수백만 건의 메시지 처리
• 전 세계 분산 처리
• 최소 한 번 전달 보장
• 메시지 순서 보장 옵션

실전 예제: 주문 처리 시스템

from google.cloud import pubsub_v1
import json

# Publisher 설정
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('my-project', 'orders')

def publish_order(order_data):
    """주문 메시지 발행"""
    
    message_json = json.dumps(order_data)
    message_bytes = message_json.encode('utf-8')
    
    # 메시지 속성 추가
    future = publisher.publish(
        topic_path,
        message_bytes,
        order_type=order_data['type'],
        priority=str(order_data.get('priority', 'normal'))
    )
    
    message_id = future.result()
    print(f'Published message ID: {message_id}')
    return message_id

# Subscriber 설정
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path('my-project', 'orders-sub')

def callback(message):
    """메시지 처리 콜백"""
    
    order_data = json.loads(message.data.decode('utf-8'))
    
    try:
        # 주문 처리
        process_order(order_data)
        
        # 처리 성공 시 ACK
        message.ack()
        print(f'Order {order_data["id"]} processed successfully!')
        
    except Exception as e:
        # 처리 실패 시 NACK (재시도)
        message.nack()
        print(f'Error processing order: {e}')

def process_order(order):
    """실제 주문 처리 로직"""
    print(f'Processing order: {order}')
    # 재고 확인, 결제 처리, 배송 준비 등...

# 구독 시작
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f'Listening for messages on {subscription_path}...')

try:
    streaming_pull_future.result()
except KeyboardInterrupt:
    streaming_pull_future.cancel()

🔄 멀티 클라우드 연동 패턴

패턴 1: API Gateway를 통한 통합
각 클라우드의 서비스를 API로 노출하고, 중앙 API Gateway에서 라우팅하는 방식이야.

# Kong, Apigee, AWS API Gateway 등을 사용한 통합 예시
# Python Flask 기반 통합 API

from flask import Flask, request, jsonify
import boto3  # AWS
from azure.storage.blob import BlobServiceClient  # Azure
from google.cloud import storage  # GCP

app = Flask(__name__)

# 각 클라우드 클라이언트 초기화
aws_s3 = boto3.client('s3')
azure_blob = BlobServiceClient.from_connection_string("connection_string")
gcp_storage = storage.Client()

@app.route('/upload', methods=['POST'])
def upload_file():
    """파일을 모든 클라우드에 동시 업로드"""
    
    file = request.files['file']
    filename = file.filename
    
    results = {}
    
    # AWS S3에 업로드
    try:
        aws_s3.upload_fileobj(file, 'my-bucket', filename)
        results['aws'] = 'success'
    except Exception as e:
        results['aws'] = f'error: {str(e)}'
    
    # Azure Blob에 업로드
    try:
        blob_client = azure_blob.get_blob_client(
            container='my-container',
            blob=filename
        )
        file.seek(0)  # 파일 포인터 리셋
        blob_client.upload_blob(file)
        results['azure'] = 'success'
    except Exception as e:
        results['azure'] = f'error: {str(e)}'
    
    # GCP Storage에 업로드
    try:
        bucket = gcp_storage.bucket('my-bucket')
        blob = bucket.blob(filename)
        file.seek(0)  # 파일 포인터 리셋
        blob.upload_from_file(file)
        results['gcp'] = 'success'
    except Exception as e:
        results['gcp'] = f'error: {str(e)}'
    
    return jsonify(results)

@app.route('/download/<cloud>/<filename>')
def download_file(cloud, filename):
    """특정 클라우드에서 파일 다운로드"""
    
    if cloud == 'aws':
        obj = aws_s3.get_object(Bucket='my-bucket', Key=filename)
        return obj['Body'].read()
    
    elif cloud == 'azure':
        blob_client = azure_blob.get_blob_client(
            container='my-container',
            blob=filename
        )
        return blob_client.download_blob().readall()
    
    elif cloud == 'gcp':
        bucket = gcp_storage.bucket('my-bucket')
        blob = bucket.blob(filename)
        return blob.download_as_bytes()
    
    return jsonify({'error': 'Invalid cloud provider'}), 400

패턴 2: 메시지 큐를 통한 비동기 통합
각 클라우드의 이벤트를 중앙 메시지 큐(Kafka, RabbitMQ 등)로 모으고, 다른 클라우드에서 소비하는 방식이야.

# Apache Kafka를 사용한 멀티 클라우드 이벤트 통합
from kafka import KafkaProducer, KafkaConsumer
import json

# Producer: AWS Lambda에서 이벤트 발행
def aws_lambda_handler(event, context):
    producer = KafkaProducer(
        bootstrap_servers=['kafka-broker:9092'],
        value_serializer=lambda v: json.dumps(v).encode('utf-8')
    )
    
    event_data = {
        'source': 'aws',
        'type': 'user_signup',
        'data': event
    }
    
    producer.send('multi-cloud-events', event_data)
    producer.flush()
    
    return {'statusCode': 200}

# Consumer: GCP Cloud Function에서 이벤트 소비
def gcp_event_consumer(request):
    consumer = KafkaConsumer(
        'multi-cloud-events',
        bootstrap_servers=['kafka-broker:9092'],
        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
    )
    
    for message in consumer:
        event = message.value
        
        if event['source'] == 'aws' and event['type'] == 'user_signup':
            # GCP에서 처리
            process_user_signup(event['data'])
    
    return 'OK'

패턴 3: 데이터 동기화
각 클라우드의 데이터베이스를 실시간으로 동기화하는 방식이야. Change Data Capture (CDC)를 사용하면 효율적이야!

# Debezium을 사용한 데이터베이스 동기화 예시
# AWS RDS → Kafka → Azure Cosmos DB

import pymongo
from kafka import KafkaConsumer
import json

# Kafka에서 변경 이벤트 수신
consumer = KafkaConsumer(
    'dbserver1.inventory.customers',
    bootstrap_servers=['kafka:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

# Azure Cosmos DB 클라이언트
cosmos_client = pymongo.MongoClient("mongodb://cosmos-connection-string")
cosmos_db = cosmos_client['inventory']
cosmos_collection = cosmos_db['customers']

for message in consumer:
    change_event = message.value
    
    if change_event['op'] == 'c':  # Create
        cosmos_collection.insert_one(change_event['after'])
    
    elif change_event['op'] == 'u':  # Update
        cosmos_collection.update_one(
            {'id': change_event['after']['id']},
            {'$set': change_event['after']}
        )
    
    elif change_event['op'] == 'd':  # Delete
        cosmos_collection.delete_one({'id': change_event['before']['id']})
    
    print(f"Synced change: {change_event['op']}")

🛠️ 멀티 클라우드 관리 도구

여러 클라우드를 동시에 관리하려면 적절한 도구가 필요해! 다음은 실무에서 많이 사용되는 도구들이야:

1. Terraform (Infrastructure as Code)
코드로 인프라를 정의하고 관리하는 도구야. 모든 주요 클라우드를 지원해! 🏗️

# Terraform으로 멀티 클라우드 인프라 정의
# main.tf

# AWS Provider
provider "aws" {
  region = "us-east-1"
}

# Azure Provider
provider "azurerm" {
  features {}
}

# GCP Provider
provider "google" {
  project = "my-project"
  region  = "us-central1"
}

# AWS S3 버킷 생성
resource "aws_s3_bucket" "data_bucket" {
  bucket = "my-multi-cloud-bucket-aws"
  
  tags = {
    Environment = "Production"
    ManagedBy   = "Terraform"
  }
}

# Azure Storage Account 생성
resource "azurerm_storage_account" "data_storage" {
  name                     = "mymulticloudstorage"
  resource_group_name      = azurerm_resource_group.main.name
  location                 = "eastus"
  account_tier             = "Standard"
  account_replication_type = "LRS"
}

# GCP Storage Bucket 생성
resource "google_storage_bucket" "data_bucket" {
  name     = "my-multi-cloud-bucket-gcp"
  location = "US"
  
  uniform_bucket_level_access = true
}

# 출력
output "aws_bucket_name" {
  value = aws_s3_bucket.data_bucket.id
}

output "azure_storage_name" {
  value = azurerm_storage_account.data_storage.name
}

output "gcp_bucket_name" {
  value = google_storage_bucket.data_bucket.name
}

2. Kubernetes (컨테이너 오케스트레이션)
컨테이너 기반 애플리케이션을 어떤 클라우드에서든 동일하게 실행할 수 있어! 진정한 클라우드 중립성이지. ☸️

3. Prometheus + Grafana (모니터링)
모든 클라우드의 메트릭을 한 곳에서 모니터링할 수 있어. 통합 대시보드를 만들 수 있지!

4. HashiCorp Vault (비밀 관리)
모든 클라우드의 인증 정보를 중앙에서 안전하게 관리할 수 있어. 🔐

💻 코드 구현

1. API Gateway (Node.js)

// server.js
const express = require('express');
const multer = require('multer');
const AWS = require('aws-sdk');
const { BlobServiceClient } = require('@azure/storage-blob');
const { Storage } = require('@google-cloud/storage');
const Redis = require('ioredis');
const { v4: uuidv4 } = require('uuid');

const app = express();
const upload = multer({ storage: multer.memoryStorage() });
const redis = new Redis();

// 클라우드 클라이언트 초기화
const s3 = new AWS.S3();
const azureBlob = BlobServiceClient.fromConnectionString(process.env.AZURE_CONNECTION);
const gcsStorage = new Storage();

app.post('/upload', upload.single('image'), async (req, res) => {
  const imageId = uuidv4();
  const file = req.file;
  
  console.log(`Processing image: ${imageId}`);
  
  try {
    // 1. AWS S3에 원본 저장
    const s3Params = {
      Bucket: 'my-images-bucket',
      Key: `originals/${imageId}.jpg`,
      Body: file.buffer,
      ContentType: file.mimetype
    };
    await s3.upload(s3Params).promise();
    console.log('✅ Uploaded to AWS S3');
    
    // 2. Azure Blob에 업로드 (얼굴 인식용)
    const azureContainer = azureBlob.getContainerClient('images');
    const azureBlob = azureContainer.getBlockBlobClient(`${imageId}.jpg`);
    await azureBlob.upload(file.buffer, file.buffer.length);
    console.log('✅ Uploaded to Azure Blob');
    
    // 3. GCP Storage에 업로드 (객체 감지용)
    const gcsBucket = gcsStorage.bucket('my-images-bucket');
    const gcsFile = gcsBucket.file(`${imageId}.jpg`);
    await gcsFile.save(file.buffer);
    console.log('✅ Uploaded to GCP Storage');
    
    // 4. 처리 작업을 큐에 추가
    await redis.lpush('image-processing-queue', JSON.stringify({
      imageId,
      timestamp: new Date().toISOString()
    }));
    
    res.json({
      success: true,
      imageId,
      message: '이미지 업로드 완료! 처리 중입니다... ⏳'
    });
    
  } catch (error) {
    console.error('Error:', error);
    res.status(500).json({ error: error.message });
  }
});

// 처리 상태 조회
app.get('/status/:imageId', async (req, res) => {
  const { imageId } = req.params;
  
  const status = await redis.hgetall(`image:${imageId}:status`);
  
  res.json({
    imageId,
    status: {
      aws: status.aws || 'pending',
      azure: status.azure || 'pending',
      gcp: status.gcp || 'pending'
    },
    results: {
      thumbnail: status.thumbnail_url,
      faces: JSON.parse(status.faces || '[]'),
      labels: JSON.parse(status.labels || '[]')
    }
  });
});

app.listen(3000, () => {
  console.log('🚀 Server running on port 3000');
});

2. AWS Lambda - 썸네일 생성

// aws-thumbnail-lambda.js
const AWS = require('aws-sdk');
const sharp = require('sharp');

const s3 = new AWS.S3();

exports.handler = async (event) => {
  const bucket = event.Records[0].s3.bucket.name;
  const key = event.Records[0].s3.object.key;
  
  console.log(`Processing: ${key}`);
  
  // 원본 이미지 다운로드
  const originalImage = await s3.getObject({
    Bucket: bucket,
    Key: key
  }).promise();
  
  // 썸네일 생성 (300x300)
  const thumbnail = await sharp(originalImage.Body)
    .resize(300, 300, { fit: 'cover' })
    .jpeg({ quality: 80 })
    .toBuffer();
  
  // 썸네일 업로드
  const thumbnailKey = key.replace('originals/', 'thumbnails/');
  await s3.putObject({
    Bucket: bucket,
    Key: thumbnailKey,
    Body: thumbnail,
    ContentType: 'image/jpeg'
  }).promise();
  
  console.log(`✅ Thumbnail created: ${thumbnailKey}`);
  
  // Redis에 상태 업데이트
  const imageId = key.split('/')[1].replace('.jpg', '');
  await updateStatus(imageId, 'aws', 'completed', {
    thumbnail_url: `https://${bucket}.s3.amazonaws.com/${thumbnailKey}`
  });
  
  return { statusCode: 200 };
};

3. Azure Function - 얼굴 인식

# azure-face-function.py
import logging
import json
import os
from azure.cognitiveservices.vision.face import FaceClient
from msrest.authentication import CognitiveServicesCredentials
import redis

# Face API 클라이언트
face_client = FaceClient(
    os.environ['FACE_ENDPOINT'],
    CognitiveServicesCredentials(os.environ['FACE_KEY'])
)

# Redis 클라이언트
redis_client = redis.Redis(
    host=os.environ['REDIS_HOST'],
    port=6379,
    decode_responses=True
)

def main(myblob: func.InputStream):
    logging.info(f"Processing blob: {myblob.name}")
    
    # 이미지 URL 생성
    image_url = f"https://{os.environ['STORAGE_ACCOUNT']}.blob.core.windows.net/images/{myblob.name}"
    
    # 얼굴 감지 및 감정 분석
    detected_faces = face_client.face.detect_with_url(
        url=image_url,
        return_face_attributes=['emotion', 'age', 'gender']
    )
    
    # 결과 처리
    faces_data = []
    for face in detected_faces:
        faces_data.append({
            'age': face.face_attributes.age,
            'gender': face.face_attributes.gender,
            'emotions': {
                'happiness': face.face_attributes.emotion.happiness,
                'sadness': face.face_attributes.emotion.sadness,
                'anger': face.face_attributes.emotion.anger,
                'surprise': face.face_attributes.emotion.surprise
            }
        })
    
    logging.info(f"✅ Detected {len(faces_data)} faces")
    
    # Redis에 결과 저장
    image_id = myblob.name.replace('.jpg', '')
    redis_client.hset(
        f"image:{image_id}:status",
        mapping={
            'azure': 'completed',
            'faces': json.dumps(faces_data)
        }
    )
    
    return f"Processed {len(faces_data)} faces"

4. GCP Cloud Function - 객체 감지

# gcp-vision-function.py
from google.cloud import vision
from google.cloud import storage
import redis
import json
import os

# Vision API 클라이언트
vision_client = vision.ImageAnnotatorClient()

# Redis 클라이언트
redis_client = redis.Redis(
    host=os.environ['REDIS_HOST'],
    port=6379,
    decode_responses=True
)

def process_image(event, context):
    """Cloud Storage 트리거 함수"""
    
    file_name = event['name']
    bucket_name = event['bucket']
    
    print(f'Processing file: {file_name}')
    
    # 이미지 분석
    image = vision.Image()
    image.source.image_uri = f'gs://{bucket_name}/{file_name}'
    
    # 라벨 감지
    label_response = vision_client.label_detection(image=image)
    labels = [
        {
            'description': label.description,
            'score': label.score,
            'confidence': f'{label.score * 100:.1f}%'
        }
        for label in label_response.label_annotations
    ]
    
    # 객체 감지
    object_response = vision_client.object_localization(image=image)
    objects = [
        {
            'name': obj.name,
            'score': obj.score
        }
        for obj in object_response.localized_object_annotations
    ]
    
    print(f'✅ Detected {len(labels)} labels and {len(objects)} objects')
    
    # Redis에 결과 저장
    image_id = file_name.replace('.jpg', '')
    redis_client.hset(
        f'image:{image_id}:status',
        mapping={
            'gcp': 'completed',
            'labels': json.dumps(labels),
            'objects': json.dumps(objects)
        }
    )
    
    return 'OK'

🎨 프론트엔드 (React)

// ImageUploader.jsx
import React, { useState } from 'react';
import axios from 'axios';

function ImageUploader() {
  const [file, setFile] = useState(null);
  const [imageId, setImageId] = useState(null);
  const [status, setStatus] = useState(null);
  const [loading, setLoading] = useState(false);

  const handleUpload = async () => {
    if (!file) return;
    
    setLoading(true);
    const formData = new FormData();
    formData.append('image', file);
    
    try {
      // 이미지 업로드
      const uploadRes = await axios.post('/upload', formData);
      const { imageId } = uploadRes.data;
      setImageId(imageId);
      
      // 처리 상태 폴링
      const interval = setInterval(async () => {
        const statusRes = await axios.get(`/status/${imageId}`);
        setStatus(statusRes.data);
        
        // 모든 처리 완료 확인
        const { aws, azure, gcp } = statusRes.data.status;
        if (aws === 'completed' && azure === 'completed' && gcp === 'completed') {
          clearInterval(interval);
          setLoading(false);
        }
      }, 2000);
      
    } catch (error) {
      console.error('Error:', error);
      setLoading(false);
    }
  };

  return (
    <div className="image-uploader">
      <h2>🖼️ 멀티 클라우드 이미지 분석</h2>
      
      <input
        type="file"
        accept="image/*"
        onChange={(e) => setFile(e.target.files[0])}
      />
      
      <button onClick={handleUpload} disabled={!file || loading}>
        {loading ? '처리 중... ⏳' : '업로드 및 분석 🚀'}
      </button>
      
      {status && (
        <div className="results">
          <h3>처리 상태</h3>
          <div className="status-grid">
            <div className={`status-card ${status.status.aws}`}>
              <h4>AWS</h4>
              <p>썸네일 생성: {status.status.aws}</p>
              {status.results.thumbnail && (
                <img src={status.results.thumbnail} alt="Thumbnail" />
              )}
            </div>
            
            <div className={`status-card ${status.status.azure}`}>
              <h4>Azure</h4>
              <p>얼굴 인식: {status.status.azure}</p>
              {status.results.faces && status.results.faces.length > 0 && (
                <ul>
                  {status.results.faces.map((face, i) => (
                    <li key={i}>
                      {face.gender}, {face.age}세
                      <br />
                      행복: {(face.emotions.happiness * 100).toFixed(1)}%
                    </li>
                  ))}
                </ul>
              )}
            </div>
            
            <div className={`status-card ${status.status.gcp}`}>
              <h4>GCP</h4>
              <p>객체 감지: {status.status.gcp}</p>
              {status.results.labels && (
                <ul>
                  {status.results.labels.slice(0, 5).map((label, i) => (
                    <li key={i}>
                      {label.description} ({label.confidence})
                    </li>
                  ))}
                </ul>
              )}
            </div>
          </div>
        </div>
      )}
    </div>
  );
}

export default ImageUploader;

🛡️ 보안 베스트 프랙티스

1. 최소 권한 원칙 (Principle of Least Privilege)
각 서비스와 사용자에게 필요한 최소한의 권한만 부여해야 해.

AWS IAM 정책 예시:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject"
      ],
      "Resource": "arn:aws:s3:::my-bucket/uploads/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:123456789:table/MyTable"
    }
  ]
}

2. 비밀 정보 관리
절대로 코드에 하드코딩하지 말고, 비밀 관리 서비스를 사용하자!

• AWS: Secrets Manager, Parameter Store
• Azure: Key Vault
• GCP: Secret Manager

Python에서 AWS Secrets Manager 사용:

import boto3
import json

def get_secret(secret_name):
    """AWS Secrets Manager에서 비밀 가져오기"""
    
    client = boto3.client('secretsmanager', region_name='us-east-1')
    
    try:
        response = client.get_secret_value(SecretId=secret_name)
        secret = json.loads(response['SecretString'])
        return secret
    except Exception as e:
        print(f"Error retrieving secret: {e}")
        raise

# 사용 예시
db_credentials = get_secret('prod/database/credentials')
db_host = db_credentials['host']
db_password = db_credentials['password']

# 이제 안전하게 데이터베이스 연결!

3. 네트워크 보안
• VPC (Virtual Private Cloud)로 네트워크 격리
• Security Group으로 방화벽 규칙 설정
• Private Subnet에 민감한 리소스 배치
• VPN이나 Direct Connect로 안전한 연결

4. 암호화
• 전송 중 암호화: HTTPS/TLS 사용
• 저장 시 암호화: S3, RDS 등에서 암호화 활성화
• 키 관리: KMS (Key Management Service) 사용

5. 로깅과 모니터링
• CloudTrail (AWS), Activity Log (Azure), Cloud Audit Logs (GCP)로 모든 활동 기록
• 이상 행동 탐지 시스템 구축
• 정기적인 보안 감사

💰 비용 최적화 전략

클라우드는 편리하지만 방심하면 청구서 폭탄을 맞을 수 있어! 💣

1. 리소스 크기 최적화
• 과도하게 큰 인스턴스 사용하지 않기
• Auto Scaling으로 필요한 만큼만 사용
• Reserved Instances나 Savings Plans로 장기 할인

2. 서버리스 활용
• Lambda, Cloud Functions 등은 실행 시간만큼만 과금
• 트래픽이 적은 서비스는 서버리스가 훨씬 저렴

3. 스토리지 최적화
• 자주 접근하지 않는 데이터는 저렴한 스토리지 클래스로 이동
• S3 Lifecycle Policy로 자동 이동 설정
• 불필요한 데이터 정기적으로 삭제

S3 Lifecycle Policy 예시:

{
  "Rules": [
    {
      "Id": "MoveToGlacier",
      "Status": "Enabled",
      "Transitions": [
        {
          "Days": 30,
          "StorageClass": "STANDARD_IA"
        },
        {
          "Days": 90,
          "StorageClass": "GLACIER"
        }
      ],
      "Expiration": {
        "Days": 365
      }
    }
  ]
}

4. 네트워크 비용 절감
• 같은 리전 내에서 통신 (리전 간 전송은 비용 발생)
• CloudFront, CDN 활용으로 전송 비용 절감
• NAT Gateway 대신 NAT Instance 고려

5. 비용 모니터링
• AWS Cost Explorer, Azure Cost Management 활용
• 예산 알림 설정
• 태그를 활용한 비용 추적

Python으로 AWS 비용 모니터링:

import boto3
from datetime import datetime, timedelta

def get_cost_report():
    """지난 30일간 비용 조회"""
    
    client = boto3.client('ce')  # Cost Explorer
    
    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=30)
    
    response = client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date.strftime('%Y-%m-%d'),
            'End': end_date.strftime('%Y-%m-%d')
        },
        Granularity='DAILY',
        Metrics=['UnblendedCost'],
        GroupBy=[
            {
                'Type': 'SERVICE',
                'Key': 'SERVICE'
            }
        ]
    )
    
    # 서비스별 비용 집계
    service_costs = {}
    for result in response['ResultsByTime']:
        for group in result['Groups']:
            service = group['Keys'][0]
            cost = float(group['Metrics']['UnblendedCost']['Amount'])
            
            if service in service_costs:
                service_costs[service] += cost
            else:
                service_costs[service] = cost
    
    # 비용 높은 순으로 정렬
    sorted_costs = sorted(
        service_costs.items(),
        key=lambda x: x[1],
        reverse=True
    )
    
    print("📊 서비스별 비용 (지난 30일)")
    print("-" * 50)
    for service, cost in sorted_costs[:10]:
        print(f"{service:30s} ${cost:10.2f}")
    
    total_cost = sum(service_costs.values())
    print("-" * 50)
    print(f"{'총 비용':30s} ${total_cost:10.2f}")
    
    return service_costs

# 실행
get_cost_report()

6. 개발 환경 최적화
• 개발/테스트 환경은 업무 시간에만 실행
• Spot Instances (AWS), Low-priority VMs (Azure) 활용
• 프리 티어 최대한 활용