Tweak object store parameters to not mention MinIO

This commit is contained in:
Cyber MacGeddon 2025-12-23 16:05:41 +00:00
parent 5304f96fe6
commit cb945ec900
5 changed files with 279 additions and 41 deletions

View file

@ -1,8 +1,8 @@
# TrustGraph Librarian API
This API provides document library management for TrustGraph. It handles document storage,
metadata management, and processing orchestration using hybrid storage (MinIO for content,
Cassandra for metadata) with multi-user support.
This API provides document library management for TrustGraph. It handles document storage,
metadata management, and processing orchestration using hybrid storage (S3-compatible object
storage for content, Cassandra for metadata) with multi-user support.
## Request/response
@ -374,13 +374,14 @@ await client.add_processing(
## Features
- **Hybrid Storage**: MinIO for content, Cassandra for metadata
- **Hybrid Storage**: S3-compatible object storage (MinIO, Ceph RGW, AWS S3, etc.) for content, Cassandra for metadata
- **Multi-user Support**: User-based document ownership and access control
- **Rich Metadata**: RDF-style metadata triples and tagging system
- **Processing Integration**: Automatic triggering of document processing workflows
- **Content Types**: Support for multiple document formats (PDF, text, etc.)
- **Collection Management**: Optional document grouping by collection
- **Metadata Search**: Query documents by metadata criteria
- **Flexible Storage Backend**: Works with any S3-compatible storage (MinIO, Ceph RADOS Gateway, AWS S3, Cloudflare R2, etc.)
## Use Cases

View file

@ -0,0 +1,237 @@
# Tech Spec: S3-Compatible Storage Backend Support
## Overview
The Librarian service uses S3-compatible object storage for document blob storage. This spec documents the implementation that enables support for any S3-compatible backend including MinIO, Ceph RADOS Gateway (RGW), AWS S3, Cloudflare R2, DigitalOcean Spaces, and others.
## Architecture
### Storage Components
- **Blob Storage**: S3-compatible object storage via `minio` Python client library
- **Metadata Storage**: Cassandra (stores object_id mapping and document metadata)
- **Affected Component**: Librarian service only
- **Storage Pattern**: Hybrid storage with metadata in Cassandra, content in S3-compatible storage
### Implementation
- **Library**: `minio` Python client (supports any S3-compatible API)
- **Location**: `trustgraph-flow/trustgraph/librarian/blob_store.py`
- **Operations**:
- `add()` - Store blob with UUID object_id
- `get()` - Retrieve blob by object_id
- `remove()` - Delete blob by object_id
- `ensure_bucket()` - Create bucket if not exists
- **Bucket**: `library`
- **Object Path**: `doc/{object_id}`
- **Supported MIME Types**: `text/plain`, `application/pdf`
### Key Files
1. `trustgraph-flow/trustgraph/librarian/blob_store.py` - BlobStore implementation
2. `trustgraph-flow/trustgraph/librarian/librarian.py` - BlobStore initialization
3. `trustgraph-flow/trustgraph/librarian/service.py` - Service configuration
4. `trustgraph-flow/pyproject.toml` - Dependencies (`minio` package)
5. `docs/apis/api-librarian.md` - API documentation
## Supported Storage Backends
The implementation works with any S3-compatible object storage system:
### Tested/Supported
- **MinIO** - Lightweight self-hosted object storage (default configuration)
- **Ceph RADOS Gateway (RGW)** - Distributed storage system with S3 API
### Should Work (S3-Compatible)
- **AWS S3** - Amazon's cloud object storage
- **Cloudflare R2** - Cloudflare's S3-compatible storage
- **DigitalOcean Spaces** - DigitalOcean's object storage
- **Wasabi** - S3-compatible cloud storage
- **Backblaze B2** - S3-compatible backup storage
- Any other service implementing the S3 REST API
## Configuration
### CLI Arguments
```bash
librarian \
--s3-endpoint <hostname:port> \
--s3-access-key <access_key> \
--s3-secret-key <secret_key>
```
### Environment Variables (Alternative)
```bash
S3_ENDPOINT=<hostname:port>
S3_ACCESS_KEY=<access_key>
S3_SECRET_KEY=<secret_key>
```
### Examples
**MinIO (default):**
```bash
--s3-endpoint minio:9000 \
--s3-access-key minioadmin \
--s3-secret-key minioadmin
```
**Ceph RADOS Gateway:**
```bash
--s3-endpoint ceph-rgw.example.com:8080 \
--s3-access-key user123 \
--s3-secret-key abcd1234secret
```
**AWS S3:**
```bash
--s3-endpoint s3.amazonaws.com \
--s3-access-key AKIAIOSFODNN7EXAMPLE \
--s3-secret-key wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
```
## Authentication
All S3-compatible backends require AWS Signature Version 4 (or v2) authentication:
- **Access Key** - Public identifier (like username)
- **Secret Key** - Private signing key (like password)
The MinIO Python client handles all signature calculation automatically.
### Creating Credentials
**For MinIO:**
```bash
# Use default credentials or create user via MinIO Console
minioadmin / minioadmin
```
**For Ceph RGW:**
```bash
radosgw-admin user create --uid="trustgraph" --display-name="TrustGraph Service"
# Returns access_key and secret_key
```
**For AWS S3:**
- Create IAM user with S3 permissions
- Generate access key in AWS Console
## Library Selection: MinIO Python Client
**Rationale:**
- Lightweight (~500KB vs boto3's ~50MB)
- S3-compatible - works with any S3 API endpoint
- Simpler API than boto3 for basic operations
- Already in use, no migration needed
- Battle-tested with MinIO and other S3 systems
## BlobStore Implementation
**Location:** `trustgraph-flow/trustgraph/librarian/blob_store.py`
```python
from minio import Minio
import io
import logging
logger = logging.getLogger(__name__)
class BlobStore:
"""
S3-compatible blob storage for document content.
Supports MinIO, Ceph RGW, AWS S3, and other S3-compatible backends.
"""
def __init__(self, endpoint, access_key, secret_key, bucket_name):
"""
Initialize S3-compatible blob storage.
Args:
endpoint: S3 endpoint (e.g., "minio:9000", "ceph-rgw:8080")
access_key: S3 access key
secret_key: S3 secret key
bucket_name: Bucket name for storage
"""
self.client = Minio(
endpoint=endpoint,
access_key=access_key,
secret_key=secret_key,
secure=False, # Set True for HTTPS
)
self.bucket_name = bucket_name
logger.info(f"Connected to S3-compatible storage at {endpoint}")
self.ensure_bucket()
def ensure_bucket(self):
"""Create bucket if it doesn't exist"""
found = self.client.bucket_exists(bucket_name=self.bucket_name)
if not found:
self.client.make_bucket(bucket_name=self.bucket_name)
logger.info(f"Created bucket {self.bucket_name}")
else:
logger.debug(f"Bucket {self.bucket_name} already exists")
async def add(self, object_id, blob, kind):
"""Store blob in S3-compatible storage"""
self.client.put_object(
bucket_name=self.bucket_name,
object_name=f"doc/{object_id}",
length=len(blob),
data=io.BytesIO(blob),
content_type=kind,
)
logger.debug("Add blob complete")
async def remove(self, object_id):
"""Delete blob from S3-compatible storage"""
self.client.remove_object(
bucket_name=self.bucket_name,
object_name=f"doc/{object_id}",
)
logger.debug("Remove blob complete")
async def get(self, object_id):
"""Retrieve blob from S3-compatible storage"""
resp = self.client.get_object(
bucket_name=self.bucket_name,
object_name=f"doc/{object_id}",
)
return resp.read()
```
## Key Benefits
1. **No Vendor Lock-in** - Works with any S3-compatible storage
2. **Lightweight** - MinIO client is only ~500KB
3. **Simple Configuration** - Just endpoint + credentials
4. **No Data Migration** - Drop-in replacement between backends
5. **Battle-Tested** - MinIO client works with all major S3 implementations
## Implementation Status
All code has been updated to use generic S3 parameter names:
- ✅ `blob_store.py` - Updated to accept `endpoint`, `access_key`, `secret_key`
- ✅ `librarian.py` - Updated parameter names
- ✅ `service.py` - Updated CLI arguments and configuration
- ✅ Documentation updated
## Future Enhancements
1. **SSL/TLS Support** - Add `--s3-use-ssl` flag for HTTPS
2. **Retry Logic** - Implement exponential backoff for transient failures
3. **Presigned URLs** - Generate temporary upload/download URLs
4. **Multi-region Support** - Replicate blobs across regions
5. **CDN Integration** - Serve blobs via CDN
6. **Storage Classes** - Use S3 storage classes for cost optimization
7. **Lifecycle Policies** - Automatic archival/deletion
8. **Versioning** - Store multiple versions of blobs
## References
- MinIO Python Client: https://min.io/docs/minio/linux/developers/python/API.html
- Ceph RGW S3 API: https://docs.ceph.com/en/latest/radosgw/s3/
- S3 API Reference: https://docs.aws.amazon.com/AmazonS3/latest/API/Welcome.html

View file

@ -14,29 +14,29 @@ class BlobStore:
def __init__(
self,
minio_host, minio_access_key, minio_secret_key, bucket_name,
endpoint, access_key, secret_key, bucket_name,
):
self.minio = Minio(
endpoint = minio_host,
access_key = minio_access_key,
secret_key = minio_secret_key,
self.client = Minio(
endpoint = endpoint,
access_key = access_key,
secret_key = secret_key,
secure = False,
)
self.bucket_name = bucket_name
logger.info("Connected to MinIO")
logger.info(f"Connected to S3-compatible storage at {endpoint}")
self.ensure_bucket()
def ensure_bucket(self):
# Make the bucket if it doesn't exist.
found = self.minio.bucket_exists(bucket_name=self.bucket_name)
found = self.client.bucket_exists(bucket_name=self.bucket_name)
if not found:
self.minio.make_bucket(bucket_name=self.bucket_name)
self.client.make_bucket(bucket_name=self.bucket_name)
logger.info(f"Created bucket {self.bucket_name}")
else:
logger.debug(f"Bucket {self.bucket_name} already exists")
@ -44,7 +44,7 @@ class BlobStore:
async def add(self, object_id, blob, kind):
# FIXME: Loop retry
self.minio.put_object(
self.client.put_object(
bucket_name = self.bucket_name,
object_name = "doc/" + str(object_id),
length = len(blob),
@ -57,7 +57,7 @@ class BlobStore:
async def remove(self, object_id):
# FIXME: Loop retry
self.minio.remove_object(
self.client.remove_object(
bucket_name = self.bucket_name,
object_name = "doc/" + str(object_id),
)
@ -68,7 +68,7 @@ class BlobStore:
async def get(self, object_id):
# FIXME: Loop retry
resp = self.minio.get_object(
resp = self.client.get_object(
bucket_name = self.bucket_name,
object_name = "doc/" + str(object_id),
)

View file

@ -17,12 +17,12 @@ class Librarian:
def __init__(
self,
cassandra_host, cassandra_username, cassandra_password,
minio_host, minio_access_key, minio_secret_key,
s3_endpoint, s3_access_key, s3_secret_key,
bucket_name, keyspace, load_document,
):
self.blob_store = BlobStore(
minio_host, minio_access_key, minio_secret_key, bucket_name
s3_endpoint, s3_access_key, s3_secret_key, bucket_name
)
self.table_store = LibraryTableStore(

View file

@ -41,9 +41,9 @@ default_collection_response_queue = collection_response_queue
default_config_request_queue = config_request_queue
default_config_response_queue = config_response_queue
default_minio_host = "minio:9000"
default_minio_access_key = "minioadmin"
default_minio_secret_key = "minioadmin"
default_s3_endpoint = "minio:9000"
default_s3_access_key = "minioadmin"
default_s3_secret_key = "minioadmin"
default_cassandra_host = "cassandra"
bucket_name = "library"
@ -80,14 +80,14 @@ class Processor(AsyncProcessor):
"config_response_queue", default_config_response_queue
)
minio_host = params.get("minio_host", default_minio_host)
minio_access_key = params.get(
"minio_access_key",
default_minio_access_key
s3_endpoint = params.get("s3_endpoint", default_s3_endpoint)
s3_access_key = params.get(
"s3_access_key",
default_s3_access_key
)
minio_secret_key = params.get(
"minio_secret_key",
default_minio_secret_key
s3_secret_key = params.get(
"s3_secret_key",
default_s3_secret_key
)
cassandra_host = params.get("cassandra_host")
@ -113,8 +113,8 @@ class Processor(AsyncProcessor):
"librarian_response_queue": librarian_response_queue,
"collection_request_queue": collection_request_queue,
"collection_response_queue": collection_response_queue,
"minio_host": minio_host,
"minio_access_key": minio_access_key,
"s3_endpoint": s3_endpoint,
"s3_access_key": s3_access_key,
"cassandra_host": self.cassandra_host,
"cassandra_username": self.cassandra_username,
"cassandra_password": self.cassandra_password,
@ -208,9 +208,9 @@ class Processor(AsyncProcessor):
cassandra_host = self.cassandra_host,
cassandra_username = self.cassandra_username,
cassandra_password = self.cassandra_password,
minio_host = minio_host,
minio_access_key = minio_access_key,
minio_secret_key = minio_secret_key,
s3_endpoint = s3_endpoint,
s3_access_key = s3_access_key,
s3_secret_key = s3_secret_key,
bucket_name = bucket_name,
keyspace = keyspace,
load_document = self.load_document,
@ -494,23 +494,23 @@ class Processor(AsyncProcessor):
)
parser.add_argument(
'--minio-host',
default=default_minio_host,
help=f'Minio hostname (default: {default_minio_host})',
'--s3-endpoint',
default=default_s3_endpoint,
help=f'S3-compatible storage endpoint (default: {default_s3_endpoint})',
)
parser.add_argument(
'--minio-access-key',
'--s3-access-key',
default='minioadmin',
help='Minio access key / username '
f'(default: {default_minio_access_key})',
help='S3 access key / username '
f'(default: {default_s3_access_key})',
)
parser.add_argument(
'--minio-secret-key',
'--s3-secret-key',
default='minioadmin',
help='Minio secret key / password '
f'(default: {default_minio_access_key})',
help='S3 secret key / password '
f'(default: {default_s3_secret_key})',
)
add_cassandra_args(parser)