Storage Tiering¶

실무 운영 가이드에서 Manage, Hot, Warm, Cold, Archive을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

개요¶

실무 운영 가이드에서 Storage을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Tier Types¶

from truthound.stores.tiering.base import TierType

TierType.HOT      # Frequently accessed, fast, expensive
TierType.WARM     # Occasionally accessed, moderate speed/cost
TierType.COLD     # Rarely accessed, slow, cheap
TierType.ARCHIVE  # Very rarely accessed, very slow, cheapest

빠른 시작¶

from truthound.stores import get_store
from truthound.stores.tiering.base import StorageTier, TierType, TieringConfig
from truthound.stores.tiering.policies import AgeBasedTierPolicy

# Create tier backends
hot_store = get_store("filesystem", base_path=".truthound/hot")
warm_store = get_store("s3", bucket="my-bucket", prefix="warm/")
cold_store = get_store("s3", bucket="archive-bucket", prefix="cold/")

# Define tiers
tiers = [
    StorageTier(
        name="hot",
        store=hot_store,
        tier_type=TierType.HOT,
        priority=1,
    ),
    StorageTier(
        name="warm",
        store=warm_store,
        tier_type=TierType.WARM,
        priority=2,
    ),
    StorageTier(
        name="cold",
        store=cold_store,
        tier_type=TierType.COLD,
        priority=3,
    ),
]

# Define migration policies
config = TieringConfig(
    policies=[
        AgeBasedTierPolicy("hot", "warm", after_days=7),
        AgeBasedTierPolicy("warm", "cold", after_days=30),
    ],
    default_tier="hot",
)

Tier Policies¶

AgeBasedTierPolicy¶

실무 운영 가이드에서 Migrate을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.stores.tiering.policies import AgeBasedTierPolicy
from truthound.stores.tiering.base import MigrationDirection

# Move to warm after 7 days
policy = AgeBasedTierPolicy(
    from_tier="hot",
    to_tier="warm",
    after_days=7,
)

# Move to cold after 30 days
policy = AgeBasedTierPolicy(
    from_tier="warm",
    to_tier="cold",
    after_days=30,
)

# Combined: 1 day 12 hours
policy = AgeBasedTierPolicy(
    from_tier="hot",
    to_tier="warm",
    after_days=1,
    after_hours=12,
)

Parameters¶

실무 운영 가이드에서 Parameter을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Default을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `from_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	소스 tier name
실무 운영 가이드에서 `to_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Destination을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `after_days`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Days before 마이그레이션
실무 운영 가이드에서 `after_hours`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Additional을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `direction`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `MigrationDirection`, MigrationDirection을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `DEMOTE`, DEMOTE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	마이그레이션 direction

AccessBasedTierPolicy¶

실무 운영 가이드에서 Migrate을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.stores.tiering.policies import AccessBasedTierPolicy
from truthound.stores.tiering.base import MigrationDirection

# Demote items not accessed in 30 days
policy = AccessBasedTierPolicy(
    from_tier="hot",
    to_tier="warm",
    inactive_days=30,
)

# Promote frequently accessed items
policy = AccessBasedTierPolicy(
    from_tier="warm",
    to_tier="hot",
    min_access_count=100,
    access_window_days=7,
    direction=MigrationDirection.PROMOTE,
)

Parameters¶

실무 운영 가이드에서 Parameter을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Default을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `from_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	소스 tier name
실무 운영 가이드에서 `to_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Destination을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `inactive_days`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `min_access_count`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `access_window_days`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `7`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Window을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `direction`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `MigrationDirection`, MigrationDirection을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `DEMOTE`, DEMOTE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	마이그레이션 direction

SizeBasedTierPolicy¶

실무 운영 가이드에서 Migrate을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.stores.tiering.policies import SizeBasedTierPolicy

# Move large items (>100MB) to cold storage
policy = SizeBasedTierPolicy(
    from_tier="hot",
    to_tier="cold",
    min_size_mb=100,
)

