Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 10 additions & 12 deletions gateway/gateway-controller/pkg/eventlistener/api_processor.go
Original file line number Diff line number Diff line change
Expand Up @@ -43,21 +43,19 @@ func (l *EventListener) processAPIEvent(event eventhub.Event) {
}
}

func (l *EventListener) updateSnapshotAsync(entityID, correlationID, failureMessage string) {
func (l *EventListener) updateSnapshot(entityID, correlationID, failureMessage string) {
if l.snapshotManager == nil {
return
}

go func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

if err := l.snapshotManager.UpdateSnapshot(ctx, correlationID); err != nil {
l.logger.Error(failureMessage,
slog.String("entity_id", entityID),
slog.Any("error", err))
}
}()
if err := l.snapshotManager.UpdateSnapshot(ctx, correlationID); err != nil {
l.logger.Error(failureMessage,
slog.String("entity_id", entityID),
slog.Any("error", err))
}
}

// handleAPICreateOrUpdate handles API create or update events
Expand Down Expand Up @@ -109,7 +107,7 @@ func (l *EventListener) handleAPICreateOrUpdate(event eventhub.Event) {

// Update xDS snapshot for REST APIs only (WebSubApi and WebBrokerApi use Policy xDS)
if storedConfig.Kind != models.KindWebSubApi && storedConfig.Kind != models.KindWebBrokerApi {
l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after replica sync")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after replica sync")
}

// Update policies
Expand Down Expand Up @@ -188,7 +186,7 @@ func (l *EventListener) handleAPIDelete(event eventhub.Event) {

// Update xDS snapshot for REST APIs only (WebSubApi and WebBrokerApi use Policy xDS)
if existingConfig == nil || (existingConfig.Kind != models.KindWebSubApi && existingConfig.Kind != models.KindWebBrokerApi) {
l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after API deletion")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after API deletion")
}

// Remove runtime config for the deleted API
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ func (l *EventListener) handleLLMProviderCreateOrUpdate(event eventhub.Event) {
slog.Any("error", err))
}

l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider replica sync")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider replica sync")
l.updatePoliciesForAPI(storedConfig, event.EventID)

l.logger.Info("Successfully processed LLM provider create/update event",
Expand Down Expand Up @@ -184,7 +184,7 @@ func (l *EventListener) handleLLMProxyCreateOrUpdate(event eventhub.Event) {
}
}

l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy replica sync")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy replica sync")
l.updatePoliciesForAPI(storedConfig, event.EventID)

l.logger.Info("Successfully processed LLM proxy create/update event",
Expand Down Expand Up @@ -243,7 +243,7 @@ func (l *EventListener) handleLLMProviderDelete(event eventhub.Event) {
}
}

l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider deletion")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM provider deletion")

if l.policyManager != nil && existingConfig != nil {
if err := l.policyManager.DeleteAPIConfig(existingConfig.Kind, existingConfig.Handle); err != nil {
Expand Down Expand Up @@ -300,7 +300,7 @@ func (l *EventListener) handleLLMProxyDelete(event eventhub.Event) {
}
}

l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy deletion")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after LLM proxy deletion")

if l.policyManager != nil && existingConfig != nil {
if err := l.policyManager.DeleteAPIConfig(existingConfig.Kind, existingConfig.Handle); err != nil {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ func (l *EventListener) handleMCPProxyCreateOrUpdate(event eventhub.Event) {
}
}

l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy replica sync")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy replica sync")
l.updatePoliciesForAPI(storedConfig, event.EventID)

