Detect consumer collision in the control plane

[spacewander]

Description

No response

Use Scenes

htnn/api/internal/consumer/consumer_manager.go

Line 92 in c8e8eeb

// TODO: find an effective way to detect collision in the control plane

Advice

No response

[Fancy-Tsang]

Hi, I would like to try this issue as a newcomer. And it looks suitable for warming up. May I take it?

[Fancy-Tsang]

Hi @spacewander, I have the following questions want to discuss with you. I am not sure whether my consideration is appropriate, so look forward to your reply:

1. Do you consider refining the detection collision mechanism to the plugin level? For example, some plugins (such as rate limiting and authentication) require unique index keys, while other plugins (such as logging) may allow duplicates.
2. Will there be a situation where two consumers have different index key calculations but the same results? The current implementation may be misjudged as a collision, but it may actually be a legal scenario.
3. If there is a collision, should the priority be distinguished for retention or abandonment?
4. Is it possible to consider using multiple go-routines for parallel processing, given that the current collision detection is serialized?

[spacewander]

Do you consider refining the detection collision mechanism to the plugin level? For example, some plugins (such as rate limiting and authentication) require unique index keys, while other plugins (such as logging) may allow duplicates.

Let the plugins implement the deduplication themselves.

Will there be a situation where two consumers have different index key calculations but the same results? The current implementation may be misjudged as a collision, but it may actually be a legal scenario.
If there is a collision, should the priority be distinguished for retention or abandonment?

Just rejecting the consumer CR is enough

Is it possible to consider using multiple go-routines for parallel processing, given that the current collision detection is serialized?

The number of consumers is small, usually < 100. So using a single goroutine is enough.

[Fancy-Tsang]

Would you consider the following question?
Multiple consumers need to reuse exactly the same plugin configuration. For example:
Multiple microservices share the same rate limiting rule (such as rate_limit: 1000qps) or multiple tenants use the same authentication policy (such as jwt: public_key=xxx)
If the index key is generated based on the configuration content, the same configuration will generate the same index key.
The current code directly rejects configurations with the same index key, causing legitimate shares to be misjudged as collisions.

[spacewander]

The users should not use the same authentication policy (such as jwt: public_key=xxx) for different consumers. Otherwise, the consumers are not distinguishable. They should add the tenant id as part as the index. As for the same rate limiting rule, it is nothing relative to the consumers, which is used for authentication.

[Fancy-Tsang]

Based on the overall architecture, I adopted the optimization scheme of centralized collision detection in controller/internal/controller/consumer_controller.go:

1. Modified harmonizer structure
   type ConsumerReconciler struct {
   component.ResourceManager
   Output component.Output
   keyIndex *KeyIndexRegistry // Add Key indexes
   }

// KeyIndexRegistry registration center
type KeyIndexRegistry struct {
mu sync.RWMutex
index map[string]map[string]map[string]string // ns -> plugin -> key -> consumerName
}

func NewKeyIndexRegistry() *KeyIndexRegistry {
return &KeyIndexRegistry{
index: make(map[string]map[string]map[string]string),
}
}

2. Implement conflict detection in the harmonizer
   func (r *ConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
   // 1. Obtain the current Consumer (used to determine deletion and other scenarios)
   var current mosniov1.Consumer
   if err := r.ResourceManager.Get(ctx, req.NamespacedName, &current); err != nil {
   return ctrl.Result{}, client.IgnoreNotFound(err)
   }

   // 2. Implement full collision detection
   if err := r.checkConsumerConflicts(ctx); err != nil {
   return ctrl.Result{}, err
   }

   // ...The original reconciliation logic...

}

3. Define the conflict detection function
   // checkConsumerConflicts Perform full conflict detection and rebuild the index
   func (r *ConsumerReconciler) checkConsumerConflicts(ctx context.Context) error {
   r.keyIndex.mu.Lock()
   defer r.keyIndex.mu.Unlock()

   // Clear old indexes
   r.keyIndex.index = make(map[string]map[string]map[string]string)

   // Get all Consumer
   var allConsumers mosniov1.ConsumerList
   if err := r.ResourceManager.List(ctx, &allConsumers); err != nil {
   return fmt.Errorf("list consumers failed: %w", err)
   }

   // Perforated detection
   for _, consumer := range allConsumers.Items {
   if !consumer.DeletionTimestamp.IsZero() {
   continue
   }

   if err := r.indexConsumer(consumer); err != nil {
      return fmt.Errorf("consumer %s/%s: %w", consumer.Namespace, consumer.Name, err)
   }
   

   }
   return nil
   }