# Limit hot tier to 10GB total
policy = SizeBasedTierPolicy(
    from_tier="hot",
    to_tier="warm",
    tier_max_size_gb=10,
)

Parameters¶

실무 운영 가이드에서 Parameter을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Default을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `from_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	소스 tier name
실무 운영 가이드에서 `to_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Destination을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `min_size_bytes`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Minimum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `min_size_kb`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Minimum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `min_size_mb`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Minimum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `min_size_gb`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Minimum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `tier_max_size_bytes`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Maximum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `tier_max_size_gb`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Maximum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `direction`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `MigrationDirection`, MigrationDirection을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `DEMOTE`, DEMOTE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	마이그레이션 direction

ScheduledTierPolicy¶

실무 운영 가이드에서 Migrate을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.stores.tiering.policies import ScheduledTierPolicy

# Migrate to cold storage on weekends at 2 AM
policy = ScheduledTierPolicy(
    from_tier="warm",
    to_tier="cold",
    on_days=[5, 6],  # Saturday, Sunday
    at_hour=2,
)

# Migrate items older than 7 days on Monday mornings
policy = ScheduledTierPolicy(
    from_tier="hot",
    to_tier="warm",
    on_days=[0],  # Monday
    at_hour=6,
    min_age_days=7,
)

Parameters¶

실무 운영 가이드에서 Parameter을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Default을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `from_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	소스 tier name
실무 운영 가이드에서 `to_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Destination을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `on_days`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `at_hour`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `min_age_days`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `0`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Minimum을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `direction`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `MigrationDirection`, MigrationDirection을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `DEMOTE`, DEMOTE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	마이그레이션 direction

CompositeTierPolicy¶

실무 운영 가이드에서 Combine, AND/OR을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.stores.tiering.policies import (
    AgeBasedTierPolicy,
    SizeBasedTierPolicy,
    AccessBasedTierPolicy,
    CompositeTierPolicy,
)

# AND logic: Migrate if old AND large (both conditions must be true)
policy = CompositeTierPolicy(
    from_tier="hot",
    to_tier="cold",
    policies=[
        AgeBasedTierPolicy("hot", "cold", after_days=30),
        SizeBasedTierPolicy("hot", "cold", min_size_mb=100),
    ],
    require_all=True,
)

# OR logic: Migrate if old OR large (either condition triggers migration)
policy = CompositeTierPolicy(
    from_tier="hot",
    to_tier="cold",
    policies=[
        AgeBasedTierPolicy("hot", "cold", after_days=90),
        SizeBasedTierPolicy("hot", "cold", min_size_mb=500),
    ],
    require_all=False,
)

Parameters¶

