mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-25 11:44:16 +00:00
clean up documentation
This commit is contained in:
Evan Jarrett
@@ -1,434 +0,0 @@
|
||||
# AppView-Mediated OAuth Architecture
|
||||
|
||||
## Overview
|
||||
|
||||
ATCR uses a two-tier authentication model to support OAuth while allowing the AppView to write manifests to users' Personal Data Servers (PDS).
|
||||
|
||||
## The Problem
|
||||
|
||||
OAuth with DPoP creates cryptographically bound tokens that cannot be delegated:
|
||||
|
||||
- **Basic Auth**: App password is a shared secret that can be forwarded from client → AppView → PDS ✅
|
||||
- **OAuth + DPoP**: Token is bound to client's keypair and cannot be reused by AppView ❌
|
||||
|
||||
This creates a challenge: How can the AppView write manifests to the user's PDS on their behalf?
|
||||
|
||||
## The Solution: Two-Tier Authentication
|
||||
|
||||
```
|
||||
┌──────────┐ ┌─────────┐ ┌────────────┐
|
||||
│ Docker │◄───────►│ AppView │◄───────►│ PDS/Auth │
|
||||
│ Client │ Auth1 │ (ATCR) │ Auth2 │ Server │
|
||||
└──────────┘ └─────────┘ └────────────┘
|
||||
```
|
||||
|
||||
**Auth Tier 1** (Docker ↔ AppView): Registry authentication
|
||||
- Client authenticates to AppView using session tokens
|
||||
- AppView issues short-lived registry JWTs
|
||||
- Standard Docker registry auth protocol
|
||||
|
||||
**Auth Tier 2** (AppView ↔ PDS): Resource access
|
||||
- AppView acts as OAuth client for each user
|
||||
- AppView stores refresh tokens per user
|
||||
- AppView gets access tokens on-demand to write manifests
|
||||
|
||||
## Complete Flows
|
||||
|
||||
### One-Time Authorization Flow
|
||||
|
||||
```
|
||||
┌────────┐ ┌──────────────┐ ┌─────────┐ ┌─────┐
|
||||
│ User │ │ Credential │ │ AppView │ │ PDS │
|
||||
│ │ │ Helper │ │ │ │ │
|
||||
└───┬────┘ └──────┬───────┘ └────┬────┘ └──┬──┘
|
||||
│ │ │ │
|
||||
│ $ docker-credential-atcr configure │ │
|
||||
│ Enter handle: evan.jarrett.net │ │
|
||||
│─────────────────────>│ │ │
|
||||
│ │ │ │
|
||||
│ │ GET /auth/oauth/authorize?handle=... │
|
||||
│ │─────────────────────>│ │
|
||||
│ │ │ │
|
||||
│ │ 302 Redirect to PDS │ │
|
||||
│ │<─────────────────────│ │
|
||||
│ │ │ │
|
||||
│ [Browser opens] │ │ │
|
||||
│<─────────────────────│ │ │
|
||||
│ │ │ │
|
||||
│ Authorize ATCR? │ │ │
|
||||
│──────────────────────────────────────────────────────────────>│
|
||||
│ │ │ │
|
||||
│ │ │<─code────────────│
|
||||
│ │ │ │
|
||||
│ │ │ POST /token │
|
||||
│ │ │ (exchange code) │
|
||||
│ │ │ + DPoP proof │
|
||||
│ │ │─────────────────>│
|
||||
│ │ │ │
|
||||
│ │ │<─refresh_token───│
|
||||
│ │ │ access_token │
|
||||
│ │ │ │
|
||||
│ │ │ [Store tokens] │
|
||||
│ │ │ DID → { │
|
||||
│ │ │ refresh_token, │
|
||||
│ │ │ dpop_key, │
|
||||
│ │ │ pds_endpoint │
|
||||
│ │ │ } │
|
||||
│ │ │ │
|
||||
│ │<─session_token───────│ │
|
||||
│ │ │ │
|
||||
│ [Store session] │ │ │
|
||||
│<─────────────────────│ │ │
|
||||
│ ~/.atcr/ │ │ │
|
||||
│ session.json │ │ │
|
||||
│ │ │ │
|
||||
│ ✓ Authorization │ │ │
|
||||
│ complete! │ │ │
|
||||
│ │ │ │
|
||||
```
|
||||
|
||||
### Docker Push Flow (Every Push)
|
||||
|
||||
```
|
||||
┌────────┐ ┌──────────┐ ┌─────────┐ ┌─────┐
|
||||
│ Docker │ │ Cred │ │ AppView │ │ PDS │
|
||||
│ │ │ Helper │ │ │ │ │
|
||||
└───┬────┘ └────┬─────┘ └────┬────┘ └──┬──┘
|
||||
│ │ │ │
|
||||
│ docker push │ │ │
|
||||
│──────────────>│ │ │
|
||||
│ │ │ │
|
||||
│ │ GET /auth/exchange │
|
||||
│ │ Authorization: Bearer │
|
||||
│ │ <session_token> │
|
||||
│ │──────────────>│ │
|
||||
│ │ │ │
|
||||
│ │ │ [Validate │
|
||||
│ │ │ session] │
|
||||
│ │ │ │
|
||||
│ │ │ [Issue JWT] │
|
||||
│ │ │ │
|
||||
│ │<──registry_jwt─│ │
|
||||
│ │ │ │
|
||||
│<─registry_jwt─│ │ │
|
||||
│ │ │ │
|
||||
│ PUT /v2/.../manifests/... │ │
|
||||
│ Authorization: Bearer │ │
|
||||
│ <registry_jwt> │ │
|
||||
│──────────────────────────────>│ │
|
||||
│ │ │
|
||||
│ │ [Validate │
|
||||
│ │ JWT] │
|
||||
│ │ │
|
||||
│ │ [Get fresh │
|
||||
│ │ access │
|
||||
│ │ token] │
|
||||
│ │ │
|
||||
│ │ POST /token │
|
||||
│ │ (refresh) │
|
||||
│ │ + DPoP │
|
||||
│ │────────────>│
|
||||
│ │ │
|
||||
│ │<access_token│
|
||||
│ │ │
|
||||
│ │ PUT record │
|
||||
│ │ (manifest) │
|
||||
│ │ + DPoP │
|
||||
│ │────────────>│
|
||||
│ │ │
|
||||
│ │<──201 OK────│
|
||||
│ │ │
|
||||
│<──────────201 OK──────────────│ │
|
||||
│ │ │
|
||||
```
|
||||
|
||||
## Components
|
||||
|
||||
### 1. OAuth Authorization Server (AppView)
|
||||
|
||||
**File**: `pkg/auth/oauth/server.go`
|
||||
|
||||
**Endpoints**:
|
||||
|
||||
#### `GET /auth/oauth/authorize`
|
||||
|
||||
Initiates OAuth flow for a user.
|
||||
|
||||
**Query Parameters**:
|
||||
- `handle` (required): User's ATProto handle (e.g., `evan.jarrett.net`)
|
||||
|
||||
**Flow**:
|
||||
1. Resolve handle → DID → PDS endpoint
|
||||
2. Discover PDS OAuth metadata
|
||||
3. Generate state + PKCE verifier
|
||||
4. Create PAR request to PDS
|
||||
5. Redirect user to PDS authorization endpoint
|
||||
|
||||
**Response**: `302 Redirect` to PDS authorization page
|
||||
|
||||
#### `GET /auth/oauth/callback`
|
||||
|
||||
Receives OAuth callback from PDS.
|
||||
|
||||
**Query Parameters**:
|
||||
- `code`: Authorization code
|
||||
- `state`: State for CSRF protection
|
||||
|
||||
**Flow**:
|
||||
1. Validate state
|
||||
2. Exchange code for tokens (POST to PDS token endpoint)
|
||||
3. Use AppView's DPoP key for the exchange
|
||||
4. Store refresh token + DPoP key for user's DID
|
||||
5. Generate AppView session token
|
||||
6. Redirect to success page with session token
|
||||
|
||||
**Response**: HTML page with session token (user copies to credential helper)
|
||||
|
||||
### 2. Refresh Token Storage
|
||||
|
||||
**File**: `pkg/auth/oauth/storage.go`
|
||||
|
||||
**Storage Format**:
|
||||
|
||||
```json
|
||||
{
|
||||
"refresh_tokens": {
|
||||
"did:plc:abc123": {
|
||||
"refresh_token": "...",
|
||||
"dpop_key_pem": "-----BEGIN EC PRIVATE KEY-----\n...",
|
||||
"pds_endpoint": "https://bsky.social",
|
||||
"handle": "evan.jarrett.net",
|
||||
"created_at": "2025-10-04T...",
|
||||
"last_refreshed": "2025-10-04T..."
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Location**:
|
||||
- Development: `~/.atcr/appview-tokens.json`
|
||||
- Production: Encrypted database or secret manager
|
||||
|
||||
**Security**:
|
||||
- File permissions: `0600` (owner read/write only)
|
||||
- Consider encrypting DPoP keys at rest
|
||||
- Rotate refresh tokens periodically
|
||||
|
||||
### 3. Token Refresher
|
||||
|
||||
**File**: `pkg/auth/oauth/refresher.go`
|
||||
|
||||
**Interface**:
|
||||
|
||||
```go
|
||||
type Refresher interface {
|
||||
// GetAccessToken gets a fresh access token for a DID
|
||||
// Returns cached token if still valid, otherwise refreshes
|
||||
GetAccessToken(ctx context.Context, did string) (token string, dpopKey *ecdsa.PrivateKey, err error)
|
||||
|
||||
// RefreshToken forces a token refresh
|
||||
RefreshToken(ctx context.Context, did string) error
|
||||
|
||||
// RevokeToken removes stored refresh token
|
||||
RevokeToken(did string) error
|
||||
}
|
||||
```
|
||||
|
||||
**Caching Strategy**:
|
||||
- Access tokens cached for 14 minutes (expire at 15min)
|
||||
- Refresh tokens stored persistently
|
||||
- Cache key: `did → {access_token, dpop_key, expires_at}`
|
||||
|
||||
### 4. Session Management
|
||||
|
||||
**File**: `pkg/auth/session/handler.go`
|
||||
|
||||
**Session Token Format**:
|
||||
```
|
||||
Base64(JSON({
|
||||
"did": "did:plc:abc123",
|
||||
"handle": "evan.jarrett.net",
|
||||
"issued_at": "2025-10-04T...",
|
||||
"expires_at": "2025-11-03T..." // 30 days
|
||||
})).HMAC-SHA256(secret)
|
||||
```
|
||||
|
||||
**Storage**: Stateless (validated by HMAC signature)
|
||||
|
||||
**Endpoints**:
|
||||
|
||||
#### `GET /auth/session/validate`
|
||||
|
||||
Validates a session token.
|
||||
|
||||
**Headers**:
|
||||
- `Authorization: Bearer <session_token>`
|
||||
|
||||
**Response**:
|
||||
```json
|
||||
{
|
||||
"did": "did:plc:abc123",
|
||||
"handle": "evan.jarrett.net",
|
||||
"valid": true
|
||||
}
|
||||
```
|
||||
|
||||
### 5. Updated Exchange Handler
|
||||
|
||||
**File**: `pkg/auth/exchange/handler.go`
|
||||
|
||||
**Changes**:
|
||||
- Accept session token instead of OAuth token
|
||||
- Validate session token → extract DID
|
||||
- Issue registry JWT with DID
|
||||
- Remove PDS token validation
|
||||
|
||||
**Request**:
|
||||
```
|
||||
POST /auth/exchange
|
||||
Authorization: Bearer <session_token>
|
||||
|
||||
{
|
||||
"scope": ["repository:*:pull,push"]
|
||||
}
|
||||
```
|
||||
|
||||
**Response**:
|
||||
```json
|
||||
{
|
||||
"token": "<registry-jwt>",
|
||||
"expires_in": 900
|
||||
}
|
||||
```
|
||||
|
||||
### 6. Credential Helper Updates
|
||||
|
||||
**File**: `cmd/credential-helper/main.go`
|
||||
|
||||
**Changes**:
|
||||
|
||||
1. **Configure command**:
|
||||
- Open browser to AppView: `http://127.0.0.1:5000/auth/oauth/authorize?handle=...`
|
||||
- User authorizes on PDS
|
||||
- AppView displays session token
|
||||
- User copies session token to helper
|
||||
- Helper stores session token
|
||||
|
||||
2. **Get command**:
|
||||
- Load session token from `~/.atcr/session.json`
|
||||
- Call `/auth/exchange` with session token
|
||||
- Return registry JWT to Docker
|
||||
|
||||
3. **Storage format**:
|
||||
```json
|
||||
{
|
||||
"session_token": "...",
|
||||
"handle": "evan.jarrett.net",
|
||||
"appview_url": "http://127.0.0.1:5000"
|
||||
}
|
||||
```
|
||||
|
||||
**Removed**:
|
||||
- DPoP key generation
|
||||
- OAuth client logic
|
||||
- Refresh token handling
|
||||
|
||||
## Security Considerations
|
||||
|
||||
### AppView as Trusted Component
|
||||
|
||||
The AppView becomes a **trusted intermediary** that:
|
||||
- Stores refresh tokens for users
|
||||
- Acts on users' behalf to write manifests
|
||||
- Issues registry authentication tokens
|
||||
|
||||
**Trust model**:
|
||||
- Users must trust the AppView operator
|
||||
- Similar to trusting a Docker registry operator
|
||||
- AppView has write access to manifests (not profile data)
|
||||
|
||||
### Scope Limitations
|
||||
|
||||
AppView OAuth tokens are requested with minimal scopes:
|
||||
- `atproto` - Basic ATProto operations
|
||||
- Only needs: `com.atproto.repo.putRecord`, `com.atproto.repo.getRecord`
|
||||
- Does NOT need: profile updates, social graph access, etc.
|
||||
|
||||
### Token Security
|
||||
|
||||
**Refresh Tokens**:
|
||||
- Stored encrypted at rest
|
||||
- File permissions: 0600
|
||||
- Rotated periodically (when used)
|
||||
- Can be revoked by user on PDS
|
||||
|
||||
**Session Tokens**:
|
||||
- 30-day expiry
|
||||
- HMAC-signed (stateless validation)
|
||||
- Can be revoked by clearing storage
|
||||
|
||||
**Access Tokens**:
|
||||
- Cached in-memory only
|
||||
- 15-minute expiry
|
||||
- Never stored persistently
|
||||
|
||||
### Audit Trail
|
||||
|
||||
AppView should log:
|
||||
- OAuth authorizations (DID, timestamp)
|
||||
- Token refreshes (DID, timestamp)
|
||||
- Manifest writes (DID, repository, timestamp)
|
||||
|
||||
## Migration from Current OAuth
|
||||
|
||||
Users currently using `docker-credential-atcr` with direct PDS OAuth will need to:
|
||||
|
||||
1. Run `docker-credential-atcr configure` again
|
||||
2. Authorize AppView (new OAuth flow)
|
||||
3. Old PDS tokens are no longer used
|
||||
|
||||
## Alternative: Bring Your Own AppView
|
||||
|
||||
Users who don't trust a shared AppView can:
|
||||
1. Run their own ATCR AppView instance
|
||||
2. Configure credential helper to point at their AppView
|
||||
3. Their AppView stores their refresh tokens locally
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
### Multi-AppView Support
|
||||
|
||||
Allow users to configure multiple AppViews:
|
||||
```json
|
||||
{
|
||||
"appviews": {
|
||||
"default": "https://atcr.io",
|
||||
"personal": "http://localhost:5000"
|
||||
},
|
||||
"sessions": {
|
||||
"https://atcr.io": {"session_token": "...", "handle": "..."},
|
||||
"http://localhost:5000": {"session_token": "...", "handle": "..."}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Refresh Token Rotation
|
||||
|
||||
Implement automatic refresh token rotation per OAuth best practices:
|
||||
- PDS issues new refresh token with each use
|
||||
- AppView updates stored token
|
||||
- Old refresh token invalidated
|
||||
|
||||
### Revocation UI
|
||||
|
||||
Add web UI for users to:
|
||||
- View active sessions
|
||||
- Revoke AppView access
|
||||
- See audit log of manifest writes
|
||||
|
||||
## References
|
||||
|
||||
- [ATProto OAuth Specification](https://atproto.com/specs/oauth)
|
||||
- [RFC 6749: OAuth 2.0](https://datatracker.ietf.org/doc/html/rfc6749)
|
||||
- [RFC 9449: DPoP](https://datatracker.ietf.org/doc/html/rfc9449)
|
||||
- [Docker Credential Helpers](https://github.com/docker/docker-credential-helpers)
|
||||
@@ -1,460 +0,0 @@
|
||||
# Hold Service Multipart Upload Architecture
|
||||
|
||||
## Overview
|
||||
|
||||
The hold service supports multipart uploads through two modes:
|
||||
1. **S3Native** - Uses S3's native multipart API with presigned URLs (optimal)
|
||||
2. **Buffered** - Buffers parts in hold service memory, assembles on completion (fallback)
|
||||
|
||||
This dual-mode approach enables the hold service to work with:
|
||||
- S3-compatible storage with presigned URL support (S3, Storj, MinIO, etc.)
|
||||
- S3-compatible storage WITHOUT presigned URL support
|
||||
- Filesystem storage
|
||||
- Any storage driver supported by distribution
|
||||
|
||||
## Current State
|
||||
|
||||
### What Works ✅
|
||||
- **S3 Native Mode with presigned URLs**: Fully working! Direct uploads to S3 via presigned URLs
|
||||
- **Buffered mode with S3**: Tested and working with `DISABLE_PRESIGNED_URLS=true`
|
||||
- **Filesystem storage**: Tested and working! Buffered mode with filesystem driver
|
||||
- **AppView multipart client**: Implements chunked uploads via multipart API
|
||||
- **MultipartManager**: Session tracking, automatic cleanup, thread-safe operations
|
||||
- **Automatic fallback**: Falls back to buffered mode when S3 unavailable or disabled
|
||||
- **ETag normalization**: Handles quoted/unquoted ETags from S3
|
||||
- **Route handler**: `/multipart-parts/{uploadID}/{partNumber}` endpoint added and tested
|
||||
|
||||
### All Implementation Complete! 🎉
|
||||
All three multipart upload modes are fully implemented, tested, and working in production.
|
||||
|
||||
### Bugs Fixed 🔧
|
||||
- **Missing S3 parts in complete**: For S3Native mode, parts uploaded directly to S3 weren't being recorded. Fixed by storing parts from request in `HandleCompleteMultipart` before calling `CompleteMultipartUploadWithManager`.
|
||||
- **Malformed XML error from S3**: S3 requires ETags to be quoted in CompleteMultipartUpload XML. Added `normalizeETag()` function to ensure quotes are present.
|
||||
- **Route missing**: `/multipart-parts/{uploadID}/{partNumber}` not registered in cmd/hold/main.go. Fixed by adding route handler with path parsing.
|
||||
- **MultipartMgr access**: Field was private, preventing route handler access. Fixed by exporting as `MultipartMgr`.
|
||||
- **DISABLE_PRESIGNED_URLS not logged**: `initS3Client()` didn't check the flag before initializing. Fixed with early return check and proper logging.
|
||||
|
||||
## Architecture
|
||||
|
||||
### Three Modes of Operation
|
||||
|
||||
#### Mode 1: S3 Native Multipart ✅ WORKING
|
||||
```
|
||||
Docker → AppView → Hold → S3 (presigned URLs)
|
||||
↓
|
||||
Returns presigned URL
|
||||
↓
|
||||
Docker ──────────→ S3 (direct upload)
|
||||
```
|
||||
|
||||
**Flow:**
|
||||
1. AppView: `POST /start-multipart` → Hold starts S3 multipart, returns uploadID
|
||||
2. AppView: `POST /part-presigned-url` → Hold returns S3 presigned URL
|
||||
3. Docker → S3: Direct upload via presigned URL
|
||||
4. AppView: `POST /complete-multipart` → Hold calls S3 CompleteMultipartUpload
|
||||
|
||||
**Advantages:**
|
||||
- No data flows through hold service
|
||||
- Minimal bandwidth usage
|
||||
- Fast uploads
|
||||
|
||||
#### Mode 2: S3 Proxy Mode (Buffered) ✅ WORKING
|
||||
```
|
||||
Docker → AppView → Hold → S3 (via driver)
|
||||
↓
|
||||
Buffers & proxies
|
||||
↓
|
||||
S3
|
||||
```
|
||||
|
||||
**Flow:**
|
||||
1. AppView: `POST /start-multipart` → Hold creates buffered session
|
||||
2. AppView: `POST /part-presigned-url` → Hold returns proxy URL
|
||||
3. Docker → Hold: `PUT /multipart-parts/{uploadID}/{part}` → Hold buffers
|
||||
4. AppView: `POST /complete-multipart` → Hold uploads to S3 via driver
|
||||
|
||||
**Use Cases:**
|
||||
- S3 provider doesn't support presigned URLs
|
||||
- S3 API fails to generate presigned URL
|
||||
- Fallback from Mode 1
|
||||
|
||||
#### Mode 3: Filesystem Mode ✅ WORKING
|
||||
```
|
||||
Docker → AppView → Hold (filesystem driver)
|
||||
↓
|
||||
Buffers & writes
|
||||
↓
|
||||
Local filesystem
|
||||
```
|
||||
|
||||
**Flow:**
|
||||
Same as Mode 2, but writes to filesystem driver instead of S3 driver.
|
||||
|
||||
**Use Cases:**
|
||||
- Development/testing with local filesystem
|
||||
- Small deployments without S3
|
||||
- Air-gapped environments
|
||||
|
||||
## Implementation: pkg/hold/multipart.go
|
||||
|
||||
### Core Components
|
||||
|
||||
#### MultipartManager
|
||||
```go
|
||||
type MultipartManager struct {
|
||||
sessions map[string]*MultipartSession
|
||||
mu sync.RWMutex
|
||||
}
|
||||
```
|
||||
|
||||
**Responsibilities:**
|
||||
- Track active multipart sessions
|
||||
- Clean up abandoned uploads (>24h inactive)
|
||||
- Thread-safe session access
|
||||
|
||||
#### MultipartSession
|
||||
```go
|
||||
type MultipartSession struct {
|
||||
UploadID string // Unique ID for this upload
|
||||
Digest string // Target blob digest
|
||||
Mode MultipartMode // S3Native or Buffered
|
||||
S3UploadID string // S3 upload ID (S3Native only)
|
||||
Parts map[int]*MultipartPart // Buffered parts (Buffered only)
|
||||
CreatedAt time.Time
|
||||
LastActivity time.Time
|
||||
}
|
||||
```
|
||||
|
||||
**State Tracking:**
|
||||
- S3Native: Tracks S3 upload ID and part ETags
|
||||
- Buffered: Stores part data in memory
|
||||
|
||||
#### MultipartPart
|
||||
```go
|
||||
type MultipartPart struct {
|
||||
PartNumber int // Part number (1-indexed)
|
||||
Data []byte // Part data (Buffered mode only)
|
||||
ETag string // S3 ETag or computed hash
|
||||
Size int64
|
||||
}
|
||||
```
|
||||
|
||||
### Key Methods
|
||||
|
||||
#### StartMultipartUploadWithManager
|
||||
```go
|
||||
func (s *HoldService) StartMultipartUploadWithManager(
|
||||
ctx context.Context,
|
||||
digest string,
|
||||
manager *MultipartManager,
|
||||
) (string, MultipartMode, error)
|
||||
```
|
||||
|
||||
**Logic:**
|
||||
1. Try S3 native multipart via `s.startMultipartUpload()`
|
||||
2. If successful → Create S3Native session
|
||||
3. If fails or no S3 client → Create Buffered session
|
||||
4. Return uploadID and mode
|
||||
|
||||
#### GetPartUploadURL
|
||||
```go
|
||||
func (s *HoldService) GetPartUploadURL(
|
||||
ctx context.Context,
|
||||
session *MultipartSession,
|
||||
partNumber int,
|
||||
did string,
|
||||
) (string, error)
|
||||
```
|
||||
|
||||
**Logic:**
|
||||
- S3Native mode: Generate S3 presigned URL via `s.getPartPresignedURL()`
|
||||
- Buffered mode: Return proxy endpoint `/multipart-parts/{uploadID}/{part}`
|
||||
|
||||
#### CompleteMultipartUploadWithManager
|
||||
```go
|
||||
func (s *HoldService) CompleteMultipartUploadWithManager(
|
||||
ctx context.Context,
|
||||
session *MultipartSession,
|
||||
manager *MultipartManager,
|
||||
) error
|
||||
```
|
||||
|
||||
**Logic:**
|
||||
- S3Native: Call `s.completeMultipartUpload()` with S3 API
|
||||
- Buffered: Assemble parts in order, write via storage driver
|
||||
|
||||
#### HandleMultipartPartUpload (New Endpoint)
|
||||
```go
|
||||
func (s *HoldService) HandleMultipartPartUpload(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
uploadID string,
|
||||
partNumber int,
|
||||
did string,
|
||||
manager *MultipartManager,
|
||||
)
|
||||
```
|
||||
|
||||
**New HTTP endpoint:** `PUT /multipart-parts/{uploadID}/{partNumber}`
|
||||
|
||||
**Purpose:** Receive part uploads in Buffered mode
|
||||
|
||||
**Logic:**
|
||||
1. Validate session exists and is in Buffered mode
|
||||
2. Authorize write access
|
||||
3. Read part data from request body
|
||||
4. Store in session with computed ETag (SHA256)
|
||||
5. Return ETag in response header
|
||||
|
||||
## Integration Plan
|
||||
|
||||
### Phase 1: Migrate to pkg/hold (COMPLETE)
|
||||
- [x] Extract code from cmd/hold/main.go to pkg/hold/
|
||||
- [x] Create isolated multipart.go implementation
|
||||
- [x] Update cmd/hold/main.go to import pkg/hold
|
||||
- [x] Test existing functionality works
|
||||
|
||||
### Phase 2: Add Buffered Mode Support (COMPLETE ✅)
|
||||
- [x] Add MultipartManager to HoldService
|
||||
- [x] Update handlers to use `*WithManager` methods
|
||||
- [x] Add DISABLE_PRESIGNED_URLS environment variable for testing
|
||||
- [x] Implement presigned URL disable checks in all methods
|
||||
- [x] **Fixed: Record S3 parts from request in HandleCompleteMultipart**
|
||||
- [x] **Fixed: ETag normalization (add quotes for S3 XML)**
|
||||
- [x] **Test S3 native mode with presigned URLs** ✅ WORKING
|
||||
- [x] **Add route in cmd/hold/main.go** ✅ COMPLETE
|
||||
- [x] **Export MultipartMgr field for route handler access** ✅ COMPLETE
|
||||
- [x] **Test DISABLE_PRESIGNED_URLS=true with S3 storage** ✅ WORKING
|
||||
- [x] **Test filesystem storage with buffered multipart** ✅ WORKING
|
||||
|
||||
### Phase 3: Update AppView
|
||||
- [ ] Detect hold capabilities (presigned vs proxy)
|
||||
- [ ] Fallback to buffered mode when presigned fails
|
||||
- [ ] Handle `/multipart-parts/` proxy URLs
|
||||
|
||||
### Phase 4: Capability Discovery
|
||||
- [ ] Add capability endpoint: `GET /capabilities`
|
||||
- [ ] Return: `{"multipart": "native|buffered|both", "storage": "s3|filesystem"}`
|
||||
- [ ] AppView uses capabilities to choose upload strategy
|
||||
|
||||
## Testing Strategy
|
||||
|
||||
### Unit Tests
|
||||
- [ ] MultipartManager session lifecycle
|
||||
- [ ] Part buffering and assembly
|
||||
- [ ] Concurrent part uploads (thread safety)
|
||||
- [ ] Session cleanup (expired uploads)
|
||||
|
||||
### Integration Tests
|
||||
|
||||
**S3 Native Mode:**
|
||||
- [x] Start multipart → get presigned URLs → upload parts → complete ✅ WORKING
|
||||
- [x] Verify no data flows through hold service (only ~1KB API calls)
|
||||
- [ ] Test abort cleanup
|
||||
|
||||
**Buffered Mode (S3 with DISABLE_PRESIGNED_URLS):**
|
||||
- [x] Start multipart → get proxy URLs → upload parts → complete ✅ WORKING
|
||||
- [x] Verify parts assembled correctly
|
||||
- [ ] Test missing part detection
|
||||
- [ ] Test abort cleanup
|
||||
|
||||
**Buffered Mode (Filesystem):**
|
||||
- [x] Start multipart → get proxy URLs → upload parts → complete ✅ WORKING
|
||||
- [x] Verify parts assembled correctly ✅ WORKING
|
||||
- [x] Verify blobs written to filesystem ✅ WORKING
|
||||
- [ ] Test missing part detection
|
||||
- [ ] Test abort cleanup
|
||||
|
||||
### Load Tests
|
||||
- [ ] Concurrent multipart uploads (multiple sessions)
|
||||
- [ ] Large blobs (100MB+, many parts)
|
||||
- [ ] Memory usage with many buffered parts
|
||||
|
||||
## Performance Considerations
|
||||
|
||||
### Memory Usage (Buffered Mode)
|
||||
- Parts stored in memory until completion
|
||||
- Docker typically uses 5MB chunks (S3 minimum)
|
||||
- 100MB image = ~20 parts = ~100MB RAM during upload
|
||||
- Multiple concurrent uploads multiply memory usage
|
||||
|
||||
**Mitigation:**
|
||||
- Session cleanup (24h timeout)
|
||||
- Consider disk-backed buffering for large parts (future optimization)
|
||||
- Monitor memory usage and set limits
|
||||
|
||||
### Network Bandwidth
|
||||
- S3Native: Minimal (only API calls)
|
||||
- Buffered: Full blob data flows through hold service
|
||||
- Filesystem: Always buffered (no presigned URL option)
|
||||
|
||||
## Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
**Current (S3 only):**
|
||||
```bash
|
||||
STORAGE_DRIVER=s3
|
||||
S3_BUCKET=my-bucket
|
||||
S3_ENDPOINT=https://s3.amazonaws.com
|
||||
AWS_ACCESS_KEY_ID=...
|
||||
AWS_SECRET_ACCESS_KEY=...
|
||||
```
|
||||
|
||||
**Filesystem:**
|
||||
```bash
|
||||
STORAGE_DRIVER=filesystem
|
||||
STORAGE_ROOT_DIR=/var/lib/atcr/hold
|
||||
```
|
||||
|
||||
### Automatic Mode Selection
|
||||
No configuration needed - hold service automatically:
|
||||
1. Tries S3 native multipart if S3 client exists
|
||||
2. Falls back to buffered mode if S3 unavailable or fails
|
||||
3. Always uses buffered mode for filesystem driver
|
||||
|
||||
## Security Considerations
|
||||
|
||||
### Authorization
|
||||
- All multipart operations require write authorization
|
||||
- Buffered mode: Check auth on every part upload
|
||||
- S3Native: Auth only on start/complete (presigned URLs have embedded auth)
|
||||
|
||||
### Resource Limits
|
||||
- Max upload size: Controlled by storage backend
|
||||
- Max concurrent uploads: Limited by memory
|
||||
- Session timeout: 24 hours (configurable)
|
||||
|
||||
### Attack Vectors
|
||||
- **Memory exhaustion**: Attacker uploads many large parts
|
||||
- Mitigation: Session limits, cleanup, auth
|
||||
- **Incomplete uploads**: Attacker starts but never completes
|
||||
- Mitigation: 24h timeout, cleanup goroutine
|
||||
- **Part flooding**: Upload many tiny parts
|
||||
- Mitigation: S3 has 10,000 part limit, could add to buffered mode
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
### Disk-Backed Buffering
|
||||
Instead of memory, buffer parts to temporary disk location:
|
||||
- Reduces memory pressure
|
||||
- Supports larger uploads
|
||||
- Requires cleanup on completion/abort
|
||||
|
||||
### Parallel Part Assembly
|
||||
For large uploads, assemble parts in parallel:
|
||||
- Stream parts to writer as they arrive
|
||||
- Reduce memory footprint
|
||||
- Faster completion
|
||||
|
||||
### Chunked Completion
|
||||
For very large assembled blobs:
|
||||
- Stream to storage driver in chunks
|
||||
- Avoid loading entire blob in memory
|
||||
- Use `io.Copy()` with buffer
|
||||
|
||||
### Multi-Backend Support
|
||||
- Azure Blob Storage multipart
|
||||
- Google Cloud Storage resumable uploads
|
||||
- Backblaze B2 large file API
|
||||
|
||||
## Implementation Complete ✅
|
||||
|
||||
The buffered multipart mode is fully implemented with the following components:
|
||||
|
||||
**Route Handler** (`cmd/hold/main.go:47-73`):
|
||||
- Endpoint: `PUT /multipart-parts/{uploadID}/{partNumber}`
|
||||
- Parses URL path to extract uploadID and partNumber
|
||||
- Delegates to `service.HandleMultipartPartUpload()`
|
||||
|
||||
**Exported Manager** (`pkg/hold/service.go:20`):
|
||||
- Field `MultipartMgr` is now exported for route handler access
|
||||
- All handlers updated to use `s.MultipartMgr`
|
||||
|
||||
**Configuration Check** (`pkg/hold/s3.go:20-25`):
|
||||
- `initS3Client()` checks `DISABLE_PRESIGNED_URLS` flag before initializing
|
||||
- Logs clear message when presigned URLs are disabled
|
||||
- Prevents misleading "S3 presigned URLs enabled" message
|
||||
|
||||
## Testing Multipart Modes
|
||||
|
||||
### Test 1: S3 Native Mode (presigned URLs) ✅ TESTED
|
||||
```bash
|
||||
export STORAGE_DRIVER=s3
|
||||
export S3_BUCKET=your-bucket
|
||||
export AWS_ACCESS_KEY_ID=...
|
||||
export AWS_SECRET_ACCESS_KEY=...
|
||||
# Do NOT set DISABLE_PRESIGNED_URLS
|
||||
|
||||
# Start hold service
|
||||
./bin/atcr-hold
|
||||
|
||||
# Push an image
|
||||
docker push atcr.io/yourdid/test:latest
|
||||
|
||||
# Expected logs:
|
||||
# "✅ S3 presigned URLs enabled"
|
||||
# "Started S3 native multipart: uploadID=... s3UploadID=..."
|
||||
# "Completed multipart upload: digest=... uploadID=... parts=..."
|
||||
```
|
||||
|
||||
**Status**: ✅ Working - Direct uploads to S3, minimal bandwidth through hold service
|
||||
|
||||
### Test 2: Buffered Mode with S3 (forced proxy) ✅ TESTED
|
||||
```bash
|
||||
export STORAGE_DRIVER=s3
|
||||
export S3_BUCKET=your-bucket
|
||||
export AWS_ACCESS_KEY_ID=...
|
||||
export AWS_SECRET_ACCESS_KEY=...
|
||||
export DISABLE_PRESIGNED_URLS=true # Force buffered mode
|
||||
|
||||
# Start hold service
|
||||
./bin/atcr-hold
|
||||
|
||||
# Push an image
|
||||
docker push atcr.io/yourdid/test:latest
|
||||
|
||||
# Expected logs:
|
||||
# "⚠️ S3 presigned URLs DISABLED by config (DISABLE_PRESIGNED_URLS=true)"
|
||||
# "Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode"
|
||||
# "Stored part: uploadID=... part=1 size=..."
|
||||
# "Assembled buffered parts: uploadID=... parts=... totalSize=..."
|
||||
# "Completed buffered multipart: uploadID=... size=... written=..."
|
||||
```
|
||||
|
||||
**Status**: ✅ Working - Parts buffered in hold service memory, assembled and written to S3 via driver
|
||||
|
||||
### Test 3: Filesystem Mode (always buffered) ✅ TESTED
|
||||
```bash
|
||||
export STORAGE_DRIVER=filesystem
|
||||
export STORAGE_ROOT_DIR=/tmp/atcr-hold-test
|
||||
# DISABLE_PRESIGNED_URLS not needed (filesystem never has presigned URLs)
|
||||
|
||||
# Start hold service
|
||||
./bin/atcr-hold
|
||||
|
||||
# Push an image
|
||||
docker push atcr.io/yourdid/test:latest
|
||||
|
||||
# Expected logs:
|
||||
# "Storage driver is filesystem (not S3), presigned URLs disabled"
|
||||
# "Started buffered multipart: uploadID=..."
|
||||
# "Stored part: uploadID=... part=1 size=..."
|
||||
# "Assembled buffered parts: uploadID=... parts=... totalSize=..."
|
||||
# "Completed buffered multipart: uploadID=... size=... written=..."
|
||||
|
||||
# Verify blobs written to:
|
||||
ls -lh /var/lib/atcr/hold/docker/registry/v2/blobs/sha256/
|
||||
# Or from outside container:
|
||||
docker exec atcr-hold ls -lh /var/lib/atcr/hold/docker/registry/v2/blobs/sha256/
|
||||
```
|
||||
|
||||
**Status**: ✅ Working - Parts buffered in memory, assembled, and written to filesystem via driver
|
||||
|
||||
**Note**: Initial HEAD requests will show "Path not found" errors - this is normal! Docker checks if blobs exist before uploading. The errors occur for blobs that haven't been uploaded yet. After upload, subsequent HEAD checks succeed.
|
||||
|
||||
## References
|
||||
|
||||
- S3 Multipart Upload API: https://docs.aws.amazon.com/AmazonS3/latest/API/API_CreateMultipartUpload.html
|
||||
- Distribution Storage Driver Interface: https://github.com/distribution/distribution/blob/main/registry/storage/driver/storagedriver.go
|
||||
- OCI Distribution Spec (Blob Upload): https://github.com/opencontainers/distribution-spec/blob/main/spec.md#pushing-a-blob-in-chunks
|
||||
@@ -1,570 +0,0 @@
|
||||
S3 Multipart Upload Implementation Plan
|
||||
|
||||
Problem Summary
|
||||
|
||||
Current implementation uses a single presigned URL with a pipe for chunked uploads (PATCH). This causes:
|
||||
- Docker PATCH requests block waiting for pipe writes
|
||||
- S3 upload happens in background via single presigned URL
|
||||
- Docker times out → "client disconnected during blob PATCH"
|
||||
- Root cause: Single presigned URLs don't support OCI's chunked upload protocol
|
||||
|
||||
Solution: S3 Multipart Upload API
|
||||
|
||||
Implement proper S3 multipart upload to support Docker's chunked PATCH operations:
|
||||
- Each PATCH → separate S3 part upload with its own presigned URL
|
||||
- On Commit → complete multipart upload
|
||||
- No buffering, no pipes, no blocking
|
||||
|
||||
---
|
||||
Architecture Changes
|
||||
|
||||
Current (Broken) Flow
|
||||
|
||||
POST /blobs/uploads/ → Create() → Single presigned URL to temp location
|
||||
PATCH → Write to pipe → [blocks] → Background goroutine uploads via single URL
|
||||
PATCH → [blocks on pipe] → Docker timeout → disconnect ❌
|
||||
|
||||
New (Multipart) Flow
|
||||
|
||||
POST /blobs/uploads/ → Create() → Initiate multipart upload, get upload ID
|
||||
PATCH #1 → Get presigned URL for part 1 → Upload part 1 to S3 → Store ETag
|
||||
PATCH #2 → Get presigned URL for part 2 → Upload part 2 to S3 → Store ETag
|
||||
PUT (commit) → Complete multipart upload with ETags → Done ✅
|
||||
|
||||
---
|
||||
Implementation Details
|
||||
|
||||
1. Hold Service: Add Multipart Upload Endpoints
|
||||
|
||||
File: cmd/hold/main.go
|
||||
|
||||
New Request/Response Types
|
||||
|
||||
// StartMultipartUploadRequest initiates a multipart upload
|
||||
type StartMultipartUploadRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
|
||||
type StartMultipartUploadResponse struct {
|
||||
UploadID string `json:"upload_id"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
|
||||
// GetPartURLRequest requests a presigned URL for a specific part
|
||||
type GetPartURLRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
UploadID string `json:"upload_id"`
|
||||
PartNumber int `json:"part_number"`
|
||||
}
|
||||
|
||||
type GetPartURLResponse struct {
|
||||
URL string `json:"url"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
|
||||
// CompleteMultipartRequest completes a multipart upload
|
||||
type CompleteMultipartRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
UploadID string `json:"upload_id"`
|
||||
Parts []CompletedPart `json:"parts"`
|
||||
}
|
||||
|
||||
type CompletedPart struct {
|
||||
PartNumber int `json:"part_number"`
|
||||
ETag string `json:"etag"`
|
||||
}
|
||||
|
||||
// AbortMultipartRequest aborts an in-progress upload
|
||||
type AbortMultipartRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
UploadID string `json:"upload_id"`
|
||||
}
|
||||
|
||||
New Endpoints
|
||||
|
||||
POST /start-multipart
|
||||
func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Build S3 key from digest
|
||||
// Call s3.CreateMultipartUploadRequest()
|
||||
// Generate presigned URL if needed, or return upload ID
|
||||
// Return upload ID to client
|
||||
}
|
||||
|
||||
POST /part-presigned-url
|
||||
func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Build S3 key from digest
|
||||
// Call s3.UploadPartRequest() with part number and upload ID
|
||||
// Generate presigned URL
|
||||
// Return presigned URL for this specific part
|
||||
}
|
||||
|
||||
POST /complete-multipart
|
||||
func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Build S3 key from digest
|
||||
// Prepare CompletedPart array with part numbers and ETags
|
||||
// Call s3.CompleteMultipartUpload()
|
||||
// Return success
|
||||
}
|
||||
|
||||
POST /abort-multipart (for cleanup)
|
||||
func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Call s3.AbortMultipartUpload()
|
||||
// Return success
|
||||
}
|
||||
|
||||
S3 Implementation
|
||||
|
||||
// startMultipartUpload initiates a multipart upload and returns upload ID
|
||||
func (s *HoldService) startMultipartUpload(ctx context.Context, digest string) (string, error) {
|
||||
if s.s3Client == nil {
|
||||
return "", fmt.Errorf("S3 not configured")
|
||||
}
|
||||
|
||||
path := blobPath(digest)
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
if s.s3PathPrefix != "" {
|
||||
s3Key = s.s3PathPrefix + "/" + s3Key
|
||||
}
|
||||
|
||||
result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return *result.UploadId, nil
|
||||
}
|
||||
|
||||
// getPartPresignedURL generates presigned URL for a specific part
|
||||
func (s *HoldService) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) {
|
||||
if s.s3Client == nil {
|
||||
return "", fmt.Errorf("S3 not configured")
|
||||
}
|
||||
|
||||
path := blobPath(digest)
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
if s.s3PathPrefix != "" {
|
||||
s3Key = s.s3PathPrefix + "/" + s3Key
|
||||
}
|
||||
|
||||
req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
PartNumber: aws.Int64(int64(partNumber)),
|
||||
})
|
||||
|
||||
return req.Presign(15 * time.Minute)
|
||||
}
|
||||
|
||||
// completeMultipartUpload finalizes the multipart upload
|
||||
func (s *HoldService) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error {
|
||||
if s.s3Client == nil {
|
||||
return fmt.Errorf("S3 not configured")
|
||||
}
|
||||
|
||||
path := blobPath(digest)
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
if s.s3PathPrefix != "" {
|
||||
s3Key = s.s3PathPrefix + "/" + s3Key
|
||||
}
|
||||
|
||||
// Convert to S3 CompletedPart format
|
||||
s3Parts := make([]*s3.CompletedPart, len(parts))
|
||||
for i, p := range parts {
|
||||
s3Parts[i] = &s3.CompletedPart{
|
||||
PartNumber: aws.Int64(int64(p.PartNumber)),
|
||||
ETag: aws.String(p.ETag),
|
||||
}
|
||||
}
|
||||
|
||||
_, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
MultipartUpload: &s3.CompletedMultipartUpload{
|
||||
Parts: s3Parts,
|
||||
},
|
||||
})
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
---
|
||||
2. AppView: Rewrite ProxyBlobStore for Multipart
|
||||
|
||||
File: pkg/storage/proxy_blob_store.go
|
||||
|
||||
Remove Current Implementation
|
||||
|
||||
- Remove pipe-based streaming
|
||||
- Remove background goroutine with single presigned URL
|
||||
- Remove global upload tracking map
|
||||
|
||||
New ProxyBlobWriter Structure
|
||||
|
||||
type ProxyBlobWriter struct {
|
||||
store *ProxyBlobStore
|
||||
options distribution.CreateOptions
|
||||
uploadID string // S3 multipart upload ID
|
||||
parts []CompletedPart // Track uploaded parts with ETags
|
||||
partNumber int // Current part number (starts at 1)
|
||||
buffer *bytes.Buffer // Buffer for current part
|
||||
size int64 // Total bytes written
|
||||
closed bool
|
||||
id string // Distribution's upload ID (for state)
|
||||
startedAt time.Time
|
||||
finalDigest string // Set on Commit
|
||||
}
|
||||
|
||||
type CompletedPart struct {
|
||||
PartNumber int
|
||||
ETag string
|
||||
}
|
||||
|
||||
New Create() - Initiate Multipart Upload
|
||||
|
||||
func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
|
||||
var opts distribution.CreateOptions
|
||||
for _, option := range options {
|
||||
if err := option.Apply(&opts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// Use temp digest for upload location
|
||||
writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano())
|
||||
tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", writerID))
|
||||
|
||||
// Start multipart upload via hold service
|
||||
uploadID, err := p.startMultipartUpload(ctx, tempDigest)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to start multipart upload: %w", err)
|
||||
}
|
||||
|
||||
writer := &ProxyBlobWriter{
|
||||
store: p,
|
||||
options: opts,
|
||||
uploadID: uploadID,
|
||||
parts: make([]CompletedPart, 0),
|
||||
partNumber: 1,
|
||||
buffer: bytes.NewBuffer(make([]byte, 0, 5*1024*1024)), // 5MB buffer
|
||||
id: writerID,
|
||||
startedAt: time.Now(),
|
||||
}
|
||||
|
||||
// Store in global map for Resume()
|
||||
globalUploadsMu.Lock()
|
||||
globalUploads[writer.id] = writer
|
||||
globalUploadsMu.Unlock()
|
||||
|
||||
return writer, nil
|
||||
}
|
||||
|
||||
New Write() - Buffer and Flush Parts
|
||||
|
||||
func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
|
||||
if w.closed {
|
||||
return 0, fmt.Errorf("writer closed")
|
||||
}
|
||||
|
||||
n, err := w.buffer.Write(p)
|
||||
w.size += int64(n)
|
||||
|
||||
// Flush if buffer reaches 5MB (S3 minimum part size)
|
||||
if w.buffer.Len() >= 5*1024*1024 {
|
||||
if err := w.flushPart(); err != nil {
|
||||
return n, err
|
||||
}
|
||||
}
|
||||
|
||||
return n, err
|
||||
}
|
||||
|
||||
func (w *ProxyBlobWriter) flushPart() error {
|
||||
if w.buffer.Len() == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
// Get presigned URL for this part
|
||||
tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id))
|
||||
url, err := w.store.getPartPresignedURL(ctx, tempDigest, w.uploadID, w.partNumber)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get part presigned URL: %w", err)
|
||||
}
|
||||
|
||||
// Upload part to S3
|
||||
req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(w.buffer.Bytes()))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
resp, err := w.store.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
|
||||
return fmt.Errorf("part upload failed: status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
// Store ETag for completion
|
||||
etag := resp.Header.Get("ETag")
|
||||
if etag == "" {
|
||||
return fmt.Errorf("no ETag in response")
|
||||
}
|
||||
|
||||
w.parts = append(w.parts, CompletedPart{
|
||||
PartNumber: w.partNumber,
|
||||
ETag: etag,
|
||||
})
|
||||
|
||||
// Reset buffer and increment part number
|
||||
w.buffer.Reset()
|
||||
w.partNumber++
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
New Commit() - Complete Multipart and Move
|
||||
|
||||
func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descriptor) (distribution.Descriptor, error) {
|
||||
if w.closed {
|
||||
return distribution.Descriptor{}, fmt.Errorf("writer closed")
|
||||
}
|
||||
w.closed = true
|
||||
|
||||
// Flush any remaining buffered data
|
||||
if w.buffer.Len() > 0 {
|
||||
if err := w.flushPart(); err != nil {
|
||||
// Try to abort multipart on error
|
||||
w.store.abortMultipartUpload(ctx, w.uploadID)
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
}
|
||||
|
||||
// Complete multipart upload at temp location
|
||||
tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id))
|
||||
if err := w.store.completeMultipartUpload(ctx, tempDigest, w.uploadID, w.parts); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
// Move from temp → final location (server-side S3 copy)
|
||||
tempPath := fmt.Sprintf("uploads/temp-%s", w.id)
|
||||
finalPath := desc.Digest.String()
|
||||
|
||||
moveURL := fmt.Sprintf("%s/move?from=%s&to=%s&did=%s",
|
||||
w.store.storageEndpoint, tempPath, finalPath, w.store.did)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", moveURL, nil)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
|
||||
resp, err := w.store.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
|
||||
bodyBytes, _ := io.ReadAll(resp.Body)
|
||||
return distribution.Descriptor{}, fmt.Errorf("move failed: %d, %s", resp.StatusCode, bodyBytes)
|
||||
}
|
||||
|
||||
// Remove from global map
|
||||
globalUploadsMu.Lock()
|
||||
delete(globalUploads, w.id)
|
||||
globalUploadsMu.Unlock()
|
||||
|
||||
return distribution.Descriptor{
|
||||
Digest: desc.Digest,
|
||||
Size: w.size,
|
||||
MediaType: desc.MediaType,
|
||||
}, nil
|
||||
}
|
||||
|
||||
Add Hold Service Client Methods
|
||||
|
||||
func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, dgst digest.Digest) (string, error) {
|
||||
reqBody := map[string]any{
|
||||
"did": p.did,
|
||||
"digest": dgst.String(),
|
||||
}
|
||||
body, _ := json.Marshal(reqBody)
|
||||
|
||||
url := fmt.Sprintf("%s/start-multipart", p.storageEndpoint)
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
var result struct {
|
||||
UploadID string `json:"upload_id"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return result.UploadID, nil
|
||||
}
|
||||
|
||||
func (p *ProxyBlobStore) getPartPresignedURL(ctx context.Context, dgst digest.Digest, uploadID string, partNumber int) (string, error) {
|
||||
reqBody := map[string]any{
|
||||
"did": p.did,
|
||||
"digest": dgst.String(),
|
||||
"upload_id": uploadID,
|
||||
"part_number": partNumber,
|
||||
}
|
||||
body, _ := json.Marshal(reqBody)
|
||||
|
||||
url := fmt.Sprintf("%s/part-presigned-url", p.storageEndpoint)
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
var result struct {
|
||||
URL string `json:"url"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return result.URL, nil
|
||||
}
|
||||
|
||||
func (p *ProxyBlobStore) completeMultipartUpload(ctx context.Context, dgst digest.Digest, uploadID string, parts []CompletedPart) error {
|
||||
reqBody := map[string]any{
|
||||
"did": p.did,
|
||||
"digest": dgst.String(),
|
||||
"upload_id": uploadID,
|
||||
"parts": parts,
|
||||
}
|
||||
body, _ := json.Marshal(reqBody)
|
||||
|
||||
url := fmt.Sprintf("%s/complete-multipart", p.storageEndpoint)
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("complete multipart failed: status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
---
|
||||
Testing Plan
|
||||
|
||||
1. Unit Tests
|
||||
|
||||
- Test multipart upload initiation
|
||||
- Test part upload with presigned URLs
|
||||
- Test completion with ETags
|
||||
- Test abort on errors
|
||||
|
||||
2. Integration Tests
|
||||
|
||||
- Push small images (< 5MB, single part)
|
||||
- Push medium images (10MB, 2 parts)
|
||||
- Push large images (100MB, 20 parts)
|
||||
- Test with Upcloud S3
|
||||
- Test with Storj S3
|
||||
|
||||
3. Validation
|
||||
|
||||
- Monitor logs for "client disconnected" errors (should be gone)
|
||||
- Check Docker push success rate
|
||||
- Verify blobs stored correctly in S3
|
||||
- Check bandwidth usage on hold service (should be minimal)
|
||||
|
||||
---
|
||||
Migration & Deployment
|
||||
|
||||
Backward Compatibility
|
||||
|
||||
- Keep /put-presigned-url endpoint for fallback
|
||||
- Keep /move endpoint (still needed)
|
||||
- New multipart endpoints are additive
|
||||
|
||||
Deployment Steps
|
||||
|
||||
1. Update hold service with new endpoints
|
||||
2. Update AppView ProxyBlobStore
|
||||
3. Deploy hold service first
|
||||
4. Deploy AppView
|
||||
5. Test with sample push
|
||||
6. Monitor logs
|
||||
|
||||
Rollback Plan
|
||||
|
||||
- Revert AppView to previous version (uses old presigned URL method)
|
||||
- Hold service keeps both old and new endpoints
|
||||
|
||||
---
|
||||
Documentation Updates
|
||||
|
||||
Update docs/PRESIGNED_URLS.md
|
||||
|
||||
- Add section "Multipart Upload for Chunked Data"
|
||||
- Explain why single presigned URLs don't work with PATCH
|
||||
- Document new endpoints and flow
|
||||
- Add S3 part size recommendations (5MB-64MB for Storj)
|
||||
|
||||
Add Troubleshooting Section
|
||||
|
||||
- "Client disconnected during PATCH" → resolved by multipart
|
||||
- Storj-specific considerations (64MB parts recommended)
|
||||
- Upcloud compatibility notes
|
||||
|
||||
---
|
||||
Performance Impact
|
||||
|
||||
Before (Broken)
|
||||
|
||||
- Docker PATCH → blocks on pipe → timeout → retry → fail
|
||||
- Unable to push large images reliably
|
||||
|
||||
After (Multipart)
|
||||
|
||||
- Each PATCH → independent part upload → immediate response
|
||||
- No blocking, no timeouts
|
||||
- Parallel part uploads possible (future optimization)
|
||||
- Reliable pushes for any image size
|
||||
|
||||
Bandwidth
|
||||
|
||||
- Hold service: Only API calls (~1KB per part)
|
||||
- Direct S3 uploads: Full blob data
|
||||
- S3 copy for move: Server-side (no hold bandwidth)
|
||||
|
||||
Estimated savings: 99.98% hold service bandwidth reduction (same as before, but now actually works!)
|
||||
@@ -1,448 +0,0 @@
|
||||
S3 Multipart Upload Implementation Plan
|
||||
Problem Summary
|
||||
Current implementation uses a single presigned URL with a pipe for chunked uploads (PATCH). This causes:
|
||||
- Docker PATCH requests block waiting for pipe writes
|
||||
- S3 upload happens in background via single presigned URL
|
||||
- Docker times out → "client disconnected during blob PATCH"
|
||||
- Root cause: Single presigned URLs don't support OCI's chunked upload protocol
|
||||
Solution: S3 Multipart Upload API
|
||||
Implement proper S3 multipart upload to support Docker's chunked PATCH operations:
|
||||
- Each PATCH → separate S3 part upload with its own presigned URL
|
||||
- On Commit → complete multipart upload
|
||||
- No buffering, no pipes, no blocking
|
||||
---
|
||||
Architecture Changes
|
||||
Current (Broken) Flow
|
||||
POST /blobs/uploads/ → Create() → Single presigned URL to temp location
|
||||
PATCH → Write to pipe → [blocks] → Background goroutine uploads via single URL
|
||||
PATCH → [blocks on pipe] → Docker timeout → disconnect ❌
|
||||
New (Multipart) Flow
|
||||
POST /blobs/uploads/ → Create() → Initiate multipart upload, get upload ID
|
||||
PATCH #1 → Get presigned URL for part 1 → Upload part 1 to S3 → Store ETag
|
||||
PATCH #2 → Get presigned URL for part 2 → Upload part 2 to S3 → Store ETag
|
||||
PUT (commit) → Complete multipart upload with ETags → Done ✅
|
||||
---
|
||||
Implementation Details
|
||||
1. Hold Service: Add Multipart Upload Endpoints
|
||||
File: cmd/hold/main.go
|
||||
New Request/Response Types
|
||||
// StartMultipartUploadRequest initiates a multipart upload
|
||||
type StartMultipartUploadRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
type StartMultipartUploadResponse struct {
|
||||
UploadID string `json:"upload_id"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
// GetPartURLRequest requests a presigned URL for a specific part
|
||||
type GetPartURLRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
UploadID string `json:"upload_id"`
|
||||
PartNumber int `json:"part_number"`
|
||||
}
|
||||
type GetPartURLResponse struct {
|
||||
URL string `json:"url"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
// CompleteMultipartRequest completes a multipart upload
|
||||
type CompleteMultipartRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
UploadID string `json:"upload_id"`
|
||||
Parts []CompletedPart `json:"parts"`
|
||||
}
|
||||
type CompletedPart struct {
|
||||
PartNumber int `json:"part_number"`
|
||||
ETag string `json:"etag"`
|
||||
}
|
||||
// AbortMultipartRequest aborts an in-progress upload
|
||||
type AbortMultipartRequest struct {
|
||||
DID string `json:"did"`
|
||||
Digest string `json:"digest"`
|
||||
UploadID string `json:"upload_id"`
|
||||
}
|
||||
New Endpoints
|
||||
POST /start-multipart
|
||||
func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Build S3 key from digest
|
||||
// Call s3.CreateMultipartUploadRequest()
|
||||
// Generate presigned URL if needed, or return upload ID
|
||||
// Return upload ID to client
|
||||
}
|
||||
POST /part-presigned-url
|
||||
func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Build S3 key from digest
|
||||
// Call s3.UploadPartRequest() with part number and upload ID
|
||||
// Generate presigned URL
|
||||
// Return presigned URL for this specific part
|
||||
}
|
||||
POST /complete-multipart
|
||||
func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Build S3 key from digest
|
||||
// Prepare CompletedPart array with part numbers and ETags
|
||||
// Call s3.CompleteMultipartUpload()
|
||||
// Return success
|
||||
}
|
||||
POST /abort-multipart (for cleanup)
|
||||
func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
// Validate DID authorization for WRITE
|
||||
// Call s3.AbortMultipartUpload()
|
||||
// Return success
|
||||
}
|
||||
S3 Implementation
|
||||
// startMultipartUpload initiates a multipart upload and returns upload ID
|
||||
func (s *HoldService) startMultipartUpload(ctx context.Context, digest string) (string, error) {
|
||||
if s.s3Client == nil {
|
||||
return "", fmt.Errorf("S3 not configured")
|
||||
}
|
||||
path := blobPath(digest)
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
if s.s3PathPrefix != "" {
|
||||
s3Key = s.s3PathPrefix + "/" + s3Key
|
||||
}
|
||||
result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
})
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return *result.UploadId, nil
|
||||
}
|
||||
// getPartPresignedURL generates presigned URL for a specific part
|
||||
func (s *HoldService) getPartPresignedURL(ctx context.Context, digest, uploadID string, partNumber int) (string, error) {
|
||||
if s.s3Client == nil {
|
||||
return "", fmt.Errorf("S3 not configured")
|
||||
}
|
||||
path := blobPath(digest)
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
if s.s3PathPrefix != "" {
|
||||
s3Key = s.s3PathPrefix + "/" + s3Key
|
||||
}
|
||||
req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
PartNumber: aws.Int64(int64(partNumber)),
|
||||
})
|
||||
return req.Presign(15 * time.Minute)
|
||||
}
|
||||
// completeMultipartUpload finalizes the multipart upload
|
||||
func (s *HoldService) completeMultipartUpload(ctx context.Context, digest, uploadID string, parts []CompletedPart) error {
|
||||
if s.s3Client == nil {
|
||||
return fmt.Errorf("S3 not configured")
|
||||
}
|
||||
path := blobPath(digest)
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
if s.s3PathPrefix != "" {
|
||||
s3Key = s.s3PathPrefix + "/" + s3Key
|
||||
}
|
||||
// Convert to S3 CompletedPart format
|
||||
s3Parts := make([]*s3.CompletedPart, len(parts))
|
||||
for i, p := range parts {
|
||||
s3Parts[i] = &s3.CompletedPart{
|
||||
PartNumber: aws.Int64(int64(p.PartNumber)),
|
||||
ETag: aws.String(p.ETag),
|
||||
}
|
||||
}
|
||||
_, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
MultipartUpload: &s3.CompletedMultipartUpload{
|
||||
Parts: s3Parts,
|
||||
},
|
||||
})
|
||||
return err
|
||||
}
|
||||
---
|
||||
2. AppView: Rewrite ProxyBlobStore for Multipart
|
||||
File: pkg/storage/proxy_blob_store.go
|
||||
Remove Current Implementation
|
||||
- Remove pipe-based streaming
|
||||
- Remove background goroutine with single presigned URL
|
||||
- Remove global upload tracking map
|
||||
New ProxyBlobWriter Structure
|
||||
type ProxyBlobWriter struct {
|
||||
store *ProxyBlobStore
|
||||
options distribution.CreateOptions
|
||||
uploadID string // S3 multipart upload ID
|
||||
parts []CompletedPart // Track uploaded parts with ETags
|
||||
partNumber int // Current part number (starts at 1)
|
||||
buffer *bytes.Buffer // Buffer for current part
|
||||
size int64 // Total bytes written
|
||||
closed bool
|
||||
id string // Distribution's upload ID (for state)
|
||||
startedAt time.Time
|
||||
finalDigest string // Set on Commit
|
||||
}
|
||||
type CompletedPart struct {
|
||||
PartNumber int
|
||||
ETag string
|
||||
}
|
||||
New Create() - Initiate Multipart Upload
|
||||
func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) {
|
||||
var opts distribution.CreateOptions
|
||||
for _, option := range options {
|
||||
if err := option.Apply(&opts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
// Use temp digest for upload location
|
||||
writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano())
|
||||
tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", writerID))
|
||||
// Start multipart upload via hold service
|
||||
uploadID, err := p.startMultipartUpload(ctx, tempDigest)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to start multipart upload: %w", err)
|
||||
}
|
||||
writer := &ProxyBlobWriter{
|
||||
store: p,
|
||||
options: opts,
|
||||
uploadID: uploadID,
|
||||
parts: make([]CompletedPart, 0),
|
||||
partNumber: 1,
|
||||
buffer: bytes.NewBuffer(make([]byte, 0, 5*1024*1024)), // 5MB buffer
|
||||
id: writerID,
|
||||
startedAt: time.Now(),
|
||||
}
|
||||
// Store in global map for Resume()
|
||||
globalUploadsMu.Lock()
|
||||
globalUploads[writer.id] = writer
|
||||
globalUploadsMu.Unlock()
|
||||
return writer, nil
|
||||
}
|
||||
New Write() - Buffer and Flush Parts
|
||||
func (w *ProxyBlobWriter) Write(p []byte) (int, error) {
|
||||
if w.closed {
|
||||
return 0, fmt.Errorf("writer closed")
|
||||
}
|
||||
n, err := w.buffer.Write(p)
|
||||
w.size += int64(n)
|
||||
// Flush if buffer reaches 5MB (S3 minimum part size)
|
||||
if w.buffer.Len() >= 5*1024*1024 {
|
||||
if err := w.flushPart(); err != nil {
|
||||
return n, err
|
||||
}
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
func (w *ProxyBlobWriter) flushPart() error {
|
||||
if w.buffer.Len() == 0 {
|
||||
return nil
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
|
||||
defer cancel()
|
||||
// Get presigned URL for this part
|
||||
tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id))
|
||||
url, err := w.store.getPartPresignedURL(ctx, tempDigest, w.uploadID, w.partNumber)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get part presigned URL: %w", err)
|
||||
}
|
||||
// Upload part to S3
|
||||
req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(w.buffer.Bytes()))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := w.store.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
|
||||
return fmt.Errorf("part upload failed: status %d", resp.StatusCode)
|
||||
}
|
||||
// Store ETag for completion
|
||||
etag := resp.Header.Get("ETag")
|
||||
if etag == "" {
|
||||
return fmt.Errorf("no ETag in response")
|
||||
}
|
||||
w.parts = append(w.parts, CompletedPart{
|
||||
PartNumber: w.partNumber,
|
||||
ETag: etag,
|
||||
})
|
||||
// Reset buffer and increment part number
|
||||
w.buffer.Reset()
|
||||
w.partNumber++
|
||||
return nil
|
||||
}
|
||||
New Commit() - Complete Multipart and Move
|
||||
func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descriptor) (distribution.Descriptor, error) {
|
||||
if w.closed {
|
||||
return distribution.Descriptor{}, fmt.Errorf("writer closed")
|
||||
}
|
||||
w.closed = true
|
||||
// Flush any remaining buffered data
|
||||
if w.buffer.Len() > 0 {
|
||||
if err := w.flushPart(); err != nil {
|
||||
// Try to abort multipart on error
|
||||
w.store.abortMultipartUpload(ctx, w.uploadID)
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
}
|
||||
// Complete multipart upload at temp location
|
||||
tempDigest := digest.Digest(fmt.Sprintf("uploads/temp-%s", w.id))
|
||||
if err := w.store.completeMultipartUpload(ctx, tempDigest, w.uploadID, w.parts); err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
// Move from temp → final location (server-side S3 copy)
|
||||
tempPath := fmt.Sprintf("uploads/temp-%s", w.id)
|
||||
finalPath := desc.Digest.String()
|
||||
moveURL := fmt.Sprintf("%s/move?from=%s&to=%s&did=%s",
|
||||
w.store.storageEndpoint, tempPath, finalPath, w.store.did)
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", moveURL, nil)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
resp, err := w.store.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return distribution.Descriptor{}, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
|
||||
bodyBytes, _ := io.ReadAll(resp.Body)
|
||||
return distribution.Descriptor{}, fmt.Errorf("move failed: %d, %s", resp.StatusCode, bodyBytes)
|
||||
}
|
||||
// Remove from global map
|
||||
globalUploadsMu.Lock()
|
||||
delete(globalUploads, w.id)
|
||||
globalUploadsMu.Unlock()
|
||||
return distribution.Descriptor{
|
||||
Digest: desc.Digest,
|
||||
Size: w.size,
|
||||
MediaType: desc.MediaType,
|
||||
}, nil
|
||||
}
|
||||
Add Hold Service Client Methods
|
||||
func (p *ProxyBlobStore) startMultipartUpload(ctx context.Context, dgst digest.Digest) (string, error) {
|
||||
reqBody := map[string]any{
|
||||
"did": p.did,
|
||||
"digest": dgst.String(),
|
||||
}
|
||||
body, _ := json.Marshal(reqBody)
|
||||
url := fmt.Sprintf("%s/start-multipart", p.storageEndpoint)
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var result struct {
|
||||
UploadID string `json:"upload_id"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return result.UploadID, nil
|
||||
}
|
||||
func (p *ProxyBlobStore) getPartPresignedURL(ctx context.Context, dgst digest.Digest, uploadID string, partNumber int) (string, error) {
|
||||
reqBody := map[string]any{
|
||||
"did": p.did,
|
||||
"digest": dgst.String(),
|
||||
"upload_id": uploadID,
|
||||
"part_number": partNumber,
|
||||
}
|
||||
body, _ := json.Marshal(reqBody)
|
||||
url := fmt.Sprintf("%s/part-presigned-url", p.storageEndpoint)
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
var result struct {
|
||||
URL string `json:"url"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return result.URL, nil
|
||||
}
|
||||
func (p *ProxyBlobStore) completeMultipartUpload(ctx context.Context, dgst digest.Digest, uploadID string, parts []CompletedPart) error {
|
||||
reqBody := map[string]any{
|
||||
"did": p.did,
|
||||
"digest": dgst.String(),
|
||||
"upload_id": uploadID,
|
||||
"parts": parts,
|
||||
}
|
||||
body, _ := json.Marshal(reqBody)
|
||||
url := fmt.Sprintf("%s/complete-multipart", p.storageEndpoint)
|
||||
req, _ := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := p.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("complete multipart failed: status %d", resp.StatusCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
---
|
||||
Testing Plan
|
||||
1. Unit Tests
|
||||
- Test multipart upload initiation
|
||||
- Test part upload with presigned URLs
|
||||
- Test completion with ETags
|
||||
- Test abort on errors
|
||||
2. Integration Tests
|
||||
- Push small images (< 5MB, single part)
|
||||
- Push medium images (10MB, 2 parts)
|
||||
- Push large images (100MB, 20 parts)
|
||||
- Test with Upcloud S3
|
||||
- Test with Storj S3
|
||||
3. Validation
|
||||
- Monitor logs for "client disconnected" errors (should be gone)
|
||||
- Check Docker push success rate
|
||||
- Verify blobs stored correctly in S3
|
||||
- Check bandwidth usage on hold service (should be minimal)
|
||||
---
|
||||
Migration & Deployment
|
||||
Backward Compatibility
|
||||
- Keep /put-presigned-url endpoint for fallback
|
||||
- Keep /move endpoint (still needed)
|
||||
- New multipart endpoints are additive
|
||||
Deployment Steps
|
||||
1. Update hold service with new endpoints
|
||||
2. Update AppView ProxyBlobStore
|
||||
3. Deploy hold service first
|
||||
4. Deploy AppView
|
||||
5. Test with sample push
|
||||
6. Monitor logs
|
||||
Rollback Plan
|
||||
- Revert AppView to previous version (uses old presigned URL method)
|
||||
- Hold service keeps both old and new endpoints
|
||||
---
|
||||
Documentation Updates
|
||||
Update docs/PRESIGNED_URLS.md
|
||||
- Add section "Multipart Upload for Chunked Data"
|
||||
- Explain why single presigned URLs don't work with PATCH
|
||||
- Document new endpoints and flow
|
||||
- Add S3 part size recommendations (5MB-64MB for Storj)
|
||||
Add Troubleshooting Section
|
||||
- "Client disconnected during PATCH" → resolved by multipart
|
||||
- Storj-specific considerations (64MB parts recommended)
|
||||
- Upcloud compatibility notes
|
||||
---
|
||||
Performance Impact
|
||||
Before (Broken)
|
||||
- Docker PATCH → blocks on pipe → timeout → retry → fail
|
||||
- Unable to push large images reliably
|
||||
After (Multipart)
|
||||
- Each PATCH → independent part upload → immediate response
|
||||
- No blocking, no timeouts
|
||||
- Parallel part uploads possible (future optimization)
|
||||
- Reliable pushes for any image size
|
||||
Bandwidth
|
||||
- Hold service: Only API calls (~1KB per part)
|
||||
- Direct S3 uploads: Full blob data
|
||||
- S3 copy for move: Server-side (no hold bandwidth)
|
||||
Estimated savings: 99.98% hold service bandwidth reduction (same as before, but now actually works!)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,824 +0,0 @@
|
||||
# S3 Presigned URLs Implementation
|
||||
|
||||
## Overview
|
||||
|
||||
Currently, ATCR's hold service acts as a proxy for all blob data, meaning every byte flows through the hold service when uploading or downloading container images. This document describes the implementation of **S3 presigned URLs** to eliminate this bottleneck, allowing direct data transfer between clients and S3-compatible storage.
|
||||
|
||||
### Current Architecture (Proxy Mode)
|
||||
|
||||
```
|
||||
Downloads: Docker → AppView → Hold Service → S3 → Hold Service → AppView → Docker
|
||||
Uploads: Docker → AppView → Hold Service → S3
|
||||
```
|
||||
|
||||
**Problems:**
|
||||
- All blob data flows through hold service
|
||||
- Hold service bandwidth = total image bandwidth
|
||||
- Latency from extra hops
|
||||
- Hold service becomes bottleneck for large images
|
||||
|
||||
### Target Architecture (Presigned URLs)
|
||||
|
||||
```
|
||||
Downloads: Docker → AppView (gets presigned URL) → S3 (direct download)
|
||||
Uploads: Docker → AppView → S3 (via presigned URL)
|
||||
Move: AppView → Hold Service → S3 (server-side CopyObject API)
|
||||
```
|
||||
|
||||
**Benefits:**
|
||||
- ✅ Hold service only orchestrates (no data transfer)
|
||||
- ✅ Blob data never touches hold service
|
||||
- ✅ Direct S3 uploads/downloads at wire speed
|
||||
- ✅ Hold service can run on minimal resources
|
||||
- ✅ Works with all S3-compatible services
|
||||
|
||||
## How Presigned URLs Work
|
||||
|
||||
### For Downloads (GET)
|
||||
|
||||
1. **Docker requests blob:** `GET /v2/alice/myapp/blobs/sha256:abc123`
|
||||
2. **AppView asks hold service:** `POST /get-presigned-url`
|
||||
```json
|
||||
{"did": "did:plc:alice123", "digest": "sha256:abc123"}
|
||||
```
|
||||
3. **Hold service generates presigned URL:**
|
||||
```go
|
||||
req, _ := s3Client.GetObjectRequest(&s3.GetObjectInput{
|
||||
Bucket: "my-bucket",
|
||||
Key: "blobs/sha256/ab/abc123.../data",
|
||||
})
|
||||
url, _ := req.Presign(15 * time.Minute)
|
||||
// Returns: https://gateway.storjshare.io/bucket/blobs/...?X-Amz-Signature=...
|
||||
```
|
||||
4. **AppView redirects Docker:** `HTTP 307 Location: <presigned-url>`
|
||||
5. **Docker downloads directly from S3** using the presigned URL
|
||||
|
||||
**Data path:** Docker → S3 (direct)
|
||||
**Hold service bandwidth:** ~1KB (API request/response)
|
||||
|
||||
### For Uploads (PUT)
|
||||
|
||||
**Small blobs (< 5MB) using Put():**
|
||||
|
||||
1. **Docker sends blob to AppView:** `PUT /v2/alice/myapp/blobs/uploads/{uuid}`
|
||||
2. **AppView asks hold service:** `POST /put-presigned-url`
|
||||
```json
|
||||
{"did": "did:plc:alice123", "digest": "sha256:abc123", "size": 1024}
|
||||
```
|
||||
3. **Hold service generates presigned URL:**
|
||||
```go
|
||||
req, _ := s3Client.PutObjectRequest(&s3.PutObjectInput{
|
||||
Bucket: "my-bucket",
|
||||
Key: "blobs/sha256/ab/abc123.../data",
|
||||
})
|
||||
url, _ := req.Presign(15 * time.Minute)
|
||||
```
|
||||
4. **AppView uploads to S3** using presigned URL
|
||||
5. **AppView confirms to Docker:** `201 Created`
|
||||
|
||||
**Data path:** Docker → AppView → S3 (via presigned URL)
|
||||
**Hold service bandwidth:** ~1KB (API request/response)
|
||||
|
||||
### For Streaming Uploads (Create/Commit)
|
||||
|
||||
**Large blobs (> 5MB) using streaming:**
|
||||
|
||||
1. **Docker starts upload:** `POST /v2/alice/myapp/blobs/uploads/`
|
||||
2. **AppView creates upload session** with UUID
|
||||
3. **AppView gets presigned URL for temp location:**
|
||||
```json
|
||||
POST /put-presigned-url
|
||||
{"did": "...", "digest": "uploads/temp-{uuid}", "size": 0}
|
||||
```
|
||||
4. **Docker streams data:** `PATCH /v2/alice/myapp/blobs/uploads/{uuid}`
|
||||
5. **AppView streams to S3** using presigned URL to `uploads/temp-{uuid}/data`
|
||||
6. **Docker finalizes:** `PUT /v2/.../uploads/{uuid}?digest=sha256:abc123`
|
||||
7. **AppView requests move:** `POST /move?from=uploads/temp-{uuid}&to=sha256:abc123`
|
||||
8. **Hold service executes S3 server-side copy:**
|
||||
```go
|
||||
s3.CopyObject(&s3.CopyObjectInput{
|
||||
Bucket: "my-bucket",
|
||||
CopySource: "/my-bucket/uploads/temp-{uuid}/data",
|
||||
Key: "blobs/sha256/ab/abc123.../data",
|
||||
})
|
||||
s3.DeleteObject(&s3.DeleteObjectInput{
|
||||
Key: "uploads/temp-{uuid}/data",
|
||||
})
|
||||
```
|
||||
|
||||
**Data path:** Docker → AppView → S3 (temp location)
|
||||
**Move path:** S3 internal copy (no data transfer!)
|
||||
**Hold service bandwidth:** ~2KB (presigned URL + CopyObject API)
|
||||
|
||||
### For Chunked Uploads (Multipart Upload)
|
||||
|
||||
**Large blobs with OCI chunked protocol (Docker PATCH requests):**
|
||||
|
||||
The OCI Distribution Spec uses chunked uploads via multiple PATCH requests. Single presigned URLs don't support this - we need **S3 Multipart Upload**.
|
||||
|
||||
1. **Docker starts upload:** `POST /v2/alice/myapp/blobs/uploads/`
|
||||
2. **AppView initiates multipart:**
|
||||
```json
|
||||
POST /start-multipart
|
||||
{"did": "...", "digest": "uploads/temp-{uuid}"}
|
||||
→ Returns: {"upload_id": "xyz123"}
|
||||
```
|
||||
3. **Docker sends chunk 1:** `PATCH /v2/.../uploads/{uuid}` (5MB data)
|
||||
4. **AppView gets part URL:**
|
||||
```json
|
||||
POST /part-presigned-url
|
||||
{"did": "...", "digest": "uploads/temp-{uuid}", "upload_id": "xyz123", "part_number": 1}
|
||||
→ Returns: {"url": "https://s3.../part?uploadId=xyz123&partNumber=1&..."}
|
||||
```
|
||||
5. **AppView uploads part 1** using presigned URL → Gets ETag
|
||||
6. **Docker sends chunk 2:** `PATCH /v2/.../uploads/{uuid}` (5MB data)
|
||||
7. **Repeat steps 4-5** for part 2 (and subsequent parts)
|
||||
8. **Docker finalizes:** `PUT /v2/.../uploads/{uuid}?digest=sha256:abc123`
|
||||
9. **AppView completes multipart:**
|
||||
```json
|
||||
POST /complete-multipart
|
||||
{"did": "...", "digest": "uploads/temp-{uuid}", "upload_id": "xyz123",
|
||||
"parts": [{"part_number": 1, "etag": "..."}, {"part_number": 2, "etag": "..."}]}
|
||||
```
|
||||
10. **AppView requests move:** `POST /move?from=uploads/temp-{uuid}&to=sha256:abc123`
|
||||
11. **Hold service executes S3 server-side copy** (same as above)
|
||||
|
||||
**Data path:** Docker → AppView (buffers 5MB) → S3 (via presigned URL per part)
|
||||
**Each PATCH:** Independent, non-blocking, immediate response
|
||||
**Hold service bandwidth:** ~1KB per part + ~1KB for completion
|
||||
|
||||
**Why This Fixes "Client Disconnected" Errors:**
|
||||
- Previous implementation: Single presigned URL + pipe → PATCH blocks → Docker timeout
|
||||
- New implementation: Each PATCH → separate part upload → immediate response → no blocking
|
||||
|
||||
## Why the Temp → Final Move is Required
|
||||
|
||||
This is **not an ATCR implementation detail** — it's required by the [OCI Distribution Specification](https://github.com/opencontainers/distribution-spec/blob/main/spec.md#push).
|
||||
|
||||
### The Problem: Unknown Digest
|
||||
|
||||
Docker doesn't know the blob's digest until **after** uploading:
|
||||
|
||||
1. **Streaming data:** Can't buffer 5GB layer in memory to calculate digest first
|
||||
2. **Stdin pipes:** `docker build . | docker push` generates data on-the-fly
|
||||
3. **Chunked uploads:** Multiple PATCH requests, digest calculated as data streams
|
||||
|
||||
### The Solution: Upload to Temp, Verify, Move
|
||||
|
||||
**All OCI registries do this:**
|
||||
|
||||
1. Client: `POST /v2/{name}/blobs/uploads/` → Get upload UUID
|
||||
2. Client: `PATCH /v2/{name}/blobs/uploads/{uuid}` → Stream data to temp location
|
||||
3. Client: `PUT /v2/{name}/blobs/uploads/{uuid}?digest=sha256:abc` → Provide digest
|
||||
4. Registry: Verify digest matches uploaded data
|
||||
5. Registry: Move `uploads/{uuid}` → `blobs/sha256/abc123...`
|
||||
|
||||
**Docker Hub, GHCR, ECR, Harbor — all use this pattern.**
|
||||
|
||||
### Why It's Efficient with S3
|
||||
|
||||
**For S3, the move is a CopyObject API call:**
|
||||
|
||||
```go
|
||||
// This happens INSIDE S3 servers - no data transfer!
|
||||
s3.CopyObject(&s3.CopyObjectInput{
|
||||
Bucket: "my-bucket",
|
||||
CopySource: "/my-bucket/uploads/temp-12345/data", // 5GB blob
|
||||
Key: "blobs/sha256/ab/abc123.../data",
|
||||
})
|
||||
// S3 copies internally, hold service only sends ~1KB API request
|
||||
```
|
||||
|
||||
**For a 5GB layer:**
|
||||
- Hold service bandwidth: **~1KB** (API request/response)
|
||||
- S3 internal copy: Instant (metadata operation on S3 side)
|
||||
- No data leaves S3, no network transfer
|
||||
|
||||
This is why the move operation is essentially free!
|
||||
|
||||
## Implementation Details
|
||||
|
||||
### 1. Add S3 Client to Hold Service
|
||||
|
||||
**File: `cmd/hold/main.go`**
|
||||
|
||||
Modify `HoldService` struct:
|
||||
```go
|
||||
type HoldService struct {
|
||||
driver storagedriver.StorageDriver
|
||||
config *Config
|
||||
s3Client *s3.S3 // NEW: S3 client for presigned URLs
|
||||
bucket string // NEW: Bucket name
|
||||
s3PathPrefix string // NEW: Path prefix (if any)
|
||||
}
|
||||
```
|
||||
|
||||
Add initialization function:
|
||||
```go
|
||||
func (s *HoldService) initS3Client() error {
|
||||
if s.config.Storage.Type() != "s3" {
|
||||
log.Printf("Storage driver is %s (not S3), presigned URLs disabled", s.config.Storage.Type())
|
||||
return nil
|
||||
}
|
||||
|
||||
params := s.config.Storage.Parameters()["s3"].(configuration.Parameters)
|
||||
|
||||
// Build AWS config
|
||||
awsConfig := &aws.Config{
|
||||
Region: aws.String(params["region"].(string)),
|
||||
Credentials: credentials.NewStaticCredentials(
|
||||
params["accesskey"].(string),
|
||||
params["secretkey"].(string),
|
||||
"",
|
||||
),
|
||||
}
|
||||
|
||||
// Add custom endpoint for S3-compatible services (Storj, MinIO, etc.)
|
||||
if endpoint, ok := params["regionendpoint"].(string); ok && endpoint != "" {
|
||||
awsConfig.Endpoint = aws.String(endpoint)
|
||||
awsConfig.S3ForcePathStyle = aws.Bool(true) // Required for MinIO, Storj
|
||||
}
|
||||
|
||||
sess, err := session.NewSession(awsConfig)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create AWS session: %w", err)
|
||||
}
|
||||
|
||||
s.s3Client = s3.New(sess)
|
||||
s.bucket = params["bucket"].(string)
|
||||
|
||||
log.Printf("S3 presigned URLs enabled for bucket: %s", s.bucket)
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
Call during service initialization:
|
||||
```go
|
||||
func NewHoldService(cfg *Config) (*HoldService, error) {
|
||||
// ... existing driver creation ...
|
||||
|
||||
service := &HoldService{
|
||||
driver: driver,
|
||||
config: cfg,
|
||||
}
|
||||
|
||||
// Initialize S3 client for presigned URLs
|
||||
if err := service.initS3Client(); err != nil {
|
||||
log.Printf("WARNING: S3 presigned URLs disabled: %v", err)
|
||||
}
|
||||
|
||||
return service, nil
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Implement Presigned URL Generation
|
||||
|
||||
**For Downloads:**
|
||||
|
||||
```go
|
||||
func (s *HoldService) getDownloadURL(ctx context.Context, digest string, did string) (string, error) {
|
||||
path := blobPath(digest)
|
||||
|
||||
// Check if blob exists
|
||||
if _, err := s.driver.Stat(ctx, path); err != nil {
|
||||
return "", fmt.Errorf("blob not found: %w", err)
|
||||
}
|
||||
|
||||
// If S3 client available, generate presigned URL
|
||||
if s.s3Client != nil {
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
|
||||
req, _ := s.s3Client.GetObjectRequest(&s3.GetObjectInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
})
|
||||
|
||||
url, err := req.Presign(15 * time.Minute)
|
||||
if err != nil {
|
||||
log.Printf("WARN: Presigned URL generation failed, falling back to proxy: %v", err)
|
||||
return s.getProxyDownloadURL(digest, did), nil
|
||||
}
|
||||
|
||||
log.Printf("Generated presigned download URL for %s (expires in 15min)", digest)
|
||||
return url, nil
|
||||
}
|
||||
|
||||
// Fallback: return proxy URL
|
||||
return s.getProxyDownloadURL(digest, did), nil
|
||||
}
|
||||
|
||||
func (s *HoldService) getProxyDownloadURL(digest, did string) string {
|
||||
return fmt.Sprintf("%s/blobs/%s?did=%s", s.config.Server.PublicURL, digest, did)
|
||||
}
|
||||
```
|
||||
|
||||
**For Uploads:**
|
||||
|
||||
```go
|
||||
func (s *HoldService) getUploadURL(ctx context.Context, digest string, size int64, did string) (string, error) {
|
||||
path := blobPath(digest)
|
||||
|
||||
// If S3 client available, generate presigned URL
|
||||
if s.s3Client != nil {
|
||||
s3Key := strings.TrimPrefix(path, "/")
|
||||
|
||||
req, _ := s.s3Client.PutObjectRequest(&s3.PutObjectInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
})
|
||||
|
||||
url, err := req.Presign(15 * time.Minute)
|
||||
if err != nil {
|
||||
log.Printf("WARN: Presigned URL generation failed, falling back to proxy: %v", err)
|
||||
return s.getProxyUploadURL(digest, did), nil
|
||||
}
|
||||
|
||||
log.Printf("Generated presigned upload URL for %s (expires in 15min)", digest)
|
||||
return url, nil
|
||||
}
|
||||
|
||||
// Fallback: return proxy URL
|
||||
return s.getProxyUploadURL(digest, did), nil
|
||||
}
|
||||
|
||||
func (s *HoldService) getProxyUploadURL(digest, did string) string {
|
||||
return fmt.Sprintf("%s/blobs/%s?did=%s", s.config.Server.PublicURL, digest, did)
|
||||
}
|
||||
```
|
||||
|
||||
### 3. Multipart Upload Endpoints (Required for Chunked Uploads)
|
||||
|
||||
**File: `cmd/hold/main.go`**
|
||||
|
||||
#### Start Multipart Upload
|
||||
|
||||
```go
|
||||
func (s *HoldService) HandleStartMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
var req StartMultipartUploadRequest // {did, digest}
|
||||
|
||||
// Validate DID authorization for WRITE
|
||||
if !s.isAuthorizedWrite(req.DID) {
|
||||
// Return 403 Forbidden
|
||||
}
|
||||
|
||||
// Initiate S3 multipart upload
|
||||
result, err := s.s3Client.CreateMultipartUploadWithContext(ctx, &s3.CreateMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
})
|
||||
|
||||
// Return upload ID
|
||||
json.NewEncoder(w).Encode(StartMultipartUploadResponse{
|
||||
UploadID: *result.UploadId,
|
||||
ExpiresAt: time.Now().Add(24 * time.Hour),
|
||||
})
|
||||
}
|
||||
```
|
||||
|
||||
**Route:** `POST /start-multipart`
|
||||
|
||||
#### Get Part Presigned URL
|
||||
|
||||
```go
|
||||
func (s *HoldService) HandleGetPartURL(w http.ResponseWriter, r *http.Request) {
|
||||
var req GetPartURLRequest // {did, digest, upload_id, part_number}
|
||||
|
||||
// Generate presigned URL for specific part
|
||||
req, _ := s.s3Client.UploadPartRequest(&s3.UploadPartInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
PartNumber: aws.Int64(int64(partNumber)),
|
||||
})
|
||||
|
||||
url, err := req.Presign(15 * time.Minute)
|
||||
|
||||
json.NewEncoder(w).Encode(GetPartURLResponse{URL: url})
|
||||
}
|
||||
```
|
||||
|
||||
**Route:** `POST /part-presigned-url`
|
||||
|
||||
#### Complete Multipart Upload
|
||||
|
||||
```go
|
||||
func (s *HoldService) HandleCompleteMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
var req CompleteMultipartRequest // {did, digest, upload_id, parts: [{part_number, etag}]}
|
||||
|
||||
// Convert parts to S3 format
|
||||
s3Parts := make([]*s3.CompletedPart, len(req.Parts))
|
||||
for i, p := range req.Parts {
|
||||
s3Parts[i] = &s3.CompletedPart{
|
||||
PartNumber: aws.Int64(int64(p.PartNumber)),
|
||||
ETag: aws.String(p.ETag),
|
||||
}
|
||||
}
|
||||
|
||||
// Complete multipart upload
|
||||
_, err := s.s3Client.CompleteMultipartUploadWithContext(ctx, &s3.CompleteMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
MultipartUpload: &s3.CompletedMultipartUpload{Parts: s3Parts},
|
||||
})
|
||||
}
|
||||
```
|
||||
|
||||
**Route:** `POST /complete-multipart`
|
||||
|
||||
#### Abort Multipart Upload
|
||||
|
||||
```go
|
||||
func (s *HoldService) HandleAbortMultipart(w http.ResponseWriter, r *http.Request) {
|
||||
var req AbortMultipartRequest // {did, digest, upload_id}
|
||||
|
||||
// Abort and cleanup parts
|
||||
_, err := s.s3Client.AbortMultipartUploadWithContext(ctx, &s3.AbortMultipartUploadInput{
|
||||
Bucket: aws.String(s.bucket),
|
||||
Key: aws.String(s3Key),
|
||||
UploadId: aws.String(uploadID),
|
||||
})
|
||||
}
|
||||
```
|
||||
|
||||
**Route:** `POST /abort-multipart`
|
||||
|
||||
### 4. Move Operation (No Changes)
|
||||
|
||||
The existing `/move` endpoint already uses `driver.Move()`, which for S3:
|
||||
- Calls `s3.CopyObject()` (server-side copy)
|
||||
- Calls `s3.DeleteObject()` (delete source)
|
||||
- No data transfer through hold service!
|
||||
|
||||
**File: `cmd/hold/main.go:393` (already exists, no changes needed)**
|
||||
|
||||
```go
|
||||
func (s *HoldService) HandleMove(w http.ResponseWriter, r *http.Request) {
|
||||
// ... existing auth and parsing ...
|
||||
|
||||
sourcePath := blobPath(fromPath) // uploads/temp-{uuid}/data
|
||||
destPath := blobPath(toDigest) // blobs/sha256/ab/abc123.../data
|
||||
|
||||
// For S3, this does CopyObject + DeleteObject (server-side)
|
||||
if err := s.driver.Move(ctx, sourcePath, destPath); err != nil {
|
||||
// ... error handling ...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 5. AppView Changes (Multipart Upload Implementation)
|
||||
|
||||
**File: `pkg/storage/proxy_blob_store.go:228`**
|
||||
|
||||
Currently streams to hold service proxy URL. Could be optimized to use presigned URL:
|
||||
|
||||
```go
|
||||
// In Create() - line 228
|
||||
go func() {
|
||||
defer pipeReader.Close()
|
||||
|
||||
tempPath := fmt.Sprintf("uploads/temp-%s", writer.id)
|
||||
|
||||
// Try to get presigned URL for temp location
|
||||
url, err := p.getUploadURL(ctx, digest.FromString(tempPath), 0)
|
||||
if err != nil {
|
||||
// Fallback to direct proxy URL
|
||||
url = fmt.Sprintf("%s/blobs/%s?did=%s", p.storageEndpoint, tempPath, p.did)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(uploadCtx, "PUT", url, pipeReader)
|
||||
// ... rest unchanged
|
||||
}()
|
||||
```
|
||||
|
||||
**Note:** This optimization is optional. The presigned URL will be returned by hold service's `getUploadURL()` anyway.
|
||||
|
||||
## S3-Compatible Service Support
|
||||
|
||||
### Storj
|
||||
|
||||
```bash
|
||||
# .env file
|
||||
STORAGE_DRIVER=s3
|
||||
AWS_ACCESS_KEY_ID=your-storj-access-key
|
||||
AWS_SECRET_ACCESS_KEY=your-storj-secret-key
|
||||
S3_BUCKET=your-bucket-name
|
||||
S3_REGION=global
|
||||
S3_ENDPOINT=https://gateway.storjshare.io
|
||||
```
|
||||
|
||||
**Presigned URL example:**
|
||||
```
|
||||
https://gateway.storjshare.io/your-bucket/blobs/sha256/ab/abc123.../data?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=...&X-Amz-Signature=...
|
||||
```
|
||||
|
||||
### MinIO
|
||||
|
||||
```bash
|
||||
STORAGE_DRIVER=s3
|
||||
AWS_ACCESS_KEY_ID=minioadmin
|
||||
AWS_SECRET_ACCESS_KEY=minioadmin
|
||||
S3_BUCKET=registry
|
||||
S3_REGION=us-east-1
|
||||
S3_ENDPOINT=http://minio.example.com:9000
|
||||
```
|
||||
|
||||
### Backblaze B2
|
||||
|
||||
```bash
|
||||
STORAGE_DRIVER=s3
|
||||
AWS_ACCESS_KEY_ID=your-b2-key-id
|
||||
AWS_SECRET_ACCESS_KEY=your-b2-application-key
|
||||
S3_BUCKET=your-bucket-name
|
||||
S3_REGION=us-west-002
|
||||
S3_ENDPOINT=https://s3.us-west-002.backblazeb2.com
|
||||
```
|
||||
|
||||
### Cloudflare R2
|
||||
|
||||
```bash
|
||||
STORAGE_DRIVER=s3
|
||||
AWS_ACCESS_KEY_ID=your-r2-access-key-id
|
||||
AWS_SECRET_ACCESS_KEY=your-r2-secret-access-key
|
||||
S3_BUCKET=your-bucket-name
|
||||
S3_REGION=auto
|
||||
S3_ENDPOINT=https://<account-id>.r2.cloudflarestorage.com
|
||||
```
|
||||
|
||||
**All these services support presigned URLs with AWS SDK v1!**
|
||||
|
||||
## Performance Impact
|
||||
|
||||
### Bandwidth Savings
|
||||
|
||||
**Before (proxy mode):**
|
||||
- 5GB layer upload: Hold service receives 5GB, sends 5GB to S3 = **10GB** bandwidth
|
||||
- 5GB layer download: S3 sends 5GB to hold, hold sends 5GB to client = **10GB** bandwidth
|
||||
- **Total for push+pull: 20GB hold service bandwidth**
|
||||
|
||||
**After (presigned URLs):**
|
||||
- 5GB layer upload: Hold generates URL (1KB), AppView → S3 direct (5GB), CopyObject API (1KB) = **~2KB** hold bandwidth
|
||||
- 5GB layer download: Hold generates URL (1KB), client → S3 direct = **~1KB** hold bandwidth
|
||||
- **Total for push+pull: ~3KB hold service bandwidth**
|
||||
|
||||
**Savings: 99.98% reduction in hold service bandwidth!**
|
||||
|
||||
### Latency Improvements
|
||||
|
||||
**Before:**
|
||||
- Download: Client → AppView → Hold → S3 → Hold → AppView → Client (4 hops)
|
||||
- Upload: Client → AppView → Hold → S3 (3 hops)
|
||||
|
||||
**After:**
|
||||
- Download: Client → AppView (redirect) → S3 (1 hop to data)
|
||||
- Upload: Client → AppView → S3 (2 hops)
|
||||
- Move: S3 internal (no network hops)
|
||||
|
||||
### Resource Requirements
|
||||
|
||||
**Before:**
|
||||
- Hold service needs bandwidth = sum of all image operations
|
||||
- For 100 concurrent 1GB pushes: 100GB/s bandwidth needed
|
||||
- Expensive, hard to scale
|
||||
|
||||
**After:**
|
||||
- Hold service needs minimal CPU for presigned URL signing
|
||||
- For 100 concurrent 1GB pushes: ~100KB/s bandwidth needed (API traffic)
|
||||
- Can run on $5/month instance!
|
||||
|
||||
## Security Considerations
|
||||
|
||||
### Presigned URL Expiration
|
||||
|
||||
- Default: **15 minutes** expiration
|
||||
- Presigned URL includes embedded credentials in query params
|
||||
- After expiry, URL becomes invalid (S3 rejects with 403)
|
||||
- No long-lived URLs floating around
|
||||
|
||||
### Authorization Flow
|
||||
|
||||
1. **AppView validates user** via ATProto OAuth
|
||||
2. **AppView passes DID to hold service** in presigned URL request
|
||||
3. **Hold service validates DID** (owner or crew member)
|
||||
4. **Hold service generates presigned URL** if authorized
|
||||
5. **Client uses presigned URL** directly with S3
|
||||
|
||||
**Security boundary:** Hold service controls who gets presigned URLs, S3 validates the URLs.
|
||||
|
||||
### Fallback Security
|
||||
|
||||
If presigned URL generation fails:
|
||||
- Falls back to proxy URLs (existing behavior)
|
||||
- Still requires hold service authorization
|
||||
- Data flows through hold service (original security model)
|
||||
|
||||
## Testing & Validation
|
||||
|
||||
### Verify Presigned URLs are Used
|
||||
|
||||
**1. Check hold service logs:**
|
||||
```bash
|
||||
docker logs atcr-hold | grep -i presigned
|
||||
# Should see: "Generated presigned download/upload URL for sha256:..."
|
||||
```
|
||||
|
||||
**2. Monitor network traffic:**
|
||||
```bash
|
||||
# Before: Large data transfers to/from hold service
|
||||
docker stats atcr-hold
|
||||
|
||||
# After: Minimal network usage on hold service
|
||||
docker stats atcr-hold
|
||||
```
|
||||
|
||||
**3. Inspect redirect responses:**
|
||||
```bash
|
||||
# Should see 307 redirect to S3 URL
|
||||
curl -v http://appview:5000/v2/alice/myapp/blobs/sha256:abc123 \
|
||||
-H "Authorization: Bearer $TOKEN"
|
||||
|
||||
# Look for:
|
||||
# < HTTP/1.1 307 Temporary Redirect
|
||||
# < Location: https://gateway.storjshare.io/...?X-Amz-Signature=...
|
||||
```
|
||||
|
||||
### Test Fallback Behavior
|
||||
|
||||
**1. With filesystem driver (should use proxy URLs):**
|
||||
```bash
|
||||
STORAGE_DRIVER=filesystem docker-compose up atcr-hold
|
||||
# Logs should show: "Storage driver is filesystem (not S3), presigned URLs disabled"
|
||||
```
|
||||
|
||||
**2. With S3 but invalid credentials (should fall back):**
|
||||
```bash
|
||||
AWS_ACCESS_KEY_ID=invalid docker-compose up atcr-hold
|
||||
# Logs should show: "WARN: Presigned URL generation failed, falling back to proxy"
|
||||
```
|
||||
|
||||
### Bandwidth Monitoring
|
||||
|
||||
**Track hold service bandwidth over time:**
|
||||
```bash
|
||||
# Install bandwidth monitoring
|
||||
docker exec atcr-hold apt-get update && apt-get install -y vnstat
|
||||
|
||||
# Monitor
|
||||
docker exec atcr-hold vnstat -l
|
||||
```
|
||||
|
||||
**Expected results:**
|
||||
- Before: Bandwidth correlates with image operations
|
||||
- After: Bandwidth stays minimal regardless of image operations
|
||||
|
||||
## Migration Guide
|
||||
|
||||
### For Existing ATCR Deployments
|
||||
|
||||
**1. Update hold service code** (this implementation)
|
||||
|
||||
**2. No configuration changes needed** if already using S3:
|
||||
```bash
|
||||
# Existing S3 config works automatically
|
||||
STORAGE_DRIVER=s3
|
||||
AWS_ACCESS_KEY_ID=...
|
||||
AWS_SECRET_ACCESS_KEY=...
|
||||
S3_BUCKET=...
|
||||
S3_ENDPOINT=...
|
||||
```
|
||||
|
||||
**3. Restart hold service:**
|
||||
```bash
|
||||
docker-compose restart atcr-hold
|
||||
```
|
||||
|
||||
**4. Verify in logs:**
|
||||
```
|
||||
S3 presigned URLs enabled for bucket: my-bucket
|
||||
```
|
||||
|
||||
**5. Test with image push/pull:**
|
||||
```bash
|
||||
docker push atcr.io/alice/myapp:latest
|
||||
docker pull atcr.io/alice/myapp:latest
|
||||
```
|
||||
|
||||
**6. Monitor bandwidth** to confirm reduction
|
||||
|
||||
### Rollback Plan
|
||||
|
||||
If issues arise:
|
||||
|
||||
**Option 1: Disable presigned URLs via env var** (if we add this feature)
|
||||
```bash
|
||||
PRESIGNED_URLS_ENABLED=false docker-compose restart atcr-hold
|
||||
```
|
||||
|
||||
**Option 2: Revert code changes** to previous hold service version
|
||||
|
||||
The implementation has automatic fallbacks, so partial failures won't break functionality.
|
||||
|
||||
## Testing with DISABLE_PRESIGNED_URLS
|
||||
|
||||
### Environment Variable
|
||||
|
||||
Set `DISABLE_PRESIGNED_URLS=true` to force proxy/buffered mode even when S3 is configured.
|
||||
|
||||
**Use cases:**
|
||||
- Testing proxy/buffered code paths with S3 storage
|
||||
- Debugging multipart uploads in buffered mode
|
||||
- Simulating S3 providers that don't support presigned URLs
|
||||
- Verifying fallback behavior works correctly
|
||||
|
||||
### How It Works
|
||||
|
||||
When `DISABLE_PRESIGNED_URLS=true`:
|
||||
|
||||
**Single blob operations:**
|
||||
- `getDownloadURL()` returns proxy URL instead of S3 presigned URL
|
||||
- `getHeadURL()` returns proxy URL instead of S3 presigned HEAD URL
|
||||
- `getUploadURL()` returns proxy URL instead of S3 presigned PUT URL
|
||||
- Client uses `/blobs/{digest}` endpoints (proxy through hold service)
|
||||
|
||||
**Multipart uploads:**
|
||||
- `StartMultipartUploadWithManager()` creates **Buffered** session instead of **S3Native**
|
||||
- `GetPartUploadURL()` returns `/multipart-parts/{uploadID}/{partNumber}` instead of S3 presigned URL
|
||||
- Parts are buffered in memory in the hold service
|
||||
- `CompleteMultipartUploadWithManager()` assembles parts and writes via storage driver
|
||||
|
||||
### Testing Example
|
||||
|
||||
```bash
|
||||
# Test S3 with forced proxy mode
|
||||
export STORAGE_DRIVER=s3
|
||||
export S3_BUCKET=my-bucket
|
||||
export AWS_ACCESS_KEY_ID=...
|
||||
export AWS_SECRET_ACCESS_KEY=...
|
||||
export DISABLE_PRESIGNED_URLS=true # Force buffered/proxy mode
|
||||
|
||||
./bin/atcr-hold
|
||||
|
||||
# Push an image - should use proxy mode
|
||||
docker push atcr.io/yourdid/test:latest
|
||||
|
||||
# Check logs for:
|
||||
# "Presigned URLs disabled, using proxy URL"
|
||||
# "Presigned URLs disabled (DISABLE_PRESIGNED_URLS=true), using buffered mode"
|
||||
# "Stored part: uploadID=... part=1 size=..."
|
||||
```
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
### 1. Configurable Expiration
|
||||
|
||||
Allow customizing presigned URL expiry:
|
||||
```bash
|
||||
PRESIGNED_URL_EXPIRY=30m # Default: 15m
|
||||
```
|
||||
|
||||
### 2. Presigned URL Caching
|
||||
|
||||
Cache presigned URLs for frequently accessed blobs (with shorter TTL).
|
||||
|
||||
### 3. CloudFront/CDN Integration
|
||||
|
||||
For downloads, use CloudFront presigned URLs instead of direct S3:
|
||||
- Better global distribution
|
||||
- Lower egress costs
|
||||
- Faster downloads
|
||||
|
||||
### 4. Multipart Upload Support
|
||||
|
||||
For very large layers (>5GB), use presigned URLs with multipart upload:
|
||||
- Generate presigned URLs for each part
|
||||
- Client uploads parts directly to S3
|
||||
- Hold service finalizes multipart upload
|
||||
|
||||
### 5. Metrics & Monitoring
|
||||
|
||||
Track presigned URL usage:
|
||||
- Count of presigned URLs generated
|
||||
- Fallback rate (proxy vs presigned)
|
||||
- Bandwidth savings metrics
|
||||
|
||||
## References
|
||||
|
||||
- [OCI Distribution Specification - Push](https://github.com/opencontainers/distribution-spec/blob/main/spec.md#push)
|
||||
- [AWS SDK Go v1 - Presigned URLs](https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/s3-example-presigned-urls.html)
|
||||
- [Storj - Using Presigned URLs](https://docs.storj.io/dcs/api-reference/s3-compatible-gateway/using-presigned-urls)
|
||||
- [MinIO - Presigned Upload via Browser](https://docs.min.io/community/minio-object-store/integrations/presigned-put-upload-via-browser.html)
|
||||
- [Cloudflare R2 - Presigned URLs](https://developers.cloudflare.com/r2/api/s3/presigned-urls/)
|
||||
- [Backblaze B2 - S3 Compatible API](https://help.backblaze.com/hc/en-us/articles/360047815993-Does-the-B2-S3-Compatible-API-support-Pre-Signed-URLs)
|
||||
|
||||
## Summary
|
||||
|
||||
Implementing S3 presigned URLs transforms ATCR's hold service from a **data proxy** to a **lightweight orchestrator**:
|
||||
|
||||
✅ **99.98% bandwidth reduction** for hold service
|
||||
✅ **Direct client → S3 transfers** for maximum speed
|
||||
✅ **Works with all S3-compatible services** (Storj, MinIO, R2, B2)
|
||||
✅ **OCI-compliant** temp → final move pattern
|
||||
✅ **Automatic fallback** to proxy mode for non-S3 drivers
|
||||
✅ **No breaking changes** to existing deployments
|
||||
|
||||
This makes BYOS (Bring Your Own Storage) truly scalable and cost-effective, as users can run hold services on minimal infrastructure while serving arbitrarily large container images.
|
||||
Reference in New Issue
Block a user