一名云计算从业者的思考

大家都在搜：

热搜词1 热搜词2 热搜词3

当前位置：首页 > 日常方案 > 正文内容

管理AWS Kinesis Video Streams：理解流生命周期与自动化清理

anycodes1年前 (2025-07-08)日常方案4830

## 引言

在使用AWS Kinesis Video Streams (KVS)服务时，许多用户会遇到一个常见的困惑：即使设置了数据保留期限，那些很久未使用的视频流资源仍然存在于控制台中。这篇博客将深入探讨KVS的流生命周期管理机制，解释为什么会出现这种情况，并提供一个Python脚本来帮助您自动化清理未使用的流资源。

## KVS流生命周期与数据保留机制

在AWS Kinesis Video Streams中，有两个关键概念需要区分：

1. **数据保留期限**：这控制流中的视频数据保存多长时间。当您设置720小时（30天）的保留期限时，这仅意味着视频数据会在30天后自动删除。

2. **流资源生命周期**：流本身作为AWS资源会一直存在，直到您明确删除它。AWS没有提供自动删除未使用流的机制。

这就解释了为什么您会看到2021年创建的流仍然显示在控制台中，尽管它们可能已经很久没有新数据了。

## 官方文档说明

AWS官方文档对此有明确说明：

> "You can also configure your Kinesis video stream to durably store media data for the specified retention period."
> — [AWS Kinesis Video Streams文档](https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/what-is-kinesis-video.html)

这表明保留期仅适用于媒体数据，而不是流资源本身。

