fix: r

Signed-off-by: ashing <[email protected]>

Author: ronething

api/adc/types.go

@@ -103,6 +103,13 @@ func (g *GlobalRule) DeepCopy() GlobalRule {
 	return GlobalRule(copied)
 }
 
+// +k8s:deepcopy-gen=true
+type GlobalRuleItem struct {
+	Metadata `json:",inline" yaml:",inline"`
+
+	Plugins Plugins `json:"plugins" yaml:"plugins"`
+}
+
 type PluginMetadata Plugins
 
 func (p *PluginMetadata) DeepCopy() PluginMetadata {

api/adc/zz_generated.deepcopy.go

@@ -181,6 +181,12 @@ func (d *adcClient) Update(ctx context.Context, tctx *provider.TranslateContext,
 		// and triggered by a timer for synchronization
 		return nil
 	case BackendModeAPI7EE:
+		// TODO: sync global rule in api7ee mode if we need
+		if _, ok := obj.(*apiv2.ApisixGlobalRule); ok {
+			log.Debugw("apisix global rule, skip sync", zap.Any("obj", obj))
+			return nil
+		}
+
 		return d.sync(ctx, Task{
 			Name:          obj.GetName(),
 			Labels:        label.GenLabel(obj),

internal/provider/adc/adc.go

@@ -26,27 +26,35 @@ type Cache interface {
 	InsertService(*types.Service) error
 	// InsertConsumer adds or updates consumer to cache.
 	InsertConsumer(*types.Consumer) error
+	// InsertGlobalRule adds or updates global rule to cache.
+	InsertGlobalRule(*types.GlobalRuleItem) error
 
 	// GetSSL finds the ssl from cache according to the primary index (id).
 	GetSSL(string) (*types.SSL, error)
 	// GetUpstream finds the upstream from cache according to the primary index (id).
 	GetService(string) (*types.Service, error)
 	// GetConsumer finds the consumer from cache according to the primary index (username).
 	GetConsumer(string) (*types.Consumer, error)
+	// GetGlobalRule finds the global rule from cache according to the primary index (id).
+	GetGlobalRule(string) (*types.GlobalRuleItem, error)
 
 	// DeleteSSL deletes the specified ssl in cache.
 	DeleteSSL(*types.SSL) error
 	// DeleteUpstream deletes the specified upstream in cache.
 	DeleteService(*types.Service) error
-	// DeleteGlobalRule deletes the specified stream_route in cache.
+	// DeleteConsumer deletes the specified consumer in cache.
 	DeleteConsumer(*types.Consumer) error
+	// DeleteGlobalRule deletes the specified global rule in cache.
+	DeleteGlobalRule(*types.GlobalRuleItem) error
 
 	// ListSSL lists all ssl objects in cache.
 	ListSSL(...ListOption) ([]*types.SSL, error)
 	// ListUpstreams lists all upstreams in cache.
 	ListServices(...ListOption) ([]*types.Service, error)
 	// ListConsumers lists all consumer objects in cache.
 	ListConsumers(...ListOption) ([]*types.Consumer, error)
+	// ListGlobalRules lists all global rule objects in cache.
+	ListGlobalRules(...ListOption) ([]*types.GlobalRuleItem, error)
 }
 
 type ListOption interface {

internal/provider/adc/cache/cache.go

@@ -50,6 +50,8 @@ func (c *dbCache) Insert(obj any) error {
 		return c.InsertService(t)
 	case *types.Consumer:
 		return c.InsertConsumer(t)
+	case *types.GlobalRuleItem:
+		return c.InsertGlobalRule(t)
 	default:
 		return errors.New("unsupported type")
 	}
@@ -65,6 +67,8 @@ func (c *dbCache) Delete(obj any) error {
 		return c.DeleteService(t)
 	case *types.Consumer:
 		return c.DeleteConsumer(t)
+	case *types.GlobalRuleItem:
+		return c.DeleteGlobalRule(t)
 	default:
 		return errors.New("unsupported type")
 	}
@@ -87,6 +91,10 @@ func (c *dbCache) InsertConsumer(consumer *types.Consumer) error {
 	return c.insert("consumer", consumer.DeepCopy())
 }
 
+func (c *dbCache) InsertGlobalRule(globalRule *types.GlobalRuleItem) error {
+	return c.insert("global_rule", globalRule.DeepCopy())
+}
+
 func (c *dbCache) insert(table string, obj any) error {
 	txn := c.db.Txn(true)
 	defer txn.Abort()
@@ -129,6 +137,14 @@ func (c *dbCache) GetConsumer(username string) (*types.Consumer, error) {
 	return obj.(*types.Consumer).DeepCopy(), nil
 }
 
+func (c *dbCache) GetGlobalRule(id string) (*types.GlobalRuleItem, error) {
+	obj, err := c.get("global_rule", id)
+	if err != nil {
+		return nil, err
+	}
+	return obj.(*types.GlobalRuleItem).DeepCopy(), nil
+}
+
 func (c *dbCache) GetStreamRoute(id string) (*types.StreamRoute, error) {
 	obj, err := c.get("stream_route", id)
 	if err != nil {
@@ -201,6 +217,18 @@ func (c *dbCache) ListConsumers(opts ...ListOption) ([]*types.Consumer, error) {
 	return consumers, nil
 }
 
+func (c *dbCache) ListGlobalRules(opts ...ListOption) ([]*types.GlobalRuleItem, error) {
+	raws, err := c.list("global_rule", opts...)
+	if err != nil {
+		return nil, err
+	}
+	globalRules := make([]*types.GlobalRuleItem, 0, len(raws))
+	for _, raw := range raws {
+		globalRules = append(globalRules, raw.(*types.GlobalRuleItem).DeepCopy())
+	}
+	return globalRules, nil
+}
+
 func (c *dbCache) list(table string, opts ...ListOption) ([]any, error) {
 	txn := c.db.Txn(false)
 	defer txn.Abort()
@@ -239,6 +267,10 @@ func (c *dbCache) DeleteConsumer(consumer *types.Consumer) error {
 	return c.delete("consumer", consumer)
 }
 
+func (c *dbCache) DeleteGlobalRule(globalRule *types.GlobalRuleItem) error {
+	return c.delete("global_rule", globalRule)
+}
+
 func (c *dbCache) delete(table string, obj any) error {
 	txn := c.db.Txn(true)
 	defer txn.Abort()

internal/provider/adc/cache/memdb.go

@@ -40,7 +40,7 @@ func (c *noopCache) InsertService(u *types.Service) error {
 	return nil
 }
 
-func (c *noopCache) InsertGlobalRule(gr *types.GlobalRule) error {
+func (c *noopCache) InsertGlobalRule(gr *types.GlobalRuleItem) error {
 	return nil
 }
 
@@ -56,7 +56,7 @@ func (c *noopCache) GetService(id string) (*types.Service, error) {
 	return nil, nil
 }
 
-func (c *noopCache) GetGlobalRule(id string) (*types.GlobalRule, error) {
+func (c *noopCache) GetGlobalRule(id string) (*types.GlobalRuleItem, error) {
 	return nil, nil
 }
 
@@ -76,7 +76,7 @@ func (c *noopCache) ListStreamRoutes(...ListOption) ([]*types.StreamRoute, error
 	return nil, nil
 }
 
-func (c *noopCache) ListGlobalRules(...ListOption) ([]*types.GlobalRule, error) {
+func (c *noopCache) ListGlobalRules(...ListOption) ([]*types.GlobalRuleItem, error) {
 	return nil, nil
 }
 
@@ -92,7 +92,7 @@ func (c *noopCache) DeleteService(u *types.Service) error {
 	return nil
 }
 
-func (c *noopCache) DeleteGlobalRule(gr *types.GlobalRule) error {
+func (c *noopCache) DeleteGlobalRule(gr *types.GlobalRuleItem) error {
 	return nil
 }
 

internal/provider/adc/cache/noop_db.go

@@ -73,6 +73,22 @@ var (
 					},
 				},
 			},
+			"global_rule": {
+				Name: "global_rule",
+				Indexes: map[string]*memdb.IndexSchema{
+					"id": {
+						Name:    "id",
+						Unique:  true,
+						Indexer: &memdb.StringFieldIndex{Field: "ID"},
+					},
+					KindLabelIndex: {
+						Name:         KindLabelIndex,
+						Unique:       false,
+						AllowMissing: true,
+						Indexer:      &KindLabelIndexer,
+					},
+				},
+			},
 		},
 	}
 )

internal/provider/adc/cache/schema.go

@@ -45,10 +45,9 @@ func (e *DefaultADCExecutor) Execute(ctx context.Context, mode string, config ad
 }
 
 func (e *DefaultADCExecutor) runADC(ctx context.Context, mode string, config adcConfig, args []string) error {
-	ctxWithTimeout, cancel := context.WithTimeout(ctx, 10*time.Second)
-	defer cancel()
-
 	for _, addr := range config.ServerAddrs {
+		ctxWithTimeout, cancel := context.WithTimeout(ctx, 15*time.Second)
+		defer cancel()
 		if err := e.runForSingleServer(ctxWithTimeout, addr, mode, config, args); err != nil {
 			return err
 		}
@@ -62,6 +61,8 @@ func (e *DefaultADCExecutor) runForSingleServer(ctx context.Context, serverAddr,
 		cmdArgs = append(cmdArgs, "--tls-skip-verify")
 	}
 
+	cmdArgs = append(cmdArgs, "--timeout", "15s")
+
 	env := e.prepareEnv(serverAddr, mode, config.Token)
 
 	var stdout, stderr bytes.Buffer

internal/provider/adc/executor.go

@@ -16,6 +16,8 @@ import (
 	"sync"
 
 	"github.com/api7/gopkg/pkg/log"
+	"github.com/google/uuid"
+	"go.uber.org/zap"
 
 	adctypes "github.com/apache/apisix-ingress-controller/api/adc"
 	"github.com/apache/apisix-ingress-controller/internal/controller/label"
@@ -24,7 +26,6 @@ import (
 
 type Store struct {
 	cacheMap          map[string]cache.Cache
-	globalruleMap     map[string]adctypes.GlobalRule
 	pluginMetadataMap map[string]adctypes.PluginMetadata
 
 	sync.Mutex
@@ -33,7 +34,6 @@ type Store struct {
 func NewStore() *Store {
 	return &Store{
 		cacheMap:          make(map[string]cache.Cache),
-		globalruleMap:     make(map[string]adctypes.GlobalRule),
 		pluginMetadataMap: make(map[string]adctypes.PluginMetadata),
 	}
 }
@@ -105,7 +105,32 @@ func (s *Store) Insert(name string, resourceTypes []string, resources adctypes.R
 				}
 			}
 		case "global_rule":
-			s.globalruleMap[name] = resources.GlobalRules
+			// List existing global rules that match the selector
+			globalRules, err := targetCache.ListGlobalRules(selector)
+			if err != nil {
+				return err
+			}
+			// Delete existing matching global rules
+			for _, globalRule := range globalRules {
+				if err := targetCache.DeleteGlobalRule(globalRule); err != nil {
+					return err
+				}
+			}
+			// Convert GlobalRule (Plugins) to GlobalRuleItem and insert
+			if len(resources.GlobalRules) > 0 {
+				id := name + "-" + uuid.NewString()
+				globalRuleItem := &adctypes.GlobalRuleItem{
+					Metadata: adctypes.Metadata{
+						ID:     id,
+						Name:   id,
+						Labels: Labels,
+					},
+					Plugins: adctypes.Plugins(resources.GlobalRules),
+				}
+				if err := targetCache.InsertGlobalRule(globalRuleItem); err != nil {
+					return err
+				}
+			}
 		case "plugin_metadata":
 			s.pluginMetadataMap[name] = resources.PluginMetadata
 		default:
@@ -160,7 +185,15 @@ func (s *Store) Delete(name string, resourceTypes []string, Labels map[string]st
 				}
 			}
 		case "global_rule":
-			delete(s.globalruleMap, name)
+			globalRules, err := targetCache.ListGlobalRules(selector)
+			if err != nil {
+				log.Errorf("failed to list global rules: %v", err)
+			}
+			for _, globalRule := range globalRules {
+				if err := targetCache.DeleteGlobalRule(globalRule); err != nil {
+					log.Errorf("failed to delete global rule %s: %v", globalRule.ID, err)
+				}
+			}
 		case "plugin_metadata":
 			delete(s.pluginMetadataMap, name)
 		}
@@ -180,9 +213,18 @@ func (s *Store) GetResources(name string) (*adctypes.Resources, error) {
 	}
 	var globalrule adctypes.GlobalRule
 	var metadata adctypes.PluginMetadata
-	if global, ok := s.globalruleMap[name]; ok {
-		globalrule = global.DeepCopy()
+	// Get all global rules from cache and merge them
+	globalRuleItems, _ := targetCache.ListGlobalRules()
+	if len(globalRuleItems) > 0 {
+		merged := make(adctypes.Plugins)
+		for _, item := range globalRuleItems {
+			for k, v := range item.Plugins {
+				merged[k] = v
+			}
+		}
+		globalrule = adctypes.GlobalRule(merged)
 	}
+	log.Debugw("get resources global rule items", zap.Any("globalRuleItems", globalRuleItems))
 	if meta, ok := s.pluginMetadataMap[name]; ok {
 		metadata = meta.DeepCopy()
 	}