leader discovery fix and code cleaning

This commit is contained in:
Pierre
2026-02-01 13:39:32 +01:00
parent d1e74e0b70
commit 9a60f0e231
14 changed files with 2095 additions and 120 deletions
+1 -1
View File
@@ -39,7 +39,7 @@ func main() {
}) })
election.OnBecomeFollower(func(info *leader.LeaderLockInfo) { election.OnBecomeFollower(func(info *leader.LeaderLockInfo) {
log.Printf("[meta-core] Became FOLLOWER - Leader at %s", info.API) log.Printf("[meta-core] Became FOLLOWER - Leader at %s (Redis: %s)", info.BaseUrl, info.RedisUrl)
}) })
election.OnLeaderLost(func() { election.OnLeaderLost(func() {
+105 -11
View File
@@ -10,6 +10,10 @@ interface Mount {
mounted: boolean; mounted: boolean;
error?: string; error?: string;
lastChecked: number; lastChecked: number;
pollingEnabled?: boolean;
pollingIntervalMs?: number;
pollingActive?: boolean;
lastPolledScan?: number;
} }
interface RcloneRemote { interface RcloneRemote {
@@ -55,7 +59,10 @@ export default function Mounts() {
smbPassword: '', smbPassword: '',
rcloneRemote: '', rcloneRemote: '',
rclonePath: '', rclonePath: '',
pollingEnabled: false,
pollingIntervalMs: 30000,
}); });
const [scanning, setScanning] = useState<string | null>(null);
const fetchMounts = async () => { const fetchMounts = async () => {
try { try {
@@ -106,6 +113,29 @@ export default function Mounts() {
fetchMounts(); fetchMounts();
}; };
const handleScan = async (id: string) => {
setScanning(id);
try {
await fetch(`/api/mounts/${id}/scan`, { method: 'POST' });
fetchMounts();
} finally {
setScanning(null);
}
};
const handleUpdatePolling = async (id: string, pollingEnabled: boolean, pollingIntervalMs?: number) => {
const body: Record<string, unknown> = { pollingEnabled };
if (pollingIntervalMs !== undefined) {
body.pollingIntervalMs = pollingIntervalMs;
}
await fetch(`/api/mounts/${id}`, {
method: 'PATCH',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body),
});
fetchMounts();
};
const handleCreate = async (e: React.FormEvent) => { const handleCreate = async (e: React.FormEvent) => {
e.preventDefault(); e.preventDefault();
const body: Record<string, unknown> = { const body: Record<string, unknown> = {
@@ -126,6 +156,12 @@ export default function Mounts() {
body.rclonePath = formData.rclonePath; body.rclonePath = formData.rclonePath;
} }
// Add polling configuration
body.pollingEnabled = formData.pollingEnabled;
if (formData.pollingEnabled && formData.pollingIntervalMs >= 5000) {
body.pollingIntervalMs = formData.pollingIntervalMs;
}
await fetch('/api/mounts', { await fetch('/api/mounts', {
method: 'POST', method: 'POST',
headers: { 'Content-Type': 'application/json' }, headers: { 'Content-Type': 'application/json' },
@@ -144,6 +180,8 @@ export default function Mounts() {
smbPassword: '', smbPassword: '',
rcloneRemote: '', rcloneRemote: '',
rclonePath: '', rclonePath: '',
pollingEnabled: false,
pollingIntervalMs: 30000,
}); });
fetchMounts(); fetchMounts();
}; };
@@ -275,6 +313,28 @@ export default function Mounts() {
</> </>
)} )}
<div style={{ marginTop: '1rem', borderTop: '1px solid #0f3460', paddingTop: '1rem' }}>
<label style={{ display: 'flex', alignItems: 'center', marginBottom: '0.5rem', cursor: 'pointer' }}>
<input
type="checkbox"
checked={formData.pollingEnabled}
onChange={(e) => setFormData({ ...formData, pollingEnabled: e.target.checked })}
style={{ marginRight: '0.5rem' }}
/>
Enable Polling (auto-scan on interval)
</label>
{formData.pollingEnabled && (
<input
style={inputStyle}
type="number"
min="5000"
placeholder="Polling Interval (ms)"
value={formData.pollingIntervalMs}
onChange={(e) => setFormData({ ...formData, pollingIntervalMs: parseInt(e.target.value) || 30000 })}
/>
)}
</div>
<button type="submit" style={buttonStyle}> <button type="submit" style={buttonStyle}>
Create Mount Create Mount
</button> </button>
@@ -301,23 +361,57 @@ export default function Mounts() {
{mount.mounted ? 'Mounted' : 'Not mounted'} {mount.mounted ? 'Mounted' : 'Not mounted'}
</span> </span>
</p> </p>
{mount.pollingEnabled && (
<p style={{ color: '#888', fontSize: '0.9rem' }}>
Polling:{' '}
<span style={{ color: mount.pollingActive ? '#4ade80' : '#f59e0b' }}>
{mount.pollingActive ? 'Active' : 'Inactive'}
</span>
{' | Interval: '}{(mount.pollingIntervalMs || 30000) / 1000}s
{mount.lastPolledScan && (
<> | Last scan: {new Date(mount.lastPolledScan).toLocaleTimeString()}</>
)}
</p>
)}
{mount.error && ( {mount.error && (
<p style={{ color: '#f87171', marginTop: '0.5rem' }}>Error: {mount.error}</p> <p style={{ color: '#f87171', marginTop: '0.5rem' }}>Error: {mount.error}</p>
)} )}
</div> </div>
<div> <div style={{ display: 'flex', flexDirection: 'column', gap: '0.5rem' }}>
{mount.mounted ? ( <div>
<button style={buttonStyle} onClick={() => handleUnmount(mount.id)}> {mount.mounted ? (
Unmount <button style={buttonStyle} onClick={() => handleUnmount(mount.id)}>
</button> Unmount
) : ( </button>
<button style={buttonStyle} onClick={() => handleMount(mount.id)}> ) : (
Mount <button style={buttonStyle} onClick={() => handleMount(mount.id)}>
Mount
</button>
)}
<button style={dangerButtonStyle} onClick={() => handleDelete(mount.id)}>
Delete
</button> </button>
</div>
{mount.mounted && (
<div style={{ display: 'flex', gap: '0.5rem', alignItems: 'center' }}>
<button
style={{ ...buttonStyle, marginRight: 0 }}
onClick={() => handleScan(mount.id)}
disabled={scanning === mount.id}
>
{scanning === mount.id ? 'Scanning...' : 'Scan Now'}
</button>
<label style={{ display: 'flex', alignItems: 'center', fontSize: '0.85rem', cursor: 'pointer' }}>
<input
type="checkbox"
checked={mount.pollingEnabled || false}
onChange={(e) => handleUpdatePolling(mount.id, e.target.checked, mount.pollingIntervalMs)}
style={{ marginRight: '0.25rem' }}
/>
Poll
</label>
</div>
)} )}
<button style={dangerButtonStyle} onClick={() => handleDelete(mount.id)}>
Delete
</button>
</div> </div>
</div> </div>
</div> </div>
+5 -2
View File
@@ -51,10 +51,13 @@ init_dirs() {
fi fi
} }
# Check if path is mounted # Check if path is mounted (uses /proc/mounts for Alpine compatibility)
is_mounted() { is_mounted() {
local mount_path="$1" local mount_path="$1"
findmnt -n "$mount_path" > /dev/null 2>&1 # Remove trailing slash for consistent matching
local clean_path="${mount_path%/}"
# Check if the path appears as a mount point in /proc/mounts (field 2)
grep -q " ${clean_path} " /proc/mounts 2>/dev/null
} }
# Clear error file for mount # Clear error file for mount
+854
View File
@@ -0,0 +1,854 @@
# Meta-Core Architecture
## Document Purpose
This document describes the architectural design of **meta-core**, the unified data and metadata access layer for MetaMesh. Meta-core is a language-agnostic sidecar binary that runs alongside each service container, providing KV storage, leader election, service discovery, and file access APIs. This document covers the sidecar pattern, leader election mechanism, API design, data model, and integration patterns.
---
## System Overview
MetaMesh consists of three services that require shared access to metadata and files:
| Service | Language | Role |
|---------|----------|------|
| **meta-sort** | TypeScript | Watches folders, extracts metadata, writes to KV |
| **meta-fuse** | TypeScript/Rust | Serves virtual filesystem from KV metadata |
| **meta-stremio** | Python | Stremio addon for HLS streaming |
### Problem Statement
Without meta-core, each service implements its own:
- Leader election logic (616 lines TS + 301 lines Python + 277 lines TS)
- Redis client configuration
- Service discovery mechanism
- File path resolution
This leads to:
1. **Code duplication** across languages
2. **Inconsistent behavior** between implementations
3. **Tight coupling** to Redis internals
4. **Complex maintenance** when adding new services
### Core Design Principle
**Provide a single, language-agnostic interface for all data and metadata operations, abstracting storage implementation from service logic.**
This design enables:
- **Language independence**: Any service language can use HTTP API
- **Single source of truth**: One leader manages all storage
- **Clean abstraction**: Services don't know about Redis, flock, or file paths
- **Operational simplicity**: One binary to deploy, monitor, and debug
---
## Architecture Design
### Sidecar Pattern
Meta-core runs as a **sidecar process** in each container, alongside the main service:
```
┌─────────────────────────────────────────────────────────────────┐
│ Container (meta-sort / meta-fuse / meta-stremio) │
│ │
│ ┌──────────────────┐ HTTP API ┌──────────────────────┐│
│ │ Main Service │◄────────────────►│ meta-core ││
│ │ │ localhost:9000 │ (Go binary) ││
│ │ - TypeScript │ │ ││
│ │ - Python │ │ - Leader election ││
│ │ - Any language │ │ - Redis management ││
│ │ │ │ - Service discovery ││
│ └──────────────────┘ │ - File access API ││
│ └──────────┬───────────┘│
│ │ │
└───────────────────────────────────────────────────┼─────────────┘
│
┌───────────────────────────────┼───────────────┐
│ /meta-core (shared volume) │
│ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ locks/ │ │
│ │ kv-leader.lock (flock marker) │ │
│ │ kv-leader.info (leader JSON) │ │
│ │ db/ │ │
│ │ redis/ (RDB + AOF) │ │
│ │ services/ │ │
│ │ meta-sort.json (service registry) │ │
│ │ meta-fuse.json │ │
│ │ meta-stremio.json │ │
│ └─────────────────────────────────────────┘ │
└──────────────────────────────────────────────┘
┌──────────────────────────────────────────────┐
│ /files (shared volume) │
│ │
│ media/ │
│ movies/ │
│ Movie.2024.mkv │
│ series/ │
│ Show.S01E01.mkv │
└──────────────────────────────────────────────┘
```
### Multi-Container Topology
Across all containers, exactly **one meta-core instance** is the leader:
```
┌─────────────────────┐ ┌─────────────────────┐ ┌─────────────────────┐
│ meta-sort │ │ meta-fuse │ │ meta-stremio │
│ container │ │ container │ │ container │
│ │ │ │ │ │
│ ┌───────────────┐ │ │ ┌───────────────┐ │ │ ┌───────────────┐ │
│ │ meta-core │ │ │ │ meta-core │ │ │ │ meta-core │ │
│ │ ★ LEADER │ │ │ │ follower │ │ │ │ follower │ │
│ │ │ │ │ │ │ │ │ │ │ │
│ │ [Redis]──────┼──┼──┼──┼───────────────┼──┼──┼──┼───────────────│ │
│ │ │ │ │ │ │ │ │ │ │ │
│ └───────────────┘ │ │ └───────────────┘ │ │ └───────────────┘ │
│ ▲ │ │ │ │ │ │ │
│ │ HTTP │ │ │ HTTP │ │ │ HTTP │
│ ▼ │ │ ▼ │ │ ▼ │
│ ┌───────────────┐ │ │ ┌───────────────┐ │ │ ┌───────────────┐ │
│ │ meta-sort │ │ │ │ meta-fuse │ │ │ │ meta-stremio │ │
│ │ service │ │ │ │ service │ │ │ │ service │ │
│ └───────────────┘ │ │ └───────────────┘ │ │ └───────────────┘ │
└─────────────────────┘ └─────────────────────┘ └─────────────────────┘
│ │ │
└────────────────────────┼────────────────────────┘
▼
┌──────────────────────────────┐
│ /meta-core (shared volume) │
│ /files (shared volume) │
└──────────────────────────────┘
```
---
## Leader Election
### Mechanism: flock-based Consensus
Meta-core uses **POSIX file locking (flock)** on a shared filesystem for leader election. This approach requires no external consensus system (etcd, Consul, Zookeeper).
```
Leader Election Flow:
meta-core startup
│
▼
┌──────────────────┐
│ Try flock() on │
│ kv-leader.lock │
│ (non-blocking) │
└────────┬─────────┘
│
┌───────┴───────┐
│ │
acquired blocked
│ │
▼ ▼
┌─────────┐ ┌──────────┐
│ LEADER │ │ FOLLOWER │
└────┬────┘ └────┬─────┘
│ │
▼ ▼
┌──────────┐ ┌────────────┐
│ Spawn │ │ Read │
│ Redis │ │ leader.info│
│ │ │ │
│ Write │ │ Connect to │
│ leader │ │ leader │
│ .info │ │ Redis │
└──────────┘ └────────────┘
```
### Lock File Structure
**`/meta-core/locks/kv-leader.lock`**
- Binary flock marker (empty file)
- Held exclusively by leader process
- Automatically released on process death (kernel handles cleanup)
**`/meta-core/locks/kv-leader.info`**
```json
{
"hostname": "meta-sort-dev",
"redis_url": "redis://10.0.1.50:6379",
"http_url": "http://10.0.1.50:9000",
"started_at": "2024-01-15T10:30:00Z",
"pid": 12345,
"version": "1.0.0"
}
```
### Failover Behavior
When the leader process dies:
1. **Kernel releases flock** automatically (no stale lock possible)
2. **Followers detect** via:
- File watch on `kv-leader.info`
- Health check failure to leader HTTP endpoint
3. **First follower acquires lock** becomes new leader
4. **New leader spawns Redis** using existing data in `/meta-core/db/redis/`
5. **Other followers reconnect** to new leader
```
Timeline: Leader Failure and Recovery
t=0 t=1 t=2 t=3 t=4
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
Leader Leader Flock Follower New leader
running crashes released acquires spawns Redis
flock & serves
```
### Leader Responsibilities
The leader meta-core instance:
1. **Spawns Redis** as a child process on port 6379
2. **Writes leader.info** with connection details
3. **Accepts all write operations** (metadata updates)
4. **Serves read operations** (metadata queries, file paths)
5. **Manages service registry** (heartbeats, stale detection)
### Follower Responsibilities
Follower meta-core instances:
1. **Monitor leader.info** for leader changes
2. **Proxy requests to leader** for writes
3. **Cache metadata locally** for fast reads (optional)
4. **Attempt leader acquisition** on leader failure
---
## Data Model
### Two-Concept Model
Meta-core manages two distinct but related concepts:
```
┌─────────────────────────────────────────────────────────────────┐
│ METADATA (KV Store) │
│ │
│ Key: midhash256 CID │
│ Value: { │
│ "title": "Movie Name", │
│ "year": 2024, │
│ "path": "movies/Movie.2024.mkv", ◄── relative to /files │
│ "size": 15032385536, │
│ "duration": 7200, │
│ "codec": "hevc", │
│ ... │
│ } │
│ │
│ ✓ Can exist WITHOUT local file (remote/deleted) │
│ ✓ Searchable by any field │
│ ✓ Stored in Redis hashes │
└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐
│ DATA (Files) │
│ │
│ Location: /files volume (shared across containers) │
│ Access: Via path from metadata │
│ │
│ ✓ Always has metadata (at minimum: hash, path, title) │
│ ✓ Accessed by path or streamed via API │
│ ✓ Never duplicated or moved │
└─────────────────────────────────────────────────────────────────┘
```
### Path Resolution
File paths in metadata are **relative to the /files volume**:
```
Metadata path: "movies/Movie.2024.mkv"
Container path: "/files/movies/Movie.2024.mkv"
Host path: "${DATA_WATCH_PATH}/movies/Movie.2024.mkv"
```
This enables:
- **Multi-host deployments**: Same metadata works across machines
- **Volume remapping**: Change mount points without updating metadata
- **Portable backups**: Metadata export/import works across systems
---
## API Design
### Endpoint Overview
Meta-core exposes an HTTP API on `localhost:9000` (configurable):
```
Health & Status
GET /health → Health check and role info
GET /status → Detailed status and metrics
Metadata Operations
GET /meta/{hash} → Get metadata by hash
PUT /meta/{hash} → Create/update metadata
DELETE /meta/{hash} → Delete metadata
GET /meta?search=... → Search metadata
Data Operations
GET /data/{hash}/path → Get file path
GET /data/{hash}/stream → Stream file content
HEAD /data/{hash} → Check if file exists
Service Discovery
GET /services → List registered services
POST /services/{name} → Register/heartbeat service
DELETE /services/{name} → Deregister service
```
### Health Endpoint
```
GET /health
Response:
{
"status": "healthy",
"role": "leader", // or "follower"
"leader": {
"hostname": "meta-sort-dev",
"http_url": "http://10.0.1.50:9000",
"redis_url": "redis://10.0.1.50:6379"
},
"uptime_seconds": 3600,
"version": "1.0.0"
}
```
### Metadata CRUD
**Get Metadata**
```
GET /meta/bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku
Response:
{
"hash": "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku",
"path": "movies/Inception.2010.mkv",
"title": "Inception",
"year": 2010,
"duration": 8880,
"size": 15032385536,
...
}
```
**Update Metadata**
```
PUT /meta/bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku
Content-Type: application/json
{
"title": "Inception",
"year": 2010,
"custom.rating": 9.5
}
Response:
{
"success": true,
"hash": "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"
}
```
**Search Metadata**
```
GET /meta?title=inception&year=2010&limit=10&offset=0
Response:
{
"results": [
{ "hash": "bafkrei...", "title": "Inception", ... }
],
"total": 1,
"limit": 10,
"offset": 0
}
```
### Data Access
**Get File Path**
```
GET /data/bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku/path
Response:
{
"path": "/files/movies/Inception.2010.mkv",
"exists": true,
"size": 15032385536
}
```
**Stream File Content**
```
GET /data/bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku/stream
Range: bytes=0-1048575
Response:
HTTP/1.1 206 Partial Content
Content-Type: application/octet-stream
Content-Range: bytes 0-1048575/15032385536
Content-Length: 1048576
[binary data]
```
### Service Discovery
**Register Service**
```
POST /services/meta-fuse
Content-Type: application/json
{
"hostname": "meta-fuse-dev",
"http_url": "http://10.0.1.51:8181",
"version": "2.0.0",
"capabilities": ["vfs", "webdav"]
}
Response:
{
"success": true,
"ttl_seconds": 60
}
```
**List Services**
```
GET /services
Response:
{
"services": [
{
"name": "meta-sort",
"hostname": "meta-sort-dev",
"http_url": "http://10.0.1.50:8180",
"status": "healthy",
"last_heartbeat": "2024-01-15T10:30:00Z"
},
{
"name": "meta-fuse",
"hostname": "meta-fuse-dev",
"http_url": "http://10.0.1.51:8181",
"status": "healthy",
"last_heartbeat": "2024-01-15T10:30:05Z"
}
]
}
```
---
## Implementation
### Technology Choice: Go
Meta-core is implemented in Go for the following reasons:
| Requirement | Go Advantage |
|-------------|--------------|
| Language agnostic | HTTP API, no runtime dependencies |
| Single binary | Static compilation, no interpreter |
| Low resource usage | Small memory footprint (~10-20MB) |
| File locking | stdlib `syscall.Flock` |
| HTTP server | stdlib `net/http` |
| Redis client | go-redis (mature, full-featured) |
| Cross-platform | Compile for Linux, macOS, Windows |
### Binary Structure
```
meta-core
├── cmd/
│ └── meta-core/
│ └── main.go # Entry point
├── internal/
│ ├── leader/
│ │ ├── election.go # flock-based election
│ │ └── redis.go # Redis process management
│ ├── api/
│ │ ├── server.go # HTTP server setup
│ │ ├── health.go # Health endpoints
│ │ ├── meta.go # Metadata endpoints
│ │ ├── data.go # Data endpoints
│ │ └── services.go # Service discovery endpoints
│ ├── storage/
│ │ ├── redis.go # Redis client wrapper
│ │ └── cache.go # Optional local cache
│ └── config/
│ └── config.go # Configuration loading
├── go.mod
└── go.sum
```
### Process Management
Meta-core manages the Redis process as a child:
```go
// Leader spawns Redis
cmd := exec.Command("redis-server",
"--port", "6379",
"--dir", "/meta-core/db/redis",
"--save", "60", "1", // RDB snapshot
"--appendonly", "yes", // AOF persistence
)
cmd.Start()
// Monitor Redis health
go func() {
for {
if err := redisClient.Ping(); err != nil {
log.Error("Redis unhealthy, restarting...")
cmd.Process.Kill()
cmd.Start()
}
time.Sleep(5 * time.Second)
}
}()
```
### Container Integration
Meta-core runs via supervisord alongside the main service:
```ini
# /etc/supervisor/conf.d/meta-core.conf
[program:meta-core]
command=/usr/local/bin/meta-core
directory=/meta-core
autostart=true
autorestart=true
priority=1
startsecs=2
stdout_logfile=/var/log/meta-core.log
stderr_logfile=/var/log/meta-core.err
[program:main-service]
command=/app/start.sh
autostart=true
autorestart=true
priority=10
startsecs=5
```
### Configuration
Environment variables configure meta-core:
```bash
# Core paths
META_CORE_PATH=/meta-core # Shared volume for locks/db
FILES_PATH=/files # Shared volume for media files
# Network
META_CORE_HTTP_PORT=9000 # HTTP API port
META_CORE_REDIS_PORT=6379 # Redis port (leader only)
# Behavior
META_CORE_SERVICE_NAME=meta-sort # Service identity for discovery
META_CORE_LEADER_TIMEOUT=30 # Seconds before leader considered dead
META_CORE_HEARTBEAT_INTERVAL=10 # Service heartbeat frequency
```
---
## Client Integration
### TypeScript Client
Services use a simple HTTP client to interact with meta-core:
```typescript
// meta-core-client.ts
class MetaCoreClient {
private baseUrl = 'http://localhost:9000';
async getMeta(hash: string): Promise<Metadata | null> {
const res = await fetch(`${this.baseUrl}/meta/${hash}`);
if (res.status === 404) return null;
return res.json();
}
async setMeta(hash: string, meta: Partial<Metadata>): Promise<void> {
await fetch(`${this.baseUrl}/meta/${hash}`, {
method: 'PUT',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(meta),
});
}
async getDataPath(hash: string): Promise<string | null> {
const res = await fetch(`${this.baseUrl}/data/${hash}/path`);
if (res.status === 404) return null;
const data = await res.json();
return data.path;
}
async search(query: SearchQuery): Promise<SearchResult> {
const params = new URLSearchParams(query as any);
const res = await fetch(`${this.baseUrl}/meta?${params}`);
return res.json();
}
async isHealthy(): Promise<boolean> {
try {
const res = await fetch(`${this.baseUrl}/health`);
return res.ok;
} catch {
return false;
}
}
}
export const metaCore = new MetaCoreClient();
```
### Python Client
```python
# meta_core_client.py
import requests
from typing import Optional, Dict, Any
class MetaCoreClient:
def __init__(self, base_url: str = "http://localhost:9000"):
self.base_url = base_url
def get_meta(self, hash: str) -> Optional[Dict[str, Any]]:
res = requests.get(f"{self.base_url}/meta/{hash}")
if res.status_code == 404:
return None
return res.json()
def set_meta(self, hash: str, meta: Dict[str, Any]) -> None:
requests.put(
f"{self.base_url}/meta/{hash}",
json=meta
)
def get_data_path(self, hash: str) -> Optional[str]:
res = requests.get(f"{self.base_url}/data/{hash}/path")
if res.status_code == 404:
return None
return res.json()["path"]
def search(self, **query) -> Dict[str, Any]:
res = requests.get(f"{self.base_url}/meta", params=query)
return res.json()
def is_healthy(self) -> bool:
try:
res = requests.get(f"{self.base_url}/health")
return res.ok
except:
return False
meta_core = MetaCoreClient()
```
---
## Search Capabilities
### Current Implementation
Basic search uses Redis SCAN with pattern matching:
```
GET /meta?title=inception
→ SCAN 0 MATCH "file:*" COUNT 100
→ For each key: HGET key title → filter matches
```
This works for small collections but scales poorly.
### Enhanced Search (Future)
For advanced search, meta-core supports Redis Search module:
```
# Create search index
FT.CREATE idx:files ON HASH PREFIX 1 "file:" SCHEMA
title TEXT WEIGHT 5.0
series_title TEXT WEIGHT 3.0
year NUMERIC SORTABLE
media_type TAG
duration NUMERIC SORTABLE
# Search query
FT.SEARCH idx:files "@title:inception @year:[2010 2010]"
```
Search query syntax:
```
GET /meta?q=title:inception AND year:2010
GET /meta?q=media_type:episode AND series_title:"Breaking Bad"
GET /meta?q=duration:[3600 7200] # 1-2 hours
```
---
## Failure Modes
### Leader Crash
1. **Detection**: Followers detect via health check failure (5s timeout)
2. **Election**: Fastest follower acquires flock
3. **Recovery**: New leader spawns Redis with existing data
4. **Reconnection**: Other followers read new leader.info, reconnect
5. **Downtime**: 5-10 seconds typical
### Redis Crash
1. **Detection**: Leader detects ping failure
2. **Recovery**: Leader restarts Redis process
3. **Data**: Recovered from RDB snapshot + AOF replay
4. **Downtime**: 1-2 seconds typical
### Network Partition
1. **Followers isolated**: Continue serving cached reads, writes fail
2. **Leader isolated**: Continues operating, followers elect new leader
3. **Resolution**: On reconnection, followers sync with true leader
4. **Risk**: Brief split-brain if partition during election (mitigated by flock atomicity)
### Shared Volume Unavailable
1. **Impact**: All meta-core instances fail
2. **Behavior**: Services operate in degraded mode (no metadata)
3. **Recovery**: Automatic when volume restored
---
## Observability
### Metrics Endpoint
```
GET /metrics
# Prometheus format
meta_core_role{service="meta-sort"} 1 # 1=leader, 0=follower
meta_core_uptime_seconds{service="meta-sort"} 3600
meta_core_requests_total{endpoint="/meta"} 15420
meta_core_request_duration_seconds{endpoint="/meta",quantile="0.99"} 0.005
meta_core_redis_commands_total 892341
meta_core_files_total 12847
meta_core_metadata_bytes 45678901
```
### Logging
Structured JSON logging:
```json
{
"time": "2024-01-15T10:30:00Z",
"level": "info",
"msg": "Acquired leader lock",
"service": "meta-sort",
"pid": 12345
}
```
Log levels: `debug`, `info`, `warn`, `error`
### Health Probes
For Kubernetes/Docker health checks:
```yaml
# docker-compose.yml
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:9000/health"]
interval: 10s
timeout: 5s
retries: 3
```
---
## Security Considerations
### Network Isolation
- Meta-core listens on `localhost:9000` only (not exposed externally)
- Redis listens on `localhost:6379` only
- Inter-container communication uses Docker network
### File Access
- Meta-core runs as non-root user
- Read-only access to /files volume (except meta-sort)
- No shell access or command execution
### Input Validation
- Hash parameters validated as valid CID format
- Path traversal prevented (paths must be relative, no `..`)
- JSON body size limited (10MB max)
---
## Configuration Reference
### Environment Variables
| Variable | Default | Description |
|----------|---------|-------------|
| `META_CORE_PATH` | `/meta-core` | Path to shared infrastructure volume |
| `FILES_PATH` | `/files` | Path to shared files volume |
| `META_CORE_HTTP_PORT` | `9000` | HTTP API port |
| `META_CORE_REDIS_PORT` | `6379` | Redis port (leader only) |
| `META_CORE_SERVICE_NAME` | hostname | Service identity |
| `META_CORE_LEADER_TIMEOUT` | `30` | Leader health timeout (seconds) |
| `META_CORE_HEARTBEAT_INTERVAL` | `10` | Heartbeat frequency (seconds) |
| `META_CORE_LOG_LEVEL` | `info` | Log verbosity |
| `META_CORE_CACHE_ENABLED` | `false` | Enable local read cache |
| `META_CORE_CACHE_TTL` | `60` | Cache TTL (seconds) |
### File Paths
| Path | Purpose |
|------|---------|
| `/meta-core/locks/kv-leader.lock` | Leader election flock file |
| `/meta-core/locks/kv-leader.info` | Leader connection info |
| `/meta-core/db/redis/dump.rdb` | Redis RDB snapshot |
| `/meta-core/db/redis/appendonly.aof` | Redis AOF log |
| `/meta-core/services/*.json` | Service registry files |
---
## Summary
Meta-core provides a unified, language-agnostic interface for MetaMesh's data and metadata operations. By consolidating leader election, storage management, and service discovery into a single sidecar binary, it:
1. **Eliminates code duplication** across TypeScript and Python services
2. **Simplifies service implementation** to pure business logic
3. **Provides operational visibility** through consistent health/metrics
4. **Enables future evolution** by abstracting storage implementation
The flock-based leader election ensures exactly-one-leader semantics without external dependencies, while the HTTP API enables integration from any programming language.
+410
View File
@@ -0,0 +1,410 @@
# Redis Metadata Storage Architecture
This document describes how MetaMesh stores file metadata in Redis, the key structure, plugin metadata contributions, and special cases like FFmpeg stream handling.
## Overview
MetaMesh uses Redis as its primary metadata storage, storing file metadata as **Redis Hashes** with flattened key-value pairs. This design enables:
- Efficient nested object storage without JSON serialization overhead
- Individual field updates without full document rewrites
- Cross-service access via Redis pub/sub notifications
- Fast VFS lookups using content-based hash IDs
## Key Structure
### Hash-Based Storage
Each file's metadata is stored as a Redis Hash:
```
Hash Key: file:{hashId}
Hash Fields: property/path (e.g., "fileinfo/duration", "titles/eng")
Hash Values: String values
```
**Example Redis commands:**
```bash
# Store file metadata
HMSET file:bafkr4ih5kapbjzq... \
"cid_midhash256" "bafkr4ih5kapbjzq..." \
"filePath" "/files/watch/movie.mkv" \
"fileinfo/duration" "7200.5" \
"fileinfo/formatName" "matroska,webm"
# Retrieve all metadata
HGETALL file:bafkr4ih5kapbjzq...
# Get single field
HGET file:bafkr4ih5kapbjzq... "fileinfo/duration"
```
### File Index
A Redis Set tracks all known file hash IDs:
```
Key: file:__index__
Type: Set
Value: {hashId1, hashId2, hashId3, ...}
```
This enables fast bulk queries like "get all files" without scanning keys.
## Primary Hash ID
The primary identifier for each file is `cid_midhash256` - an IPFS-compatible CID generated from:
1. SHA-256 hash of the middle 1MB of the file
2. File size in bytes
3. IPFS CID v1 encoding (multicodec 0x1000)
**Example:** `bafkr4ih5kapbjzqvylwj7kx7zmkpxn6qj5xw...`
This provides:
- Content-based addressing (same file = same ID)
- Fast computation (only reads 1MB)
- IPFS compatibility for future distributed storage
### Collision Handling
If two different files produce the same midhash (rare), the system falls back to full SHA-256:
```typescript
// Normal case
hashId = midHash256; // "bafkr4ih5kapbjzq..."
// Collision detected (different filePath with same midhash)
hashId = metadata['cid_sha2-256']; // Full hash fallback
```
## Metadata Flattening
Nested objects are flattened into path-based keys for efficient Redis storage.
### Flattening Rules
```typescript
// Input (nested object)
{
title: "Inception",
video: {
codec: "h265",
width: "1920"
},
streams: [
{ language: "eng" },
{ language: "jpn" }
]
}
// Output (flattened for Redis)
title = "Inception"
video/codec = "h265"
video/width = "1920"
streams/0/language = "eng"
streams/1/language = "jpn"
```
**Key rules:**
- Nested objects use `/` as path separator
- Arrays use numeric indices (0, 1, 2...)
- All values stored as strings
- `null`/`undefined` stored as empty strings
- Numbers and booleans converted via `String()`
### Reconstruction
When reading, the flattened structure is reconstructed into nested objects:
```typescript
const metadata = await kvClient.getMetadataFlat(hashId);
// Returns: { title: "Inception", video: { codec: "h265", width: "1920" }, ... }
```
## Plugin Metadata Structure
Each plugin writes metadata using a prefixed key pattern:
```
{plugin-id}/{field} # Simple field
{plugin-id}/{group}/{field} # Grouped field
{plugin-id}/{group}/{index}/{field} # Indexed field (for arrays)
```
### Plugin Contribution Flow
```
File Discovered
↓
[FileProcessorPiscina]
├─ Compute midhash256
├─ Store preliminary metadata
└─ Dispatch plugin tasks
↓
[Plugin Processing]
├─ file-info → fileType, mimeType, sizeByte
├─ ffmpeg → fileinfo/duration, fileinfo/streamdetails/...
├─ filename-parser → titles/eng, season, episode, videoType
├─ anime-detector → anime, titles/jpn, titles/jpl
├─ tmdb → tmdbid, plot/eng, genres
└─ full-hash → cid_sha2-256, cid_sha1
↓
[Plugin Callbacks]
└─ Merge metadata into Redis via HMSET
```
## Core Metadata Types
### HashMeta - Content Identification
```
cid_crc32 = "bafkr..." # CRC32 (multicodec 0x0132)
cid_md5 = "bafkr..." # MD5 (multicodec 0xd5)
cid_sha1 = "bafkr..." # SHA-1 (multicodec 0x11)
cid_sha2-256 = "bafkr..." # SHA-256 (multicodec 0x12)
cid_midhash256 = "bafkr..." # Middle-hash (PRIMARY ID, multicodec 0x1000)
```
### FileStatMeta - File Properties
```
fileType = "video" # video, audio, image, text
mimeType = "video/x-matroska" # MIME type
sizeByte = "4294967296" # File size in bytes
```
### VideoMeta - Classification
```
videoType = "movie" # movie, episode, unknown
originalTitle = "Inception"
titles/eng = "Inception" # Localized titles
titles/jpn = "インセプション"
season = "1"
episode = "5"
movieYear = "2010"
```
### AnimeMeta - Anime Detection
```
anime = "true" # "true" or "false"
titles/jpn = "進撃の巨人" # Japanese title
titles/jpl = "Shingeki no Kyojin" # Romaji
```
### SubtitleMeta - Subtitle References
```
subtitles/eng/0 = "bafkr..." # Reference to subtitle file CID
subtitles/eng/1 = "bafks..."
subtitles/jpn/0 = "bafkt..."
```
### JellyfinMeta - Comprehensive Metadata
```
imdbid = "tt1375666"
tmdbid = "27205"
anidbid = "12345"
plot/eng = "A movie about..."
genre/0 = "Action"
genre/1 = "Sci-Fi"
rating = "8.8"
mpaa = "PG-13"
releasedate = "2010-07-16"
art/poster = "bafkr..." # Reference to poster image CID
art/fanart = "bafks..." # Reference to backdrop image CID
```
## FFmpeg Plugin - Special Case
The FFmpeg plugin extracts detailed stream information with a specific nested structure.
### Stream Details Structure
```
fileinfo/duration = "7200.5"
fileinfo/formatName = "matroska,webm"
fileinfo/streamdetails/video/0/codec = "h264"
fileinfo/streamdetails/video/0/width = "1920"
fileinfo/streamdetails/video/0/height = "1080"
fileinfo/streamdetails/video/0/aspect = "16:9"
fileinfo/streamdetails/video/0/bitrate = "8000000"
fileinfo/streamdetails/audio/0/codec = "aac"
fileinfo/streamdetails/audio/0/language = "eng"
fileinfo/streamdetails/audio/0/channels = "6"
fileinfo/streamdetails/audio/1/codec = "dts"
fileinfo/streamdetails/audio/1/language = "fra"
fileinfo/streamdetails/audio/1/channels = "6"
fileinfo/streamdetails/subtitle/0/language = "eng"
fileinfo/streamdetails/subtitle/0/format = "subrip"
fileinfo/streamdetails/subtitle/1/language = "jpn"
fileinfo/streamdetails/subtitle/1/format = "ass"
fileinfo/streamdetails/embeddedimage/0/type = "cover"
```
### Stream Type Keys
| Stream Type | Key Pattern | Description |
|------------|-------------|-------------|
| Video | `fileinfo/streamdetails/video/{n}/...` | Video streams (0, 1, 2...) |
| Audio | `fileinfo/streamdetails/audio/{n}/...` | Audio tracks |
| Subtitle | `fileinfo/streamdetails/subtitle/{n}/...` | Embedded subtitles |
| Embedded Image | `fileinfo/streamdetails/embeddedimage/{n}/...` | Cover art, thumbnails |
### Common Stream Fields
**Video streams:**
- `codec` - h264, hevc, av1, vp9, etc.
- `width`, `height` - Resolution
- `aspect` - Aspect ratio
- `bitrate` - Video bitrate
- `profile` - Codec profile (main, high, etc.)
- `level` - Codec level
**Audio streams:**
- `codec` - aac, dts, ac3, flac, etc.
- `language` - ISO 639-2 code (eng, jpn, fra)
- `channels` - Channel count (2, 6, 8)
- `bitrate` - Audio bitrate
- `samplerate` - Sample rate
**Subtitle streams:**
- `language` - ISO 639-2 code
- `format` - subrip, ass, pgs, vobsub
## Redis Pub/Sub Notifications
MetaMesh uses Redis pub/sub to notify other services of metadata changes.
### Channels
```
meta-sort:file:batch # Batch file updates (every 5 seconds)
meta-sort:scan:reset # Full rescan notification
```
### Batch Update Message Format
```json
{
"timestamp": 1704067200000,
"changes": [
{ "action": "add", "hashId": "bafkr4ih5..." },
{ "action": "update", "hashId": "bafkr4ih5..." },
{ "action": "remove", "hashId": "bafkr4ih5..." }
]
}
```
This enables meta-fuse to update its VFS in near real-time without polling.
## Complete Example
Here's a complete metadata example for a movie file:
```bash
# Query all metadata
redis-cli HGETALL file:bafkr4ih5kapbjzq...
# Result (reconstructed as nested object)
{
# Content identification
"cid_midhash256": "bafkr4ih5kapbjzq...",
"cid_sha2-256": "bafks7ij2kbpck...",
"filePath": "/files/watch/movies/Inception.2010.1080p.BluRay.x264.mkv",
# File properties
"fileType": "video",
"mimeType": "video/x-matroska",
"sizeByte": "4294967296",
# Classification
"videoType": "movie",
"originalTitle": "Inception",
"movieYear": "2010",
# Localized titles
"titles": {
"eng": "Inception",
"fra": "Inception",
"jpn": "インセプション"
},
# FFmpeg stream details
"fileinfo": {
"duration": "8878.5",
"formatName": "matroska,webm",
"streamdetails": {
"video": {
"0": {
"codec": "h264",
"width": "1920",
"height": "1080",
"aspect": "16:9",
"bitrate": "8000000"
}
},
"audio": {
"0": { "codec": "dts", "language": "eng", "channels": "6" },
"1": { "codec": "ac3", "language": "fra", "channels": "6" }
},
"subtitle": {
"0": { "language": "eng", "format": "subrip" },
"1": { "language": "fra", "format": "subrip" }
}
}
},
# External metadata (TMDB)
"tmdbid": "27205",
"imdbid": "tt1375666",
"plot": {
"eng": "A thief who steals corporate secrets through dream-sharing technology..."
},
"genre": ["Action", "Science Fiction", "Adventure"],
"rating": "8.4",
"releasedate": "2010-07-16",
# Media assets
"art": {
"poster": "bafkr4poster...",
"fanart": "bafkr4fanart..."
},
# Detection flags
"anime": "false"
}
```
## Key Implementation Files
| Component | File Path |
|-----------|-----------|
| Redis Client | `packages/meta-sort/packages/meta-sort-core/src/kv/RedisClient.ts` |
| Metadata Utils | `packages/meta-sort/packages/meta-sort-core/src/kv/MetadataUtils.ts` |
| KV Interface | `packages/meta-sort/packages/meta-sort-core/src/kv/IKVClient.ts` |
| File Processor | `packages/meta-sort/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts` |
| Pub/Sub Pipeline | `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts` |
| Type Definitions | `packages/meta-interface/src/lib/metadata-type/*.ts` |
## Debugging
```bash
# List all file keys
docker exec meta-sort-dev redis-cli KEYS 'file:*' | head -20
# Get all metadata for a file
docker exec meta-sort-dev redis-cli HGETALL 'file:bafkr4ih5...'
# Get specific field
docker exec meta-sort-dev redis-cli HGET 'file:bafkr4ih5...' 'fileinfo/duration'
# Count total files
docker exec meta-sort-dev redis-cli SCARD 'file:__index__'
# Monitor metadata changes in real-time
docker exec meta-sort-dev redis-cli SUBSCRIBE 'meta-sort:file:batch'
```
+31
View File
@@ -134,6 +134,37 @@ func (s *Server) handleRole(w http.ResponseWriter, r *http.Request) {
}) })
} }
// URLsResponse is the response for /urls
type URLsResponse struct {
Hostname string `json:"hostname"`
BaseUrl string `json:"baseUrl"`
ApiUrl string `json:"apiUrl"`
RedisUrl string `json:"redisUrl"`
WebdavUrl string `json:"webdavUrl"`
IsLeader bool `json:"isLeader"`
}
// handleURLs handles GET /urls
// Returns URLs for the current leader (or this instance if leader)
func (s *Server) handleURLs(w http.ResponseWriter, r *http.Request) {
info := s.election.LeaderInfo()
if info == nil {
writeError(w, http.StatusServiceUnavailable, "no leader available")
return
}
response := URLsResponse{
Hostname: info.Hostname,
BaseUrl: info.BaseUrl,
ApiUrl: info.ApiUrl,
RedisUrl: info.RedisUrl,
WebdavUrl: info.WebdavUrl,
IsLeader: s.election.IsLeader(),
}
writeJSON(w, http.StatusOK, response)
}
// handleGetMeta handles GET /meta/{hash} // handleGetMeta handles GET /meta/{hash}
func (s *Server) handleGetMeta(w http.ResponseWriter, r *http.Request) { func (s *Server) handleGetMeta(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r) vars := mux.Vars(r)
+43 -20
View File
@@ -18,17 +18,18 @@ import (
// Server is the HTTP API server for meta-core // Server is the HTTP API server for meta-core
type Server struct { type Server struct {
config *config.Config config *config.Config
election *leader.Election election *leader.Election
discovery *discovery.Service discovery *discovery.Service
storage *storage.Client storage *storage.Client
mountsManager *mounts.Manager mountsManager *mounts.Manager
mountsHandlers *mounts.Handlers mountsHandlers *mounts.Handlers
mountPoller *mounts.Poller
watcherDispatcher *watcher.Dispatcher watcherDispatcher *watcher.Dispatcher
fileWatcher *watcher.Watcher fileWatcher *watcher.Watcher
watcherHandlers *watcher.Handlers watcherHandlers *watcher.Handlers
router *mux.Router router *mux.Router
server *http.Server server *http.Server
} }
// NewServer creates a new API server // NewServer creates a new API server
@@ -46,16 +47,8 @@ func NewServer(
router: mux.NewRouter(), router: mux.NewRouter(),
} }
// Initialize mounts manager // Initialize file watcher (if enabled) - before mounts so we can pass scanFunc
mountsManager, err := mounts.NewManager(cfg) var scanFunc mounts.ScanFunc
if err != nil {
log.Printf("[API] Warning: failed to initialize mounts manager: %v", err)
} else {
s.mountsManager = mountsManager
s.mountsHandlers = mounts.NewHandlers(mountsManager)
}
// Initialize file watcher (if enabled)
if cfg.EnableFileWatcher && len(cfg.WatchFolderList) > 0 { if cfg.EnableFileWatcher && len(cfg.WatchFolderList) > 0 {
s.watcherDispatcher = watcher.NewDispatcher() s.watcherDispatcher = watcher.NewDispatcher()
fileWatcher, err := watcher.NewWatcher(cfg, s.watcherDispatcher) fileWatcher, err := watcher.NewWatcher(cfg, s.watcherDispatcher)
@@ -64,9 +57,24 @@ func NewServer(
} else { } else {
s.fileWatcher = fileWatcher s.fileWatcher = fileWatcher
s.watcherHandlers = watcher.NewHandlers(fileWatcher, s.watcherDispatcher) s.watcherHandlers = watcher.NewHandlers(fileWatcher, s.watcherDispatcher)
// Create scan function from watcher
scanFunc = fileWatcher.ScanMountPath
} }
} }
// Initialize mounts manager
mountsManager, err := mounts.NewManager(cfg)
if err != nil {
log.Printf("[API] Warning: failed to initialize mounts manager: %v", err)
} else {
s.mountsManager = mountsManager
s.mountsHandlers = mounts.NewHandlers(mountsManager, scanFunc)
// Create and configure mount poller
s.mountPoller = mounts.NewPoller(mountsManager, scanFunc)
mountsManager.SetPoller(s.mountPoller)
}
s.setupRoutes() s.setupRoutes()
return s return s
} }
@@ -78,6 +86,7 @@ func (s *Server) setupRoutes() {
s.router.HandleFunc("/status", s.handleStatus).Methods("GET") s.router.HandleFunc("/status", s.handleStatus).Methods("GET")
s.router.HandleFunc("/leader", s.handleLeader).Methods("GET") s.router.HandleFunc("/leader", s.handleLeader).Methods("GET")
s.router.HandleFunc("/role", s.handleRole).Methods("GET") s.router.HandleFunc("/role", s.handleRole).Methods("GET")
s.router.HandleFunc("/urls", s.handleURLs).Methods("GET")
// Metadata Editor API routes (must be before /meta/{hash} routes) // Metadata Editor API routes (must be before /meta/{hash} routes)
s.router.HandleFunc("/api/metadata/hash-ids", s.handleGetHashIds).Methods("GET") s.router.HandleFunc("/api/metadata/hash-ids", s.handleGetHashIds).Methods("GET")
@@ -170,11 +179,25 @@ func (s *Server) Start() error {
} }
} }
// Start mount poller (if initialized)
if s.mountPoller != nil {
if err := s.mountPoller.Start(); err != nil {
log.Printf("[API] Warning: failed to start mount poller: %v", err)
}
}
return nil return nil
} }
// Stop gracefully stops the HTTP server // Stop gracefully stops the HTTP server
func (s *Server) Stop() error { func (s *Server) Stop() error {
// Stop mount poller
if s.mountPoller != nil {
if err := s.mountPoller.Stop(); err != nil {
log.Printf("[API] Warning: failed to stop mount poller: %v", err)
}
}
// Stop file watcher // Stop file watcher
if s.fileWatcher != nil { if s.fileWatcher != nil {
if err := s.fileWatcher.Stop(); err != nil { if err := s.fileWatcher.Stop(); err != nil {
+56 -55
View File
@@ -13,18 +13,14 @@ import (
"github.com/metazla/meta-core/internal/config" "github.com/metazla/meta-core/internal/config"
) )
// ServiceInfo matches the TypeScript ServiceInfo interface // ServiceInfo contains simplified service registration data
// URLs are obtained via the /urls API endpoint
type ServiceInfo struct { type ServiceInfo struct {
Name string `json:"name"` Name string `json:"name"`
Version string `json:"version"` Hostname string `json:"hostname"`
API string `json:"api"` BaseUrl string `json:"baseUrl"`
Status string `json:"status"` Status string `json:"status"`
PID int `json:"pid"` LastHeartbeat string `json:"lastHeartbeat"`
Hostname string `json:"hostname"`
StartedAt string `json:"startedAt"`
LastHeartbeat string `json:"lastHeartbeat"`
Capabilities []string `json:"capabilities"`
Endpoints map[string]string `json:"endpoints"`
} }
// Service handles service registration and discovery // Service handles service registration and discovery
@@ -41,10 +37,11 @@ type Service struct {
// NewService creates a new service discovery instance // NewService creates a new service discovery instance
func NewService(cfg *config.Config) *Service { func NewService(cfg *config.Config) *Service {
hostname, _ := os.Hostname()
return &Service{ return &Service{
config: cfg, config: cfg,
servicesDir: cfg.ServicesDir(), servicesDir: cfg.ServicesDir(),
serviceFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+".json"), serviceFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+"-"+hostname+".json"),
stopChan: make(chan struct{}), stopChan: make(chan struct{}),
} }
} }
@@ -90,40 +87,20 @@ func (s *Service) buildServiceInfo() *ServiceInfo {
hostname, _ := os.Hostname() hostname, _ := os.Hostname()
ip := getLocalIP() ip := getLocalIP()
// Use BASE_URL for external access (dashboard navigation), fall back to internal IP // Use BASE_URL for external access, fall back to internal IP
var apiBase string var baseUrl string
if s.config.BaseURL != "" { if s.config.BaseURL != "" {
apiBase = s.config.BaseURL baseUrl = s.config.BaseURL
} else { } else {
apiBase = fmt.Sprintf("http://%s:%d", ip, s.config.APIPort) baseUrl = fmt.Sprintf("http://%s:%d", ip, s.config.APIPort)
} }
// Internal URLs always use container IP for container-to-container communication
internalBase := fmt.Sprintf("http://%s:%d", ip, s.config.APIPort)
metaCoreBase := fmt.Sprintf("http://%s:%d", ip, s.config.HTTPPort)
return &ServiceInfo{ return &ServiceInfo{
Name: s.config.ServiceName, Name: s.config.ServiceName,
Version: s.config.ServiceVersion,
API: apiBase,
Status: "running",
PID: os.Getpid(),
Hostname: hostname, Hostname: hostname,
StartedAt: time.Now().UTC().Format(time.RFC3339), BaseUrl: baseUrl,
Status: "running",
LastHeartbeat: time.Now().UTC().Format(time.RFC3339), LastHeartbeat: time.Now().UTC().Format(time.RFC3339),
Capabilities: []string{"meta-core"},
Endpoints: map[string]string{
// meta-core sidecar endpoints (port 9000) - always internal
"health": metaCoreBase + "/health",
"meta": metaCoreBase + "/meta",
"leader": metaCoreBase + "/leader",
"services": metaCoreBase + "/services",
// main service endpoints - external for dashboard
"api": apiBase + "/api",
"webdav": apiBase + "/webdav",
// callback uses internal IP for container-to-container plugin communication
"callback": internalBase + "/api/plugins/callback",
},
} }
} }
@@ -198,23 +175,44 @@ func (s *Service) UpdateStatus(status string) error {
} }
// Discover finds a service by name // Discover finds a service by name
// Looks for files matching pattern: name-*.json (hostname-based naming)
func (s *Service) Discover(name string) (*ServiceInfo, error) { func (s *Service) Discover(name string) (*ServiceInfo, error) {
filePath := filepath.Join(s.servicesDir, name+".json") // First try exact match for backward compatibility
exactPath := filepath.Join(s.servicesDir, name+".json")
data, err := os.ReadFile(filePath) if data, err := os.ReadFile(exactPath); err == nil {
if err != nil { var info ServiceInfo
if os.IsNotExist(err) { if err := json.Unmarshal(data, &info); err == nil {
return nil, nil return s.checkStale(&info), nil
} }
}
// Search for hostname-based files: name-*.json
pattern := filepath.Join(s.servicesDir, name+"-*.json")
matches, err := filepath.Glob(pattern)
if err != nil {
return nil, err return nil, err
} }
var info ServiceInfo // Return the first valid service found
if err := json.Unmarshal(data, &info); err != nil { for _, match := range matches {
return nil, err data, err := os.ReadFile(match)
if err != nil {
continue
}
var info ServiceInfo
if err := json.Unmarshal(data, &info); err != nil {
continue
}
return s.checkStale(&info), nil
} }
// Check if service is stale return nil, nil
}
// checkStale marks a service as stale if heartbeat is too old
func (s *Service) checkStale(info *ServiceInfo) *ServiceInfo {
lastHeartbeat, err := time.Parse(time.RFC3339, info.LastHeartbeat) lastHeartbeat, err := time.Parse(time.RFC3339, info.LastHeartbeat)
if err == nil { if err == nil {
staleThreshold := time.Duration(s.config.StaleThresholdMS) * time.Millisecond staleThreshold := time.Duration(s.config.StaleThresholdMS) * time.Millisecond
@@ -222,8 +220,7 @@ func (s *Service) Discover(name string) (*ServiceInfo, error) {
info.Status = "stale" info.Status = "stale"
} }
} }
return info
return &info, nil
} }
// DiscoverAll finds all registered services // DiscoverAll finds all registered services
@@ -242,16 +239,20 @@ func (s *Service) DiscoverAll() ([]*ServiceInfo, error) {
continue continue
} }
name := entry.Name()[:len(entry.Name())-5] // Remove .json extension filePath := filepath.Join(s.servicesDir, entry.Name())
info, err := s.Discover(name) data, err := os.ReadFile(filePath)
if err != nil { if err != nil {
log.Printf("[Discovery] Failed to read service %s: %v", name, err) log.Printf("[Discovery] Failed to read service file %s: %v", entry.Name(), err)
continue continue
} }
if info != nil { var info ServiceInfo
services = append(services, info) if err := json.Unmarshal(data, &info); err != nil {
log.Printf("[Discovery] Failed to parse service file %s: %v", entry.Name(), err)
continue
} }
services = append(services, s.checkStale(&info))
} }
return services, nil return services, nil
+20 -12
View File
@@ -14,12 +14,13 @@ import (
"github.com/metazla/meta-core/internal/config" "github.com/metazla/meta-core/internal/config"
) )
// LeaderLockInfo matches the TypeScript LeaderLockInfo interface // LeaderLockInfo contains all URLs and metadata for the leader
type LeaderLockInfo struct { type LeaderLockInfo struct {
Host string `json:"host"` Hostname string `json:"hostname"`
API string `json:"api"` BaseUrl string `json:"baseUrl"`
HTTP string `json:"http"` ApiUrl string `json:"apiUrl"` // meta-core API (port 9000)
BaseURL string `json:"baseUrl,omitempty"` RedisUrl string `json:"redisUrl"`
WebdavUrl string `json:"webdavUrl"`
Timestamp int64 `json:"timestamp"` Timestamp int64 `json:"timestamp"`
PID int `json:"pid"` PID int `json:"pid"`
} }
@@ -236,7 +237,7 @@ func (e *Election) transitionToLeader() error {
// Connect storage to local Redis // Connect storage to local Redis
if e.storage != nil { if e.storage != nil {
if err := e.storage.Connect(info.API); err != nil { if err := e.storage.Connect(info.RedisUrl); err != nil {
log.Printf("[Election] Warning: failed to connect storage: %v", err) log.Printf("[Election] Warning: failed to connect storage: %v", err)
} }
} }
@@ -269,11 +270,11 @@ func (e *Election) transitionToFollower() error {
e.mu.Unlock() e.mu.Unlock()
if info != nil { if info != nil {
log.Printf("[Election] Leader is at %s", info.API) log.Printf("[Election] Leader is at %s (Redis: %s)", info.ApiUrl, info.RedisUrl)
// Connect storage to leader's Redis // Connect storage to leader's Redis
if e.storage != nil { if e.storage != nil {
if err := e.storage.Connect(info.API); err != nil { if err := e.storage.Connect(info.RedisUrl); err != nil {
log.Printf("[Election] Warning: failed to connect to leader: %v", err) log.Printf("[Election] Warning: failed to connect to leader: %v", err)
} }
} }
@@ -293,11 +294,18 @@ func (e *Election) buildLeaderInfo() *LeaderLockInfo {
ip := getLocalIP() ip := getLocalIP()
hostname, _ := os.Hostname() hostname, _ := os.Hostname()
// Use BaseURL if configured, otherwise construct from IP
baseUrl := e.config.BaseURL
if baseUrl == "" {
baseUrl = fmt.Sprintf("http://%s:%d", ip, e.config.APIPort)
}
return &LeaderLockInfo{ return &LeaderLockInfo{
Host: hostname, Hostname: hostname,
API: fmt.Sprintf("redis://%s:%d", ip, e.config.RedisPort), BaseUrl: baseUrl,
HTTP: fmt.Sprintf("http://%s:%d", ip, e.config.APIPort), ApiUrl: fmt.Sprintf("http://%s:%d", ip, e.config.HTTPPort),
BaseURL: e.config.BaseURL, RedisUrl: fmt.Sprintf("redis://%s:%d", ip, e.config.RedisPort),
WebdavUrl: baseUrl + "/webdav",
Timestamp: time.Now().UnixMilli(), Timestamp: time.Now().UnixMilli(),
PID: os.Getpid(), PID: os.Getpid(),
} }
+93 -3
View File
@@ -9,12 +9,16 @@ import (
// Handlers provides HTTP handlers for mount operations // Handlers provides HTTP handlers for mount operations
type Handlers struct { type Handlers struct {
manager *Manager manager *Manager
scanFunc ScanFunc
} }
// NewHandlers creates new mount handlers // NewHandlers creates new mount handlers
func NewHandlers(manager *Manager) *Handlers { func NewHandlers(manager *Manager, scanFunc ScanFunc) *Handlers {
return &Handlers{manager: manager} return &Handlers{
manager: manager,
scanFunc: scanFunc,
}
} }
// RegisterRoutes registers all mount-related routes // RegisterRoutes registers all mount-related routes
@@ -24,10 +28,12 @@ func (h *Handlers) RegisterRoutes(r *mux.Router) {
r.HandleFunc("/api/mounts", h.handleCreateMount).Methods("POST") r.HandleFunc("/api/mounts", h.handleCreateMount).Methods("POST")
r.HandleFunc("/api/mounts/rclone/remotes", h.handleListRcloneRemotes).Methods("GET") r.HandleFunc("/api/mounts/rclone/remotes", h.handleListRcloneRemotes).Methods("GET")
r.HandleFunc("/api/mounts/{id}", h.handleGetMount).Methods("GET") r.HandleFunc("/api/mounts/{id}", h.handleGetMount).Methods("GET")
r.HandleFunc("/api/mounts/{id}", h.handleUpdateMount).Methods("PUT", "PATCH")
r.HandleFunc("/api/mounts/{id}", h.handleDeleteMount).Methods("DELETE") r.HandleFunc("/api/mounts/{id}", h.handleDeleteMount).Methods("DELETE")
r.HandleFunc("/api/mounts/{id}/mount", h.handleRequestMount).Methods("POST") r.HandleFunc("/api/mounts/{id}/mount", h.handleRequestMount).Methods("POST")
r.HandleFunc("/api/mounts/{id}/unmount", h.handleRequestUnmount).Methods("POST") r.HandleFunc("/api/mounts/{id}/unmount", h.handleRequestUnmount).Methods("POST")
r.HandleFunc("/api/mounts/{id}/safe-unmount", h.handleSafeUnmount).Methods("POST") r.HandleFunc("/api/mounts/{id}/safe-unmount", h.handleSafeUnmount).Methods("POST")
r.HandleFunc("/api/mounts/{id}/scan", h.handleTriggerMountScan).Methods("POST")
} }
// handleListMounts handles GET /api/mounts // handleListMounts handles GET /api/mounts
@@ -194,6 +200,90 @@ func (h *Handlers) handleListRcloneRemotes(w http.ResponseWriter, r *http.Reques
writeJSON(w, http.StatusOK, RcloneRemotesResponse{Remotes: remotes}) writeJSON(w, http.StatusOK, RcloneRemotesResponse{Remotes: remotes})
} }
// handleUpdateMount handles PUT/PATCH /api/mounts/{id}
func (h *Handlers) handleUpdateMount(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
id := vars["id"]
var updates map[string]interface{}
if err := json.NewDecoder(r.Body).Decode(&updates); err != nil {
writeError(w, http.StatusBadRequest, "invalid JSON body")
return
}
mount, err := h.manager.UpdateMount(id, updates)
if err != nil {
if err.Error() == "mount not found" {
writeError(w, http.StatusNotFound, err.Error())
return
}
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, MountResponse{Mount: mount})
}
// handleTriggerMountScan handles POST /api/mounts/{id}/scan
func (h *Handlers) handleTriggerMountScan(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
id := vars["id"]
// Get mount info
mount, err := h.manager.GetMount(id)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
if mount == nil {
writeError(w, http.StatusNotFound, "mount not found")
return
}
// Check if mount is mounted
if !mount.Mounted {
writeJSON(w, http.StatusBadRequest, MountScanResponse{
Status: "error",
Message: "mount is not currently mounted",
MountID: mount.ID,
MountPath: mount.MountPath,
})
return
}
// Check if scan function is available
if h.scanFunc == nil {
writeJSON(w, http.StatusServiceUnavailable, MountScanResponse{
Status: "error",
Message: "file watcher not available",
MountID: mount.ID,
MountPath: mount.MountPath,
})
return
}
// Trigger scan
fileCount, err := h.scanFunc(mount.MountPath)
if err != nil {
writeJSON(w, http.StatusInternalServerError, MountScanResponse{
Status: "error",
Message: err.Error(),
MountID: mount.ID,
MountPath: mount.MountPath,
})
return
}
writeJSON(w, http.StatusOK, MountScanResponse{
Status: "ok",
Message: "scan triggered successfully",
MountID: mount.ID,
MountPath: mount.MountPath,
FileCount: fileCount,
})
}
// Helper functions // Helper functions
func writeJSON(w http.ResponseWriter, status int, data interface{}) { func writeJSON(w http.ResponseWriter, status int, data interface{}) {
+150 -13
View File
@@ -23,6 +23,7 @@ type Manager struct {
mu sync.RWMutex mu sync.RWMutex
filesPath string filesPath string
mountsDir string mountsDir string
poller *Poller
} }
// NewManager creates a new mount manager // NewManager creates a new mount manager
@@ -41,6 +42,11 @@ func NewManager(cfg *config.Config) (*Manager, error) {
return m, nil return m, nil
} }
// SetPoller sets the poller for the manager
func (m *Manager) SetPoller(poller *Poller) {
m.poller = poller
}
// ensureDirs creates required directories // ensureDirs creates required directories
func (m *Manager) ensureDirs() error { func (m *Manager) ensureDirs() error {
dirs := []string{ dirs := []string{
@@ -95,14 +101,30 @@ func (m *Manager) writeConfig(mountsFile *MountsFile) error {
return os.WriteFile(m.config.MountsFilePath(), data, 0644) return os.WriteFile(m.config.MountsFilePath(), data, 0644)
} }
// IsMounted checks if a path is currently mounted // IsMounted checks if a path is currently mounted by parsing /proc/mounts
// Note: Uses /proc/mounts instead of findmnt for Alpine compatibility
func (m *Manager) IsMounted(mountPath string) bool { func (m *Manager) IsMounted(mountPath string) bool {
cmd := exec.Command("findmnt", "-n", mountPath) data, err := os.ReadFile("/proc/mounts")
output, err := cmd.CombinedOutput()
if err != nil { if err != nil {
return false return false
} }
return strings.TrimSpace(string(output)) != ""
// Clean the mount path (remove trailing slash)
cleanPath := strings.TrimSuffix(mountPath, "/")
// Parse each line of /proc/mounts
lines := strings.Split(string(data), "\n")
for _, line := range lines {
fields := strings.Fields(line)
if len(fields) >= 2 {
// Field 1 is the mount point
mountPoint := strings.TrimSuffix(fields[1], "/")
if mountPoint == cleanPath {
return true
}
}
}
return false
} }
// ReadError reads the error file for a mount // ReadError reads the error file for a mount
@@ -164,6 +186,26 @@ func ObscurePassword(password string) (string, error) {
// ListMounts returns all mounts with their current status // ListMounts returns all mounts with their current status
func (m *Manager) ListMounts() ([]MountStatus, error) { func (m *Manager) ListMounts() ([]MountStatus, error) {
statuses, err := m.listMountsWithoutPollingStatus()
if err != nil {
return nil, err
}
// Add polling status from poller (done separately to avoid deadlock)
if m.poller != nil {
for i := range statuses {
pollingStatus := m.poller.GetPollingStatus(statuses[i].ID)
statuses[i].PollingActive = pollingStatus.Active
statuses[i].LastPolledScan = pollingStatus.LastScan
}
}
return statuses, nil
}
// listMountsWithoutPollingStatus returns mounts without querying the poller
// This is used internally by the poller to avoid deadlocks
func (m *Manager) listMountsWithoutPollingStatus() ([]MountStatus, error) {
mountsFile, err := m.readConfig() mountsFile, err := m.readConfig()
if err != nil { if err != nil {
return nil, err return nil, err
@@ -197,12 +239,21 @@ func (m *Manager) GetMount(id string) (*MountStatus, error) {
mounted := m.IsMounted(mount.MountPath) mounted := m.IsMounted(mount.MountPath)
errMsg, _ := m.ReadError(mount.ID) errMsg, _ := m.ReadError(mount.ID)
return &MountStatus{ status := &MountStatus{
MountConfig: mount, MountConfig: mount,
Mounted: mounted, Mounted: mounted,
Error: errMsg, Error: errMsg,
LastChecked: NowMS(), LastChecked: NowMS(),
}, nil }
// Add polling status from poller (safe since GetMount isn't called from poller with lock held)
if m.poller != nil {
pollingStatus := m.poller.GetPollingStatus(mount.ID)
status.PollingActive = pollingStatus.Active
status.LastPolledScan = pollingStatus.LastScan
}
return status, nil
} }
} }
@@ -258,14 +309,26 @@ func (m *Manager) CreateMount(req *CreateMountRequest) (*MountStatus, error) {
enabled = *req.Enabled enabled = *req.Enabled
} }
// Default polling settings
pollingEnabled := false
if req.PollingEnabled != nil {
pollingEnabled = *req.PollingEnabled
}
pollingIntervalMs := DefaultPollingIntervalMs
if req.PollingIntervalMs != nil && *req.PollingIntervalMs >= MinPollingIntervalMs {
pollingIntervalMs = *req.PollingIntervalMs
}
mount := MountConfig{ mount := MountConfig{
ID: id, ID: id,
Name: req.Name, Name: req.Name,
Type: req.Type, Type: req.Type,
Enabled: enabled, Enabled: enabled,
DesiredMounted: enabled, // Auto-mount if enabled DesiredMounted: enabled, // Auto-mount if enabled
MountPath: mountPath, MountPath: mountPath,
Options: req.Options, Options: req.Options,
PollingEnabled: pollingEnabled,
PollingIntervalMs: pollingIntervalMs,
} }
// Type-specific fields // Type-specific fields
@@ -297,6 +360,11 @@ func (m *Manager) CreateMount(req *CreateMountRequest) (*MountStatus, error) {
log.Printf("[Mounts] Created mount config: %s (%s) -> %s", mount.Name, mount.Type, mount.MountPath) log.Printf("[Mounts] Created mount config: %s (%s) -> %s", mount.Name, mount.Type, mount.MountPath)
// Notify poller about new mount
if m.poller != nil {
m.poller.NotifyMountChanged()
}
return &MountStatus{ return &MountStatus{
MountConfig: mount, MountConfig: mount,
Mounted: false, Mounted: false,
@@ -304,6 +372,63 @@ func (m *Manager) CreateMount(req *CreateMountRequest) (*MountStatus, error) {
}, nil }, nil
} }
// UpdateMount updates mount configuration (for polling settings)
func (m *Manager) UpdateMount(id string, updates map[string]interface{}) (*MountStatus, error) {
mountsFile, err := m.readConfig()
if err != nil {
return nil, err
}
index := -1
for i, mount := range mountsFile.Mounts {
if mount.ID == id {
index = i
break
}
}
if index == -1 {
return nil, fmt.Errorf("mount not found")
}
// Apply updates
mount := &mountsFile.Mounts[index]
if val, ok := updates["pollingEnabled"]; ok {
if enabled, ok := val.(bool); ok {
mount.PollingEnabled = enabled
}
}
if val, ok := updates["pollingIntervalMs"]; ok {
// Handle both int and float64 (JSON numbers are float64)
switch v := val.(type) {
case int:
if v >= MinPollingIntervalMs {
mount.PollingIntervalMs = v
}
case float64:
if int(v) >= MinPollingIntervalMs {
mount.PollingIntervalMs = int(v)
}
}
}
if err := m.writeConfig(mountsFile); err != nil {
return nil, err
}
log.Printf("[Mounts] Updated mount config: %s", mount.Name)
// Notify poller about config change
if m.poller != nil {
m.poller.NotifyMountChanged()
}
// Return updated status
return m.GetMount(id)
}
// RequestMount sets desiredMounted to true // RequestMount sets desiredMounted to true
func (m *Manager) RequestMount(id string) error { func (m *Manager) RequestMount(id string) error {
mountsFile, err := m.readConfig() mountsFile, err := m.readConfig()
@@ -318,6 +443,12 @@ func (m *Manager) RequestMount(id string) error {
return err return err
} }
log.Printf("[Mounts] Mount requested: %s", mount.Name) log.Printf("[Mounts] Mount requested: %s", mount.Name)
// Notify poller about state change
if m.poller != nil {
m.poller.NotifyMountChanged()
}
return nil return nil
} }
} }
@@ -339,6 +470,12 @@ func (m *Manager) RequestUnmount(id string) error {
return err return err
} }
log.Printf("[Mounts] Unmount requested: %s", mount.Name) log.Printf("[Mounts] Unmount requested: %s", mount.Name)
// Notify poller about state change
if m.poller != nil {
m.poller.NotifyMountChanged()
}
return nil return nil
} }
} }
+269
View File
@@ -0,0 +1,269 @@
package mounts
import (
"log"
"sync"
"time"
)
// ScanFunc is a function type for scanning a mount path
type ScanFunc func(path string) (int, error)
// mountPoller manages polling for a single mount
type mountPoller struct {
mountID string
stopChan chan struct{}
stopped bool
mu sync.Mutex
}
// PollingStatus holds the polling state for a mount
type PollingStatus struct {
Active bool `json:"active"`
LastScan int64 `json:"lastScan,omitempty"`
}
// Poller manages per-mount polling goroutines
type Poller struct {
manager *Manager
scanFunc ScanFunc
pollers map[string]*mountPoller // key: mount ID
status map[string]*PollingStatus
running bool
mu sync.RWMutex
}
// NewPoller creates a new mount poller
func NewPoller(manager *Manager, scanFunc ScanFunc) *Poller {
return &Poller{
manager: manager,
scanFunc: scanFunc,
pollers: make(map[string]*mountPoller),
status: make(map[string]*PollingStatus),
}
}
// Start begins the polling service
func (p *Poller) Start() error {
p.mu.Lock()
defer p.mu.Unlock()
if p.running {
return nil
}
p.running = true
log.Println("[Poller] Starting mount polling service")
// Initial sync with mount configurations
go p.syncWithMounts()
return nil
}
// Stop stops all polling goroutines
func (p *Poller) Stop() error {
p.mu.Lock()
defer p.mu.Unlock()
if !p.running {
return nil
}
p.running = false
log.Println("[Poller] Stopping mount polling service")
// Stop all pollers
for id, poller := range p.pollers {
p.stopPoller(poller)
delete(p.pollers, id)
}
return nil
}
// SyncWithMounts synchronizes pollers with current mount configurations
func (p *Poller) SyncWithMounts() {
p.syncWithMounts()
}
// syncWithMounts is the internal implementation
func (p *Poller) syncWithMounts() {
p.mu.RLock()
running := p.running
p.mu.RUnlock()
if !running {
return
}
// Get current mounts BEFORE acquiring the write lock to avoid deadlock
// (ListMounts calls GetPollingStatus which needs RLock)
mounts, err := p.manager.listMountsWithoutPollingStatus()
if err != nil {
log.Printf("[Poller] Failed to list mounts: %v", err)
return
}
p.mu.Lock()
defer p.mu.Unlock()
// Re-check running state after acquiring lock
if !p.running {
return
}
// Track which mounts we've seen
seenMounts := make(map[string]bool)
for _, mount := range mounts {
seenMounts[mount.ID] = true
// Check if mount should be polled
shouldPoll := mount.PollingEnabled && mount.Mounted
existingPoller, exists := p.pollers[mount.ID]
if shouldPoll {
// Calculate interval
interval := mount.PollingIntervalMs
if interval < MinPollingIntervalMs {
interval = DefaultPollingIntervalMs
}
if !exists {
// Start new poller
log.Printf("[Poller] Starting poller for mount %s (interval: %dms)", mount.Name, interval)
p.startPollerForMount(mount.ID, mount.MountPath, interval)
}
} else if exists {
// Stop existing poller
log.Printf("[Poller] Stopping poller for mount %s", mount.Name)
p.stopPoller(existingPoller)
delete(p.pollers, mount.ID)
}
}
// Stop pollers for mounts that no longer exist
for id, poller := range p.pollers {
if !seenMounts[id] {
log.Printf("[Poller] Stopping poller for removed mount %s", id)
p.stopPoller(poller)
delete(p.pollers, id)
}
}
}
// startPollerForMount starts a polling goroutine for a mount
func (p *Poller) startPollerForMount(mountID, mountPath string, intervalMs int) {
poller := &mountPoller{
mountID: mountID,
stopChan: make(chan struct{}),
}
p.pollers[mountID] = poller
// Initialize status
if _, exists := p.status[mountID]; !exists {
p.status[mountID] = &PollingStatus{}
}
p.status[mountID].Active = true
go func() {
ticker := time.NewTicker(time.Duration(intervalMs) * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-poller.stopChan:
return
case <-ticker.C:
p.runPollScan(mountID, mountPath)
}
}
}()
}
// stopPoller stops a mount poller
func (p *Poller) stopPoller(poller *mountPoller) {
poller.mu.Lock()
defer poller.mu.Unlock()
if poller.stopped {
return
}
poller.stopped = true
close(poller.stopChan)
// Update status
if status, exists := p.status[poller.mountID]; exists {
status.Active = false
}
}
// runPollScan executes a scan for a mount
func (p *Poller) runPollScan(mountID, mountPath string) {
if p.scanFunc == nil {
return
}
fileCount, err := p.scanFunc(mountPath)
if err != nil {
log.Printf("[Poller] Scan failed for mount %s: %v", mountID, err)
return
}
// Update status
p.mu.Lock()
if status, exists := p.status[mountID]; exists {
status.LastScan = NowMS()
}
p.mu.Unlock()
log.Printf("[Poller] Poll scan complete for mount %s: %d files", mountID, fileCount)
}
// TriggerScan triggers an immediate scan for a mount
func (p *Poller) TriggerScan(mountID, mountPath string) (int, error) {
if p.scanFunc == nil {
return 0, nil
}
fileCount, err := p.scanFunc(mountPath)
if err != nil {
return 0, err
}
// Update status
p.mu.Lock()
if status, exists := p.status[mountID]; exists {
status.LastScan = NowMS()
} else {
p.status[mountID] = &PollingStatus{
LastScan: NowMS(),
}
}
p.mu.Unlock()
return fileCount, nil
}
// GetPollingStatus returns the polling status for a mount
func (p *Poller) GetPollingStatus(mountID string) *PollingStatus {
p.mu.RLock()
defer p.mu.RUnlock()
if status, exists := p.status[mountID]; exists {
return &PollingStatus{
Active: status.Active,
LastScan: status.LastScan,
}
}
return &PollingStatus{}
}
// NotifyMountChanged notifies the poller that a mount's state has changed
func (p *Poller) NotifyMountChanged() {
go p.syncWithMounts()
}
+28 -3
View File
@@ -11,6 +11,12 @@ const (
MountTypeRclone MountType = "rclone" MountTypeRclone MountType = "rclone"
) )
// Polling constants
const (
DefaultPollingIntervalMs = 30000 // 30 seconds
MinPollingIntervalMs = 5000 // 5 seconds
)
// MountConfig represents a mount configuration // MountConfig represents a mount configuration
type MountConfig struct { type MountConfig struct {
// Common fields // Common fields
@@ -36,14 +42,20 @@ type MountConfig struct {
// rclone-specific fields // rclone-specific fields
RcloneRemote string `json:"rcloneRemote,omitempty"` RcloneRemote string `json:"rcloneRemote,omitempty"`
RclonePath string `json:"rclonePath,omitempty"` RclonePath string `json:"rclonePath,omitempty"`
// Polling configuration
PollingEnabled bool `json:"pollingEnabled"`
PollingIntervalMs int `json:"pollingIntervalMs,omitempty"` // Default: 30000
} }
// MountStatus represents the runtime status of a mount // MountStatus represents the runtime status of a mount
type MountStatus struct { type MountStatus struct {
MountConfig MountConfig
Mounted bool `json:"mounted"` Mounted bool `json:"mounted"`
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
LastChecked int64 `json:"lastChecked"` LastChecked int64 `json:"lastChecked"`
PollingActive bool `json:"pollingActive,omitempty"`
LastPolledScan int64 `json:"lastPolledScan,omitempty"`
} }
// MountsFile represents the mounts.json file structure // MountsFile represents the mounts.json file structure
@@ -79,6 +91,19 @@ type CreateMountRequest struct {
// rclone // rclone
RcloneRemote string `json:"rcloneRemote,omitempty"` RcloneRemote string `json:"rcloneRemote,omitempty"`
RclonePath string `json:"rclonePath,omitempty"` RclonePath string `json:"rclonePath,omitempty"`
// Polling
PollingEnabled *bool `json:"pollingEnabled,omitempty"`
PollingIntervalMs *int `json:"pollingIntervalMs,omitempty"`
}
// MountScanResponse is the response for mount scan operations
type MountScanResponse struct {
Status string `json:"status"`
Message string `json:"message,omitempty"`
MountID string `json:"mountId"`
MountPath string `json:"mountPath"`
FileCount int `json:"fileCount,omitempty"`
} }
// MountResponse is the response for mount operations // MountResponse is the response for mount operations
+30
View File
@@ -3,6 +3,7 @@ package watcher
import ( import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"fmt"
"io" "io"
"log" "log"
"os" "os"
@@ -308,6 +309,35 @@ func (w *Watcher) scanDirectory(root string) int {
return count return count
} }
// ScanMountPath scans a specific mount path and returns file count
// This is used by the mount poller to scan individual mounts
func (w *Watcher) ScanMountPath(mountPath string) (int, error) {
// Validate path is under filesPath
if !strings.HasPrefix(mountPath, w.filesPath) {
return 0, fmt.Errorf("mount path %s is not under files path %s", mountPath, w.filesPath)
}
// Check directory exists
info, err := os.Stat(mountPath)
if err != nil {
if os.IsNotExist(err) {
return 0, fmt.Errorf("mount path %s does not exist", mountPath)
}
return 0, fmt.Errorf("failed to stat mount path %s: %w", mountPath, err)
}
if !info.IsDir() {
return 0, fmt.Errorf("mount path %s is not a directory", mountPath)
}
// Scan the directory
log.Printf("[Watcher] Scanning mount path: %s", mountPath)
count := w.scanDirectory(mountPath)
log.Printf("[Watcher] Mount scan complete: %s (%d files)", mountPath, count)
return count, nil
}
// GetRecentEvents returns events since a given timestamp // GetRecentEvents returns events since a given timestamp
func (w *Watcher) GetRecentEvents(sinceMS int64, limit int) []FileEvent { func (w *Watcher) GetRecentEvents(sinceMS int64, limit int) []FileEvent {
w.mu.RLock() w.mu.RLock()