// indexConsumer Index a single Consumer and detect conflicts
func (r *ConsumerReconciler) indexConsumer(consumer mosniov1.Consumer) error {
for pluginName, plugin := range consumer.Spec.Auth {
key, err := extractPluginKey(plugin.Config.Raw)
if err != nil {
return fmt.Errorf("invalid plugin %s config: %w", pluginName, err)
}

   // Inertly initialize the three-level map
   if r.keyIndex.index[consumer.Namespace] == nil {
      r.keyIndex.index[consumer.Namespace] = make(map[string]map[string]string)
   }
   if r.keyIndex.index[consumer.Namespace][pluginName] == nil {
      r.keyIndex.index[consumer.Namespace][pluginName] = make(map[string]string)
   }

   //  collision detection 
   if existing, exists := r.keyIndex.index[consumer.Namespace][pluginName][key]; exists {
      return fmt.Errorf(
         "key conflict: plugin=%s key=%s (already used by %s)",
         pluginName, key, existing,
      )
   }
   r.keyIndex.index[consumer.Namespace][pluginName][key] = consumer.Name
}
return nil

}

// Auxiliary function: Extract Key from RawExtension
func extractPluginKey(raw []byte) (string, error) {
var cfg map[string]interface{}
if err := json.Unmarshal(raw, &cfg); err != nil {
return "", err
}

// Supports the Key field name for different plug-ins
if key, ok := cfg["key"].(string); ok {
   return key, nil
}
if iss, ok := cfg["issuer"].(string); ok {
   return iss, nil
}
return "", errors.New("no valid key field found")

}

Do you agree with what I have done above? Can you give me some suggestions based on the code logic above? Looking forward to your reply.

[spacewander]

You can write some unit tests, and run make test or go test under controller/ directory to see if it works.

[Fancy-Tsang]

func TestConsumerCollisionDetection(t *testing.T) {

testCases := []struct {
	name          string
	consumers     []mosniov1.Consumer
	req           ctrl.Request 
	wantErr       string
	wantConflicts int
}{
	{
		name: "no conflict",
		consumers: []mosniov1.Consumer{
			buildConsumer("ns1", "c1", "key-auth", `{"key": "k1"}`),
			buildConsumer("ns1", "c2", "jwt-auth", `{"issuer": "i1"}`),
		},
		req:           ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "ns1", Name: "c1"}},
		wantConflicts: 0,
	},
	{
		name: "same plugin key conflict",
		consumers: []mosniov1.Consumer{
			buildConsumer("ns1", "c1", "key-auth", `{"key": "k1"}`),
			buildConsumer("ns1", "c2", "key-auth", `{"key": "k1"}`), // collision
		},
		wantErr:       "key conflict: plugin=key-auth key=k1 (already used by c1)",
		wantConflicts: 1,
	},
	{
		name: "cross-namespace no conflict",
		consumers: []mosniov1.Consumer{
			buildConsumer("ns1", "c1", "key-auth", `{"key": "k1"}`),
			buildConsumer("ns2", "c2", "key-auth", `{"key": "k1"}`), // different namespace
		},
		wantConflicts: 0,
	},
}

for _, tc := range testCases {
	t.Run(tc.name, func(t *testing.T) {
		cli := pkg.FakeK8sClient(t)
		output := component.NewK8sOutput(cli)
		rm := component.NewK8sResourceManager(cli)
		reconciler := &ConsumerReconciler{
			ResourceManager: rm,
			Output:          output,
			keyIndex:        NewKeyIndexRegistry(),
		}

		ctx := context.Background()
		
		_, err := reconciler.Reconcile(ctx, tc.req)

		if tc.wantErr != "" {
			require.ErrorContains(t, err, tc.wantErr)
		} else {
			require.NoError(t, err)
		}
		assert.Len(t, 0, tc.wantConflicts)
	})
}

}

// Auxiliary functions: Build test Consumer
func buildConsumer(ns, name, plugin, config string) mosniov1.Consumer {

return mosniov1.Consumer{
	ObjectMeta: metav1.ObjectMeta{
		Namespace: ns,
		Name:      name,
	},
	Spec: mosniov1.ConsumerSpec{
		Auth: map[string]mosniov1.ConsumerPlugin{
			plugin: {Config: runtime.RawExtension{Raw: []byte(config)}},
		},
	},
}

}
unit tests works.