删除流必须通过[DeleteStream API](https://docs.aws.amazon.com/kinesisvideostreams/latest/dg/API_DeleteStream.html)手动执行：

> "Deletes a Kinesis video stream and the data contained in the stream. This method marks the stream for deletion, and makes the data in the stream inaccessible immediately."

## 自动化清理解决方案：Python脚本

为了帮助管理和清理未使用的Kinesis Video Streams，以下是一个Python脚本，它可以识别旧的或未使用的流，并根据您的指示删除它们：

```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AWS Kinesis Video Streams Cleanup Script

This script helps identify and delete unused Kinesis Video Streams.
It filters streams based on creation date and provides a dry run option.
"""

import boto3
import argparse
import datetime
from dateutil import parser as date_parser
from colorama import init, Fore, Style

# Initialize colorama for cross-platform colored output
init()

def parse_arguments():
    """Parse command line arguments."""
    parser = argparse.ArgumentParser(description='Clean up unused Kinesis Video Streams')
    parser.add_argument('--region', default='us-east-2', help='AWS region name (default: us-east-2)')
    parser.add_argument('--older-than', default='365', type=int, help='Delete streams older than specified days (default: 365)')
    parser.add_argument('--prefix', default='', help='Only consider streams with this name prefix')
    parser.add_argument('--execute', action='store_true', help='Execute deletion (without this flag, script runs in dry-run mode)')
    return parser.parse_args()

def list_streams(client, prefix=''):
    """List all Kinesis Video Streams, optionally filtered by prefix."""
    all_streams = []
    
    # Prepare filter if prefix is provided
    name_filter = {}
    if prefix:
        name_filter = {
            'StreamNameCondition': {
                'ComparisonOperator': 'BEGINS_WITH',
                'ComparisonValue': prefix
            }
        }
    
    # Get all streams using pagination
    paginator = client.get_paginator('list_streams')
    for page in paginator.paginate(**name_filter):
        all_streams.extend(page['StreamInfoList'])
    
    return all_streams

def get_stream_details(client, streams, older_than_days):
    """Analyze streams and identify ones older than the threshold."""
    current_time = datetime.datetime.now(datetime.timezone.utc)
    threshold_date = current_time - datetime.timedelta(days=older_than_days)
    
    old_streams = []
    
    print(f"\n{Fore.BLUE}Analyzing {len(streams)} Kinesis Video Streams...{Style.RESET_ALL}")
    
    for stream in streams:
        stream_name = stream['StreamName']
        stream_arn = stream['StreamARN']
        creation_time = stream['CreationTime']
        
        # Format creation time for display
        creation_time_str = creation_time.strftime("%Y-%m-%d %H:%M:%S")
        
        # Calculate stream age
        stream_age = current_time - creation_time
        days_old = stream_age.days
        
        # Print stream info
        if creation_time < threshold_date:
            age_color = Fore.RED
        else:
            age_color = Fore.GREEN
            
        print(f"Stream: {Fore.CYAN}{stream_name}{Style.RESET_ALL}")
        print(f"  ARN: {stream_arn}")
        print(f"  Creation Date: {creation_time_str} ({age_color}{days_old} days old{Style.RESET_ALL})")
        
        # Add to deletion candidates if older than threshold
        if creation_time < threshold_date:
            old_streams.append({
                'name': stream_name,
                'arn': stream_arn,
                'creation_time': creation_time,
                'days_old': days_old
            })
    
    return old_streams

def delete_streams(client, streams, execute=False):
    """Delete the specified streams (or simulate deletion in dry-run mode)."""
    if not streams:
        print(f"\n{Fore.GREEN}No streams found that meet the deletion criteria.{Style.RESET_ALL}")
        return
    
    print(f"\n{Fore.YELLOW}Found {len(streams)} streams eligible for deletion:{Style.RESET_ALL}")
    
    for idx, stream in enumerate(streams, 1):
        print(f"{idx}. {Fore.CYAN}{stream['name']}{Style.RESET_ALL} (Created: {stream['creation_time'].strftime('%Y-%m-%d')}, {stream['days_old']} days old)")
    
    if execute:
        print(f"\n{Fore.RED}EXECUTION MODE: Will delete {len(streams)} streams{Style.RESET_ALL}")
        confirm = input("Type 'yes' to confirm deletion: ")
        
        if confirm.lower() == 'yes':
            for stream in streams:
                try:
                    print(f"Deleting stream: {Fore.CYAN}{stream['name']}{Style.RESET_ALL}...")
                    client.delete_stream(StreamARN=stream['arn'])
                    print(f"  {Fore.GREEN}Successfully deleted{Style.RESET_ALL}")
                except Exception as e:
                    print(f"  {Fore.RED}Error deleting stream: {str(e)}{Style.RESET_ALL}")
        else:
            print("Deletion cancelled.")
    else:
        print(f"\n{Fore.YELLOW}DRY-RUN MODE: No streams will be deleted.{Style.RESET_ALL}")
        print("To delete these streams, run the script with the --execute flag.")

def main():
    """Main function to run the script."""
    args = parse_arguments()
    
    print(f"{Fore.BLUE}AWS Kinesis Video Streams Cleanup{Style.RESET_ALL}")
    print(f"Region: {args.region}")
    print(f"Age threshold: {args.older_than} days")
    if args.prefix:
        print(f"Stream name prefix filter: {args.prefix}")
    print(f"Mode: {'EXECUTION' if args.execute else 'DRY-RUN'}")
    
    # Initialize AWS client
    kvs_client = boto3.client('kinesisvideo', region_name=args.region)
    
    # List streams
    all_streams = list_streams(kvs_client, args.prefix)
    
    # Get stream details and filter old streams
    old_streams = get_stream_details(kvs_client, all_streams, args.older_than)
    
    # Delete or simulate deletion
    delete_streams(kvs_client, old_streams, args.execute)
    
    print(f"\n{Fore.BLUE}Script completed.{Style.RESET_ALL}")

if __name__ == '__main__':
    main()
```

## 脚本使用说明

### 前提条件

1. 安装必要的Python包：
   ```
   pip install boto3 colorama python-dateutil
   ```

2. 确保您已配置AWS CLI凭证，并且具有足够的权限来列出和删除Kinesis Video Streams。

### 运行脚本

基本用法：

```bash
python kvs_cleanup.py
```

默认情况下，脚本会在`us-east-2`区域中查找所有超过365天的流，但不会实际删除它们（干运行模式）。

高级用法：

```bash
# 删除us-west-2区域中超过90天未使用的流
python kvs_cleanup.py --region us-west-2 --older-than 90 --execute

# 只考虑名称以"test-"开头的流
python kvs_cleanup.py --prefix test- --older-than 30 --execute
```

### 参数说明

- `--region`：指定AWS区域（默认：us-east-2）
- `--older-than`：指定删除多少天前创建的流（默认：365天）
- `--prefix`：只考虑以特定前缀开头的流
- `--execute`：实际执行删除操作（不带此参数则为干运行模式）

## 最佳实践建议

为了更好地管理您的Kinesis Video Streams资源，建议：

1. **使用标签**：为您的流添加有意义的标签，包括项目名称、用途和预期寿命等信息。

2. **定期审查**：定期运行此脚本（可以设置为每月或每季度的自动化任务）来识别和清理未使用的资源。

3. **更精细的判断逻辑**：根据您的业务需求，您可以扩展此脚本，增加更复杂的判断逻辑，例如检查最近的数据活动。

4. **保留记录**：在删除操作之前，确保记录将要删除的流信息，以便需要时可以恢复。

## 结论

理解AWS Kinesis Video Streams的流生命周期管理机制对于有效控制资源使用和成本至关重要。AWS不会自动删除未使用的流资源，这需要您主动进行管理。本文提供的Python脚本可以帮助您自动化这个过程，识别并清理那些不再使用的流资源。

记住，良好的资源管理不仅可以降低成本，还能提高AWS控制台的可用性，使您更容易找到和管理当前活跃的重要资源。

扫描二维码推送至手机访问。

本文链接：https://runor.cn/?id=22

分享给朋友：

返回列表

上一篇：AWS EC2 自动镜像创建与弹性伸缩部署指南

下一篇：在AWS EKS上安装Karpenter的完整指南

“管理AWS Kinesis Video Streams：理解流生命周期与自动化清理” 的相关文章

刘博说云

大家都在搜：

管理AWS Kinesis Video Streams：理解流生命周期与自动化清理

“管理AWS Kinesis Video Streams：理解流生命周期与自动化清理” 的相关文章

AWS账号分账的悖论：为什么"最佳实践"总是失效

OpenSearch向量搜索实践：使用Flow Framework快速部署

在AWS EKS上安装Karpenter的完整指南

AWS EC2 自动镜像创建与弹性伸缩部署指南

AWS与Azure云环境互联的三种主要方案

发表评论

Powered By Z-BlogPHP. Theme by TOYEAN.

刘博说云

大家都在搜：

管理AWS Kinesis Video Streams：理解流生命周期与自动化清理

“管理AWS Kinesis Video Streams：理解流生命周期与自动化清理” 的相关文章

AWS账号分账的悖论：为什么"最佳实践"总是失效

OpenSearch向量搜索实践：使用Flow Framework快速部署

在AWS EKS上安装Karpenter的完整指南

AWS EC2 自动镜像创建与弹性伸缩部署指南

AWS与Azure云环境互联的三种主要方案

发表评论取消回复

Powered By Z-BlogPHP. Theme by TOYEAN.

发表评论