feat: add upstream support in service transfer and diff logic

Author: zhu327

internal/adc/client/kind_executor.go

@@ -187,6 +187,7 @@ func (e *KindExecutor) convertADCTypesToKineTypes(adcTypes []string) []string {
 			// ADC Service transfers to both Kine Service and Route
 			kineTypesSet[string(kine.ResourceTypeService)] = true
 			kineTypesSet[string(kine.ResourceTypeRoute)] = true
+			kineTypesSet[string(kine.ResourceTypeUpstream)] = true
 		case adctypes.TypeSSL:
 			kineTypesSet[string(kine.ResourceTypeSSL)] = true
 		case adctypes.TypeGlobalRule:

internal/adc/kine/diff.go

@@ -57,6 +57,7 @@ type Differ interface {
 type TransferredResources struct {
 	Routes      []*Route
 	Services    []*Service
+	Upstreams   []*Upstream
 	SSLs        []*SSL
 	GlobalRules []*GlobalRule
 }
@@ -114,6 +115,15 @@ func (d *differ) Diff(newResources *TransferredResources, opts *DiffOptions) ([]
 		events = append(events, serviceEvents...)
 	}
 
+	// Diff upstreams
+	if len(typesToDiff) == 0 || typesToDiff[string(ResourceTypeUpstream)] {
+		upstreamEvents, err := d.diffUpstreams(newResources.Upstreams, listOpts)
+		if err != nil {
+			return nil, fmt.Errorf("failed to diff upstreams: %w", err)
+		}
+		events = append(events, upstreamEvents...)
+	}
+
 	// Diff SSLs
 	if len(typesToDiff) == 0 || typesToDiff[string(ResourceTypeSSL)] {
 		sslEvents, err := d.diffSSLs(newResources.SSLs, listOpts)
@@ -264,6 +274,69 @@ func (d *differ) diffServices(newServices []*Service, listOpts []ListOption) ([]
 	return events, nil
 }
 
+// diffUpstreams compares new upstreams with cached upstreams
+func (d *differ) diffUpstreams(newUpstreams []*Upstream, listOpts []ListOption) ([]Event, error) {
+	// Get cached upstreams
+	cachedUpstreams, err := d.cache.ListUpstreams(listOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list cached upstreams: %w", err)
+	}
+
+	// Build maps for comparison
+	newMap := make(map[string]*Upstream)
+	for _, upstream := range newUpstreams {
+		newMap[upstream.ID] = upstream
+	}
+
+	cachedMap := make(map[string]*Upstream)
+	for _, upstream := range cachedUpstreams {
+		cachedMap[upstream.ID] = upstream
+	}
+
+	var events []Event
+
+	// Find CREATE and UPDATE events
+	for id, newUpstream := range newMap {
+		if cachedUpstream, exists := cachedMap[id]; exists {
+			// Check if update is needed
+			if !areUpstreamsEqual(cachedUpstream, newUpstream) {
+				events = append(events, Event{
+					Type:         EventTypeUpdate,
+					ResourceType: ResourceTypeUpstream,
+					ResourceID:   id,
+					ResourceName: newUpstream.Name,
+					OldValue:     cachedUpstream,
+					NewValue:     newUpstream,
+				})
+			}
+		} else {
+			// Create new upstream
+			events = append(events, Event{
+				Type:         EventTypeCreate,
+				ResourceType: ResourceTypeUpstream,
+				ResourceID:   id,
+				ResourceName: newUpstream.Name,
+				NewValue:     newUpstream,
+			})
+		}
+	}
+
+	// Find DELETE events
+	for id, cachedUpstream := range cachedMap {
+		if _, exists := newMap[id]; !exists {
+			events = append(events, Event{
+				Type:         EventTypeDelete,
+				ResourceType: ResourceTypeUpstream,
+				ResourceID:   id,
+				ResourceName: cachedUpstream.Name,
+				OldValue:     cachedUpstream,
+			})
+		}
+	}
+
+	return events, nil
+}
+
 // diffSSLs compares new SSLs with cached SSLs
 func (d *differ) diffSSLs(newSSLs []*SSL, listOpts []ListOption) ([]Event, error) {
 	// Get cached SSLs
@@ -402,6 +475,11 @@ func areServicesEqual(a, b *Service) bool {
 	return cmp.Equal(a, b)
 }
 
+// areUpstreamsEqual compares two upstreams for equality using go-cmp
+func areUpstreamsEqual(a, b *Upstream) bool {
+	return cmp.Equal(a, b)
+}
+
 // areSSLsEqual compares two SSLs for equality using go-cmp
 func areSSLsEqual(a, b *SSL) bool {
 	return cmp.Equal(a, b)
@@ -475,14 +553,16 @@ func TransferResources(resources *adc.Resources) (*TransferredResources, error)
 
 	// Transfer services (which includes routes and upstream)
 	for _, adcService := range resources.Services {
-		kineService, kineRoutes, err := TransferService(adcService)
+		kineService, kineRoutes, kineUpstreams, err := TransferService(adcService)
 		if err != nil {
 			return nil, fmt.Errorf("failed to transfer service %s: %w", adcService.Name, err)
 		}
 		if kineService != nil {
 			result.Services = append(result.Services, kineService)
 		}
 		result.Routes = append(result.Routes, kineRoutes...)
+
+		result.Upstreams = append(result.Upstreams, kineUpstreams...)
 	}
 
 	// Transfer SSLs

internal/adc/kine/transfer.go

@@ -10,14 +10,14 @@ import (
 )
 
 // TransferService converts an ADC Service to Kine Service and Routes
-func TransferService(adcSvc *adc.Service) (*Service, []*Route, error) {
+func TransferService(adcSvc *adc.Service) (*Service, []*Route, []*Upstream, error) {
 	if adcSvc == nil {
-		return nil, nil, fmt.Errorf("adc service is nil")
+		return nil, nil, nil, fmt.Errorf("adc service is nil")
 	}
 
 	// Ignore Upstreams, only use Upstream
 	if adcSvc.Upstream == nil {
-		return nil, nil, fmt.Errorf("adc service upstream is nil")
+		return nil, nil, nil, fmt.Errorf("adc service upstream is nil")
 	}
 
 	// Convert ADC Service to Kine Service
@@ -29,7 +29,7 @@ func TransferService(adcSvc *adc.Service) (*Service, []*Route, error) {
 			Labels: copyLabels(adcSvc.Labels),
 		},
 		Plugins:  convertPlugins(adcSvc.Plugins),
-		Upstream: convertUpstream(adcSvc.Upstream),
+		Upstream: convertUpstream(adcSvc.Upstream, adcSvc),
 		Hosts:    copyStringSlice(adcSvc.Hosts),
 	}
 
@@ -38,12 +38,21 @@ func TransferService(adcSvc *adc.Service) (*Service, []*Route, error) {
 	for _, adcRoute := range adcSvc.Routes {
 		kineRoute, err := convertRoute(adcRoute, adcSvc)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to convert route: %w", err)
+			return nil, nil, nil, fmt.Errorf("failed to convert route: %w", err)
 		}
 		kineRoutes = append(kineRoutes, kineRoute)
 	}
 
-	return kineSvc, kineRoutes, nil
+	// Convert ADC Upstream to Kine Upstream
+	kineUpstreams := make([]*Upstream, 0, len(adcSvc.Upstreams))
+	if adcSvc.Upstreams != nil {
+		for _, adcUpstream := range adcSvc.Upstreams {
+			kineUpstream := convertUpstream(adcUpstream, adcSvc)
+			kineUpstreams = append(kineUpstreams, kineUpstream)
+		}
+	}
+
+	return kineSvc, kineRoutes, kineUpstreams, nil
 }
 
 // generateServiceID generates service ID from name using SHA1
@@ -103,7 +112,7 @@ func convertRoute(adcRoute *adc.Route, adcSvc *adc.Service) (*Route, error) {
 }
 
 // convertUpstream converts ADC Upstream to Kine Upstream
-func convertUpstream(adcUpstream *adc.Upstream) *Upstream {
+func convertUpstream(adcUpstream *adc.Upstream, adcSvc *adc.Service) *Upstream {
 	if adcUpstream == nil {
 		return nil
 	}
@@ -119,7 +128,7 @@ func convertUpstream(adcUpstream *adc.Upstream) *Upstream {
 			ID:     upstreamID,
 			Name:   adcUpstream.Name,
 			Desc:   adcUpstream.Desc,
-			Labels: copyLabels(adcUpstream.Labels),
+			Labels: copyLabels(adcSvc.Labels),
 		},
 		Nodes:    convertNodes(adcUpstream.Nodes),
 		Type:     convertUpstreamType(adcUpstream.Type),

internal/adc/kine/transfer_test.go

@@ -67,7 +67,7 @@ func TestTransferService(t *testing.T) {
 	}
 
 	// Test TransferService
-	kineSvc, kineRoutes, err := TransferService(adcSvc)
+	kineSvc, kineRoutes, _, err := TransferService(adcSvc)
 	if err != nil {
 		t.Fatalf("TransferService failed: %v", err)
 	}
@@ -175,7 +175,7 @@ func TestTransferServiceWithCustomID(t *testing.T) {
 		},
 	}
 
-	kineSvc, kineRoutes, err := TransferService(adcSvc)
+	kineSvc, kineRoutes, _, err := TransferService(adcSvc)
 	if err != nil {
 		t.Fatalf("TransferService failed: %v", err)
 	}
@@ -199,7 +199,7 @@ func TestTransferServiceNilUpstream(t *testing.T) {
 		Upstream: nil,
 	}
 
-	_, _, err := TransferService(adcSvc)
+	_, _, _, err := TransferService(adcSvc)
 	if err == nil {
 		t.Error("Expected error for nil upstream")
 	}
@@ -357,7 +357,14 @@ func TestConvertUpstreamWithoutID(t *testing.T) {
 		Scheme: "http",
 	}
 
-	result := convertUpstream(adcUpstream)
+	// Create a service to pass to convertUpstream
+	adcSvc := &adc.Service{
+		Metadata: adc.Metadata{
+			Name: "test-service",
+		},
+	}
+
+	result := convertUpstream(adcUpstream, adcSvc)
 
 	if result == nil {
 		t.Fatal("Result should not be nil")
@@ -387,7 +394,14 @@ func TestConvertUpstreamWithID(t *testing.T) {
 		Scheme: "http",
 	}
 
-	result := convertUpstream(adcUpstream)
+	// Create a service to pass to convertUpstream
+	adcSvc := &adc.Service{
+		Metadata: adc.Metadata{
+			Name: "test-service",
+		},
+	}
+
+	result := convertUpstream(adcUpstream, adcSvc)
 
 	if result == nil {
 		t.Fatal("Result should not be nil")
@@ -437,7 +451,14 @@ func TestConvertUpstreamWithHealthCheck(t *testing.T) {
 		},
 	}
 
-	result := convertUpstream(adcUpstream)
+	// Create a service to pass to convertUpstream
+	adcSvc := &adc.Service{
+		Metadata: adc.Metadata{
+			Name: "test-service",
+		},
+	}
+
+	result := convertUpstream(adcUpstream, adcSvc)
 
 	if result == nil {
 		t.Fatal("Result should not be nil")