replace file discovery by udp discovery
This commit is contained in:
Worph
+1
-22
@@ -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)
|
||||
}
|
||||
|
||||
@@ -1,155 +1,53 @@
|
||||
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 <meta-service-menu>,
|
||||
* 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.
|
||||
*/
|
||||
|
||||
declare global {
|
||||
// eslint-disable-next-line @typescript-eslint/no-namespace
|
||||
namespace JSX {
|
||||
interface IntrinsicElements {
|
||||
'meta-service-menu': React.DetailedHTMLProps<
|
||||
React.HTMLAttributes<HTMLElement>,
|
||||
HTMLElement
|
||||
> & { current?: string; endpoint?: string; label?: string };
|
||||
}
|
||||
|
||||
interface ServicesResponse {
|
||||
services: ServiceInfo[];
|
||||
current: string;
|
||||
}
|
||||
|
||||
const serviceIcons: Record<string, string> = {
|
||||
'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(' ');
|
||||
}
|
||||
|
||||
function ServiceNav() {
|
||||
const [services, setServices] = useState<ServiceInfo[]>([]);
|
||||
const [loading, setLoading] = useState(true);
|
||||
const currentService = 'meta-core';
|
||||
|
||||
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);
|
||||
// Import above is for the side effect of defining the element; this
|
||||
// keeps bundlers from tree-shaking it away in production builds.
|
||||
}, []);
|
||||
|
||||
if (loading) return null;
|
||||
|
||||
const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name));
|
||||
|
||||
if (sortedServices.length === 0) return null;
|
||||
|
||||
return (
|
||||
<nav className="services-nav">
|
||||
<span className="services-nav-label">Services:</span>
|
||||
<div className="services-nav-items">
|
||||
{sortedServices.map(service => {
|
||||
const icon = serviceIcons[service.name] || serviceIcons.default;
|
||||
const isActive = service.name === currentService;
|
||||
return (
|
||||
<a
|
||||
key={service.name}
|
||||
href={isActive ? '#' : service.baseUrl}
|
||||
className={`service-link${isActive ? ' active' : ''}`}
|
||||
onClick={(e) => {
|
||||
e.preventDefault();
|
||||
if (!isActive) {
|
||||
window.location.href = service.baseUrl;
|
||||
}
|
||||
}}
|
||||
>
|
||||
<span className="service-icon">{icon}</span>
|
||||
<span>{formatServiceName(service.name)}</span>
|
||||
<span className="service-status"></span>
|
||||
</a>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
|
||||
<>
|
||||
<meta-service-menu current="meta-core" />
|
||||
<style>{`
|
||||
.services-nav {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: 0.5rem;
|
||||
}
|
||||
|
||||
.services-nav-label {
|
||||
color: #888;
|
||||
font-size: 0.8rem;
|
||||
white-space: nowrap;
|
||||
}
|
||||
|
||||
.services-nav-items {
|
||||
display: flex;
|
||||
gap: 0.35rem;
|
||||
}
|
||||
|
||||
.service-link {
|
||||
display: inline-flex;
|
||||
align-items: center;
|
||||
gap: 0.3rem;
|
||||
padding: 0.35rem 0.6rem;
|
||||
border-radius: 6px;
|
||||
text-decoration: none;
|
||||
font-size: 0.8rem;
|
||||
font-weight: 500;
|
||||
transition: all 0.2s;
|
||||
color: #ccc;
|
||||
background: #0f3460;
|
||||
border: 1px solid #1a4a7a;
|
||||
}
|
||||
|
||||
.service-link:hover {
|
||||
background: #1a4a7a;
|
||||
border-color: rgba(78, 205, 196, 0.5);
|
||||
text-decoration: none;
|
||||
color: #fff;
|
||||
}
|
||||
|
||||
.service-link.active {
|
||||
background: linear-gradient(135deg, rgba(78, 205, 196, 0.2), rgba(68, 160, 141, 0.2));
|
||||
border-color: rgba(78, 205, 196, 0.5);
|
||||
color: #fff;
|
||||
}
|
||||
|
||||
.service-link .service-icon {
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
|
||||
.service-link .service-status {
|
||||
width: 5px;
|
||||
height: 5px;
|
||||
border-radius: 50%;
|
||||
background: #4ecdc4;
|
||||
}
|
||||
|
||||
@media (max-width: 800px) {
|
||||
.services-nav-label {
|
||||
display: none;
|
||||
}
|
||||
meta-service-menu {
|
||||
--mm-nav-fg: var(--text-primary, #e0e0e0);
|
||||
--mm-nav-bg: var(--bg-tertiary, var(--bg-secondary, #1a1a2e));
|
||||
--mm-nav-border: var(--border-color, rgba(255,255,255,0.14));
|
||||
--mm-nav-accent: var(--accent-primary, #4ecdc4);
|
||||
}
|
||||
`}</style>
|
||||
</nav>
|
||||
</>
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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..."
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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=<service-name> 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")
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user