diff --git a/cmd/registry/main.go b/cmd/registry/main.go index 3eee551..ed68abf 100644 --- a/cmd/registry/main.go +++ b/cmd/registry/main.go @@ -57,15 +57,31 @@ func main() { // Initialize configuration cfg := config.NewConfig() - // Create a context with timeout for PostgreSQL connection - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - - // Connect to PostgreSQL - db, err = database.NewPostgreSQL(ctx, cfg.DatabaseURL) - if err != nil { - log.Printf("Failed to connect to PostgreSQL: %v", err) - return + // 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 diff --git a/deploy/pkg/k8s/registry.go b/deploy/pkg/k8s/registry.go index 4655203..579f740 100644 --- a/deploy/pkg/k8s/registry.go +++ b/deploy/pkg/k8s/registry.go @@ -10,6 +10,7 @@ import ( "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/helm/v3" metav1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/meta/v1" networkingv1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/networking/v1" + policyv1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/policy/v1" "github.com/pulumi/pulumi/sdk/v3/go/pulumi" "github.com/pulumi/pulumi/sdk/v3/go/pulumi/config" @@ -95,6 +96,22 @@ func DeployMCPRegistry(ctx *pulumi.Context, cluster *providers.ProviderInfo, env }, }, Spec: &corev1.PodSpecArgs{ + // Soft-spread pods across nodes so the PodDisruptionBudget below can't + // deadlock a node drain when both pods happen to co-locate. + // `ScheduleAnyway` keeps single-node clusters (e.g. constrained dev + // envs) schedulable while still preferring spread under normal load. + TopologySpreadConstraints: corev1.TopologySpreadConstraintArray{ + &corev1.TopologySpreadConstraintArgs{ + MaxSkew: pulumi.Int(1), + TopologyKey: pulumi.String("kubernetes.io/hostname"), + WhenUnsatisfiable: pulumi.String("ScheduleAnyway"), + LabelSelector: &metav1.LabelSelectorArgs{ + MatchLabels: pulumi.StringMap{ + "app": pulumi.String("mcp-registry"), + }, + }, + }, + }, Containers: corev1.ContainerArray{ &corev1.ContainerArgs{ Name: pulumi.String("mcp-registry"), @@ -169,6 +186,18 @@ func DeployMCPRegistry(ctx *pulumi.Context, cluster *providers.ProviderInfo, env Value: pulumi.String("*"), }, }, + // StartupProbe protects the DB-retry budget in cmd/registry/main.go: + // 30 × 5s = 150s allowed before liveness/readiness take over, which + // covers the worst-case retry budget (8 attempts × 10s + ~40s sleep). + StartupProbe: &corev1.ProbeArgs{ + HttpGet: &corev1.HTTPGetActionArgs{ + Path: pulumi.String("/v0/health"), + Port: pulumi.Int(8080), + }, + PeriodSeconds: pulumi.Int(5), + FailureThreshold: pulumi.Int(30), + TimeoutSeconds: pulumi.Int(3), + }, LivenessProbe: &corev1.ProbeArgs{ HttpGet: &corev1.HTTPGetActionArgs{ Path: pulumi.String("/v0/health"), @@ -187,11 +216,11 @@ func DeployMCPRegistry(ctx *pulumi.Context, cluster *providers.ProviderInfo, env }, Resources: &corev1.ResourceRequirementsArgs{ Requests: pulumi.StringMap{ - "memory": pulumi.String("128Mi"), + "memory": pulumi.String("256Mi"), "cpu": pulumi.String("100m"), }, Limits: pulumi.StringMap{ - "memory": pulumi.String("256Mi"), + "memory": pulumi.String("512Mi"), }, }, }, @@ -204,6 +233,31 @@ func DeployMCPRegistry(ctx *pulumi.Context, cluster *providers.ProviderInfo, env return nil, err } + // Keep at least one registry pod available during voluntary disruptions + // (node drains, evictions, etc.) so a synchronized eviction can't take both + // pods down at once. + _, err = policyv1.NewPodDisruptionBudget(ctx, "mcp-registry", &policyv1.PodDisruptionBudgetArgs{ + Metadata: &metav1.ObjectMetaArgs{ + Name: pulumi.String("mcp-registry"), + Namespace: pulumi.String("default"), + Labels: pulumi.StringMap{ + "app": pulumi.String("mcp-registry"), + "environment": pulumi.String(environment), + }, + }, + Spec: &policyv1.PodDisruptionBudgetSpecArgs{ + MinAvailable: pulumi.Int(1), + Selector: &metav1.LabelSelectorArgs{ + MatchLabels: pulumi.StringMap{ + "app": pulumi.String("mcp-registry"), + }, + }, + }, + }, pulumi.Provider(cluster.Provider)) + if err != nil { + return nil, err + } + // Create Service service, err := corev1.NewService(ctx, "mcp-registry", &corev1.ServiceArgs{ Metadata: &metav1.ObjectMetaArgs{ diff --git a/deploy/pkg/providers/gcp/provider.go b/deploy/pkg/providers/gcp/provider.go index 108a209..6a51e27 100644 --- a/deploy/pkg/providers/gcp/provider.go +++ b/deploy/pkg/providers/gcp/provider.go @@ -3,9 +3,12 @@ package gcp import ( "encoding/base64" "fmt" + "strings" "github.com/pulumi/pulumi-gcp/sdk/v8/go/gcp" "github.com/pulumi/pulumi-gcp/sdk/v8/go/gcp/container" + "github.com/pulumi/pulumi-gcp/sdk/v8/go/gcp/organizations" + "github.com/pulumi/pulumi-gcp/sdk/v8/go/gcp/projects" "github.com/pulumi/pulumi-gcp/sdk/v8/go/gcp/storage" "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes" corev1 "github.com/pulumi/pulumi-kubernetes/sdk/v4/go/kubernetes/core/v1" @@ -58,6 +61,49 @@ func createGCPProvider(ctx *pulumi.Context, name string) (*gcp.Provider, error) return nil, nil } +// grantNodeServiceAccountRoles grants the default compute service account the standard +// GKE node roles required for log shipping (fluentbit-gke) and metrics scraping +// (managed Prometheus). New GCP projects no longer auto-grant Editor to the default +// compute SA, so these have to be set explicitly or every pod log/metric is dropped. +func grantNodeServiceAccountRoles(ctx *pulumi.Context, projectID string, gcpProvider *gcp.Provider) error { + invokeOpts := []pulumi.InvokeOption{} + resourceOpts := []pulumi.ResourceOption{} + if gcpProvider != nil { + invokeOpts = append(invokeOpts, pulumi.Provider(gcpProvider)) + resourceOpts = append(resourceOpts, pulumi.Provider(gcpProvider)) + } + + project, err := organizations.LookupProject(ctx, &organizations.LookupProjectArgs{ + ProjectId: pulumi.StringRef(projectID), + }, invokeOpts...) + if err != nil { + return fmt.Errorf("failed to look up project number for node SA bindings: %w", err) + } + + nodeSA := fmt.Sprintf("serviceAccount:%s-compute@developer.gserviceaccount.com", project.Number) + + roles := []string{ + "roles/logging.logWriter", + "roles/monitoring.metricWriter", + "roles/monitoring.viewer", + "roles/stackdriver.resourceMetadata.writer", + } + + for _, role := range roles { + name := fmt.Sprintf("node-sa-%s", strings.ReplaceAll(strings.TrimPrefix(role, "roles/"), ".", "-")) + _, err := projects.NewIAMMember(ctx, name, &projects.IAMMemberArgs{ + Project: pulumi.String(projectID), + Role: pulumi.String(role), + Member: pulumi.String(nodeSA), + }, resourceOpts...) + if err != nil { + return fmt.Errorf("failed to grant %s to node SA: %w", role, err) + } + } + + return nil +} + // CreateCluster creates a Google Kubernetes Engine cluster func (p *Provider) CreateCluster(ctx *pulumi.Context, environment string) (*providers.ProviderInfo, error) { // Get configuration @@ -152,6 +198,13 @@ func (p *Provider) CreateCluster(ctx *pulumi.Context, environment string) (*prov return nil, fmt.Errorf("failed to create node pool: %w", err) } + // Grant the default compute service account the standard GKE node roles so the + // fluentbit-gke and managed Prometheus collectors can ship logs and metrics. + // Without these, every pod log line and time series fails with IAM_PERMISSION_DENIED. + if err := grantNodeServiceAccountRoles(ctx, projectID, gcpProvider); err != nil { + return nil, err + } + // Create Kubernetes provider using the cluster directly k8sProvider, err := kubernetes.NewProvider(ctx, "k8s-provider", &kubernetes.ProviderArgs{ Kubeconfig: pulumi.All(cluster.Endpoint, cluster.MasterAuth).ApplyT(func(args []any) (string, error) { diff --git a/internal/service/registry_service.go b/internal/service/registry_service.go index 4f4b8a0..4403136 100644 --- a/internal/service/registry_service.go +++ b/internal/service/registry_service.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "log/slog" "time" "github.com/jackc/pgx/v5" @@ -85,15 +86,29 @@ func (s *registryServiceImpl) CreateServer(ctx context.Context, req *apiv0.Serve }) } -// createServerInTransaction contains the actual CreateServer logic within a transaction +// createServerInTransaction contains the actual CreateServer logic within a transaction. func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx pgx.Tx, req *apiv0.ServerJSON) (*apiv0.ServerResponse, error) { - // Validate the request - if err := validators.ValidatePublishRequest(ctx, *req, s.cfg); err != nil { - return nil, err + serverJSON := *req + + // Validate the request. ValidatePublishRequest fans out to npm/PyPI/OCI for + // registry-ownership checks with 10s per-host timeouts, so it's the most likely + // contributor to publish latency. Log validate_ms on both success and failure + // so the next /v0/publish latency alert is diagnostic — a slow upstream that + // times out into a validation error is exactly the case we need to see. + validateStart := time.Now() + validateErr := validators.ValidatePublishRequest(ctx, serverJSON, s.cfg) + validateMs := time.Since(validateStart).Milliseconds() + if validateErr != nil { + slog.WarnContext(ctx, "publish validate failed", + "server_name", serverJSON.Name, + "version", serverJSON.Version, + "validate_ms", validateMs, + "error", validateErr.Error(), + ) + return nil, validateErr } publishTime := time.Now() - serverJSON := *req // Acquire advisory lock to prevent concurrent publishes of the same server if err := s.db.AcquirePublishLock(ctx, tx, serverJSON.Name); err != nil { @@ -161,7 +176,17 @@ func (s *registryServiceImpl) createServerInTransaction(ctx context.Context, tx } // Insert new server version - return s.db.CreateServer(ctx, tx, &serverJSON, officialMeta) + resp, err := s.db.CreateServer(ctx, tx, &serverJSON, officialMeta) + if err != nil { + return nil, err + } + + slog.InfoContext(ctx, "publish complete", + "server_name", serverJSON.Name, + "version", serverJSON.Version, + "validate_ms", validateMs, + ) + return resp, nil } // recalculateLatest picks the highest non-deleted version of the given server and flags it