From 0403a6f3bc2d3e822721d22a40aee04cd35e58d7 Mon Sep 17 00:00:00 2001 From: Worph Date: Mon, 14 Sep 2026 11:24:46 +0200 Subject: [PATCH] replace file discovery by udp discovery --- cmd/meta-core/main.go | 23 +- dashboard/src/components/ServiceNav.tsx | 190 ++++------------ docker/entrypoint.sh | 1 - docker/leader-election.sh | 36 +-- docker/nginx.conf | 12 - internal/api/handlers.go | 70 +----- internal/api/server.go | 14 -- internal/config/config.go | 16 +- internal/discovery/cleaner.go | 196 ---------------- internal/discovery/service.go | 286 ------------------------ test/config_test.go | 6 - test/election_test.go | 144 ------------ 12 files changed, 61 insertions(+), 933 deletions(-) delete mode 100644 internal/discovery/cleaner.go delete mode 100644 internal/discovery/service.go delete mode 100644 test/election_test.go diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go index f3e1b88..d9c6bf0 100644 --- a/cmd/meta-core/main.go +++ b/cmd/meta-core/main.go @@ -10,7 +10,6 @@ import ( "github.com/metazla/meta-core/internal/api" "github.com/metazla/meta-core/internal/config" - "github.com/metazla/meta-core/internal/discovery" "github.com/metazla/meta-core/internal/leader" "github.com/metazla/meta-core/internal/meshdisco" "github.com/metazla/meta-core/internal/storage" @@ -61,18 +60,6 @@ func main() { } log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion) - // Create and start service discovery - disc := discovery.NewService(cfg) - if err := disc.Start(); err != nil { - log.Fatalf("[meta-core] Failed to start service discovery: %v", err) - } - - // Create and start dead service cleaner - cleaner := discovery.NewCleaner(cfg) - if err := cleaner.Start(); err != nil { - log.Printf("[meta-core] Warning: failed to start service cleaner: %v", err) - } - // UDP service discovery (meta-discovery v1). Runs alongside the file // registry above for now — nothing consumes the wire yet, so this cannot // regress the existing path. The announce payload is rebuilt from @@ -112,7 +99,7 @@ func main() { } // Create and start API server - apiServer := api.NewServer(cfg, leaderProvider, disc, cleaner, storageClient) + apiServer := api.NewServer(cfg, leaderProvider, storageClient) apiServer.SetMeshNode(mesh) if err := apiServer.Start(); err != nil { log.Fatalf("[meta-core] Failed to start API server: %v", err) @@ -136,14 +123,6 @@ func main() { } } - if err := cleaner.Stop(); err != nil { - log.Printf("[meta-core] Error stopping service cleaner: %v", err) - } - - if err := disc.Stop(); err != nil { - log.Printf("[meta-core] Error stopping service discovery: %v", err) - } - if err := storageClient.Close(); err != nil { log.Printf("[meta-core] Error closing storage client: %v", err) } diff --git a/dashboard/src/components/ServiceNav.tsx b/dashboard/src/components/ServiceNav.tsx index c83bd3d..1a3d11f 100644 --- a/dashboard/src/components/ServiceNav.tsx +++ b/dashboard/src/components/ServiceNav.tsx @@ -1,156 +1,54 @@ -import { useState, useEffect } from 'react'; +import { useEffect } from 'react'; +// eslint-disable-next-line @typescript-eslint/ban-ts-comment +// @ts-ignore -- plain ES module, no .d.ts; imported for its side effect +// of defining the custom element. +import '../meta-service-menu.js'; -interface ServiceInfo { - name: string; - hostname: string; - baseUrl: string; - status: string; - lastHeartbeat: string; -} +/** + * ServiceNav — the neighbouring meta-* services, as a burger menu. + * + * Since meta-discovery v1 this is a thin wrapper around , + * a framework-free custom element shared by every meta-* UI (see + * scripts/check-mirrors.sh). The old implementation was ~165 lines of React + * that existed in four near-identical copies and polled meta-core's + * /api/services; the element polls this service's own /api/neighbors instead, + * so the nav still renders when meta-core is down. + * + * The element lives in shadow DOM and cannot see this app's CSS variables, so + * the four --mm-nav-* tokens below map our palette onto it. + */ -interface ServicesResponse { - services: ServiceInfo[]; - current: string; -} - -const serviceIcons: Record = { - 'meta-core': '🗄️', - 'meta-sort': '📁', - 'meta-fuse': '🗂️', - 'meta-stremio': '🎬', - 'meta-orbit': '🌐', - 'default': '📦' -}; - -function formatServiceName(name: string): string { - return name.split('-').map(word => - word.charAt(0).toUpperCase() + word.slice(1) - ).join(' '); +declare global { + // eslint-disable-next-line @typescript-eslint/no-namespace + namespace JSX { + interface IntrinsicElements { + 'meta-service-menu': React.DetailedHTMLProps< + React.HTMLAttributes, + HTMLElement + > & { current?: string; endpoint?: string; label?: string }; + } + } } function ServiceNav() { - const [services, setServices] = useState([]); - const [loading, setLoading] = useState(true); - const currentService = 'meta-core'; + useEffect(() => { + // Import above is for the side effect of defining the element; this + // keeps bundlers from tree-shaking it away in production builds. + }, []); - useEffect(() => { - const fetchServices = async () => { - try { - const response = await fetch('/services'); - if (response.ok) { - const data: ServicesResponse = await response.json(); - setServices(data.services || []); - } - } catch (error) { - console.error('Failed to fetch services:', error); - } finally { - setLoading(false); - } - }; - - fetchServices(); - const interval = setInterval(fetchServices, 30000); - return () => clearInterval(interval); - }, []); - - if (loading) return null; - - const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name)); - - if (sortedServices.length === 0) return null; - - return ( - - ); + `} + + ); } export default ServiceNav; diff --git a/docker/entrypoint.sh b/docker/entrypoint.sh index 6c97864..ce0904d 100644 --- a/docker/entrypoint.sh +++ b/docker/entrypoint.sh @@ -10,7 +10,6 @@ set -e # Create required directories mkdir -p /meta-core/locks mkdir -p /meta-core/db/redis -mkdir -p /meta-core/services mkdir -p /meta-core/mounts/errors mkdir -p /files mkdir -p /var/log/supervisor diff --git a/docker/leader-election.sh b/docker/leader-election.sh index f92a819..12c7135 100644 --- a/docker/leader-election.sh +++ b/docker/leader-election.sh @@ -2,9 +2,13 @@ # ============================================================================= # 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. +# The flock here is a MUTEX, not service discovery. It is the only thing +# stopping two containers that mount the same /meta-core volume from putting +# two Redis writers on one RDB/AOF. Sibling services no longer read anything +# from this directory — they locate meta-core over UDP (meta-discovery v1, see +# docs/project-architecture/service-discovery.md), so the kv-leader.info file +# this script used to publish is gone. Keep the lock; it costs nothing and the +# corruption it prevents is unrecoverable. # # Environment variables: # META_CORE_PATH - Path to meta-core data directory (default: /meta-core) @@ -20,7 +24,6 @@ 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 @@ -40,35 +43,13 @@ get_local_ip() { 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" @@ -79,9 +60,6 @@ while true; do if flock -n 200; then echo "[election] Acquired flock - transitioning to LEADER" - # Write leader info before starting services - write_leader_info - # exec supervisord - replaces this process # When supervisord exits, the flock is released automatically echo "[election] Starting supervisord..." diff --git a/docker/nginx.conf b/docker/nginx.conf index c7bfd54..9a94d64 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -123,18 +123,6 @@ http { # ========================================================================= # Service discovery # ========================================================================= - location /services { - proxy_pass http://meta-core-api/services; - proxy_http_version 1.1; - proxy_set_header Host $host; - } - - location /services/ { - proxy_pass http://meta-core-api/services/; - proxy_http_version 1.1; - proxy_set_header Host $host; - } - # meta-discovery v1 neighbours (supersedes /services, which stays # while consumers migrate). location /neighbors { diff --git a/internal/api/handlers.go b/internal/api/handlers.go index d01afdf..fbb0f9b 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -548,62 +548,13 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) { http.ServeContent(w, r, fileInfo.Name(), fileInfo.ModTime(), file) } -// handleListServices handles GET /services and GET /api/services -// Supports optional ?current= query parameter -func (s *Server) handleListServices(w http.ResponseWriter, r *http.Request) { - services, err := s.discovery.DiscoverAll() - if err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) - return - } - - response := map[string]interface{}{ - "services": services, - "count": len(services), - } - - // Add 'current' field if query parameter is provided - current := r.URL.Query().Get("current") - if current != "" { - response["current"] = current - } - - writeJSON(w, http.StatusOK, response) -} - -// handleGetService handles GET /services/{name} -func (s *Server) handleGetService(w http.ResponseWriter, r *http.Request) { - vars := mux.Vars(r) - name := vars["name"] - - if name == "" { - writeError(w, http.StatusBadRequest, "service name is required") - return - } - - service, err := s.discovery.Discover(name) - if err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) - return - } - - if service == nil { - writeError(w, http.StatusNotFound, "service not found") - return - } - - writeJSON(w, http.StatusOK, service) -} - // handleListNeighbors handles GET /neighbors and GET /api/neighbors. // -// This is the meta-discovery v1 replacement for /api/services: the list is -// built from UDP announces held in memory, not from files under -// /meta-core/services, so it works across stacks that share a network but no -// volume. Instances of the same service are collapsed to the most recent, which -// preserves the one-row-per-service contract the old endpoint had. -// -// A `?all=1` query returns every instance uncollapsed, for debugging. +// meta-discovery v1. The list is built from UDP announces held in memory, not +// from files under /meta-core/services, so it works across stacks that share a +// network but no volume. Instances of the same service are collapsed to the +// most recent, which preserves the one-row-per-service shape the old +// /api/services had. `?all=1` returns every instance, for debugging. func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) { response := map[string]interface{}{ "current": s.config.ServiceName, @@ -633,17 +584,6 @@ func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, response) } -// handleCleanupStats handles GET /services/cleanup/stats -func (s *Server) handleCleanupStats(w http.ResponseWriter, r *http.Request) { - if s.cleaner == nil { - writeError(w, http.StatusServiceUnavailable, "cleaner not initialized") - return - } - - stats := s.cleaner.Stats() - writeJSON(w, http.StatusOK, stats) -} - // writeJSON writes a JSON response func writeJSON(w http.ResponseWriter, status int, data interface{}) { w.Header().Set("Content-Type", "application/json") diff --git a/internal/api/server.go b/internal/api/server.go index 2480e4f..7d032a5 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -9,7 +9,6 @@ import ( "github.com/gorilla/mux" "github.com/metazla/meta-core/internal/config" - "github.com/metazla/meta-core/internal/discovery" "github.com/metazla/meta-core/internal/events" "github.com/metazla/meta-core/internal/identity" "github.com/metazla/meta-core/internal/leader" @@ -26,8 +25,6 @@ import ( type Server struct { config *config.Config leaderProvider *leader.LeaderInfoProvider - discovery *discovery.Service - cleaner *discovery.Cleaner storage *storage.Client mountsManager *mounts.Manager mountsHandlers *mounts.Handlers @@ -65,15 +62,11 @@ func (s *Server) SetMeshNode(n *meshdisco.Node) { func NewServer( cfg *config.Config, leaderProvider *leader.LeaderInfoProvider, - disc *discovery.Service, - cleaner *discovery.Cleaner, stor *storage.Client, ) *Server { s := &Server{ config: cfg, leaderProvider: leaderProvider, - discovery: disc, - cleaner: cleaner, storage: stor, search: &searchIndex{}, router: mux.NewRouter(), @@ -290,13 +283,6 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/api/events/files", s.handleEventsFiles).Methods("GET") s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).Methods("GET") - // Service discovery - // Both /services and /api/services are supported for consistency with other services - s.router.HandleFunc("/services", s.handleListServices).Methods("GET") - s.router.HandleFunc("/api/services", s.handleListServices).Methods("GET") - s.router.HandleFunc("/services/cleanup/stats", s.handleCleanupStats).Methods("GET") - s.router.HandleFunc("/services/{name}", s.handleGetService).Methods("GET") - // meta-discovery v1: neighbours heard over UDP. Supersedes /api/services, // which is still served above while consumers migrate. s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET") diff --git a/internal/config/config.go b/internal/config/config.go index 42139a9..43921b1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -42,7 +42,7 @@ type Config struct { MountsDir string // Path to mounts configuration (default: /meta-core/mounts) // UDP service discovery (meta-discovery v1). Replaces the file-based - // registry under ServicesDir(); see + // registry that used to live under /meta-core/services; see // docs/project-architecture/service-discovery.md. EnableUDPDiscovery bool // Announce + listen on the multicast group (default: true) DiscoveryGroup string // IPv4 multicast group (default: 239.255.77.1) @@ -83,26 +83,18 @@ func Load() *Config { return cfg } -// LockFilePath returns the path to the leader lock file +// LockFilePath returns the path to the Redis-volume mutex. Held by +// docker/leader-election.sh; nothing reads it, and it is NOT service +// discovery — see docs/project-architecture/service-discovery.md. func (c *Config) LockFilePath() string { return c.MetaCorePath + "/locks/kv-leader.lock" } -// InfoFilePath returns the path to the leader info file -func (c *Config) InfoFilePath() string { - return c.MetaCorePath + "/locks/kv-leader.info" -} - // RedisDataDir returns the path to Redis data directory func (c *Config) RedisDataDir() string { return c.MetaCorePath + "/db/redis" } -// ServicesDir returns the path to services directory -func (c *Config) ServicesDir() string { - return c.MetaCorePath + "/services" -} - // MountsFilePath returns the path to the mounts configuration file func (c *Config) MountsFilePath() string { return c.MountsDir + "/mounts.json" diff --git a/internal/discovery/cleaner.go b/internal/discovery/cleaner.go deleted file mode 100644 index 388661b..0000000 --- a/internal/discovery/cleaner.go +++ /dev/null @@ -1,196 +0,0 @@ -package discovery - -import ( - "encoding/json" - "log" - "os" - "path/filepath" - "sync" - "time" - - "github.com/metazla/meta-core/internal/config" -) - -// CleanupStats holds statistics about the cleanup process -type CleanupStats struct { - LastRunTime string `json:"lastRunTime,omitempty"` - ServicesRemoved int `json:"servicesRemoved"` - TotalRuns int `json:"totalRuns"` - Errors int `json:"errors"` -} - -// Cleaner handles periodic cleanup of dead service registration files -type Cleaner struct { - config *config.Config - servicesDir string - ownFile string // Our own service file (never delete) - - stats CleanupStats - stopChan chan struct{} - wg sync.WaitGroup - mu sync.RWMutex -} - -// NewCleaner creates a new service cleaner -func NewCleaner(cfg *config.Config) *Cleaner { - hostname, _ := os.Hostname() - return &Cleaner{ - config: cfg, - servicesDir: cfg.ServicesDir(), - ownFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+"-"+hostname+".json"), - stopChan: make(chan struct{}), - } -} - -// Start begins the cleanup loop -func (c *Cleaner) Start() error { - log.Printf("[Cleaner] Starting dead service cleanup (interval: %dms, threshold: %dms)", - c.config.CleanupIntervalMS, c.config.DeadServiceThresholdMS) - - c.wg.Add(1) - go c.cleanupLoop() - - return nil -} - -// Stop stops the cleanup loop -func (c *Cleaner) Stop() error { - log.Println("[Cleaner] Stopping dead service cleanup...") - close(c.stopChan) - c.wg.Wait() - return nil -} - -// cleanupLoop runs the periodic cleanup -func (c *Cleaner) cleanupLoop() { - defer c.wg.Done() - - // Run an initial cleanup shortly after starting - initialDelay := time.NewTimer(30 * time.Second) - select { - case <-c.stopChan: - initialDelay.Stop() - return - case <-initialDelay.C: - c.runCleanup() - } - - ticker := time.NewTicker(time.Duration(c.config.CleanupIntervalMS) * time.Millisecond) - defer ticker.Stop() - - for { - select { - case <-c.stopChan: - return - case <-ticker.C: - c.runCleanup() - } - } -} - -// runCleanup scans the services directory and removes dead services -func (c *Cleaner) runCleanup() { - c.mu.Lock() - c.stats.TotalRuns++ - c.stats.LastRunTime = time.Now().UTC().Format(time.RFC3339) - c.mu.Unlock() - - entries, err := os.ReadDir(c.servicesDir) - if err != nil { - if !os.IsNotExist(err) { - log.Printf("[Cleaner] Failed to read services directory: %v", err) - c.mu.Lock() - c.stats.Errors++ - c.mu.Unlock() - } - return - } - - threshold := time.Duration(c.config.DeadServiceThresholdMS) * time.Millisecond - removedCount := 0 - - for _, entry := range entries { - if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" { - continue - } - - filePath := filepath.Join(c.servicesDir, entry.Name()) - - // Never delete our own service file - if filePath == c.ownFile { - continue - } - - // Check if service is dead - isDead, serviceName, age := c.isServiceDead(filePath, threshold) - if isDead { - if err := os.Remove(filePath); err != nil { - log.Printf("[Cleaner] Failed to remove dead service file %s: %v", entry.Name(), err) - c.mu.Lock() - c.stats.Errors++ - c.mu.Unlock() - } else { - ageMinutes := age.Minutes() - log.Printf("[Cleaner] Removed dead service: %s (last heartbeat: %.1f minutes ago)", serviceName, ageMinutes) - removedCount++ - } - } - } - - if removedCount > 0 { - c.mu.Lock() - c.stats.ServicesRemoved += removedCount - c.mu.Unlock() - log.Printf("[Cleaner] Cleanup complete: removed %d dead service(s)", removedCount) - } -} - -// isServiceDead checks if a service file represents a dead service -// Returns: isDead, serviceName, age since last heartbeat -func (c *Cleaner) isServiceDead(filePath string, threshold time.Duration) (bool, string, time.Duration) { - data, err := os.ReadFile(filePath) - if err != nil { - // Can't read file - treat as dead to clean up corrupted files - log.Printf("[Cleaner] Cannot read service file %s: %v (treating as dead)", filepath.Base(filePath), err) - return true, filepath.Base(filePath), threshold - } - - var info ServiceInfo - if err := json.Unmarshal(data, &info); err != nil { - // Corrupted JSON - skip rather than delete (might be recoverable) - log.Printf("[Cleaner] Skipping corrupted service file %s: %v", filepath.Base(filePath), err) - return false, "", 0 - } - - // Parse last heartbeat timestamp - if info.LastHeartbeat == "" { - // No heartbeat timestamp - treat as dead - return true, info.Name, threshold - } - - lastHeartbeat, err := time.Parse(time.RFC3339, info.LastHeartbeat) - if err != nil { - // Invalid timestamp - treat as dead - log.Printf("[Cleaner] Invalid timestamp in service file %s: %v (treating as dead)", filepath.Base(filePath), err) - return true, info.Name, threshold - } - - age := time.Since(lastHeartbeat) - if age > threshold { - return true, info.Name, age - } - - return false, info.Name, age -} - -// Stats returns a copy of the current cleanup statistics -func (c *Cleaner) Stats() CleanupStats { - c.mu.RLock() - defer c.mu.RUnlock() - return CleanupStats{ - LastRunTime: c.stats.LastRunTime, - ServicesRemoved: c.stats.ServicesRemoved, - TotalRuns: c.stats.TotalRuns, - Errors: c.stats.Errors, - } -} diff --git a/internal/discovery/service.go b/internal/discovery/service.go deleted file mode 100644 index 84d42d8..0000000 --- a/internal/discovery/service.go +++ /dev/null @@ -1,286 +0,0 @@ -package discovery - -import ( - "encoding/json" - "fmt" - "log" - "net" - "os" - "path/filepath" - "sync" - "time" - - "github.com/metazla/meta-core/internal/config" -) - -// ServiceInfo contains simplified service registration data -// URLs are obtained via the /urls API endpoint -type ServiceInfo struct { - Name string `json:"name"` - Hostname string `json:"hostname"` - BaseUrl string `json:"baseUrl"` - Status string `json:"status"` - LastHeartbeat string `json:"lastHeartbeat"` -} - -// Service handles service registration and discovery -type Service struct { - config *config.Config - servicesDir string - serviceFile string - info *ServiceInfo - mu sync.RWMutex - - stopChan chan struct{} - wg sync.WaitGroup -} - -// NewService creates a new service discovery instance -func NewService(cfg *config.Config) *Service { - hostname, _ := os.Hostname() - return &Service{ - config: cfg, - servicesDir: cfg.ServicesDir(), - serviceFile: filepath.Join(cfg.ServicesDir(), cfg.ServiceName+"-"+hostname+".json"), - stopChan: make(chan struct{}), - } -} - -// Start begins service registration and heartbeat -func (s *Service) Start() error { - log.Printf("[Discovery] Starting service discovery for %s", s.config.ServiceName) - - // Ensure services directory exists - if err := os.MkdirAll(s.servicesDir, 0755); err != nil { - return fmt.Errorf("failed to create services directory: %w", err) - } - - // Build and register service info - s.info = s.buildServiceInfo() - if err := s.register(); err != nil { - return fmt.Errorf("failed to register service: %w", err) - } - - // Start heartbeat loop - s.wg.Add(1) - go s.heartbeatLoop() - - return nil -} - -// Stop stops service discovery and unregisters -func (s *Service) Stop() error { - log.Println("[Discovery] Stopping service discovery...") - close(s.stopChan) - s.wg.Wait() - - // Unregister by removing service file - if err := os.Remove(s.serviceFile); err != nil && !os.IsNotExist(err) { - log.Printf("[Discovery] Failed to remove service file: %v", err) - } - - return nil -} - -// buildServiceInfo creates the service info for this instance -func (s *Service) buildServiceInfo() *ServiceInfo { - hostname, _ := os.Hostname() - ip := getLocalIP() - - // Browser-facing nav URL: prefer META_CORE_PUBLIC_URL (the reachable URL - // when there's no Caddy perimeter in front — e.g. a debug-direct port), - // then BASE_URL (the Caddy perimeter URL), then the internal IP. - var baseUrl string - if s.config.PublicURL != "" { - baseUrl = s.config.PublicURL - } else if s.config.BaseURL != "" { - baseUrl = s.config.BaseURL - } else { - baseUrl = fmt.Sprintf("http://%s:%d", ip, s.config.APIPort) - } - - info := &ServiceInfo{ - Name: s.config.ServiceName, - Hostname: hostname, - BaseUrl: baseUrl, - Status: "running", - LastHeartbeat: time.Now().UTC().Format(time.RFC3339), - } - - return info -} - -// register writes service info to file -func (s *Service) register() error { - s.mu.Lock() - defer s.mu.Unlock() - - return s.writeServiceInfo(s.info) -} - -// writeServiceInfo atomically writes service info to file -func (s *Service) writeServiceInfo(info *ServiceInfo) error { - tempPath := s.serviceFile + ".tmp" - - data, err := json.MarshalIndent(info, "", " ") - if err != nil { - return err - } - - if err := os.WriteFile(tempPath, data, 0644); err != nil { - return err - } - - return os.Rename(tempPath, s.serviceFile) -} - -// heartbeatLoop periodically updates the last heartbeat -func (s *Service) heartbeatLoop() { - defer s.wg.Done() - - ticker := time.NewTicker(time.Duration(s.config.HeartbeatIntervalMS) * time.Millisecond) - defer ticker.Stop() - - for { - select { - case <-s.stopChan: - return - case <-ticker.C: - s.heartbeat() - } - } -} - -// heartbeat updates the last heartbeat timestamp -func (s *Service) heartbeat() { - s.mu.Lock() - defer s.mu.Unlock() - - if s.info == nil { - return - } - - s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339) - - if err := s.writeServiceInfo(s.info); err != nil { - log.Printf("[Discovery] Failed to update heartbeat: %v", err) - } -} - -// UpdateStatus updates the service status -func (s *Service) UpdateStatus(status string) error { - s.mu.Lock() - defer s.mu.Unlock() - - if s.info == nil { - return fmt.Errorf("service not registered") - } - - s.info.Status = status - s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339) - return s.writeServiceInfo(s.info) -} - -// Discover finds a service by name -// Looks for files matching pattern: name-*.json (hostname-based naming) -func (s *Service) Discover(name string) (*ServiceInfo, error) { - // First try exact match for backward compatibility - exactPath := filepath.Join(s.servicesDir, name+".json") - if data, err := os.ReadFile(exactPath); err == nil { - var info ServiceInfo - if err := json.Unmarshal(data, &info); err == 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 the first valid service found - for _, match := range matches { - 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 - } - - 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) - if err == nil { - staleThreshold := time.Duration(s.config.StaleThresholdMS) * time.Millisecond - if time.Since(lastHeartbeat) > staleThreshold { - info.Status = "stale" - } - } - return info -} - -// DiscoverAll finds all registered services -func (s *Service) DiscoverAll() ([]*ServiceInfo, error) { - entries, err := os.ReadDir(s.servicesDir) - if err != nil { - if os.IsNotExist(err) { - return []*ServiceInfo{}, nil - } - return nil, err - } - - var services []*ServiceInfo - for _, entry := range entries { - if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" { - continue - } - - filePath := filepath.Join(s.servicesDir, entry.Name()) - data, err := os.ReadFile(filePath) - if err != nil { - log.Printf("[Discovery] Failed to read service file %s: %v", entry.Name(), err) - continue - } - - var info ServiceInfo - 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 -} - -// getLocalIP returns the local IP address -func getLocalIP() string { - addrs, err := net.InterfaceAddrs() - if err != nil { - hostname, _ := os.Hostname() - return hostname - } - - for _, addr := range addrs { - if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() { - if ipnet.IP.To4() != nil { - return ipnet.IP.String() - } - } - } - - hostname, _ := os.Hostname() - return hostname -} diff --git a/test/config_test.go b/test/config_test.go index a75b115..eb64c83 100644 --- a/test/config_test.go +++ b/test/config_test.go @@ -77,15 +77,9 @@ func TestConfigPaths(t *testing.T) { t.Errorf("Unexpected LockFilePath: %s", cfg.LockFilePath()) } - if cfg.InfoFilePath() != "/test/locks/kv-leader.info" { - t.Errorf("Unexpected InfoFilePath: %s", cfg.InfoFilePath()) - } if cfg.RedisDataDir() != "/test/db/redis" { t.Errorf("Unexpected RedisDataDir: %s", cfg.RedisDataDir()) } - if cfg.ServicesDir() != "/test/services" { - t.Errorf("Unexpected ServicesDir: %s", cfg.ServicesDir()) - } } diff --git a/test/election_test.go b/test/election_test.go deleted file mode 100644 index 740da8e..0000000 --- a/test/election_test.go +++ /dev/null @@ -1,144 +0,0 @@ -package test - -import ( - "encoding/json" - "os" - "path/filepath" - "testing" - - "github.com/metazla/meta-core/internal/leader" -) - -func TestLeaderLockInfoJSON(t *testing.T) { - info := &leader.LeaderLockInfo{ - Host: "test-host", - API: "redis://10.0.0.1:6379", - HTTP: "http://10.0.0.1:8180", - BaseURL: "http://localhost:8180", - Timestamp: 1704067200000, - PID: 12345, - } - - // Marshal to JSON - data, err := json.MarshalIndent(info, "", " ") - if err != nil { - t.Fatalf("Failed to marshal: %v", err) - } - - // Verify JSON structure matches TypeScript format - var parsed map[string]interface{} - if err := json.Unmarshal(data, &parsed); err != nil { - t.Fatalf("Failed to unmarshal: %v", err) - } - - if parsed["host"] != "test-host" { - t.Errorf("Expected host 'test-host', got '%v'", parsed["host"]) - } - - if parsed["api"] != "redis://10.0.0.1:6379" { - t.Errorf("Expected api 'redis://10.0.0.1:6379', got '%v'", parsed["api"]) - } - - if parsed["http"] != "http://10.0.0.1:8180" { - t.Errorf("Expected http 'http://10.0.0.1:8180', got '%v'", parsed["http"]) - } - - if parsed["baseUrl"] != "http://localhost:8180" { - t.Errorf("Expected baseUrl 'http://localhost:8180', got '%v'", parsed["baseUrl"]) - } - - // Timestamp should be a number - if _, ok := parsed["timestamp"].(float64); !ok { - t.Errorf("Expected timestamp to be a number, got %T", parsed["timestamp"]) - } - - // PID should be a number - if _, ok := parsed["pid"].(float64); !ok { - t.Errorf("Expected pid to be a number, got %T", parsed["pid"]) - } -} - -func TestLeaderLockInfoJSONOmitEmpty(t *testing.T) { - info := &leader.LeaderLockInfo{ - Host: "test-host", - API: "redis://10.0.0.1:6379", - HTTP: "http://10.0.0.1:8180", - BaseURL: "", // Empty - should be omitted - Timestamp: 1704067200000, - PID: 12345, - } - - data, err := json.Marshal(info) - if err != nil { - t.Fatalf("Failed to marshal: %v", err) - } - - // baseUrl should not be present when empty - var parsed map[string]interface{} - if err := json.Unmarshal(data, &parsed); err != nil { - t.Fatalf("Failed to unmarshal: %v", err) - } - - if _, exists := parsed["baseUrl"]; exists { - t.Error("Expected baseUrl to be omitted when empty") - } -} - -func TestLeaderInfoFileFormat(t *testing.T) { - // Create a temp directory - tmpDir, err := os.MkdirTemp("", "meta-core-test") - if err != nil { - t.Fatalf("Failed to create temp dir: %v", err) - } - defer os.RemoveAll(tmpDir) - - info := &leader.LeaderLockInfo{ - Host: "test-host", - API: "redis://10.0.0.1:6379", - HTTP: "http://10.0.0.1:8180", - Timestamp: 1704067200000, - PID: 12345, - } - - // Write to file - infoPath := filepath.Join(tmpDir, "kv-leader.info") - data, err := json.MarshalIndent(info, "", " ") - if err != nil { - t.Fatalf("Failed to marshal: %v", err) - } - - if err := os.WriteFile(infoPath, data, 0644); err != nil { - t.Fatalf("Failed to write file: %v", err) - } - - // Read and parse - readData, err := os.ReadFile(infoPath) - if err != nil { - t.Fatalf("Failed to read file: %v", err) - } - - var parsed leader.LeaderLockInfo - if err := json.Unmarshal(readData, &parsed); err != nil { - t.Fatalf("Failed to unmarshal: %v", err) - } - - if parsed.Host != info.Host { - t.Errorf("Host mismatch: expected '%s', got '%s'", info.Host, parsed.Host) - } - - if parsed.API != info.API { - t.Errorf("API mismatch: expected '%s', got '%s'", info.API, parsed.API) - } - - if parsed.HTTP != info.HTTP { - t.Errorf("HTTP mismatch: expected '%s', got '%s'", info.HTTP, parsed.HTTP) - } - - if parsed.Timestamp != info.Timestamp { - t.Errorf("Timestamp mismatch: expected %d, got %d", info.Timestamp, parsed.Timestamp) - } - - if parsed.PID != info.PID { - t.Errorf("PID mismatch: expected %d, got %d", info.PID, parsed.PID) - } -}