sivchari/kumo
GitHub: sivchari/kumo
一款基于 Go 语言的高性能 AWS 服务模拟器，通过轻量化本地环境替代云端资源，支持 S3、Lambda 等 74 种核心服务的模拟测试。
Stars: 671 | Forks: 29
一款用 Go 编写的轻量级 AWS 服务模拟器。
既可作为 CI/CD 测试工具，也可作为本地开发服务器，并支持可选的数据持久化。
## 功能特性
- **无需身份验证** - 非常适合 CI 环境
- **单一二进制文件** - 易于分发和部署
- **Docker 支持** - 作为容器运行
- **轻量级** - 启动快速，资源占用极低
- **兼容 AWS SDK v2** - 与 Go AWS SDK v2 无缝协作
- **可选数据持久化** - 通过 `KUMO_DATA_DIR` 在重启后保留数据
## 支持的服务（74 项服务）
### 存储
| Service | Description |
|---------|-------------|
| S3 | 对象存储 |
| S3 Control | S3 账户级操作 |
| S3 Tables | S3 表存储桶 |
| DynamoDB | NoSQL 数据库 |
| ElastiCache | 内存中缓存 |
| MemoryDB | 兼容 Redis 的数据库 |
| Glacier | 归档存储 |
| EBS | 块存储 |
### 计算
| Service | Description |
|---------|-------------|
| Lambda | 无服务器函数 |
| Batch | 批处理计算 |
| EC2 | 虚拟机 |
| Elastic Beanstalk | 应用程序部署 |
### 容器
| Service | Description |
|---------|-------------|
| ECS | 容器编排 |
| ECR | 容器镜像仓库 |
| EKS | Kubernetes 服务 |
### 数据库
| Service | Description |
|---------|-------------|
| RDS | 关系型数据库服务 |
| Neptune | 图数据库 |
### 消息传递与集成
| Service | Description |
|---------|-------------|
| SQS | 消息队列 |
| SNS | 发布/订阅消息 |
| EventBridge | 事件总线 |
| Kinesis | 实时流式传输 |
| Firehose | 数据投递 |
| MQ | 消息代理 |
| Pipes | 事件驱动集成 |
| MSK (Kafka) | Kafka 托管流式处理 |
### 安全与身份
| Service | Description |
|---------|-------------|
| IAM | 身份与访问管理 |
| KMS | 密钥管理 |
| Secrets Manager | 密钥存储 |
| ACM | 证书管理 |
| Cognito | 用户身份验证 |
| Security Lake | 安全数据湖 |
| STS | 安全令牌服务 |
| Macie | 数据安全与隐私 |
### 监控与日志
| Service | Description |
|---------|-------------|
| CloudWatch | 指标与告警 |
| CloudWatch Logs | 日志管理 |
| X-Ray | 分布式追踪 |
| CloudTrail | API 审计日志 |
### 网络与内容分发
| Service | Description |
|---------|-------------|
| CloudFront | CDN |
| Global Accelerator | 网络加速 |
| API Gateway | API 管理 |
| Route 53 | DNS 服务 |
| Route 53 Resolver | DNS 解析器 |
| ELBv2 | 负载均衡 |
| App Mesh | 服务网格 |
| Location | 基于位置的服务 |
### 应用集成
| Service | Description |
|---------|-------------|
| Step Functions | 工作流编排 |
| AppSync | GraphQL API |
| SES v2 | 电子邮件服务 |
| Scheduler | 任务调度 |
| Amplify | 全栈应用程序托管 |
### 管理与配置
| Service | Description |
|---------|-------------|
| SSM | Systems Manager |
| Config | 资源配置 |
| CloudFormation | 基础设施即代码 |
| Organizations | 多账户管理 |
| Service Quotas | 服务限额管理 |
| CodeConnections | 源代码连接 |
| Backup | 集中备份服务 |
### 分析与机器学习
| Service | Description |
|---------|-------------|
| Athena | SQL 查询服务 |
| Glue | ETL 服务 |
| Comprehend | 自然语言处理服务 |
| Rekognition | 图像/视频分析 |
| SageMaker | 机器学习 |
| Forecast | 时间序列预测 |
| Data Exchange | 数据市场 |
| Entity Resolution | 实体匹配 |
### 开发者工具
| Service | Description |
|---------|-------------|
| CodeGuru Profiler | 应用程序性能分析 |
| CodeGuru Reviewer | 自动代码审查 |
### 其他服务
| Service | Description |
|---------|-------------|
| Cost Explorer | 成本分析 |
| DLM | 数据生命周期管理器 |
| Directory Service | Microsoft AD |
| EMR Serverless | 大数据处理 |
| FinSpace | 金融数据管理 |
| GameLift | 游戏服务器托管 |
| Resilience Hub | 应用程序韧性 |
## 快速开始
### Docker
```
docker run -p 4566:4566 ghcr.io/sivchari/kumo:latest
```
启用数据持久化：
```
docker run -p 4566:4566 \
-e KUMO_DATA_DIR=/data \
-v kumo-data:/data \
ghcr.io/sivchari/kumo:latest
```
### 二进制文件
```
# Build
make build
# Run
./bin/kumo
# Run with data persistence
KUMO_DATA_DIR=./data ./bin/kumo
```
### Docker Compose
```
services:
kumo:
image: ghcr.io/sivchari/kumo:latest
ports:
- "4566:4566"
```
启用数据持久化：
```
services:
kumo:
image: ghcr.io/sivchari/kumo:latest
ports:
- "4566:4566"
environment:
- KUMO_DATA_DIR=/data
volumes:
- kumo-data:/data
volumes:
kumo-data:
```
## 使用示例
### S3
```
package main
import (
"context"
"strings"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/s3"
)
func main() {
cfg, _ := config.LoadDefaultConfig(context.TODO(),
config.WithRegion("us-east-1"),
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
)
client := s3.NewFromConfig(cfg, func(o *s3.Options) {
o.BaseEndpoint = aws.String("http://localhost:4566")
o.UsePathStyle = true
})
// Create bucket
client.CreateBucket(context.TODO(), &s3.CreateBucketInput{
Bucket: aws.String("my-bucket"),
})
// Put object
client.PutObject(context.TODO(), &s3.PutObjectInput{
Bucket: aws.String("my-bucket"),
Key: aws.String("hello.txt"),
Body: strings.NewReader("Hello, World!"),
})
}
```
### SQS
```
package main
import (
"context"
"fmt"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/sqs"
)
func main() {
cfg, _ := config.LoadDefaultConfig(context.TODO(),
config.WithRegion("us-east-1"),
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
)
client := sqs.NewFromConfig(cfg, func(o *sqs.Options) {
o.BaseEndpoint = aws.String("http://localhost:4566")
})
// Create queue
result, _ := client.CreateQueue(context.TODO(), &sqs.CreateQueueInput{
QueueName: aws.String("my-queue"),
})
// Send message
client.SendMessage(context.TODO(), &sqs.SendMessageInput{
QueueUrl: result.QueueUrl,
MessageBody: aws.String("Hello from SQS!"),
})
// Receive message
messages, _ := client.ReceiveMessage(context.TODO(), &sqs.ReceiveMessageInput{
QueueUrl: result.QueueUrl,
})
for _, msg := range messages.Messages {
fmt.Println(*msg.Body)
}
}
```
### DynamoDB
```
package main
import (
"context"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/dynamodb"
"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)
func main() {
cfg, _ := config.LoadDefaultConfig(context.TODO(),
config.WithRegion("us-east-1"),
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
)
client := dynamodb.NewFromConfig(cfg, func(o *dynamodb.Options) {
o.BaseEndpoint = aws.String("http://localhost:4566")
})
// Create table
client.CreateTable(context.TODO(), &dynamodb.CreateTableInput{
TableName: aws.String("users"),
KeySchema: []types.KeySchemaElement{
{AttributeName: aws.String("id"), KeyType: types.KeyTypeHash},
},
AttributeDefinitions: []types.AttributeDefinition{
{AttributeName: aws.String("id"), AttributeType: types.ScalarAttributeTypeS},
},
BillingMode: types.BillingModePayPerRequest,
})
// Put item
client.PutItem(context.TODO(), &dynamodb.PutItemInput{
TableName: aws.String("users"),
Item: map[string]types.AttributeValue{
"id": &types.AttributeValueMemberS{Value: "user-1"},
"name": &types.AttributeValueMemberS{Value: "Alice"},
},
})
}
```
### Secrets Manager
```
package main
import (
"context"
"fmt"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/credentials"
"github.com/aws/aws-sdk-go-v2/service/secretsmanager"
)
func main() {
cfg, _ := config.LoadDefaultConfig(context.TODO(),
config.WithRegion("us-east-1"),
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
)
client := secretsmanager.NewFromConfig(cfg, func(o *secretsmanager.Options) {
o.BaseEndpoint = aws.String("http://localhost:4566")
})
// Create secret
client.CreateSecret(context.TODO(), &secretsmanager.CreateSecretInput{
Name: aws.String("my-secret"),
SecretString: aws.String(`{"username":"admin","password":"secret123"}`),
})
// Get secret
result, _ := client.GetSecretValue(context.TODO(), &secretsmanager.GetSecretValueInput{
SecretId: aws.String("my-secret"),
})
fmt.Println(*result.SecretString)
}
```
## 配置
环境变量：
| Variable | Default | Description |
|----------|---------|-------------|
| `KUMO_HOST` | `0.0.0.0` | 服务器绑定地址 |
| `KUMO_PORT` | `4566` | 服务器端口 |
| `KUMO_LOG_LEVEL` | `info` | 日志级别 |
| `KUMO_DATA_DIR` | (未设置) | 持久化存储目录。未设置时，数据仅存储在内存中。 |
## 日志记录
kumo 使用结构化字段记录所有请求。日志级别控制详细的程度：
### INFO（默认）
每个请求都会记录方法、路径、状态、持续时间和 API 操作名称：
```
level=INFO msg=request method=POST path=/ status=200 duration=61µs request_id=... target=secretsmanager.CreateSecret
level=INFO msg=request method=PUT path=/my-bucket pattern=/{bucket} status=200 duration=30µs request_id=...
```
- `target` -- 出现于 JSON/Query 协议服务
- `action` -- 出现于 Query 协议服务，当 Action 位于 URL 查询字符串中时
### DEBUG
除了 INFO 输出外，还会记录完整的请求体：
```
level=DEBUG msg="request body" request_id=... body={"Name":"my-secret","SecretString":"..."}
```
启用方式：
```
KUMO_LOG_LEVEL=debug ./bin/kumo
```
## 数据持久化
默认情况下，kumo 作为纯内存模拟器运行 —— 进程停止时所有数据都会丢失。这对于每次测试运行都从干净状态开始的 CI/CD 流水线来说是理想的选择。
对于本地开发，请设置 `KUMO_DATA_DIR` 以启用持久化存储：
```
KUMO_DATA_DIR=./data ./bin/kumo
```
启用后：
- 启动时，每个服务会从 `$KUMO_DATA_DIR/{service}.json` 加载其之前的状态。
- 正常关闭时，每个服务会保存其当前状态。
- 如果数据目录不存在，则会自动创建。
- 写入操作是原子的（临时文件 + 重命名），以防止崩溃导致损坏。
- 临时状态（SQS 传输中的消息、S3 分段上传）不会被持久化。
```
$KUMO_DATA_DIR/
s3.json
sqs.json
dynamodb.json
iam.json
...
```
## 开发
```
# Run tests
make test
# Run integration tests
make test-integration
# Lint
make lint
# Build
make build
```
## 贡献
欢迎贡献！请查看 issues 以了解计划的功能和改进。
## 许可证
MIT License
标签：AWS, AWS SDK, Docker, DPI, DynamoDB, EC2, ElastiCache, EVTX分析, Go, Golang, IaC, Lambda, LangChain, LNA, Ruby工具, S3, Serverless, 云服务, 单元测试, 存储, 安全编程, 安全防御评估, 容器, 开发环境, 开源, 开源框架, 持续集成, 无服务器, 日志审计, 本地开发, 模拟, 测试工具, 计算, 请求拦截, 轻量级, 集成测试