Google Cloud Storage 사용하기 - 셋업 및 테스트

이전 포스팅에서 Google Cloud Functions을 셋업하고 간단하게 사용하는 방법에 대해 알아봤습니다.

 

https://android-developer.tistory.com/105

Google Cloud Functions 사용하기 - 기초 Cloud CLI 셋업

 

Cloud Functions에서는 함수를 실행할수만 있고 실행 결과 데이터 값을 장기간 저장할 수 없기 때문에

 

Cloud Storage를 사용해서 저장하는 방법을 알아보겠습니다.

---

Google Cloud Storage 비용 확인

먼저 Google Cloud Storage의 비용이 얼마인지 알아보겠습니다.

 

https://cloud.google.com/storage/pricing?hl=ko#cloud-storage-always-free

가격 책정  |  Cloud Storage  |  Google Cloud

 

Stadard Storage 기준으로 한 달에 5GB까지 저장은 무료입니다.

 

클라우드 Storage는 JSON API를 사용하여 데이터를 처리하는데

 

API에 따라 A 클래스 작업 / B 클래스 작업 / 무료 작업으로 나뉩니다.

 

각 클래스에 대한 작업은 다음과 같습니다.

 

 

Cloud Storage 버킷에서 같은 리전의 Google Cloud 내부 서비스로 데이터를 전송하는 경우에는 무료입니다.

 

(그렇기 때문에 Cloud Storage와 Cloud Functions을 같이 이용하실 분은 같은 리전으로 설정하는게 좋습니다.)

 

예를 들어

• Cloud Functions(미국) → Cloud Storage(미국)
  • 같은 리전에 배포된 서비스에서 읽으면 비용이 없음 

 

단, 외부 다운로드는 당연히 비용이 발생합니다.

 

• Cloud Functions(미국) → 휴대폰 앱 or 개별 웹사이트에서 HTTP 통신으로 데이터 다운로드

---

Google Cloud Storage 사용하기 - 기초 Cloud CLI 셋업

자세하게는 이미 Cloud Functions 셋업 부분에서 했기 때문에 간단하게 하고 넘어가겠습니다.

 

1. Cloud 프로젝트를 사용할 수 있도록 셋업합니다.

(이전 포스팅 참조)

 

https://android-developer.tistory.com/105

Google Cloud Functions 사용하기 - 기초 Cloud CLI 셋업

 

 

또는 공식 문서 참조

 

https://cloud.google.com/functions/docs/tutorials/storage?authuser=2&hl=ko

Cloud Storage 가이드  |  Cloud Run functions Documentation  |  Google Cloud

---

2. Google Cloud CLI를 셋업합니다.

(이전 포스팅 참조)

---

3. google-cloud-storage 패키지를 설치합니다.

pip install google-cloud-storage

---

4. Google Cloud CLI를 사용해서 Cloud Storage를 셋업합니다.

 

먼저 리전 버킷을 만들어야합니다.

 

버킷은 Storage의 저장 단위를 일컫습니다.

 

gcloud storage buckets create gs://YOUR_BUCKET_NAME --location=us-west1

 

버킷 이름은 전역으로 고유한 이름이어야하며 영문과 하이푼(-)만 사용할 수 있습니다.

 

location은 다음 페이지를 참조해서 지정하시면됩니다.

(us-west1이 가장 저렴하기 때문에 저는 이걸 추천합니다.)

 

https://cloud.google.com/about/locations?hl=ko#americas

글로벌 위치 - 리전 및 영역  |  Google Cloud

 

정상적으로 커맨드가 실행되었다면

 

Cloud Storage 콘솔에서 생성된 버킷을 확인할 수 있습니다.

 

그리고 다음 커맨드를 사용해서 현재 Cloud 서비스 계정이 Cloud Storage 함수를 사용할 수 있게 권한을 부여합니다.

 

PROJECT_ID=$(gcloud config get-value project)
PROJECT_NUMBER=$(gcloud projects list --filter="project_id:$PROJECT_ID" --format='value(project_number)')

