diff --git a/cmd/meta-core/main.go b/cmd/meta-core/main.go
index 57b134d..452d8d5 100644
--- a/cmd/meta-core/main.go
+++ b/cmd/meta-core/main.go
@@ -53,12 +53,19 @@ func main() {
// Create and start service discovery
disc := discovery.NewService(cfg)
+ disc.SetRoleProvider(&roleProviderAdapter{election: election})
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)
+ }
+
// Create and start API server
- apiServer := api.NewServer(cfg, election, disc, storageClient)
+ apiServer := api.NewServer(cfg, election, disc, cleaner, storageClient)
if err := apiServer.Start(); err != nil {
log.Fatalf("[meta-core] Failed to start API server: %v", err)
}
@@ -75,6 +82,10 @@ func main() {
log.Printf("[meta-core] Error stopping API server: %v", err)
}
+ 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)
}
@@ -109,3 +120,12 @@ func (a *storageConnectorAdapter) Connect(url string) error {
func (a *storageConnectorAdapter) Close() error {
return a.client.Close()
}
+
+// roleProviderAdapter adapts leader.Election to discovery.RoleProvider
+type roleProviderAdapter struct {
+ election *leader.Election
+}
+
+func (a *roleProviderAdapter) Role() string {
+ return string(a.election.Role())
+}
diff --git a/dashboard/dist/index.html b/dashboard/dist/index.html
index 76d200c..137e840 100644
--- a/dashboard/dist/index.html
+++ b/dashboard/dist/index.html
@@ -21,7 +21,7 @@
min-height: 100vh;
}
-
+
diff --git a/dashboard/src/components/ServiceNav.tsx b/dashboard/src/components/ServiceNav.tsx
index f040543..f8ab3fa 100644
--- a/dashboard/src/components/ServiceNav.tsx
+++ b/dashboard/src/components/ServiceNav.tsx
@@ -2,10 +2,11 @@ import { useState, useEffect } from 'react';
interface ServiceInfo {
name: string;
- api: string;
+ hostname: string;
+ baseUrl: string;
status: string;
- capabilities: string[];
- version: string;
+ lastHeartbeat: string;
+ role?: string; // "leader", "follower", or undefined (for non-meta-core services)
}
interface ServicesResponse {
@@ -55,7 +56,11 @@ function ServiceNav() {
if (loading) return null;
- const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name));
+ // Filter out follower meta-core instances (only show leader)
+ const filteredServices = services.filter(
+ s => s.name !== 'meta-core' || s.role === 'leader'
+ );
+ const sortedServices = [...filteredServices].sort((a, b) => a.name.localeCompare(b.name));
if (sortedServices.length === 0) return null;
@@ -69,12 +74,12 @@ function ServiceNav() {
return (
{
e.preventDefault();
if (!isActive) {
- window.location.href = service.api;
+ window.location.href = service.baseUrl;
}
}}
>
diff --git a/internal/api/handlers.go b/internal/api/handlers.go
index 6573724..19eb92a 100644
--- a/internal/api/handlers.go
+++ b/internal/api/handlers.go
@@ -449,6 +449,17 @@ func (s *Server) handleGetService(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, service)
}
+// 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 1733dfc..42722fb 100644
--- a/internal/api/server.go
+++ b/internal/api/server.go
@@ -21,6 +21,7 @@ type Server struct {
config *config.Config
election *leader.Election
discovery *discovery.Service
+ cleaner *discovery.Cleaner
storage *storage.Client
mountsManager *mounts.Manager
mountsHandlers *mounts.Handlers
@@ -37,12 +38,14 @@ func NewServer(
cfg *config.Config,
election *leader.Election,
disc *discovery.Service,
+ cleaner *discovery.Cleaner,
stor *storage.Client,
) *Server {
s := &Server{
config: cfg,
election: election,
discovery: disc,
+ cleaner: cleaner,
storage: stor,
router: mux.NewRouter(),
}
@@ -130,6 +133,7 @@ func (s *Server) setupRoutes() {
// Service discovery
s.router.HandleFunc("/services", s.handleListServices).Methods("GET")
+ s.router.HandleFunc("/services/cleanup/stats", s.handleCleanupStats).Methods("GET")
s.router.HandleFunc("/services/{name}", s.handleGetService).Methods("GET")
// Mount management routes (if manager initialized)
diff --git a/internal/config/config.go b/internal/config/config.go
index 3f2be70..c3081a5 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -26,9 +26,11 @@ type Config struct {
HTTPHost string // HTTP API host (default: "127.0.0.1")
// Timing configuration
- HealthCheckIntervalMS int // Health check interval in ms (default: 5000)
- HeartbeatIntervalMS int // Service heartbeat interval in ms (default: 30000)
- StaleThresholdMS int // Stale service threshold in ms (default: 60000)
+ HealthCheckIntervalMS int // Health check interval in ms (default: 5000)
+ HeartbeatIntervalMS int // Service heartbeat interval in ms (default: 30000)
+ StaleThresholdMS int // Stale service threshold in ms (default: 60000)
+ CleanupIntervalMS int // Dead service cleanup interval in ms (default: 600000)
+ DeadServiceThresholdMS int // Dead service threshold in ms (default: 600000)
// File watcher configuration
WatchFolderList []string // List of folders to watch for file changes
@@ -52,9 +54,11 @@ func Load() *Config {
RedisPort: getEnvInt("REDIS_PORT", 6379),
HTTPPort: getEnvInt("META_CORE_HTTP_PORT", 9000),
HTTPHost: getEnv("META_CORE_HTTP_HOST", "127.0.0.1"),
- HealthCheckIntervalMS: getEnvInt("HEALTH_CHECK_INTERVAL_MS", 5000),
+ HealthCheckIntervalMS: getEnvInt("HEALTH_CHECK_INTERVAL_MS", 5000),
HeartbeatIntervalMS: getEnvInt("HEARTBEAT_INTERVAL_MS", 30000),
StaleThresholdMS: getEnvInt("STALE_THRESHOLD_MS", 60000),
+ CleanupIntervalMS: getEnvInt("CLEANUP_INTERVAL_MS", 600000),
+ DeadServiceThresholdMS: getEnvInt("DEAD_SERVICE_THRESHOLD_MS", 600000),
WatchIntervalMS: getEnvInt("WATCH_INTERVAL_MS", 1000),
DebounceMS: getEnvInt("DEBOUNCE_MS", 30000),
EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true),
diff --git a/internal/discovery/cleaner.go b/internal/discovery/cleaner.go
new file mode 100644
index 0000000..388661b
--- /dev/null
+++ b/internal/discovery/cleaner.go
@@ -0,0 +1,196 @@
+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
index c011df2..c951994 100644
--- a/internal/discovery/service.go
+++ b/internal/discovery/service.go
@@ -21,15 +21,22 @@ type ServiceInfo struct {
BaseUrl string `json:"baseUrl"`
Status string `json:"status"`
LastHeartbeat string `json:"lastHeartbeat"`
+ Role string `json:"role,omitempty"` // "leader", "follower", or empty (for non-meta-core services)
+}
+
+// RoleProvider interface for getting the current role
+type RoleProvider interface {
+ Role() string
}
// Service handles service registration and discovery
type Service struct {
- config *config.Config
- servicesDir string
- serviceFile string
- info *ServiceInfo
- mu sync.RWMutex
+ config *config.Config
+ servicesDir string
+ serviceFile string
+ info *ServiceInfo
+ roleProvider RoleProvider
+ mu sync.RWMutex
stopChan chan struct{}
wg sync.WaitGroup
@@ -46,6 +53,11 @@ func NewService(cfg *config.Config) *Service {
}
}
+// SetRoleProvider sets the role provider for this service
+func (s *Service) SetRoleProvider(rp RoleProvider) {
+ s.roleProvider = rp
+}
+
// Start begins service registration and heartbeat
func (s *Service) Start() error {
log.Printf("[Discovery] Starting service discovery for %s", s.config.ServiceName)
@@ -95,13 +107,20 @@ func (s *Service) buildServiceInfo() *ServiceInfo {
baseUrl = fmt.Sprintf("http://%s:%d", ip, s.config.APIPort)
}
- return &ServiceInfo{
+ info := &ServiceInfo{
Name: s.config.ServiceName,
Hostname: hostname,
BaseUrl: baseUrl,
Status: "running",
LastHeartbeat: time.Now().UTC().Format(time.RFC3339),
}
+
+ // Add role if role provider is set (e.g., for meta-core)
+ if s.roleProvider != nil {
+ info.Role = s.roleProvider.Role()
+ }
+
+ return info
}
// register writes service info to file
@@ -145,7 +164,7 @@ func (s *Service) heartbeatLoop() {
}
}
-// heartbeat updates the last heartbeat timestamp
+// heartbeat updates the last heartbeat timestamp and role
func (s *Service) heartbeat() {
s.mu.Lock()
defer s.mu.Unlock()
@@ -155,6 +174,12 @@ func (s *Service) heartbeat() {
}
s.info.LastHeartbeat = time.Now().UTC().Format(time.RFC3339)
+
+ // Update role in case it changed (e.g., leader election)
+ if s.roleProvider != nil {
+ s.info.Role = s.roleProvider.Role()
+ }
+
if err := s.writeServiceInfo(s.info); err != nil {
log.Printf("[Discovery] Failed to update heartbeat: %v", err)
}
diff --git a/internal/leader/election.go b/internal/leader/election.go
index e96b250..51c424f 100644
--- a/internal/leader/election.go
+++ b/internal/leader/election.go
@@ -1,12 +1,16 @@
package leader
import (
+ "context"
"encoding/json"
"fmt"
+ "io"
"log"
"net"
+ "net/http"
"os"
"path/filepath"
+ "strings"
"sync"
"syscall"
"time"
@@ -311,17 +315,24 @@ func (e *Election) buildLeaderInfo() *LeaderLockInfo {
}
}
-// writeLeaderInfo atomically writes leader info to file
+// 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()
- tempPath := infoPath + ".tmp"
- data, err := json.MarshalIndent(info, "", " ")
- if err != nil {
- return err
+ // 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
}
- if err := os.WriteFile(tempPath, data, 0644); err != nil {
+ // Content changed or file doesn't exist, write atomically
+ tempPath := infoPath + ".tmp"
+ if err := os.WriteFile(tempPath, newData, 0644); err != nil {
return err
}
@@ -329,11 +340,11 @@ func (e *Election) writeLeaderInfo(info *LeaderLockInfo) error {
return err
}
- log.Printf("[Election] Wrote leader info to %s", infoPath)
+ log.Printf("[Election] Wrote leader API URL to %s: %s", infoPath, info.ApiUrl)
return nil
}
-// readLeaderInfo reads leader info from file
+// readLeaderInfo reads leader API URL from file and fetches full info via /urls API
func (e *Election) readLeaderInfo() (*LeaderLockInfo, error) {
infoPath := e.config.InfoFilePath()
@@ -345,12 +356,65 @@ func (e *Election) readLeaderInfo() (*LeaderLockInfo, error) {
return nil, err
}
- var info LeaderLockInfo
- if err := json.Unmarshal(data, &info); err != nil {
- return nil, err
+ // File contains plain text API URL
+ apiUrl := strings.TrimSpace(string(data))
+ if apiUrl == "" {
+ return nil, nil
}
- return &info, 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