From fb55663c3a7a7056b14b04ad4e0ec6f9440ec24d Mon Sep 17 00:00:00 2001 From: Pierre Date: Mon, 16 Feb 2026 00:44:59 +0100 Subject: [PATCH] change election leader mechanism --- cmd/meta-core/main.go | 63 ++--- docker/Dockerfile | 17 +- docker/leader-election.sh | 97 ++++++++ docker/nginx.conf | 12 + docker/supervisord.conf | 19 +- internal/api/handlers.go | 16 +- internal/api/server.go | 10 +- internal/leader/election.go | 462 ++++-------------------------------- internal/leader/redis.go | 185 --------------- internal/watchers/poller.go | 9 + 10 files changed, 226 insertions(+), 664 deletions(-) create mode 100644 docker/leader-election.sh delete mode 100644 internal/leader/redis.go diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go index 452d8d5..3cedc6c 100644 --- a/cmd/meta-core/main.go +++ b/cmd/meta-core/main.go @@ -1,6 +1,7 @@ package main import ( + "fmt" "log" "os" "os/signal" @@ -24,36 +25,23 @@ func main() { cfg := config.Load() log.Printf("[meta-core] Service: %s, HTTP Port: %d", cfg.ServiceName, cfg.HTTPPort) - // Create storage client + // Create role provider (reads META_CORE_ROLE from environment, set by leader-election.sh) + roleProvider := leader.NewLocalRoleProvider(cfg) + log.Printf("[meta-core] Role: %s", roleProvider.Role()) + + // Connect to local Redis (supervisord starts Redis before meta-core) + redisURL := fmt.Sprintf("redis://localhost:%d", cfg.RedisPort) + log.Printf("[meta-core] Connecting to Redis at %s", redisURL) + storageClient := storage.NewClient("") - - // Create leader election - election := leader.NewElection(cfg) - - // Wire up storage connector to election - election.SetStorageConnector(&storageConnectorAdapter{client: storageClient}) - - // Set up leader/follower callbacks - election.OnBecomeLeader(func() { - log.Println("[meta-core] Became LEADER - Redis is running") - }) - - election.OnBecomeFollower(func(info *leader.LeaderLockInfo) { - log.Printf("[meta-core] Became FOLLOWER - Leader at %s (Redis: %s)", info.BaseUrl, info.RedisUrl) - }) - - election.OnLeaderLost(func() { - log.Println("[meta-core] Lost leadership") - }) - - // Start leader election - if err := election.Start(); err != nil { - log.Fatalf("[meta-core] Failed to start leader election: %v", err) + if err := storageClient.Connect(redisURL); err != nil { + log.Fatalf("[meta-core] Failed to connect to Redis: %v", err) } + log.Println("[meta-core] Connected to Redis") // Create and start service discovery disc := discovery.NewService(cfg) - disc.SetRoleProvider(&roleProviderAdapter{election: election}) + disc.SetRoleProvider(&roleProviderAdapter{provider: roleProvider}) if err := disc.Start(); err != nil { log.Fatalf("[meta-core] Failed to start service discovery: %v", err) } @@ -65,7 +53,7 @@ func main() { } // Create and start API server - apiServer := api.NewServer(cfg, election, disc, cleaner, storageClient) + apiServer := api.NewServer(cfg, roleProvider, disc, cleaner, storageClient) if err := apiServer.Start(); err != nil { log.Fatalf("[meta-core] Failed to start API server: %v", err) } @@ -90,10 +78,6 @@ func main() { log.Printf("[meta-core] Error stopping service discovery: %v", err) } - if err := election.Stop(); err != nil { - log.Printf("[meta-core] Error stopping leader election: %v", err) - } - if err := storageClient.Close(); err != nil { log.Printf("[meta-core] Error closing storage client: %v", err) } @@ -108,24 +92,11 @@ func waitForShutdown() { <-sigChan } -// storageConnectorAdapter adapts storage.Client to leader.StorageConnector -type storageConnectorAdapter struct { - client *storage.Client -} - -func (a *storageConnectorAdapter) Connect(url string) error { - return a.client.Connect(url) -} - -func (a *storageConnectorAdapter) Close() error { - return a.client.Close() -} - -// roleProviderAdapter adapts leader.Election to discovery.RoleProvider +// roleProviderAdapter adapts leader.RoleProvider to discovery.RoleProvider type roleProviderAdapter struct { - election *leader.Election + provider leader.RoleProvider } func (a *roleProviderAdapter) Role() string { - return string(a.election.Role()) + return string(a.provider.Role()) } diff --git a/docker/Dockerfile b/docker/Dockerfile index bb6aafc..9d51679 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -3,11 +3,17 @@ # Multi-stage build for meta-core as a standalone service # # Includes: -# - meta-core Go binary (leader election, Redis, HTTP API, Mount API, Watcher) +# - Bash-based leader election script (flock on shared volume) +# - meta-core Go binary (HTTP API, Mount API, Watcher) # - nginx (dashboard, WebDAV, API proxy) -# - Redis (leader-managed) +# - Redis (managed by supervisord on leader) # - rclone (remote mount daemon) -# - supervisord (process management) +# - supervisord (process management - runs on leader only) +# +# Architecture: +# - leader-election.sh handles flock, starts supervisord on leader +# - Followers remain dormant, retry flock every 5s +# - Only leader runs Redis, meta-core, nginx, rclone, mount-watcher # ============================================================================= # ----------------------------------------------------------------------------- @@ -103,6 +109,7 @@ RUN mkdir -p \ COPY docker/supervisord.conf /etc/supervisor/conf.d/supervisord.conf COPY docker/nginx.conf /etc/nginx/nginx.conf COPY docker/mount-watcher.sh /app/scripts/mount-watcher.sh +COPY docker/leader-election.sh /app/scripts/leader-election.sh COPY docker/entrypoint.sh /entrypoint.sh # Copy dashboard static files from builder @@ -112,7 +119,7 @@ COPY --from=dashboard-builder /app/dist /app/dashboard/dist COPY --from=editor-builder /app/dist /app/editor/dist # Set permissions -RUN chmod +x /app/scripts/mount-watcher.sh /entrypoint.sh +RUN chmod +x /app/scripts/mount-watcher.sh /app/scripts/leader-election.sh /entrypoint.sh # Create htpasswd for rclone basic auth (admin:admin) RUN htpasswd -bc /etc/nginx/.htpasswd admin admin @@ -137,4 +144,4 @@ HEALTHCHECK --interval=10s --timeout=5s --start-period=10s --retries=3 \ CMD curl -sf http://localhost/health || exit 1 ENTRYPOINT ["/entrypoint.sh"] -CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"] +CMD ["/app/scripts/leader-election.sh"] diff --git a/docker/leader-election.sh b/docker/leader-election.sh new file mode 100644 index 0000000..85f10b7 --- /dev/null +++ b/docker/leader-election.sh @@ -0,0 +1,97 @@ +#!/bin/bash +# ============================================================================= +# Leader Election Script for meta-core +# +# Uses flock for leader election. Only the leader runs the full application +# stack (supervisord with Redis, meta-core, nginx, rclone, mount-watcher). +# Followers remain dormant, periodically retrying to acquire the lock. +# +# Environment variables: +# META_CORE_PATH - Path to meta-core data directory (default: /meta-core) +# META_CORE_HTTP_PORT - Port for meta-core HTTP API (default: 9000) +# ELECTION_RETRY_SECS - Seconds between flock retry attempts (default: 5) +# ============================================================================= + +set -e + +# Configuration +META_CORE_PATH="${META_CORE_PATH:-/meta-core}" +META_CORE_HTTP_PORT="${META_CORE_HTTP_PORT:-9000}" +ELECTION_RETRY_SECS="${ELECTION_RETRY_SECS:-5}" + +LOCK_FILE="${META_CORE_PATH}/locks/kv-leader.lock" +INFO_FILE="${META_CORE_PATH}/locks/kv-leader.info" +SUPERVISORD_CONF="/etc/supervisor/conf.d/supervisord.conf" + +# Get the container's IP address +# Note: Alpine uses BusyBox hostname which uses -i (lowercase) instead of -I +get_local_ip() { + # Try hostname -i first (Alpine/BusyBox), then fallback to hostname -I (GNU) + local ip + ip=$(hostname -i 2>/dev/null | awk '{print $1}') + if [ -z "$ip" ] || [ "$ip" = "127.0.0.1" ]; then + # Fallback: parse /etc/hosts or use hostname + ip=$(getent hosts "$(hostname)" 2>/dev/null | awk '{print $1}' | head -1) + fi + if [ -z "$ip" ]; then + # Last resort: use hostname + ip=$(hostname) + fi + echo "$ip" +} + +# Write leader info file (plain text API URL) +write_leader_info() { + local ip=$(get_local_ip) + local api_url="http://${ip}:${META_CORE_HTTP_PORT}" + + # Write atomically using temp file + rename + local temp_file="${INFO_FILE}.tmp" + echo "$api_url" > "$temp_file" + mv "$temp_file" "$INFO_FILE" + + echo "[election] Wrote leader info: $api_url" +} + +# Clean up leader info file on shutdown +cleanup() { + echo "[election] Cleaning up..." + rm -f "$INFO_FILE" + exit 0 +} + +# Set up signal handlers for graceful shutdown +trap cleanup SIGTERM SIGINT + +# Ensure lock directory exists +mkdir -p "$(dirname "$LOCK_FILE")" + +echo "[election] Starting leader election..." +echo "[election] Lock file: $LOCK_FILE" +echo "[election] Info file: $INFO_FILE" + +# Open lock file for flock (file descriptor 200) +exec 200>"$LOCK_FILE" + +# Main election loop +while true; do + # Try non-blocking exclusive flock + if flock -n 200; then + echo "[election] Acquired flock - transitioning to LEADER" + + # Write leader info before starting services + write_leader_info + + # Export role for meta-core Go binary to read + export META_CORE_ROLE="leader" + + # exec supervisord - replaces this process + # When supervisord exits, the flock is released automatically + echo "[election] Starting supervisord..." + exec /usr/bin/supervisord -c "$SUPERVISORD_CONF" + else + echo "[election] Lock held by another process - acting as FOLLOWER" + echo "[election] Waiting ${ELECTION_RETRY_SECS}s before retry..." + sleep "$ELECTION_RETRY_SECS" + fi +done diff --git a/docker/nginx.conf b/docker/nginx.conf index 1373a4f..5dd1dc3 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -139,6 +139,18 @@ http { proxy_set_header Host $host; } + location /urls { + proxy_pass http://meta-core-api/urls; + proxy_http_version 1.1; + proxy_set_header Host $host; + } + + location /status { + proxy_pass http://meta-core-api/status; + proxy_http_version 1.1; + proxy_set_header Host $host; + } + # ========================================================================= # rclone Web UI (proxied with basic auth) # ========================================================================= diff --git a/docker/supervisord.conf b/docker/supervisord.conf index c25ab8b..40fa834 100644 --- a/docker/supervisord.conf +++ b/docker/supervisord.conf @@ -16,11 +16,26 @@ serverurl=unix:///var/run/supervisor.sock supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface # ============================================================================= -# meta-core: Leader election, Redis management, HTTP API, Mount API, File Watcher +# Redis: Key-value store (managed by supervisord, started before meta-core) +# Previously spawned by Go code, now managed by bash-based leader election +# ============================================================================= +[program:redis] +command=/usr/bin/redis-server --port %(ENV_REDIS_PORT)s --bind 0.0.0.0 --protected-mode no --dir /meta-core/db/redis --appendonly yes --appendfilename appendonly.aof --dbfilename dump.rdb --save 60 1 --loglevel warning --notify-keyspace-events Kg$ +priority=1 +autostart=true +autorestart=true +stdout_logfile=/dev/stdout +stdout_logfile_maxbytes=0 +stderr_logfile=/dev/stderr +stderr_logfile_maxbytes=0 +startsecs=2 + +# ============================================================================= +# meta-core: HTTP API, Mount API, File Watcher (connects to local Redis) # ============================================================================= [program:meta-core] command=/usr/local/bin/meta-core -priority=5 +priority=10 autostart=true autorestart=true stdout_logfile=/dev/stdout diff --git a/internal/api/handlers.go b/internal/api/handlers.go index 19eb92a..e34fa4e 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -77,10 +77,10 @@ var startTime = time.Now() func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { response := HealthResponse{ Status: "ok", - Role: string(s.election.Role()), + Role: string(s.roleProvider.Role()), Redis: s.storage.Health(), Timestamp: time.Now().UTC().Format(time.RFC3339), - Leader: s.election.LeaderInfo(), + Leader: s.roleProvider.LeaderInfo(), } if !response.Redis { @@ -100,13 +100,13 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { response := StatusResponse{ Status: "ok", - Role: string(s.election.Role()), + Role: string(s.roleProvider.Role()), Redis: s.storage.Health(), ServiceName: s.config.ServiceName, Version: s.config.ServiceVersion, Uptime: int64(time.Since(startTime).Seconds()), FileCount: fileCount, - Leader: s.election.LeaderInfo(), + Leader: s.roleProvider.LeaderInfo(), } if !response.Redis { @@ -118,7 +118,7 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { // handleLeader handles GET /leader func (s *Server) handleLeader(w http.ResponseWriter, r *http.Request) { - info := s.election.LeaderInfo() + info := s.roleProvider.LeaderInfo() if info == nil { writeError(w, http.StatusServiceUnavailable, "no leader available") return @@ -130,7 +130,7 @@ func (s *Server) handleLeader(w http.ResponseWriter, r *http.Request) { // handleRole handles GET /role func (s *Server) handleRole(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, map[string]string{ - "role": string(s.election.Role()), + "role": string(s.roleProvider.Role()), }) } @@ -147,7 +147,7 @@ type URLsResponse struct { // 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() + info := s.roleProvider.LeaderInfo() if info == nil { writeError(w, http.StatusServiceUnavailable, "no leader available") return @@ -159,7 +159,7 @@ func (s *Server) handleURLs(w http.ResponseWriter, r *http.Request) { ApiUrl: info.ApiUrl, RedisUrl: info.RedisUrl, WebdavUrl: info.WebdavUrl, - IsLeader: s.election.IsLeader(), + IsLeader: s.roleProvider.IsLeader(), } writeJSON(w, http.StatusOK, response) diff --git a/internal/api/server.go b/internal/api/server.go index c2e7895..089cccc 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -23,7 +23,7 @@ import ( // Server is the HTTP API server for meta-core type Server struct { config *config.Config - election *leader.Election + roleProvider leader.RoleProvider discovery *discovery.Service cleaner *discovery.Cleaner storage *storage.Client @@ -48,15 +48,15 @@ type Server struct { // NewServer creates a new API server func NewServer( cfg *config.Config, - election *leader.Election, + roleProvider leader.RoleProvider, disc *discovery.Service, cleaner *discovery.Cleaner, stor *storage.Client, ) *Server { s := &Server{ - config: cfg, - election: election, - discovery: disc, + config: cfg, + roleProvider: roleProvider, + discovery: disc, cleaner: cleaner, storage: stor, router: mux.NewRouter(), diff --git a/internal/leader/election.go b/internal/leader/election.go index 5e6390a..07dee54 100644 --- a/internal/leader/election.go +++ b/internal/leader/election.go @@ -1,18 +1,9 @@ package leader import ( - "context" - "encoding/json" "fmt" - "io" - "log" "net" - "net/http" "os" - "path/filepath" - "strings" - "sync" - "syscall" "time" "github.com/metazla/meta-core/internal/config" @@ -38,449 +29,94 @@ const ( RoleFollower Role = "follower" ) -// Election handles leader election using flock on shared filesystem -type Election struct { - config *config.Config +// RoleProvider provides role and leader info +// This interface is used by the API server to get role information +// without needing the full election logic (which is now in bash) +type RoleProvider interface { + Role() Role + LeaderInfo() *LeaderLockInfo + IsLeader() bool +} - lockFile *os.File +// LocalRoleProvider provides role info for this instance +// It reads the role from environment and builds leader info locally +// This is used when running as the leader (started by leader-election.sh) +type LocalRoleProvider struct { + config *config.Config role Role leaderInfo *LeaderLockInfo - mu sync.RWMutex - - redisManager *RedisManager - storage StorageConnector - - // Callbacks - onBecomeLeader func() - onBecomeFollower func(info *LeaderLockInfo) - onLeaderLost func() - - // Lifecycle - stopChan chan struct{} - isShuttingDown bool - wg sync.WaitGroup } -// StorageConnector interface for connecting to Redis -type StorageConnector interface { - Connect(url string) error - Close() error -} - -// NewElection creates a new leader election instance -func NewElection(cfg *config.Config) *Election { - return &Election{ - config: cfg, - role: RoleUnknown, - redisManager: NewRedisManager(cfg), - stopChan: make(chan struct{}), - } -} - -// SetStorageConnector sets the storage connector for Redis connections -func (e *Election) SetStorageConnector(s StorageConnector) { - e.storage = s -} - -// OnBecomeLeader sets callback for when this instance becomes leader -func (e *Election) OnBecomeLeader(fn func()) { - e.onBecomeLeader = fn -} - -// OnBecomeFollower sets callback for when this instance becomes follower -func (e *Election) OnBecomeFollower(fn func(info *LeaderLockInfo)) { - e.onBecomeFollower = fn -} - -// OnLeaderLost sets callback for when leadership is lost -func (e *Election) OnLeaderLost(fn func()) { - e.onLeaderLost = fn -} - -// Start begins the leader election process -func (e *Election) Start() error { - log.Println("[Election] Starting leader election...") - - // Ensure lock directory exists - lockDir := filepath.Dir(e.config.LockFilePath()) - if err := os.MkdirAll(lockDir, 0755); err != nil { - return fmt.Errorf("failed to create lock directory: %w", err) +// NewLocalRoleProvider creates a RoleProvider for the local instance +// It reads META_CORE_ROLE from environment (set by leader-election.sh) +func NewLocalRoleProvider(cfg *config.Config) *LocalRoleProvider { + roleStr := os.Getenv("META_CORE_ROLE") + var role Role + switch roleStr { + case "leader": + role = RoleLeader + case "follower": + role = RoleFollower + default: + // Default to leader if not set (for backward compatibility) + role = RoleLeader } - // Try to acquire the lock - acquired, err := e.tryAcquireLock() - if err != nil { - return fmt.Errorf("failed to acquire lock: %w", err) + provider := &LocalRoleProvider{ + config: cfg, + role: role, } - if acquired { - if err := e.transitionToLeader(); err != nil { - return err - } - } else { - if err := e.transitionToFollower(); err != nil { - return err - } - } + // Build leader info (always, since we're the leader when running) + provider.leaderInfo = provider.buildLeaderInfo() - // Start health check loop - e.wg.Add(1) - go e.healthCheckLoop() - - return nil + return provider } -// Stop gracefully shuts down the election -func (e *Election) Stop() error { - log.Println("[Election] Stopping leader election...") - e.isShuttingDown = true - close(e.stopChan) - - e.wg.Wait() - - // Stop Redis if we were leader - if e.Role() == RoleLeader { - if err := e.redisManager.Stop(); err != nil { - log.Printf("[Election] Error stopping Redis: %v", err) - } - } - - // Release the lock - e.releaseLock() - - return nil +// Role returns the current role +func (p *LocalRoleProvider) Role() Role { + return p.role } -// Role returns the current role (thread-safe) -func (e *Election) Role() Role { - e.mu.RLock() - defer e.mu.RUnlock() - return e.role -} - -// LeaderInfo returns the current leader info (thread-safe) -func (e *Election) LeaderInfo() *LeaderLockInfo { - e.mu.RLock() - defer e.mu.RUnlock() - if e.leaderInfo == nil { +// LeaderInfo returns the leader info +func (p *LocalRoleProvider) LeaderInfo() *LeaderLockInfo { + if p.leaderInfo == nil { return nil } // Return a copy - info := *e.leaderInfo + info := *p.leaderInfo + // Update timestamp on each call + info.Timestamp = time.Now().UnixMilli() return &info } // IsLeader returns true if this instance is the leader -func (e *Election) IsLeader() bool { - return e.Role() == RoleLeader -} - -// tryAcquireLock attempts to acquire exclusive flock on the lock file -func (e *Election) tryAcquireLock() (bool, error) { - lockPath := e.config.LockFilePath() - - // Open or create the lock file - f, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0666) - if err != nil { - return false, fmt.Errorf("failed to open lock file: %w", err) - } - - // Try non-blocking exclusive flock - err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) - if err != nil { - f.Close() - if err == syscall.EWOULDBLOCK { - log.Println("[Election] Lock is held by another process") - return false, nil - } - return false, fmt.Errorf("flock failed: %w", err) - } - - e.lockFile = f - log.Printf("[Election] Acquired flock on %s", lockPath) - return true, nil -} - -// releaseLock releases the flock -func (e *Election) releaseLock() { - if e.lockFile != nil { - syscall.Flock(int(e.lockFile.Fd()), syscall.LOCK_UN) - e.lockFile.Close() - e.lockFile = nil - log.Println("[Election] Released flock") - } -} - -// transitionToLeader handles becoming the leader -func (e *Election) transitionToLeader() error { - log.Println("[Election] Transitioning to LEADER role") - - e.mu.Lock() - e.role = RoleLeader - e.mu.Unlock() - - // Start Redis - if err := e.redisManager.Start(); err != nil { - return fmt.Errorf("failed to start Redis: %w", err) - } - - // Wait for Redis to be ready - if err := e.redisManager.WaitForReady(30 * time.Second); err != nil { - return fmt.Errorf("Redis not ready: %w", err) - } - - // Build and write leader info - info := e.buildLeaderInfo() - e.mu.Lock() - e.leaderInfo = info - e.mu.Unlock() - - if err := e.writeLeaderInfo(info); err != nil { - return fmt.Errorf("failed to write leader info: %w", err) - } - - // Connect storage to local Redis - if e.storage != nil { - if err := e.storage.Connect(info.RedisUrl); err != nil { - log.Printf("[Election] Warning: failed to connect storage: %v", err) - } - } - - // Notify callback - if e.onBecomeLeader != nil { - e.onBecomeLeader() - } - - log.Println("[Election] Now acting as LEADER") - return nil -} - -// transitionToFollower handles becoming a follower -func (e *Election) transitionToFollower() error { - log.Println("[Election] Transitioning to FOLLOWER role") - - e.mu.Lock() - e.role = RoleFollower - e.mu.Unlock() - - // Read leader info - info, err := e.readLeaderInfo() - if err != nil { - return fmt.Errorf("failed to read leader info: %w", err) - } - - e.mu.Lock() - e.leaderInfo = info - e.mu.Unlock() - - if info != nil { - log.Printf("[Election] Leader is at %s (Redis: %s)", info.ApiUrl, info.RedisUrl) - - // Connect storage to leader's Redis - if e.storage != nil { - if err := e.storage.Connect(info.RedisUrl); err != nil { - log.Printf("[Election] Warning: failed to connect to leader: %v", err) - } - } - - // Notify callback - if e.onBecomeFollower != nil { - e.onBecomeFollower(info) - } - } - - log.Println("[Election] Now acting as FOLLOWER") - return nil +func (p *LocalRoleProvider) IsLeader() bool { + return p.role == RoleLeader } // buildLeaderInfo creates leader info for this instance -func (e *Election) buildLeaderInfo() *LeaderLockInfo { +func (p *LocalRoleProvider) buildLeaderInfo() *LeaderLockInfo { ip := getLocalIP() hostname, _ := os.Hostname() // Use BaseURL if configured, otherwise construct from IP - baseUrl := e.config.BaseURL + baseUrl := p.config.BaseURL if baseUrl == "" { - baseUrl = fmt.Sprintf("http://%s:%d", ip, e.config.APIPort) + baseUrl = fmt.Sprintf("http://%s:%d", ip, p.config.APIPort) } return &LeaderLockInfo{ Hostname: hostname, BaseUrl: baseUrl, - ApiUrl: fmt.Sprintf("http://%s:%d", ip, e.config.HTTPPort), - RedisUrl: fmt.Sprintf("redis://%s:%d", ip, e.config.RedisPort), - WebdavUrl: fmt.Sprintf("http://%s:%d/webdav", ip, e.config.APIPort), + ApiUrl: fmt.Sprintf("http://%s:%d", ip, p.config.HTTPPort), + RedisUrl: fmt.Sprintf("redis://%s:%d", ip, p.config.RedisPort), + WebdavUrl: fmt.Sprintf("http://%s:%d/webdav", ip, p.config.APIPort), Timestamp: time.Now().UnixMilli(), PID: os.Getpid(), } } -// writeLeaderInfo atomically writes leader API URL to file (plain text) -// Only writes if the content has changed to avoid triggering file watchers unnecessarily -func (e *Election) writeLeaderInfo(info *LeaderLockInfo) error { - infoPath := e.config.InfoFilePath() - - // Write just the API URL as plain text - newData := []byte(info.ApiUrl) - - // Check if file already has the same content - existingData, err := os.ReadFile(infoPath) - if err == nil && string(existingData) == string(newData) { - // Content unchanged, skip write to avoid triggering file watchers - return nil - } - - // Content changed or file doesn't exist, write atomically - tempPath := infoPath + ".tmp" - if err := os.WriteFile(tempPath, newData, 0644); err != nil { - return err - } - - if err := os.Rename(tempPath, infoPath); err != nil { - return err - } - - log.Printf("[Election] Wrote leader API URL to %s: %s", infoPath, info.ApiUrl) - return nil -} - -// readLeaderInfo reads leader API URL from file and fetches full info via /urls API -func (e *Election) readLeaderInfo() (*LeaderLockInfo, error) { - infoPath := e.config.InfoFilePath() - - data, err := os.ReadFile(infoPath) - if err != nil { - if os.IsNotExist(err) { - return nil, nil - } - return nil, err - } - - // File contains plain text API URL - apiUrl := strings.TrimSpace(string(data)) - if apiUrl == "" { - return nil, nil - } - - // Fetch full info from /urls API - return e.fetchLeaderInfoFromAPI(apiUrl) -} - -// fetchLeaderInfoFromAPI calls the /urls API to get full leader info -func (e *Election) fetchLeaderInfoFromAPI(apiUrl string) (*LeaderLockInfo, error) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - req, err := http.NewRequestWithContext(ctx, "GET", apiUrl+"/urls", nil) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - req.Header.Set("Accept", "application/json") - - resp, err := http.DefaultClient.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to call /urls API: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("/urls API returned status %d", resp.StatusCode) - } - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read response body: %w", err) - } - - // Parse URLsResponse format - var urlsResp 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"` - } - if err := json.Unmarshal(body, &urlsResp); err != nil { - return nil, fmt.Errorf("failed to parse /urls response: %w", err) - } - - // Convert to LeaderLockInfo - return &LeaderLockInfo{ - Hostname: urlsResp.Hostname, - BaseUrl: urlsResp.BaseUrl, - ApiUrl: urlsResp.ApiUrl, - RedisUrl: urlsResp.RedisUrl, - WebdavUrl: urlsResp.WebdavUrl, - Timestamp: time.Now().UnixMilli(), - PID: 0, // Unknown for remote leader - }, nil -} - -// healthCheckLoop periodically checks health and updates timestamps -func (e *Election) healthCheckLoop() { - defer e.wg.Done() - - ticker := time.NewTicker(time.Duration(e.config.HealthCheckIntervalMS) * time.Millisecond) - defer ticker.Stop() - - for { - select { - case <-e.stopChan: - return - case <-ticker.C: - e.performHealthCheck() - } - } -} - -// performHealthCheck checks health based on current role -func (e *Election) performHealthCheck() { - switch e.Role() { - case RoleLeader: - // Update timestamp in leader info - e.updateLeaderTimestamp() - - // Check if Redis is still running - if !e.redisManager.IsRunning() { - log.Println("[Election] Redis not running, attempting restart...") - if err := e.redisManager.Start(); err != nil { - log.Printf("[Election] Failed to restart Redis: %v", err) - } - } - - case RoleFollower: - // Re-read leader info in case it changed - info, err := e.readLeaderInfo() - if err != nil { - log.Printf("[Election] Failed to read leader info: %v", err) - return - } - - if info != nil { - e.mu.Lock() - e.leaderInfo = info - e.mu.Unlock() - } - } -} - -// updateLeaderTimestamp updates the timestamp in leader info -func (e *Election) updateLeaderTimestamp() { - e.mu.Lock() - if e.leaderInfo != nil { - e.leaderInfo.Timestamp = time.Now().UnixMilli() - info := *e.leaderInfo - e.mu.Unlock() - - if err := e.writeLeaderInfo(&info); err != nil { - log.Printf("[Election] Failed to update timestamp: %v", err) - } - } else { - e.mu.Unlock() - } -} - // getLocalIP returns the local IP address func getLocalIP() string { addrs, err := net.InterfaceAddrs() diff --git a/internal/leader/redis.go b/internal/leader/redis.go deleted file mode 100644 index 785bd8b..0000000 --- a/internal/leader/redis.go +++ /dev/null @@ -1,185 +0,0 @@ -package leader - -import ( - "context" - "fmt" - "log" - "os" - "os/exec" - "sync" - "syscall" - "time" - - "github.com/metazla/meta-core/internal/config" - "github.com/redis/go-redis/v9" -) - -// RedisManager handles spawning and managing the Redis process -type RedisManager struct { - config *config.Config - cmd *exec.Cmd - running bool - mu sync.RWMutex -} - -// NewRedisManager creates a new Redis manager -func NewRedisManager(cfg *config.Config) *RedisManager { - return &RedisManager{ - config: cfg, - } -} - -// Start spawns the Redis server process -func (rm *RedisManager) Start() error { - rm.mu.Lock() - defer rm.mu.Unlock() - - if rm.running { - return nil - } - - // Ensure Redis data directory exists - dataDir := rm.config.RedisDataDir() - if err := os.MkdirAll(dataDir, 0755); err != nil { - return fmt.Errorf("failed to create Redis data directory: %w", err) - } - - log.Printf("[Redis] Spawning Redis on port %d...", rm.config.RedisPort) - - // Build Redis arguments matching TypeScript implementation - args := []string{ - "--port", fmt.Sprintf("%d", rm.config.RedisPort), - "--bind", "0.0.0.0", - "--protected-mode", "no", - "--dir", dataDir, - "--appendonly", "yes", - "--appendfilename", "appendonly.aof", - "--dbfilename", "dump.rdb", - "--save", "60", "1", // Save after 60 seconds if at least 1 key changed - "--loglevel", "warning", - // Enable keyspace notifications for meta:events stream - // K = Keyspace events (publish to __keyspace@0__:keyname) - // g = Generic commands (DEL, EXPIRE, RENAME) - // $ = String commands (SET, MSET, etc.) - "--notify-keyspace-events", "Kg$", - } - - rm.cmd = exec.Command("redis-server", args...) - rm.cmd.Stdout = os.Stdout - rm.cmd.Stderr = os.Stderr - - if err := rm.cmd.Start(); err != nil { - return fmt.Errorf("failed to start Redis: %w", err) - } - - rm.running = true - log.Printf("[Redis] Started with PID %d", rm.cmd.Process.Pid) - - // Monitor the process in background - go rm.monitorProcess() - - return nil -} - -// Stop gracefully stops the Redis server -func (rm *RedisManager) Stop() error { - rm.mu.Lock() - defer rm.mu.Unlock() - - if !rm.running || rm.cmd == nil || rm.cmd.Process == nil { - return nil - } - - log.Println("[Redis] Stopping Redis...") - - // Try graceful shutdown first (SIGTERM) - if err := rm.cmd.Process.Signal(syscall.SIGTERM); err != nil { - log.Printf("[Redis] Failed to send SIGTERM: %v", err) - } - - // Wait up to 10 seconds for graceful shutdown - done := make(chan error, 1) - go func() { - done <- rm.cmd.Wait() - }() - - select { - case <-done: - log.Println("[Redis] Stopped gracefully") - case <-time.After(10 * time.Second): - log.Println("[Redis] Timeout, sending SIGKILL...") - rm.cmd.Process.Kill() - <-done - } - - rm.running = false - rm.cmd = nil - return nil -} - -// IsRunning returns true if Redis is currently running -func (rm *RedisManager) IsRunning() bool { - rm.mu.RLock() - defer rm.mu.RUnlock() - return rm.running -} - -// WaitForReady waits for Redis to be ready to accept connections -func (rm *RedisManager) WaitForReady(timeout time.Duration) error { - log.Println("[Redis] Waiting for Redis to be ready...") - - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - - client := redis.NewClient(&redis.Options{ - Addr: fmt.Sprintf("localhost:%d", rm.config.RedisPort), - }) - defer client.Close() - - ticker := time.NewTicker(100 * time.Millisecond) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return fmt.Errorf("timeout waiting for Redis") - case <-ticker.C: - if err := client.Ping(ctx).Err(); err == nil { - log.Println("[Redis] Ready to accept connections") - return nil - } - } - } -} - -// Ping checks if Redis is responding -func (rm *RedisManager) Ping() error { - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - defer cancel() - - client := redis.NewClient(&redis.Options{ - Addr: fmt.Sprintf("localhost:%d", rm.config.RedisPort), - }) - defer client.Close() - - return client.Ping(ctx).Err() -} - -// monitorProcess monitors the Redis process and updates state when it exits -func (rm *RedisManager) monitorProcess() { - if rm.cmd == nil { - return - } - - err := rm.cmd.Wait() - - rm.mu.Lock() - rm.running = false - rm.mu.Unlock() - - if err != nil { - log.Printf("[Redis] Process exited with error: %v", err) - } else { - log.Println("[Redis] Process exited normally") - } -} diff --git a/internal/watchers/poller.go b/internal/watchers/poller.go index f02fbf4..100fab8 100644 --- a/internal/watchers/poller.go +++ b/internal/watchers/poller.go @@ -68,6 +68,15 @@ func (p *Poller) Start() error { log.Println("[WatcherPoller] Starting file watcher polling service") + // Clear file:events stream on startup to prevent duplicate events from accumulating + // across container restarts. Since in-memory state is empty, all files will be + // re-emitted as "add" events - we need a fresh stream to match. + if p.dispatcher != nil { + if err := p.dispatcher.EmitReset("startup"); err != nil { + log.Printf("[WatcherPoller] Warning: failed to clear stream on startup: %v", err) + } + } + // Initial sync with watcher configurations go p.syncWithWatchers()