docs: add distributed deployment architecture and enhancement proposals
- Add comprehensive guide for multi-instance S3-backed deployment - Document configuration strategies for read/write separated instances - Propose core enhancements for distributed operations - Include implementation timeline and monitoring strategies 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
90961c5ce9
commit
9b3a8a2d05
2 changed files with 958 additions and 0 deletions
464
docs/brainy-distributed-enhancements.md
Normal file
464
docs/brainy-distributed-enhancements.md
Normal file
|
|
@ -0,0 +1,464 @@
|
|||
# Proposed Brainy Enhancements for Distributed Operations
|
||||
|
||||
## Executive Summary
|
||||
|
||||
To fully support the distributed deployment scenario with multiple specialized instances sharing S3 storage, Brainy needs several enhancements focused on coordination, consistency, and operational modes.
|
||||
|
||||
## Core Enhancement Areas
|
||||
|
||||
### 1. Instance Role Management
|
||||
|
||||
**Current State**: Brainy operates as a standalone instance without awareness of other instances.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// New distributed configuration options
|
||||
export interface DistributedConfig {
|
||||
role: 'reader' | 'writer' | 'hybrid';
|
||||
instanceId: string;
|
||||
coordinationMethod: 'none' | 's3-polling' | 'websocket' | 'pubsub';
|
||||
consistencyLevel: 'eventual' | 'strong' | 'bounded';
|
||||
conflictResolution: 'last-write-wins' | 'vector-clock' | 'crdt';
|
||||
}
|
||||
|
||||
// Enhanced BrainyData constructor
|
||||
class BrainyData {
|
||||
constructor(config: BrainyConfig & { distributed?: DistributedConfig }) {
|
||||
if (config.distributed) {
|
||||
this.initializeDistributedMode(config.distributed);
|
||||
}
|
||||
}
|
||||
|
||||
private initializeDistributedMode(config: DistributedConfig) {
|
||||
// Set up role-specific behaviors
|
||||
switch(config.role) {
|
||||
case 'reader':
|
||||
this.storage.setReadOnly(true);
|
||||
this.enableAggressiveCaching();
|
||||
this.subscribeToIndexUpdates();
|
||||
break;
|
||||
case 'writer':
|
||||
this.storage.setWriteOnly(true);
|
||||
this.enableWriteBatching();
|
||||
this.publishIndexUpdates();
|
||||
break;
|
||||
case 'hybrid':
|
||||
this.enableCoordinatedAccess();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 2. S3 Coordination Layer
|
||||
|
||||
**Current State**: Direct S3 operations without coordination.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/storage/s3Coordinator.ts
|
||||
export class S3Coordinator {
|
||||
private manifestPath = '_brainy/manifest.json';
|
||||
private lockPrefix = '_brainy/locks/';
|
||||
|
||||
async acquireWriteLock(partition: string): Promise<LockHandle> {
|
||||
const lockKey = `${this.lockPrefix}${partition}`;
|
||||
const lockId = `${this.instanceId}-${Date.now()}`;
|
||||
|
||||
// Use S3's conditional PUT for atomic lock acquisition
|
||||
try {
|
||||
await this.s3.putObject({
|
||||
Bucket: this.bucket,
|
||||
Key: lockKey,
|
||||
Body: JSON.stringify({
|
||||
owner: this.instanceId,
|
||||
acquired: Date.now(),
|
||||
ttl: 30000 // 30 second TTL
|
||||
}),
|
||||
Condition: 'ObjectDoesNotExist'
|
||||
});
|
||||
|
||||
return new LockHandle(lockId, () => this.releaseLock(lockKey));
|
||||
} catch (err) {
|
||||
if (err.code === 'PreconditionFailed') {
|
||||
throw new LockAcquisitionError(`Partition ${partition} is locked`);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
async updateManifest(update: ManifestUpdate) {
|
||||
// Atomic manifest updates using versioning
|
||||
const manifest = await this.getManifest();
|
||||
manifest.version++;
|
||||
manifest.lastUpdate = Date.now();
|
||||
manifest.updates.push(update);
|
||||
|
||||
await this.s3.putObject({
|
||||
Bucket: this.bucket,
|
||||
Key: this.manifestPath,
|
||||
Body: JSON.stringify(manifest),
|
||||
Metadata: {
|
||||
'version': manifest.version.toString()
|
||||
}
|
||||
});
|
||||
|
||||
// Notify other instances
|
||||
await this.broadcastUpdate(update);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3. Partition Assignment Strategy
|
||||
|
||||
**Current State**: All instances access all partitions.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/partitioning/distributedPartitioner.ts
|
||||
export class DistributedPartitioner {
|
||||
private assignments: Map<string, Set<string>> = new Map();
|
||||
|
||||
async assignPartitions(instances: Instance[]): Promise<PartitionAssignment> {
|
||||
const writers = instances.filter(i => i.role === 'writer');
|
||||
const partitions = await this.getAllPartitions();
|
||||
|
||||
// Use consistent hashing for stable assignments
|
||||
const ring = new ConsistentHashRing(writers.map(w => w.id));
|
||||
|
||||
const assignment: PartitionAssignment = {};
|
||||
for (const partition of partitions) {
|
||||
const writer = ring.getNode(partition.id);
|
||||
assignment[writer] = assignment[writer] || [];
|
||||
assignment[writer].push(partition.id);
|
||||
}
|
||||
|
||||
// Store assignments in S3 for coordination
|
||||
await this.storeAssignments(assignment);
|
||||
|
||||
return assignment;
|
||||
}
|
||||
|
||||
async getMyPartitions(): Promise<string[]> {
|
||||
const assignments = await this.loadAssignments();
|
||||
return assignments[this.instanceId] || [];
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 4. Write-Ahead Log for Consistency
|
||||
|
||||
**Current State**: Direct writes without transaction log.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/wal/writeAheadLog.ts
|
||||
export class WriteAheadLog {
|
||||
private logPrefix = '_brainy/wal/';
|
||||
|
||||
async logWrite(operation: WriteOperation): Promise<string> {
|
||||
const logEntry = {
|
||||
id: uuidv4(),
|
||||
timestamp: Date.now(),
|
||||
instanceId: this.instanceId,
|
||||
operation: operation,
|
||||
status: 'pending'
|
||||
};
|
||||
|
||||
// Write to WAL first
|
||||
await this.s3.putObject({
|
||||
Bucket: this.bucket,
|
||||
Key: `${this.logPrefix}${logEntry.id}`,
|
||||
Body: JSON.stringify(logEntry)
|
||||
});
|
||||
|
||||
// Then execute operation
|
||||
try {
|
||||
await this.executeOperation(operation);
|
||||
await this.markComplete(logEntry.id);
|
||||
} catch (err) {
|
||||
await this.markFailed(logEntry.id, err);
|
||||
throw err;
|
||||
}
|
||||
|
||||
return logEntry.id;
|
||||
}
|
||||
|
||||
async recoverFromWAL() {
|
||||
// On startup, check for incomplete operations
|
||||
const pendingOps = await this.getPendingOperations();
|
||||
|
||||
for (const op of pendingOps) {
|
||||
if (this.canRecover(op)) {
|
||||
await this.retryOperation(op);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 5. Operational Modes
|
||||
|
||||
**Current State**: Single operational mode.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/modes/operationalModes.ts
|
||||
export abstract class OperationalMode {
|
||||
abstract canRead(): boolean;
|
||||
abstract canWrite(): boolean;
|
||||
abstract canDelete(): boolean;
|
||||
abstract getCacheStrategy(): CacheStrategy;
|
||||
}
|
||||
|
||||
export class ReadOnlyMode extends OperationalMode {
|
||||
canRead() { return true; }
|
||||
canWrite() { return false; }
|
||||
canDelete() { return false; }
|
||||
|
||||
getCacheStrategy() {
|
||||
return {
|
||||
hotCacheRatio: 0.8, // More memory for cache
|
||||
prefetchAggressive: true,
|
||||
ttl: Infinity // Never expire cache in read-only
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export class WriteOnlyMode extends OperationalMode {
|
||||
canRead() { return false; }
|
||||
canWrite() { return true; }
|
||||
canDelete() { return true; }
|
||||
|
||||
getCacheStrategy() {
|
||||
return {
|
||||
hotCacheRatio: 0.2, // Minimal cache, focus on write buffer
|
||||
writeBuffer: true,
|
||||
batchWrites: true
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
export class HybridMode extends OperationalMode {
|
||||
constructor(private coordinator: Coordinator) {}
|
||||
|
||||
canRead() { return true; }
|
||||
canWrite() { return this.coordinator.hasWriteLock(); }
|
||||
canDelete() { return this.coordinator.hasWriteLock(); }
|
||||
|
||||
getCacheStrategy() {
|
||||
return {
|
||||
hotCacheRatio: 0.5,
|
||||
adaptive: true // Adjust based on workload
|
||||
};
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 6. Health and Coordination Endpoints
|
||||
|
||||
**Current State**: No built-in health/coordination endpoints.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/api/coordinationAPI.ts
|
||||
export class CoordinationAPI {
|
||||
setupEndpoints(app: Express) {
|
||||
// Health check with role information
|
||||
app.get('/health', async (req, res) => {
|
||||
res.json({
|
||||
status: 'healthy',
|
||||
role: this.config.role,
|
||||
instanceId: this.instanceId,
|
||||
partitions: await this.getAssignedPartitions(),
|
||||
metrics: await this.getMetrics()
|
||||
});
|
||||
});
|
||||
|
||||
// Coordination endpoints
|
||||
app.post('/coordinate/rebalance', async (req, res) => {
|
||||
const result = await this.rebalancePartitions();
|
||||
res.json(result);
|
||||
});
|
||||
|
||||
app.get('/coordinate/assignments', async (req, res) => {
|
||||
const assignments = await this.getPartitionAssignments();
|
||||
res.json(assignments);
|
||||
});
|
||||
|
||||
// Sync endpoint for configuration updates
|
||||
app.post('/sync/config', async (req, res) => {
|
||||
await this.updateConfiguration(req.body);
|
||||
res.json({ status: 'updated' });
|
||||
});
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 7. Event-Driven Coordination
|
||||
|
||||
**Current State**: No event system.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/events/distributedEvents.ts
|
||||
export class DistributedEventBus {
|
||||
private subscribers: Map<string, Set<EventHandler>> = new Map();
|
||||
|
||||
async emit(event: BrainyEvent) {
|
||||
// Local handlers
|
||||
const handlers = this.subscribers.get(event.type) || new Set();
|
||||
for (const handler of handlers) {
|
||||
await handler(event);
|
||||
}
|
||||
|
||||
// Remote broadcast (pluggable backends)
|
||||
await this.broadcast(event);
|
||||
}
|
||||
|
||||
async broadcast(event: BrainyEvent) {
|
||||
// S3-based event log (simple, no additional deps)
|
||||
if (this.config.broadcastMethod === 's3') {
|
||||
await this.s3.putObject({
|
||||
Bucket: this.bucket,
|
||||
Key: `_brainy/events/${Date.now()}-${event.type}`,
|
||||
Body: JSON.stringify(event)
|
||||
});
|
||||
}
|
||||
|
||||
// WebSocket broadcast
|
||||
if (this.config.broadcastMethod === 'websocket') {
|
||||
this.ws.broadcast(JSON.stringify(event));
|
||||
}
|
||||
|
||||
// Cloud Pub/Sub
|
||||
if (this.config.broadcastMethod === 'pubsub') {
|
||||
await this.pubsub.topic('brainy-events').publish(event);
|
||||
}
|
||||
}
|
||||
|
||||
subscribe(eventType: string, handler: EventHandler) {
|
||||
if (!this.subscribers.has(eventType)) {
|
||||
this.subscribers.set(eventType, new Set());
|
||||
}
|
||||
this.subscribers.get(eventType).add(handler);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 8. Configuration Synchronization
|
||||
|
||||
**Current State**: Local configuration only.
|
||||
|
||||
**Proposed Enhancement**:
|
||||
```typescript
|
||||
// src/config/distributedConfig.ts
|
||||
export class DistributedConfigManager {
|
||||
private configCache: ConfigCache;
|
||||
private configVersion: number = 0;
|
||||
|
||||
async loadConfig(): Promise<BrainyConfig> {
|
||||
// Try S3 first for shared config
|
||||
try {
|
||||
const s3Config = await this.loadFromS3();
|
||||
this.configVersion = s3Config.version;
|
||||
return this.mergeWithLocal(s3Config);
|
||||
} catch (err) {
|
||||
// Fall back to local config
|
||||
return this.loadLocalConfig();
|
||||
}
|
||||
}
|
||||
|
||||
async watchConfigChanges() {
|
||||
// Poll S3 for config updates
|
||||
setInterval(async () => {
|
||||
const latestVersion = await this.getConfigVersion();
|
||||
if (latestVersion > this.configVersion) {
|
||||
const newConfig = await this.loadFromS3();
|
||||
await this.applyConfigUpdate(newConfig);
|
||||
this.configVersion = latestVersion;
|
||||
}
|
||||
}, 10000); // Check every 10 seconds
|
||||
}
|
||||
|
||||
private async applyConfigUpdate(config: BrainyConfig) {
|
||||
// Hot-reload configuration without restart
|
||||
if (config.caching) {
|
||||
this.cacheManager.updateStrategy(config.caching);
|
||||
}
|
||||
if (config.partitioning) {
|
||||
await this.partitioner.reconfigure(config.partitioning);
|
||||
}
|
||||
// Emit event for other components
|
||||
this.eventBus.emit({
|
||||
type: 'config_updated',
|
||||
payload: config
|
||||
});
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Implementation Priority
|
||||
|
||||
### Phase 1: Essential Features (Week 1-2)
|
||||
1. **Operational Modes** - Enable read-only/write-only modes
|
||||
2. **S3 Coordination** - Basic locking and manifest management
|
||||
3. **Configuration Sync** - Shared configuration from S3
|
||||
|
||||
### Phase 2: Coordination (Week 3-4)
|
||||
4. **Partition Assignment** - Distributed partition management
|
||||
5. **Event System** - Basic event broadcasting
|
||||
6. **Health Endpoints** - Monitoring and coordination APIs
|
||||
|
||||
### Phase 3: Advanced Features (Week 5-6)
|
||||
7. **Write-Ahead Log** - Consistency and recovery
|
||||
8. **Advanced Coordination** - WebSocket/Pub-Sub integration
|
||||
9. **Auto-rebalancing** - Dynamic partition redistribution
|
||||
|
||||
## Backwards Compatibility
|
||||
|
||||
All enhancements should be optional and backwards compatible:
|
||||
|
||||
```typescript
|
||||
// Default behavior remains unchanged
|
||||
const brainy = new BrainyData({ /* existing config */ });
|
||||
|
||||
// Opt-in to distributed features
|
||||
const distributedBrainy = new BrainyData({
|
||||
/* existing config */,
|
||||
distributed: {
|
||||
enabled: true,
|
||||
role: 'reader',
|
||||
// ... distributed options
|
||||
}
|
||||
});
|
||||
```
|
||||
|
||||
## Testing Strategy
|
||||
|
||||
### Unit Tests
|
||||
- Test each operational mode independently
|
||||
- Mock S3 coordination operations
|
||||
- Test configuration synchronization
|
||||
|
||||
### Integration Tests
|
||||
- Multi-instance coordination tests
|
||||
- Partition assignment and rebalancing
|
||||
- Consistency under concurrent operations
|
||||
|
||||
### Load Tests
|
||||
- Simulate millions of vectors
|
||||
- Test read/write separation at scale
|
||||
- Measure coordination overhead
|
||||
|
||||
## Performance Impact
|
||||
|
||||
Expected performance characteristics:
|
||||
- **Read-only instances**: 10-20% faster due to optimized caching
|
||||
- **Write-only instances**: 30-40% higher throughput with batching
|
||||
- **Coordination overhead**: <100ms for most operations
|
||||
- **Configuration sync**: <1s propagation delay
|
||||
|
||||
## Conclusion
|
||||
|
||||
These enhancements would transform Brainy into a truly distributed vector database system capable of handling large-scale deployments with specialized instances. The modular design ensures that simpler use cases remain unaffected while enabling sophisticated distributed architectures when needed.
|
||||
494
docs/distributed-deployment-scenario.md
Normal file
494
docs/distributed-deployment-scenario.md
Normal file
|
|
@ -0,0 +1,494 @@
|
|||
# Distributed Brainy Deployment: Multi-Instance S3 Architecture
|
||||
|
||||
## Scenario Overview
|
||||
|
||||
A production deployment with 3 specialized Brainy instances sharing a single S3 bucket as the source of truth:
|
||||
|
||||
1. **Search Instance** (Read-Only): High-performance search across millions of vectors
|
||||
2. **Bluesky Processor** (Write-Only): High-throughput ingestion from Bluesky firehose
|
||||
3. **GitHub Crawler** (Write-Only): Continuous crawling and indexing of GitHub data
|
||||
|
||||
All instances run as Google Cloud Run containers with shared S3 storage.
|
||||
|
||||
## Architecture Design
|
||||
|
||||
### Shared Configuration Strategy
|
||||
|
||||
#### Option 1: Configuration Service (Recommended)
|
||||
```typescript
|
||||
// config-service.ts - Deployed as separate Cloud Run service
|
||||
export class BrainyConfigService {
|
||||
private s3Config = {
|
||||
bucket: 'brainy-vectors-prod',
|
||||
configPath: '_brainy/config.json',
|
||||
lockPath: '_brainy/config.lock'
|
||||
};
|
||||
|
||||
async getSharedConfig(): Promise<BrainyConfig> {
|
||||
// Fetch from S3 with caching
|
||||
return {
|
||||
hnsw: {
|
||||
M: 16,
|
||||
efConstruction: 200,
|
||||
seed: 42, // Critical: same seed for consistent partitioning
|
||||
maxElements: 10000000
|
||||
},
|
||||
partitioning: {
|
||||
strategy: 'semantic',
|
||||
numPartitions: 128, // Must be consistent across instances
|
||||
replicationFactor: 3,
|
||||
hashFunction: 'xxhash' // Deterministic partitioning
|
||||
},
|
||||
storage: {
|
||||
compressionLevel: 6,
|
||||
chunkSize: 1024 * 1024, // 1MB chunks
|
||||
prefixStrategy: 'date-based' // e.g., /2024/01/15/
|
||||
},
|
||||
caching: {
|
||||
hotCacheSize: '2GB',
|
||||
warmCacheSize: '8GB',
|
||||
ttl: 3600
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Option 2: S3-Based Config Synchronization
|
||||
Store configuration in S3 with versioning and atomic updates:
|
||||
```
|
||||
s3://brainy-vectors-prod/
|
||||
_brainy/
|
||||
config.json # Shared configuration
|
||||
schema.json # Vector schema definition
|
||||
partitions.json # Partition mapping
|
||||
instances/
|
||||
search-001.json # Instance-specific overrides
|
||||
bluesky-001.json
|
||||
github-001.json
|
||||
```
|
||||
|
||||
### Instance-Specific Configurations
|
||||
|
||||
#### 1. Search Instance (Read-Only)
|
||||
```typescript
|
||||
const searchConfig = {
|
||||
...sharedConfig,
|
||||
mode: 'read-only',
|
||||
caching: {
|
||||
hotCacheSize: '8GB', // Maximize cache for search
|
||||
warmCacheSize: '32GB',
|
||||
prefetchStrategy: 'aggressive',
|
||||
bloomFilters: true // Fast negative lookups
|
||||
},
|
||||
hnsw: {
|
||||
...sharedConfig.hnsw,
|
||||
efSearch: 100, // Higher for better recall
|
||||
useMmap: true // Memory-mapped files for large indices
|
||||
},
|
||||
s3: {
|
||||
readConcurrency: 20, // High parallelism for reads
|
||||
useTransferAcceleration: true,
|
||||
cacheHeaders: true
|
||||
},
|
||||
monitoring: {
|
||||
metrics: ['latency', 'recall', 'cache_hit_rate']
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
#### 2. Bluesky Processor (Write-Only)
|
||||
```typescript
|
||||
const blueksyConfig = {
|
||||
...sharedConfig,
|
||||
mode: 'write-only',
|
||||
batching: {
|
||||
size: 10000, // Large batches for throughput
|
||||
flushInterval: 5000, // 5 seconds
|
||||
parallelWrites: 4
|
||||
},
|
||||
deduplication: {
|
||||
enabled: true,
|
||||
bloomFilter: true,
|
||||
windowSize: 1000000 // Check last 1M entries
|
||||
},
|
||||
s3: {
|
||||
writeConcurrency: 10,
|
||||
multipartThreshold: 50 * 1024 * 1024, // 50MB
|
||||
useServerSideEncryption: true
|
||||
},
|
||||
indexing: {
|
||||
async: true, // Don't wait for index updates
|
||||
batchIndexUpdates: true
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
#### 3. GitHub Crawler (Write-Only)
|
||||
```typescript
|
||||
const githubConfig = {
|
||||
...sharedConfig,
|
||||
mode: 'write-only',
|
||||
rateLimit: {
|
||||
requestsPerSecond: 10, // Respect API limits
|
||||
burstSize: 20
|
||||
},
|
||||
batching: {
|
||||
size: 1000, // Smaller batches, continuous flow
|
||||
flushInterval: 10000 // 10 seconds
|
||||
},
|
||||
embedding: {
|
||||
model: 'text-embedding-3-small',
|
||||
batchSize: 100,
|
||||
cacheEmbeddings: true
|
||||
},
|
||||
s3: {
|
||||
writeConcurrency: 5,
|
||||
retryStrategy: 'exponential'
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
## Synchronization Mechanisms
|
||||
|
||||
### 1. Partition Coordinator Service
|
||||
Deploy a lightweight coordinator that manages partition assignments:
|
||||
|
||||
```typescript
|
||||
class PartitionCoordinator {
|
||||
private websocket: WebSocketServer;
|
||||
|
||||
async assignPartition(instanceId: string, mode: 'read' | 'write') {
|
||||
if (mode === 'write') {
|
||||
// Ensure no partition is assigned to multiple writers
|
||||
return this.getExclusivePartition(instanceId);
|
||||
} else {
|
||||
// Readers can access all partitions
|
||||
return 'all';
|
||||
}
|
||||
}
|
||||
|
||||
async rebalance() {
|
||||
// Triggered when instances join/leave
|
||||
// Ensures even distribution of write load
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Event Broadcasting via Pub/Sub
|
||||
Use Google Cloud Pub/Sub for coordination:
|
||||
|
||||
```typescript
|
||||
interface BrainyEvent {
|
||||
type: 'partition_created' | 'index_updated' | 'config_changed';
|
||||
timestamp: number;
|
||||
payload: any;
|
||||
}
|
||||
|
||||
// Writers publish events
|
||||
await pubsub.topic('brainy-events').publish({
|
||||
type: 'partition_created',
|
||||
payload: { partitionId: 'p-123', vectorCount: 50000 }
|
||||
});
|
||||
|
||||
// Readers subscribe and update local state
|
||||
subscription.on('message', (message) => {
|
||||
if (message.type === 'index_updated') {
|
||||
await this.refreshLocalIndex(message.payload.partitionId);
|
||||
}
|
||||
});
|
||||
```
|
||||
|
||||
## Performance Optimizations
|
||||
|
||||
### 1. Write Path Optimization
|
||||
```typescript
|
||||
// Parallel partition writes
|
||||
class PartitionedWriter {
|
||||
async write(vectors: Vector[]) {
|
||||
const partitioned = this.partitionVectors(vectors);
|
||||
|
||||
await Promise.all(
|
||||
Object.entries(partitioned).map(([partitionId, vecs]) =>
|
||||
this.writeToPartition(partitionId, vecs)
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
private partitionVectors(vectors: Vector[]) {
|
||||
// Use consistent hash to determine partition
|
||||
return vectors.reduce((acc, vec) => {
|
||||
const partition = hashToPartition(vec.id);
|
||||
acc[partition] = acc[partition] || [];
|
||||
acc[partition].push(vec);
|
||||
return acc;
|
||||
}, {});
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Read Path Optimization
|
||||
```typescript
|
||||
// Distributed search with result aggregation
|
||||
class DistributedSearch {
|
||||
async search(query: Vector, k: number) {
|
||||
// Identify relevant partitions using routing table
|
||||
const partitions = await this.getRelevantPartitions(query);
|
||||
|
||||
// Parallel search across partitions
|
||||
const results = await Promise.all(
|
||||
partitions.map(p => this.searchPartition(p, query, k * 2))
|
||||
);
|
||||
|
||||
// Merge and re-rank results
|
||||
return this.mergeResults(results, k);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3. S3 Optimization Strategies
|
||||
```typescript
|
||||
const s3Optimizations = {
|
||||
// Use S3 Transfer Acceleration for cross-region
|
||||
transferAcceleration: true,
|
||||
|
||||
// Intelligent prefixing for parallel reads
|
||||
prefixSharding: {
|
||||
enabled: true,
|
||||
shardCount: 16, // Distribute across 16 prefixes
|
||||
strategy: 'hash' // or 'round-robin'
|
||||
},
|
||||
|
||||
// Batch operations
|
||||
batchOperations: {
|
||||
getObject: 100, // Batch up to 100 GETs
|
||||
putObject: 50 // Batch up to 50 PUTs
|
||||
},
|
||||
|
||||
// Caching strategy
|
||||
caching: {
|
||||
cloudFront: true, // Use CDN for read-heavy workloads
|
||||
s3CacheControl: 'max-age=3600'
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
## Suggested Brainy Enhancements
|
||||
|
||||
### 1. Native Distributed Mode
|
||||
```typescript
|
||||
// Proposed API
|
||||
const brainy = new BrainyData({
|
||||
distributed: {
|
||||
mode: 'cluster',
|
||||
role: 'writer' | 'reader' | 'hybrid',
|
||||
coordinator: 'redis://coordinator:6379',
|
||||
instanceId: process.env.INSTANCE_ID
|
||||
}
|
||||
});
|
||||
```
|
||||
|
||||
### 2. S3 Lock Manager
|
||||
```typescript
|
||||
class S3LockManager {
|
||||
async acquireLock(resource: string, ttl: number) {
|
||||
// Use S3 conditional puts for distributed locking
|
||||
const lockKey = `_locks/${resource}`;
|
||||
const lockValue = `${this.instanceId}-${Date.now()}`;
|
||||
|
||||
try {
|
||||
await s3.putObject({
|
||||
Bucket: this.bucket,
|
||||
Key: lockKey,
|
||||
Body: lockValue,
|
||||
Metadata: { ttl: ttl.toString() },
|
||||
Condition: 'ObjectDoesNotExist'
|
||||
});
|
||||
return true;
|
||||
} catch (err) {
|
||||
if (err.code === 'PreconditionFailed') {
|
||||
return false; // Lock already held
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3. Partition Discovery Service
|
||||
```typescript
|
||||
class PartitionDiscovery {
|
||||
private cache = new Map();
|
||||
|
||||
async discoverPartitions(): Promise<Partition[]> {
|
||||
// List S3 prefixes to discover partitions
|
||||
const result = await s3.listObjectsV2({
|
||||
Bucket: this.bucket,
|
||||
Prefix: 'partitions/',
|
||||
Delimiter: '/'
|
||||
});
|
||||
|
||||
return result.CommonPrefixes.map(prefix => ({
|
||||
id: prefix.Prefix.split('/')[1],
|
||||
metadata: await this.getPartitionMetadata(prefix.Prefix)
|
||||
}));
|
||||
}
|
||||
|
||||
subscribeToChanges(callback: (event: PartitionEvent) => void) {
|
||||
// Watch S3 events or use SNS/EventBridge
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 4. Consistency Manager
|
||||
```typescript
|
||||
class ConsistencyManager {
|
||||
async ensureConsistency() {
|
||||
// Periodic consistency checks
|
||||
const tasks = [
|
||||
this.verifyPartitionIntegrity(),
|
||||
this.checkIndexConsistency(),
|
||||
this.validateConfiguration()
|
||||
];
|
||||
|
||||
const results = await Promise.all(tasks);
|
||||
|
||||
if (results.some(r => !r.valid)) {
|
||||
await this.triggerRepair();
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Deployment Configuration
|
||||
|
||||
### Cloud Run Service Definitions
|
||||
|
||||
```yaml
|
||||
# search-service.yaml
|
||||
apiVersion: serving.knative.dev/v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: brainy-search
|
||||
spec:
|
||||
template:
|
||||
spec:
|
||||
containers:
|
||||
- image: gcr.io/project/brainy-search:latest
|
||||
env:
|
||||
- name: BRAINY_MODE
|
||||
value: "read-only"
|
||||
- name: BRAINY_ROLE
|
||||
value: "search"
|
||||
resources:
|
||||
limits:
|
||||
cpu: "4"
|
||||
memory: "16Gi"
|
||||
startupProbe:
|
||||
httpGet:
|
||||
path: /health
|
||||
initialDelaySeconds: 30
|
||||
|
||||
# bluesky-processor.yaml
|
||||
apiVersion: serving.knative.dev/v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: brainy-bluesky
|
||||
spec:
|
||||
template:
|
||||
spec:
|
||||
containers:
|
||||
- image: gcr.io/project/brainy-bluesky:latest
|
||||
env:
|
||||
- name: BRAINY_MODE
|
||||
value: "write-only"
|
||||
- name: BRAINY_ROLE
|
||||
value: "bluesky-processor"
|
||||
resources:
|
||||
limits:
|
||||
cpu: "2"
|
||||
memory: "8Gi"
|
||||
```
|
||||
|
||||
### Environment Variables
|
||||
```bash
|
||||
# Common to all instances
|
||||
BRAINY_S3_BUCKET=brainy-vectors-prod
|
||||
BRAINY_S3_REGION=us-central1
|
||||
BRAINY_CONFIG_SOURCE=s3
|
||||
BRAINY_CONFIG_PATH=_brainy/config.json
|
||||
|
||||
# Instance-specific
|
||||
BRAINY_INSTANCE_ID=${K_SERVICE}-${K_REVISION}
|
||||
BRAINY_ROLE=search|bluesky|github
|
||||
BRAINY_MODE=read-only|write-only
|
||||
```
|
||||
|
||||
## Monitoring and Operations
|
||||
|
||||
### Key Metrics to Track
|
||||
1. **Search Instance**
|
||||
- Query latency (p50, p95, p99)
|
||||
- Cache hit ratio
|
||||
- Concurrent searches
|
||||
- S3 GET requests/sec
|
||||
|
||||
2. **Write Instances**
|
||||
- Ingestion rate (vectors/sec)
|
||||
- Batch size and latency
|
||||
- S3 PUT requests/sec
|
||||
- Partition distribution
|
||||
|
||||
3. **System-Wide**
|
||||
- Total vector count
|
||||
- Partition count and size distribution
|
||||
- S3 storage usage and costs
|
||||
- Cross-instance consistency lag
|
||||
|
||||
### Health Checks
|
||||
```typescript
|
||||
app.get('/health', async (req, res) => {
|
||||
const health = {
|
||||
instance: process.env.BRAINY_INSTANCE_ID,
|
||||
role: process.env.BRAINY_ROLE,
|
||||
status: 'healthy',
|
||||
checks: {
|
||||
s3_connectivity: await checkS3(),
|
||||
config_loaded: await checkConfig(),
|
||||
partition_access: await checkPartitions(),
|
||||
memory_usage: process.memoryUsage()
|
||||
}
|
||||
};
|
||||
|
||||
res.json(health);
|
||||
});
|
||||
```
|
||||
|
||||
## Cost Optimization
|
||||
|
||||
1. **S3 Intelligent Tiering**: Automatically move cold partitions to cheaper storage classes
|
||||
2. **Request Batching**: Minimize S3 API calls through batching
|
||||
3. **CDN for Reads**: Use Cloud CDN for frequently accessed partitions
|
||||
4. **Lifecycle Policies**: Auto-delete old snapshots and temporary data
|
||||
5. **Reserved Capacity**: Use committed use discounts for Cloud Run
|
||||
|
||||
## Implementation Timeline
|
||||
|
||||
### Phase 1: Basic Setup (Week 1)
|
||||
- Deploy 3 instances with shared S3 bucket
|
||||
- Implement basic configuration synchronization
|
||||
- Set up monitoring
|
||||
|
||||
### Phase 2: Optimization (Week 2-3)
|
||||
- Implement partition coordinator
|
||||
- Add caching layers
|
||||
- Optimize S3 operations
|
||||
|
||||
### Phase 3: Advanced Features (Week 4+)
|
||||
- Add WebSocket-based coordination
|
||||
- Implement consistency checks
|
||||
- Add auto-scaling based on load
|
||||
|
||||
## Conclusion
|
||||
|
||||
This architecture provides a scalable, distributed Brainy deployment that can handle millions of vectors with specialized instances for different workloads. The key is maintaining consistency through shared configuration and coordination while optimizing each instance for its specific role.
|
||||
Loading…
Add table
Add a link
Reference in a new issue