feat(server): add CONNECTOR_DISCOVERY_ENABLED flag to disable registry polling

server/conf/app.yaml

@@ -27,6 +27,16 @@ SESSION_ON: true
 TEMPORAL_ADDRESS: temporal:7233
 CONTAINER_REGISTRY_BASE: registry-1.docker.io
 
+# Set to false in Fusion-only or air-gapped deployments to disable upstream
+# container registry queries (Docker Hub / ECR / GCR) for connector version
+# discovery. When false, /versions and /spec return DEFAULT_CONNECTOR_VERSION
+# instead of polling the registry.
+CONNECTOR_DISCOVERY_ENABLED: true
+
+# Fallback connector version used when CONNECTOR_DISCOVERY_ENABLED is false.
+# Ignored when discovery is enabled. Example: "v0.3.18".
+DEFAULT_CONNECTOR_VERSION: ""
+
 # Prefer setting this via environment.
 OLAKE_SECRET_KEY: ""
 

server/internal/appconfig/appconfig.go

@@ -35,6 +35,17 @@ type Config struct {
 	OptimizationBaseURL   string
 	OptimizationUsername  string
 	OptimizationPassword  string
+
+	// ConnectorDiscoveryEnabled controls whether the server queries the
+	// upstream container registry (Docker Hub / ECR / GCR Artifact Registry)
+	// to enumerate available connector image tags. Defaults to true.
+	// Disable in Fusion-only or air-gapped deployments that don't need
+	// CDC connector discovery and want to avoid registry rate limits.
+	ConnectorDiscoveryEnabled bool
+
+	// DefaultConnectorVersion is returned by /versions and used by /spec
+	// when ConnectorDiscoveryEnabled is false. Ignored when discovery is on.
+	DefaultConnectorVersion string
 }
 
 var cfg = loadConfig()
@@ -93,5 +104,8 @@ func loadConfig() Config {
 		OptimizationBaseURL:  strings.TrimSpace(v.GetString("OPTIMIZATION_BASE_URL")),
 		OptimizationUsername: strings.TrimSpace(v.GetString("USERNAME")),
 		OptimizationPassword: strings.TrimSpace(v.GetString("PASSWORD")),
+
+		ConnectorDiscoveryEnabled: v.GetBool("CONNECTOR_DISCOVERY_ENABLED"),
+		DefaultConnectorVersion:   strings.TrimSpace(v.GetString("DEFAULT_CONNECTOR_VERSION")),
 	}
 }

server/internal/utils/docker_utils.go

@@ -73,8 +73,24 @@ func GetWorkerEnvVars() map[string]string {
 
 // GetDriverImageTags returns image tags from ECR, Artifact Registry, or Docker Hub with fallback to cached images
 func GetDriverImageTags(ctx context.Context, imageName string, cachedTags bool) ([]string, string, error) {
+	cfg := appconfig.Load()
+
+	// Short-circuit when connector discovery is disabled (e.g. Fusion-only or
+	// air-gapped deployments). Return the configured default version without
+	// contacting any container registry.
+	if !cfg.ConnectorDiscoveryEnabled {
+		if cfg.DefaultConnectorVersion == "" {
+			return nil, "", fmt.Errorf("connector discovery is disabled (CONNECTOR_DISCOVERY_ENABLED=false) and DEFAULT_CONNECTOR_VERSION is unset")
+		}
+		driverImage := imageName
+		if driverImage == "" {
+			driverImage = defaultImages[0]
+		}
+		return []string{cfg.DefaultConnectorVersion}, strings.TrimPrefix(driverImage, "olakego/source-"), nil
+	}
+
 	// TODO: make constants file and validate all env vars in start of server
-	repositoryBase := appconfig.Load().ContainerRegistryBase
+	repositoryBase := cfg.ContainerRegistryBase
 	if repositoryBase == "" {
 		return nil, "", fmt.Errorf("failed to get CONTAINER_REGISTRY_BASE")
 	}
@@ -100,7 +116,7 @@ func GetDriverImageTags(ctx context.Context, imageName string, cachedTags bool)
 		// Fallback to cached if online fetch fails or explicitly requested
 		if err != nil && cachedTags {
 			if constants.ExecutorEnvironment == "kubernetes" {
-				logger.Warn("failed to fetch image tags online for %s: %s. Cached fallback unavailable on Kubernetes (no Docker daemon)", imageName, err)
+				logger.Warn("failed to fetch image tags online for %s: %s. No local image cache is available on Kubernetes; set CONNECTOR_DISCOVERY_ENABLED=false with DEFAULT_CONNECTOR_VERSION to skip registry queries if connectors are unused", imageName, err)
 				continue
 			}
 			logger.Warn("failed to fetch image tags online for %s: %s, falling back to cached tags", imageName, err)