in router/core/router.go [712:942]
func (r *Router) bootstrap(ctx context.Context) error {
if !r.bootstrapped.CompareAndSwap(false, true) {
return fmt.Errorf("router is already bootstrapped")
}
cosmoCloudTracingEnabled := r.traceConfig.Enabled && rtrace.DefaultExporter(r.traceConfig) != nil
artInProductionEnabled := r.engineExecutionConfiguration.EnableRequestTracing && !r.developmentMode
needsRegistration := cosmoCloudTracingEnabled || artInProductionEnabled
if needsRegistration && r.selfRegister != nil {
r.logger.Info("Registering router with control plane because you opted in to send telemetry to Cosmo Cloud or advanced request tracing (ART) in production")
ri, registerErr := r.selfRegister.Register(ctx)
if registerErr != nil {
r.logger.Warn("Failed to register router on the control plane. If this warning persists, please contact support.")
} else {
r.registrationInfo = ri
// Only ensure sampling rate if the user exports traces to Cosmo Cloud
if cosmoCloudTracingEnabled {
if r.traceConfig.Sampler > float64(r.registrationInfo.AccountLimits.TraceSamplingRate) {
r.logger.Warn("Trace sampling rate is higher than account limit. Using account limit instead. Please contact support to increase your account limit.",
zap.Float64("limit", r.traceConfig.Sampler),
zap.String("account_limit", fmt.Sprintf("%.2f", r.registrationInfo.AccountLimits.TraceSamplingRate)),
)
r.traceConfig.Sampler = float64(r.registrationInfo.AccountLimits.TraceSamplingRate)
}
}
}
}
if r.traceConfig.Enabled {
tp, err := rtrace.NewTracerProvider(ctx, &rtrace.ProviderConfig{
Logger: r.logger,
Config: r.traceConfig,
ServiceInstanceID: r.instanceID,
IPAnonymization: &rtrace.IPAnonymizationConfig{
Enabled: r.ipAnonymization.Enabled,
Method: rtrace.IPAnonymizationMethod(r.ipAnonymization.Method),
},
MemoryExporter: r.traceConfig.TestMemoryExporter,
})
if err != nil {
return fmt.Errorf("failed to start trace agent: %w", err)
}
r.tracerProvider = tp
}
// Prometheus metrics rely on OTLP metrics
if r.metricConfig.IsEnabled() {
if r.metricConfig.Prometheus.Enabled {
mp, registry, err := rmetric.NewPrometheusMeterProvider(ctx, r.metricConfig, r.instanceID)
if err != nil {
return fmt.Errorf("failed to create Prometheus exporter: %w", err)
}
r.promMeterProvider = mp
r.prometheusServer = rmetric.NewPrometheusServer(r.logger, r.metricConfig.Prometheus.ListenAddr, r.metricConfig.Prometheus.Path, registry)
go func() {
if err := r.prometheusServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
r.logger.Error("Failed to start Prometheus server", zap.Error(err))
}
}()
}
if r.metricConfig.OpenTelemetry.Enabled {
mp, err := rmetric.NewOtlpMeterProvider(ctx, r.logger, r.metricConfig, r.instanceID)
if err != nil {
return fmt.Errorf("failed to start trace agent: %w", err)
}
r.otlpMeterProvider = mp
}
}
if r.graphqlMetricsConfig.Enabled {
client := graphqlmetricsv1connect.NewGraphQLMetricsServiceClient(
http.DefaultClient,
r.graphqlMetricsConfig.CollectorEndpoint,
connect.WithSendGzip(),
)
ge, err := graphqlmetrics.NewExporter(
r.logger,
client,
r.graphApiToken,
graphqlmetrics.NewDefaultExporterSettings(),
)
if err != nil {
return fmt.Errorf("failed to validate graphql metrics exporter: %w", err)
}
r.gqlMetricsExporter = ge
r.logger.Info("GraphQL schema coverage metrics enabled")
}
if r.Config.rateLimit != nil && r.Config.rateLimit.Enabled {
var err error
r.redisClient, err = rd.NewRedisCloser(&rd.RedisCloserOptions{
URLs: r.Config.rateLimit.Storage.URLs,
ClusterEnabled: r.Config.rateLimit.Storage.ClusterEnabled,
Logger: r.logger,
})
if err != nil {
return fmt.Errorf("failed to create redis client: %w", err)
}
}
if r.mcp.Enabled {
var operationsDir string
// If storage provider ID is set, resolve it to a directory path
if r.mcp.Storage.ProviderID != "" {
r.logger.Debug("Resolving storage provider for MCP operations",
zap.String("provider_id", r.mcp.Storage.ProviderID))
// Find the provider in storage_providers
found := false
// Check for file_system providers
for _, provider := range r.storageProviders.FileSystem {
if provider.ID == r.mcp.Storage.ProviderID {
r.logger.Debug("Found file_system storage provider for MCP",
zap.String("id", provider.ID),
zap.String("path", provider.Path))
// Use the resolved file system path
operationsDir = provider.Path
found = true
break
}
}
if !found {
return fmt.Errorf("storage provider with id '%s' for mcp server not found", r.mcp.Storage.ProviderID)
}
}
logFields := []zap.Field{
zap.String("storage_provider_id", r.mcp.Storage.ProviderID),
}
// Initialize the MCP server with the resolved operations directory
mcpOpts := []func(*mcpserver.Options){
mcpserver.WithGraphName(r.mcp.GraphName),
mcpserver.WithOperationsDir(operationsDir),
mcpserver.WithListenAddr(r.mcp.Server.ListenAddr),
mcpserver.WithBaseURL(r.mcp.Server.BaseURL),
mcpserver.WithLogger(r.logger.With(logFields...)),
mcpserver.WithExcludeMutations(r.mcp.ExcludeMutations),
mcpserver.WithEnableArbitraryOperations(r.mcp.EnableArbitraryOperations),
mcpserver.WithExposeSchema(r.mcp.ExposeSchema),
}
// Determine the router GraphQL endpoint
var routerGraphQLEndpoint string
// Use the custom URL if provided
if r.mcp.RouterURL != "" {
routerGraphQLEndpoint = r.mcp.RouterURL
} else {
routerGraphQLEndpoint = path.Join(r.listenAddr, r.graphqlPath)
}
mcpss, err := mcpserver.NewGraphQLSchemaServer(
routerGraphQLEndpoint,
mcpOpts...,
)
if err != nil {
return fmt.Errorf("failed to create mcp server: %w", err)
}
err = mcpss.Start()
if err != nil {
return fmt.Errorf("failed to start MCP server: %w", err)
}
r.mcpServer = mcpss
}
if r.metricConfig.OpenTelemetry.EngineStats.Enabled() || r.metricConfig.Prometheus.EngineStats.Enabled() || r.engineExecutionConfiguration.Debug.ReportWebSocketConnections {
r.EngineStats = statistics.NewEngineStats(ctx, r.logger, r.engineExecutionConfiguration.Debug.ReportWebSocketConnections)
}
if r.engineExecutionConfiguration.Debug.ReportMemoryUsage {
debug.ReportMemoryUsage(ctx, r.logger)
}
if r.playgroundConfig.Enabled {
playgroundUrl, err := url.JoinPath(r.baseURL, r.playgroundConfig.Path)
if err != nil {
return fmt.Errorf("failed to join playground url: %w", err)
}
r.logger.Info("Serving GraphQL playground", zap.String("url", playgroundUrl))
r.playgroundHandler = graphiql.NewPlayground(&graphiql.PlaygroundOptions{
Html: graphiql.PlaygroundHTML(),
GraphqlURL: r.graphqlWebURL,
PlaygroundPath: r.playgroundPath,
ConcurrencyLimit: int64(r.playgroundConfig.ConcurrencyLimit),
})
}
if r.executionConfig != nil && r.executionConfig.Path != "" {
executionConfig, err := execution_config.FromFile(r.executionConfig.Path)
if err != nil {
return fmt.Errorf("failed to read execution config: %w", err)
}
r.staticExecutionConfig = executionConfig
}
if err := r.buildClients(); err != nil {
return err
}
// Modules are only initialized once and not on every config change
if err := r.initModules(ctx); err != nil {
return fmt.Errorf("failed to init user modules: %w", err)
}
if r.traceConfig.Enabled && len(r.tracePropagators) > 0 {
r.compositePropagator = propagation.NewCompositeTextMapPropagator(r.tracePropagators...)
// Don't set it globally when we use the router in tests.
// In practice, setting it globally only makes sense for module development.
if r.traceConfig.TestMemoryExporter == nil {
otel.SetTextMapPropagator(r.compositePropagator)
}
}
return nil
}