실무 운영 가이드에서 Parameter을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Default을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `from_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	소스 tier name
실무 운영 가이드에서 `to_tier`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Destination을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `policies`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `list[TierPolicy]`, TierPolicy을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Child을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `require_all`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `bool`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `True`, True을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `True`, `False`, True, AND, False을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `direction`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `MigrationDirection`, MigrationDirection을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `DEMOTE`, DEMOTE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	마이그레이션 direction

Advanced 예시¶

실무 운영 가이드에서 Combining, Three, More, Policies을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

# Migrate if: old AND large AND inactive
policy = CompositeTierPolicy(
    from_tier="hot",
    to_tier="archive",
    policies=[
        AgeBasedTierPolicy("hot", "archive", after_days=180),
        SizeBasedTierPolicy("hot", "archive", min_size_mb=500),
        AccessBasedTierPolicy("hot", "archive", inactive_days=90),
    ],
    require_all=True,
)

실무 운영 가이드에서 Nested, Composite, Policies을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

# Complex rule: (old AND large) OR (very old)
age_and_size = CompositeTierPolicy(
    from_tier="hot",
    to_tier="cold",
    policies=[
        AgeBasedTierPolicy("hot", "cold", after_days=30),
        SizeBasedTierPolicy("hot", "cold", min_size_mb=100),
    ],
    require_all=True,
)

very_old = AgeBasedTierPolicy("hot", "cold", after_days=365)

combined = CompositeTierPolicy(
    from_tier="hot",
    to_tier="cold",
    policies=[age_and_size, very_old],
    require_all=False,  # Either nested condition triggers
)

실무 운영 가이드에서 Batch, Processing을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

# CompositeTierPolicy delegates prepare_batch() to all child policies
# This enables efficient batch evaluation
policy.prepare_batch(tier_items)  # Prepares all child policies

# Description shows combined logic
print(policy.description)
# Output:
# Migrate from hot to cold when all of:
#   - Migrate from hot to cold if older than 30 days
#   - Migrate from hot to cold if size >= 100 MB

실무 운영 가이드에서 Serialization을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

# Serialize to dictionary (useful for configuration persistence)
config = policy.to_dict()
# {
#     "type": "CompositeTierPolicy",
#     "from_tier": "hot",
#     "to_tier": "cold",
#     "direction": "DEMOTE",
#     "policies": [
#         {"type": "AgeBasedTierPolicy", "from_tier": "hot", ...},
#         {"type": "SizeBasedTierPolicy", "from_tier": "hot", ...},
#     ],
#     "require_all": true
# }

CustomTierPolicy¶

Define custom 마이그레이션 logic.

from truthound.stores.tiering.policies import CustomTierPolicy
from truthound.stores.tiering.base import TierInfo

def large_and_idle(info: TierInfo) -> bool:
    """Migrate large items that are rarely accessed."""
    return info.size_bytes > 1024 * 1024 and info.access_count < 5

policy = CustomTierPolicy(
    from_tier="hot",
    to_tier="warm",
    predicate=large_and_idle,
    description="Large but rarely accessed items",
)

설정¶

StorageTier¶

실무 운영 가이드에서 Define을(를) 다루는 항목입니다:

from truthound.stores.tiering.base import StorageTier, TierType

tier = StorageTier(
    name="hot",                  # Unique identifier
    store=hot_store,             # Store backend
    tier_type=TierType.HOT,      # Tier classification
    priority=1,                  # Read order (lower = higher priority)
    cost_per_gb=0.023,           # For cost analysis
    retrieval_time_ms=10,        # Expected latency
    metadata={"region": "us-east-1"},
)

TieringConfig¶

실무 운영 가이드에서 Configure을(를) 다루는 항목입니다:

from truthound.stores.tiering.base import TieringConfig

config = TieringConfig(
    policies=[...],                    # Migration policies
    default_tier="hot",                # Default for new items
    enable_promotion=True,             # Promote on frequent access
    promotion_threshold=10,            # Accesses to trigger promotion
    check_interval_hours=24,           # Hours between auto-checks
    batch_size=100,                    # Items per migration batch
    enable_parallel_migration=False,   # Parallel migration
    max_parallel_migrations=4,         # Max concurrent migrations
)

TierInfo¶

실무 운영 가이드에서 Metadata을(를) 다루는 항목입니다:

from truthound.stores.tiering.base import TierInfo
from datetime import datetime

info = TierInfo(
    item_id="run-123",
    tier_name="hot",
    created_at=datetime.now(),
    migrated_at=None,              # When last migrated
    access_count=5,                # Number of accesses
    last_accessed=datetime.now(),  # Last access time
    size_bytes=1024,               # Item size
    next_migration=None,           # Scheduled migration time
)

TierInfo Fields¶

실무 운영 가이드에서 Field을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `item_id`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Item을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `tier_name`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Current을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `created_at`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `datetime`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Creation을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `migrated_at`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `access_count`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Access을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `last_accessed`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `size_bytes`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `int`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Item을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `next_migration`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

TieringResult¶

결과 of a tiering operation:

from truthound.stores.tiering.base import TieringResult

result = TieringResult(
    start_time=datetime.now(),
    end_time=datetime.now(),
    items_scanned=1000,
    items_migrated=50,
    bytes_migrated=1024 * 1024 * 100,  # 100 MB
    migrations=[
        {"item_id": "run-1", "from": "hot", "to": "warm"},
        ...
    ],
    errors=[],
    dry_run=False,
)

print(f"Migrated: {result.items_migrated}")
print(f"Duration: {result.duration_seconds}s")
print(f"Bytes moved: {result.bytes_migrated}")

마이그레이션 Direction¶

from truthound.stores.tiering.base import MigrationDirection

MigrationDirection.DEMOTE   # Move to cheaper/slower tier
MigrationDirection.PROMOTE  # Move to faster/more expensive tier

Tier Metadata Store¶

실무 운영 가이드에서 Track을(를) 다루는 항목입니다:

from truthound.stores.tiering.base import InMemoryTierMetadataStore

store = InMemoryTierMetadataStore()

# Save tier info
store.save_info(tier_info)

# Get tier info
info = store.get_info("run-123")

# List items in a tier
hot_items = store.list_by_tier("hot")

# Update access stats (called on read)
store.update_access("run-123")

# Delete info
store.delete_info("run-123")

Error Handling¶

from truthound.stores.tiering.base import (
    TieringError,
    TierNotFoundError,
    TierMigrationError,
    TierAccessError,
)

try:
    # Access non-existent tier
    ...
except TierNotFoundError as e:
    print(f"Tier not found: {e.tier_name}")

try:
    # Migration failure
    ...
except TierMigrationError as e:
    print(f"Migration failed: {e.item_id} from {e.from_tier} to {e.to_tier}")

try:
    # Tier access error
    ...
except TierAccessError as e:
    print(f"Access error on tier {e.tier_name}")

Real-World 예시¶

Cost-Optimized Tiering¶

# Define tiers with cost info
tiers = [
    StorageTier(
        name="hot",
        store=ssd_store,
        tier_type=TierType.HOT,
        cost_per_gb=0.10,
        retrieval_time_ms=10,
    ),
    StorageTier(
        name="warm",
        store=s3_standard_store,
        tier_type=TierType.WARM,
        cost_per_gb=0.023,
        retrieval_time_ms=100,
    ),
    StorageTier(
        name="cold",
        store=s3_glacier_store,
        tier_type=TierType.COLD,
        cost_per_gb=0.004,
        retrieval_time_ms=180000,  # 3 hours
    ),
]

config = TieringConfig(
    policies=[
        AgeBasedTierPolicy("hot", "warm", after_days=7),
        AgeBasedTierPolicy("warm", "cold", after_days=90),
        AccessBasedTierPolicy("cold", "warm", min_access_count=3,
                              direction=MigrationDirection.PROMOTE),
    ],
)

Access-Pattern Tiering¶

config = TieringConfig(
    policies=[
        # Demote inactive items
        AccessBasedTierPolicy("hot", "warm", inactive_days=14),
        AccessBasedTierPolicy("warm", "cold", inactive_days=60),
        # Promote frequently accessed items
        AccessBasedTierPolicy(
            "warm", "hot",
            min_access_count=50,
            access_window_days=7,
            direction=MigrationDirection.PROMOTE,
        ),
    ],
    enable_promotion=True,
    promotion_threshold=50,
)

Size-Constrained Tiering¶

config = TieringConfig(
    policies=[
        # Keep hot tier under 100GB
        SizeBasedTierPolicy("hot", "warm", tier_max_size_gb=100),
        # Move very large items directly to cold
        SizeBasedTierPolicy("hot", "cold", min_size_gb=1),
    ],
)

다음 단계¶

실무 운영 가이드에서 Retention, TTL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 Caching, In-memory을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
관측성 - 감사, metrics, tracing