l.logger.Info("Successfully processed MCP proxy create/update event",
Expand All @@ -119,7 +119,7 @@ func (l *EventListener) handleMCPProxyDelete(event eventhub.Event) {
return
}

l.updateSnapshotAsync(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy deletion")
l.updateSnapshot(entityID, event.EventID, "Failed to update xDS snapshot after MCP proxy deletion")

if l.policyManager != nil && existingConfig != nil {
if err := l.policyManager.DeleteAPIConfig(existingConfig.Kind, existingConfig.Handle); err != nil {
Expand Down
12 changes: 9 additions & 3 deletions gateway/gateway-controller/pkg/xds/snapshot.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"context"
"fmt"
"log/slog"
"sync"
"time"

"github.com/envoyproxy/go-control-plane/pkg/cache/types"
Expand Down Expand Up @@ -62,13 +63,15 @@ type StatusUpdateCallback func(configID string, success bool, correlationID stri

// SnapshotManager manages xDS snapshots for Envoy
type SnapshotManager struct {
mu sync.Mutex
cache cache.SnapshotCache
translator *Translator
store *storage.ConfigStore
logger *slog.Logger
nodeID string // Node ID for Envoy (default: "router-node")
statusCallback StatusUpdateCallback
sdsSecretManager *SDSSecretManager
afterGetAll func() // nil in production; test hook for deterministic race testing
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

// NewSnapshotManager creates a new snapshot manager
Expand Down Expand Up @@ -100,6 +103,9 @@ func (sm *SnapshotManager) SetStatusCallback(callback StatusUpdateCallback) {
// UpdateSnapshot generates a new xDS snapshot from all configurations and updates the cache
// The correlationID parameter is optional and used for request tracing in logs
func (sm *SnapshotManager) UpdateSnapshot(ctx context.Context, correlationID string) error {
sm.mu.Lock()
defer sm.mu.Unlock()

startTime := time.Now()
trigger := "manual"
if correlationID != "" {
Expand All @@ -113,10 +119,10 @@ func (sm *SnapshotManager) UpdateSnapshot(ctx context.Context, correlationID str
}
// Get all configurations from in-memory store
configs := sm.store.GetAll()
if sm.afterGetAll != nil {
sm.afterGetAll()
}

// Translate configurations to Envoy resources if this is not event gw
//resources, err := sm.translator.TranslateConfigs(configs, correlationID)
// If event gw,
resources, err := sm.translator.TranslateConfigs(configs, correlationID)
if err != nil {
log.Error("Failed to translate configurations", slog.Any("error", err))
Expand Down
182 changes: 182 additions & 0 deletions gateway/gateway-controller/pkg/xds/snapshot_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
/*
* Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com).
*
* WSO2 LLC. licenses this file to you under the Apache License,
* Version 2.0 (the "License"); you may not use this file except
* in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/

package xds

import (
"context"
"strings"
"sync/atomic"
"testing"
"time"

route "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
api "github.com/wso2/api-platform/gateway/gateway-controller/pkg/api/management"
"github.com/wso2/api-platform/gateway/gateway-controller/pkg/metrics"
"github.com/wso2/api-platform/gateway/gateway-controller/pkg/models"
"github.com/wso2/api-platform/gateway/gateway-controller/pkg/storage"
)

func makeRestAPI(uuid, name, ctx string) *models.StoredConfig {
cfg := api.RestAPI{
Kind: api.RestAPIKindRestApi,
Metadata: api.Metadata{Name: name},
Spec: api.APIConfigData{
DisplayName: name,
Version: "v1.0",
Context: ctx,
Upstream: struct {
Main api.Upstream `json:"main" yaml:"main"`
Sandbox *api.Upstream `json:"sandbox,omitempty" yaml:"sandbox,omitempty"`
}{
Main: api.Upstream{Url: api.Ptr("http://backend:8080")},
},
Operations: []api.Operation{
{Method: api.Ptr(api.OperationMethodGET), Path: api.Ptr("/resource")},
},
},
}
return &models.StoredConfig{
UUID: uuid,
Kind: models.KindRestApi,
Handle: name,
DisplayName: name,
Version: "v1.0",
DesiredState: models.StateDeployed,
Configuration: cfg,
SourceConfiguration: cfg,
}
}

// TestConcurrentUpdateSnapshot reproduces a race where two goroutines call
// UpdateSnapshot and the final snapshot only contains api-one instead of
// both api-one and api-two. The slower goroutine reads the store before
// api-two is added, but writes last with a higher version, overwriting
// the correct snapshot.
//
// A test hook forces this interleaving:
//
// Step 1: A: GetAll() → [api-one]
// ↓ hook blocks A
// Step 2: Main: store.Add(api-two)
// Step 3: B: GetAll() → [api-one, api-two]
// SetSnapshot(v2, both) ✓
// ↓ hook releases A
// Step 4: A: SetSnapshot(v3, [api-one]) ✗ api-two lost
//
// Without mutex: A wins (higher version, stale data) → FAIL
// With mutex: B can't run while A holds the lock. A finishes first,
// then B reads the full store and writes the final snapshot → PASS
func TestConcurrentUpdateSnapshot(t *testing.T) {
t.Run("stale GetAll cannot overwrite a newer complete snapshot", func(t *testing.T) {
metrics.Init()
store := storage.NewConfigStore()

if err := store.Add(makeRestAPI("uuid-api-1", "api-one", "/api-one")); err != nil {
t.Fatalf("Add api-one: %v", err)
}

sm := NewSnapshotManager(store, createTestLogger(), testRouterConfig(), nil, testConfig())

// Step 1: A reads store, blocks in hook
aGotAll := make(chan struct{})
bDone := make(chan struct{})

// Only block the first caller (A); let B pass through.
var hooked atomic.Bool
sm.afterGetAll = func() {
if !hooked.CompareAndSwap(false, true) {
return
}
close(aGotAll)
select {
case <-bDone:
// pre-fix path: B raced past A → stale overwrite will occur
case <-time.After(200 * time.Millisecond):
// post-fix path: mutex held, B is queued behind A
}
}

errA := make(chan error, 1)
go func() {
errA <- sm.UpdateSnapshot(context.Background(), "corr-A")
}()
<-aGotAll

// Step 2: add api-two behind A's back
if err := store.Add(makeRestAPI("uuid-api-2", "api-two", "/api-two")); err != nil {
t.Fatalf("Add api-two: %v", err)
}

// Step 3: B sees [api-one, api-two], completes, releases A
errB := make(chan error, 1)
go func() {
err := sm.UpdateSnapshot(context.Background(), "corr-B")
close(bDone)
errB <- err
}()

// Step 4: A resumes with stale [api-one], overwrites B's snapshot
if err := <-errA; err != nil {
t.Fatalf("goroutine A UpdateSnapshot: %v", err)
}
if err := <-errB; err != nil {
t.Fatalf("goroutine B UpdateSnapshot: %v", err)
}

assertSnapshotContainsAPIs(t, sm, []string{"/api-one", "/api-two"})
})
}

func assertSnapshotContainsAPIs(t *testing.T, sm *SnapshotManager, expectedContexts []string) {
t.Helper()

snap, err := sm.GetCache().GetSnapshot("router-node")
if err != nil {
t.Fatalf("GetSnapshot: %v", err)
}

var routePaths []string
for _, res := range snap.GetResources(resource.RouteType) {
routeCfg, ok := res.(*route.RouteConfiguration)
if !ok {
continue
}
for _, vh := range routeCfg.GetVirtualHosts() {
for _, r := range vh.GetRoutes() {
if regex, ok := r.GetMatch().GetPathSpecifier().(*route.RouteMatch_SafeRegex); ok {
routePaths = append(routePaths, regex.SafeRegex.GetRegex())
}
}
}
}

for _, ctx := range expectedContexts {
found := false
for _, p := range routePaths {
if strings.Contains(p, ctx) {
found = true
break
}
}
if !found {
t.Errorf("snapshot after concurrent UpdateSnapshot is missing %s; got routes: %v", ctx, routePaths)
}
}
}