registry/cmd/registry/main.go
Radoslav Dimitrov b24d6c7657
fix: harden registry availability and add publish phase timings (#1211)
## Summary

Daily ~17:00 UTC `Publish Endpoint Latency` and `Availability < 95%`
alerts are caused by both registry pods exit-0'ing within seconds of
each other and racing with the still-recovering `registry-pg-rw`
endpoint on restart. Separately, the default compute SA had no project
IAM roles, so `fluentbit-gke` and the GMP collector were silently
dropping every log line and time series — which is why none of this was
visible in Cloud Logging.

All Kubernetes/IAM changes apply uniformly to staging and prod.

## Changes

- **GCP IAM** (`deploy/pkg/providers/gcp/provider.go`) — grant the
default compute SA `logging.logWriter`, `monitoring.metricWriter`,
`monitoring.viewer`, and `stackdriver.resourceMetadata.writer`. Already
applied live; this just encodes it in Pulumi.
- **Pod resilience** (`deploy/pkg/k8s/registry.go`) —
- `PodDisruptionBudget` (`minAvailable: 1`) prevents a synchronized
voluntary disruption.
- `TopologySpreadConstraints` (`ScheduleAnyway`, hostname) keeps the PDB
from deadlocking node drains when both pods co-locate.
- `StartupProbe` (30 × 5s = 150s) protects the DB-retry budget from the
liveness probe.
  - Memory bump 128→256Mi request, 256→512Mi limit.
- **Startup retry** (`cmd/registry/main.go`) — replace silent `return`
on initial DB-ping failure with bounded retry-with-backoff (8 attempts,
1→8s capped, 10s per-attempt). A brief endpoint flap no longer cascades
into a restart loop.
- **Diagnostic log** (`internal/service/registry_service.go`) — emit
`validate_ms` on each publish (success and failure). Validate fans out
to npm/PyPI/OCI with 10s per-host timeouts and is the most likely
contributor to publish latency. Total request duration is already
covered by the existing `http.request.duration` histogram.

## Calibration

The IAM, PDB, topology-spread, startup-probe, and DB-retry pieces are
mitigations — they reduce blast radius regardless of root cause. The
`validate_ms` log is the diagnostic for identifying the trigger when the
alert fires next. Single-PG SPOF (`deploy/pkg/k8s/postgres.go:57`)
remains a follow-up; today's evidence shows PG was a victim, not the
cause.

## Test plan

- [x] `go build ./...` and `go vet ./...` clean for both modules
- [x] `make lint` clean (12 pre-existing issues under `deploy/` are not
in `make lint`)
- [x] `go test -race -count=1 ./internal/... ./cmd/...` green
- [x] IAM applied live; `fluentbit-gke` and GMP no longer hitting
`IAM_PERMISSION_DENIED`
- [ ] After deploy, verify the next 17:00 UTC firing surfaces
`validate_ms` in Cloud Logging

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-27 09:09:50 -04:00

157 lines
4.4 KiB
Go

package main
import (
"context"
"errors"
"flag"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/modelcontextprotocol/registry/internal/api"
v0 "github.com/modelcontextprotocol/registry/internal/api/handlers/v0"
"github.com/modelcontextprotocol/registry/internal/config"
"github.com/modelcontextprotocol/registry/internal/database"
"github.com/modelcontextprotocol/registry/internal/importer"
"github.com/modelcontextprotocol/registry/internal/service"
"github.com/modelcontextprotocol/registry/internal/telemetry"
)
// Version info for the MCP Registry application
// These variables are injected at build time via ldflags
var (
// Version is the current version of the MCP Registry application
Version = "dev"
// BuildTime is the time at which the binary was built
BuildTime = "unknown"
// GitCommit is the git commit that was compiled
GitCommit = "unknown"
)
func main() {
// Parse command line flags
showVersion := flag.Bool("version", false, "Display version information")
flag.Parse()
// Show version information if requested
if *showVersion {
log.Printf("MCP Registry %s\n", Version)
log.Printf("Git commit: %s\n", GitCommit)
log.Printf("Build time: %s\n", BuildTime)
return
}
log.Printf("Starting MCP Registry Application v%s (commit: %s)", Version, GitCommit)
var (
registryService service.RegistryService
db database.Database
err error
)
// Initialize configuration
cfg := config.NewConfig()
// Connect to PostgreSQL with bounded retry. The DB endpoint can flap briefly
// during voluntary disruptions (node drains, rollouts) and without retry the
// pod exits cleanly, restarts, then races with the still-recovering service
// — turning a transient blip into a multi-minute outage.
const (
maxStartupAttempts = 8
perAttemptTimeout = 10 * time.Second
initialBackoff = 1 * time.Second
maxBackoff = 8 * time.Second
)
backoff := initialBackoff
for attempt := 1; ; attempt++ {
attemptCtx, attemptCancel := context.WithTimeout(context.Background(), perAttemptTimeout)
db, err = database.NewPostgreSQL(attemptCtx, cfg.DatabaseURL)
attemptCancel()
if err == nil {
break
}
if attempt >= maxStartupAttempts {
log.Printf("Failed to connect to PostgreSQL after %d attempts: %v", attempt, err)
return
}
log.Printf("PostgreSQL not reachable (attempt %d/%d): %v, retrying in %s", attempt, maxStartupAttempts, err, backoff)
time.Sleep(backoff)
backoff = min(backoff*2, maxBackoff)
}
// Store the PostgreSQL instance for later cleanup
defer func() {
if err := db.Close(); err != nil {
log.Printf("Error closing PostgreSQL connection: %v", err)
} else {
log.Println("PostgreSQL connection closed successfully")
}
}()
registryService = service.NewRegistryService(db, cfg)
// Import seed data if seed source is provided
if cfg.SeedFrom != "" {
log.Printf("Importing data from %s...", cfg.SeedFrom)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
importerService := importer.NewService(registryService)
if err := importerService.ImportFromPath(ctx, cfg.SeedFrom); err != nil {
log.Printf("Failed to import seed data: %v", err)
}
}
shutdownTelemetry, metrics, err := telemetry.InitMetrics(cfg.Version)
if err != nil {
log.Printf("Failed to initialize metrics: %v", err)
return
}
defer func() {
if err := shutdownTelemetry(context.Background()); err != nil {
log.Printf("Failed to shutdown telemetry: %v", err)
}
}()
// Prepare version information
versionInfo := &v0.VersionBody{
Version: Version,
GitCommit: GitCommit,
BuildTime: BuildTime,
}
// Initialize HTTP server
server := api.NewServer(cfg, registryService, metrics, versionInfo)
// Start server in a goroutine so it doesn't block signal handling
go func() {
if err := server.Start(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Printf("Failed to start server: %v", err)
os.Exit(1)
}
}()
// Wait for interrupt signal to gracefully shutdown the server
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("Shutting down server...")
// Create context with timeout for shutdown
sctx, scancel := context.WithTimeout(context.Background(), 10*time.Second)
defer scancel()
// Gracefully shutdown the server
if err := server.Shutdown(sctx); err != nil {
log.Printf("Server forced to shutdown: %v", err)
}
log.Println("Server exiting")
}