SERVICE_ACCOUNT=$(gcloud storage service-agent --project=$PROJECT_ID)

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member serviceAccount:$SERVICE_ACCOUNT \
  --role roles/pubsub.publisher

 

혹시 위 커맨드를 실행 했는데 다음과 같이 에러 메시지가 나온다면

 

Service account service-xxxxxxxxxx@gs-project-accounts.iam.gserviceaccount.com does not exist.

 

 

IAM 서비스 어카운트를 새롭게 만들어야합니다.

 

다음과 같은 방법으로 IAM 서비스 계정을 새롭게 만들 수 있습니다.

 

https://cloud.google.com/iam/docs/service-accounts-create?hl=ko#iam-service-accounts-create-gcloud

서비스 계정 만들기  |  IAM Documentation  |  Google Cloud

 

권한 부여가 정상적으로 되었다면 커맨드 실행 후 다음과 같은 화면을 볼 수 있습니다.

 

---

파이썬으로 로컬 환경에서 Cloud Storage 테스트하기

먼저 PC에서 로컬 환경의 Cloud Storage로 HTTP 요청을 통해 데이터를 저장하는 방법에 대해 알아보겠습니다.

 

(Cloud Storage는 Pub/Sub에 의해서만 동작하기 때문에 외부에서 직접 호출하여 테스트해볼 수 없습니다.)

 

로컬 환경에서 테스트 해보기 위해서는 다음과 같은 파이썬 패키지 설치가 필요합니다.

 

pip install functions-framework

 

pip install cloudevents

 

 

그리고 다음과 같이 파이썬 함수를 정의합니다.

 

from cloudevents.http import CloudEvent
import functions_framework
from google.cloud import storage


# 객체 완료(=finalize) 함수 정의 - Cloud Storage에서 버킷에 finalize 이벤트가 발생하면 다음과 같은 객체를 반환한다.
# finalize 이벤트는 버킷에 새로운 파일이 업로드되거나 기존 파일이 덮어써질 때 발생하는 이벤트임
@functions_framework.cloud_event
def hello_gcs(cloud_event: CloudEvent) -> tuple:
    """This function is triggered by a change in a storage bucket.

    Args:
        cloud_event: The CloudEvent that triggered this function.
    Returns:
        The event ID, event type, bucket, name, metageneration, and timeCreated.
    """
    data = cloud_event.data

    event_id = cloud_event["id"]
    event_type = cloud_event["type"]

    bucket = data["bucket"]
    name = data["name"]
    metageneration = data["metageneration"]
    timeCreated = data["timeCreated"]
    updated = data["updated"]

    print(f"Event ID: {event_id}")
    print(f"Event type: {event_type}")
    print(f"Bucket: {bucket}")
    print(f"File: {name}")
    print(f"Metageneration: {metageneration}")
    print(f"Created: {timeCreated}")
    print(f"Updated: {updated}")

    return event_id, event_type, bucket, name, metageneration, timeCreated, updated


# CloudEvent 객체를 사용하여 Cloud Storage에 finalize 이벤트를 호출하는 함수
def test_hello_gcs():
    # 예시 CloudEvent 데이터
    event_data = {
        "bucket": "my-bucket",
        "name": "example.txt",
        "metageneration": 1,
        "timeCreated": "2023-01-01T12:00:00Z",
        "updated": "2023-01-02T12:00:00Z"
    }

    # CloudEvent finalize 타입의 객체 생성
    cloud_event = CloudEvent(
        {
            "id": "1234-5678-91011",
            "source": "/my-source",
            "type": "google.cloud.storage.object.finalize",
            "data": event_data,
        }
    )

    # 함수 호출 - finalize 타입 이벤트를 발생시켰으니 hello_gcs 함수가 호출되야한다.
    result = hello_gcs(cloud_event)
    print(f"Returned result: {result}")


# Cloud Storage에 새로운 데이터 저장 테스트 함수
def test_gcs_store_and_read():
    """테스트: GCS에 데이터 저장 및 읽기"""
    # GCS 클라이언트 초기화
    storage_client = storage.Client()

    # 로컬 테스트용 버킷 이름 (미리 GCS에서 생성 필요)
    bucket_name = "자신의 Storage 이름"
    file_name = "test-file.txt"
    file_content = "This is a test content for GCS."

    # 버킷 및 객체 참조
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)

    # 1. 데이터 저장
    blob.upload_from_string(file_content)
    print(f"File '{file_name}' uploaded to bucket '{bucket_name}'.")

    # 2. 데이터 읽기
    downloaded_content = blob.download_as_text()
    print(f"Downloaded content from '{file_name}': {downloaded_content}")

    # 결과 확인
    assert downloaded_content == file_content, "Downloaded content does not match the uploaded content!"
    print("Test passed: Uploaded and downloaded contents match!")

# 로컬 서버 실행
if __name__ == "__main__":
    print("Running hello_gcs test...")
    print("\nRunning GCS store and read test...")
    test_gcs_store_and_read()

 

hello_gcs 함수는 Cloud Storage의 버킷이 변경되었을 때 출력되는 객체 완료 이벤트 출력을 위한 함수입니다.

 

test_hello_gcs 함수는 로컬 환경에서 Cloud Storage를 변경하는 테스트 함수입니다.

 

테스트가 정상적으로 동작했다면 hello_gcs 함수를 사용해서 객체 완료 이벤트를 출력합니다.

 

test_gcs_store_and_read 함수는 데이터를 실제로 Cloud Storage에 넣어보는 저장하고 다운로드한 뒤 

 

해당 데이터가 올바르게 다운로드 되었는지 확인하는 테스트 함수입니다.

 

(자신의 Cloud Storage 버킷 이름을 지정해야하는 것에 주의해야합니다.)

---

먼저 hello_gcs 함수에 대한 테스트를 진행하기 위해서는

 

다음과 같은 커맨드를 사용해서 위 코드를 실행하여 로컬 테스트 환경을 만들어야합니다.

 

(커맨드 실행 위치는 Python 프로젝트의 루트 디렉토리여야 합니다.)

 

functions-framework --target=hello_gcs

 

혹시 파일이 main.py가 아니라 다른 파일에 코드를 적었다면 다음과 같이 --source 옵션을 사용하여 파일 위치를 지정합니다.

 

functions-framework --target=hello_gcs --source=google_cloud/storage/StorageTest.py

 

그리고 새롭게 터미널을 열어서 다음과 같은 방법으로 HTTP 통신을 실시합니다.

 

 curl -X POST http://localhost:8080/ \
-H "Content-Type: application/json" \
-d '{
  "id": "1234-5678-91011",
  "source": "/my-source",
  "type": "google.cloud.storage.object.finalize",
  "specversion": "1.0",
  "data": {
    "bucket": "my-bucket",
    "name": "example.txt",
    "metageneration": 1,
    "timeCreated": "2023-01-01T12:00:00Z",
    "updated": "2023-01-02T12:00:00Z"
  }
}'

 

위 방법으로 통해 로컬 환경에서 Pub/Sub 이벤트를 사용해 Cloud Storage의 트리거를 발동하는듯한 테스트를 할 수 있습니다.

 

정상적으로 실행되었다면 기존 터미널(프로젝트 루트 디렉토리에서 열었던)에서 다음과 같이 출력되는 것을 확인할 수 있습니다.

 

 

test_gcs_store_and_read 함수에 대한 테스트는 그냥 해당 파일을 실행하면됩니다.

 

그럼 다음과 같은 결과를 확인할 수 있습니다.

 

---

Google Cloud Storage 사용하기 - 셋업 및 테스트 정리

이번 포스팅에서는 다음과 같은 내용을 알아봤습니다. 

 

• Cloud Storage의 셋업방법
• Cloud Storage에 데이터를 넣고 다운로드 테스트
• Cloud Storage의 객체완료(finalize) 이벤트 수신 로컬 테스트

다음 포스팅에서는 Cloud Functions와 연계하여 JSON 형태의 데이터를 업로드하고 다운로드 해서 알맞은 데이터가 업로드 되었는지 확인해보겠습니다.