enable circuitbreaker and fix ut

Author: Eikykun

config/crd/bases/ctrlmesh.kusionstack.io_circuitbreakers.yaml

@@ -123,7 +123,6 @@ spec:
                             type: array
                         required:
                         - apiGroups
-                        - namespaces
                         - resources
                         - verbs
                         type: object
@@ -155,7 +154,6 @@ spec:
                   required:
                   - bucket
                   - name
-                  - recoverPolicy
                   - triggerPolicy
                   type: object
                 type: array

pkg/apis/ctrlmesh/utils/conv/conv.go

@@ -103,15 +103,17 @@ func ConvertLimiting(limit *ctrlmeshv1alpha1.Limiting) *ctrlmeshproto.RateLimiti
 	case ctrlmeshv1alpha1.TriggerPolicyForceOpened:
 		protoLimit.TriggerPolicy = ctrlmeshproto.RateLimiting_TRIGGER_POLICY_FORCE_OPENED
 	}
-	protoLimit.RecoverPolicy = &ctrlmeshproto.RateLimiting_RecoverPolicy{}
-	switch limit.RecoverPolicy.RecoverType {
-	case ctrlmeshv1alpha1.RecoverPolicyManual:
-		protoLimit.RecoverPolicy.Type = ctrlmeshproto.RateLimiting_RECOVER_POLICY_MANUAL
-	case ctrlmeshv1alpha1.RecoverPolicySleepingWindow:
-		protoLimit.RecoverPolicy.Type = ctrlmeshproto.RateLimiting_RECOVER_POLICY_SLEEPING_WINDOW
-	}
-	if limit.RecoverPolicy.SleepingWindowSize != nil {
-		protoLimit.RecoverPolicy.SleepingWindowSize = *limit.RecoverPolicy.SleepingWindowSize
+	if limit.RecoverPolicy != nil {
+		protoLimit.RecoverPolicy = &ctrlmeshproto.RateLimiting_RecoverPolicy{}
+		switch limit.RecoverPolicy.RecoverType {
+		case ctrlmeshv1alpha1.RecoverPolicyManual:
+			protoLimit.RecoverPolicy.Type = ctrlmeshproto.RateLimiting_RECOVER_POLICY_MANUAL
+		case ctrlmeshv1alpha1.RecoverPolicySleepingWindow:
+			protoLimit.RecoverPolicy.Type = ctrlmeshproto.RateLimiting_RECOVER_POLICY_SLEEPING_WINDOW
+		}
+		if limit.RecoverPolicy.SleepingWindowSize != nil {
+			protoLimit.RecoverPolicy.SleepingWindowSize = *limit.RecoverPolicy.SleepingWindowSize
+		}
 	}
 	if limit.Properties != nil {
 		protoLimit.Properties = make(map[string]string, len(limit.Properties))

pkg/apis/ctrlmesh/v1alpha1/throttlingconfig_types.go

@@ -31,7 +31,7 @@ type ResourceRule struct {
 	// Verb is a list of kubernetes resource API verbs, like: get, list, watch, create, update, delete, proxy.  "*" means all.
 	Verbs []string `json:"verbs"`
 	// Namespaces is a list of namespaces the rule applies to. "*" means all.
-	Namespaces []string `json:"namespaces"`
+	Namespaces []string `json:"namespaces,omitempty"`
 }
 
 // RestRule defines the target rest resource of the limiting policy
@@ -45,7 +45,7 @@ type RestRule struct {
 // TriggerPolicy defines how the circuit-breaking policy triggered from 'Closed' to 'Opened'
 type TriggerPolicy string
 
-var (
+const (
 	TriggerPolicyNormal      TriggerPolicy = "Normal"
 	TriggerPolicyLimiterOnly TriggerPolicy = "LimiterOnly"
 	TriggerPolicyForceOpened TriggerPolicy = "ForceOpened"
@@ -60,33 +60,27 @@ type RecoverPolicy struct {
 
 type RecoverType string
 
-var (
+const (
 	RecoverPolicyManual         RecoverType = "Manual"
 	RecoverPolicySleepingWindow RecoverType = "SleepingWindow"
 )
 
 // InterceptType defines how the circuit-breaking traffic intercept from 'White' to 'Black'
 type InterceptType string
 
-var (
+const (
 	InterceptTypeWhitelist InterceptType = "Whitelist"
 	InterceptTypeBlacklist InterceptType = "Blacklist"
 )
 
 // ContentType defines how the circuit-breaking traffic intercept content type from 'Normal' to 'Regexp'
 type ContentType string
 
-var (
+const (
 	ContentTypeNormal ContentType = "Normal"
 	ContentTypeRegexp ContentType = "Regexp"
 )
 
-type ValidatePolicy string
-
-var (
-	AfterHttpSuccess ValidatePolicy = "AfterHttpSuccess"
-)
-
 // Bucket defines the whole token bucket of the policy
 type Bucket struct {
 	// Burst is the max token number of the bucket
@@ -110,7 +104,7 @@ type Limiting struct {
 	// TriggerPolicy defines how the circuit-breaking policy triggered from 'Closed' to 'Opened'
 	TriggerPolicy TriggerPolicy `json:"triggerPolicy"`
 	// RecoverPolicy defines how the circuit-breaking policy recovered from 'Opened' to 'Closed'
-	RecoverPolicy RecoverPolicy `json:"recoverPolicy"`
+	RecoverPolicy *RecoverPolicy `json:"recoverPolicy,omitempty"`
 	// ValidatePolicy determine the opportunity to validate req
 	//ValidatePolicy ValidatePolicy `json:"validatePolicy,omitempty"`
 	// Properties defines the additional properties of the policy, like: SleepingWindowSize
@@ -162,7 +156,7 @@ type LimitingSnapshot struct {
 // CircuitBreakerStatus defines the observed state of CircuitBreaker
 type CircuitBreakerStatus struct {
 	ObservedGeneration int64           `json:"observedGeneration,omitempty"`
-	LastUpdatedTime    metav1.Time     `json:"lastUpdatedTime,omitempty"`
+	LastUpdatedTime    *metav1.Time    `json:"lastUpdatedTime,omitempty"`
 	CurrentSpecHash    string          `json:"currentSpecHash,omitempty"`
 	TargetStatus       []*TargetStatus `json:"targetStatus,omitempty"`
 }
@@ -202,16 +196,3 @@ type CircuitBreakerList struct {
 func init() {
 	SchemeBuilder.Register(&CircuitBreaker{}, &CircuitBreakerList{})
 }
-
-//func (in *CircuitBreakerStatus) UpdateSnapshots(podIp string, snapshot []*LimitingSnapshot) {
-//	var newSnapshot []*LimitingSnapshot
-//	for _, sp := range snapshot {
-//		newSnapshot = append(newSnapshot, sp.DeepCopy())
-//	}
-//	for _, sp := range in.LimitingSnapshots {
-//		if sp.Endpoint != podIp {
-//			newSnapshot = append(newSnapshot, sp.DeepCopy())
-//		}
-//	}
-//	in.LimitingSnapshots = newSnapshot
-//}

pkg/apis/ctrlmesh/v1alpha1/zz_generated.deepcopy.go

@@ -39,6 +39,7 @@ import (
 	"github.com/KusionStack/controller-mesh/pkg/cachelimiter"
 	"github.com/KusionStack/controller-mesh/pkg/client"
 	"github.com/KusionStack/controller-mesh/pkg/grpcregistry"
+	"github.com/KusionStack/controller-mesh/pkg/manager/controllers/circuitbreaker"
 	"github.com/KusionStack/controller-mesh/pkg/manager/controllers/managerstate"
 	"github.com/KusionStack/controller-mesh/pkg/manager/controllers/patchrunnable"
 	"github.com/KusionStack/controller-mesh/pkg/manager/controllers/shardingconfigserver"
@@ -154,6 +155,12 @@ func main() {
 			setupLog.Error(err, "unable to create controller", "controller", "ManagerState")
 			os.Exit(1)
 		}
+		if err = (&circuitbreaker.CircuitBreakerReconciler{
+			Client: mgr.GetClient(),
+		}).SetupWithManager(mgr); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", "CircuitBreaker")
+			os.Exit(1)
+		}
 
 		if err = patchrunnable.SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to create runnable", "runnable", "PatchRunnable")

pkg/cmd/manager/main.go

@@ -17,89 +17,122 @@ limitations under the License.
 package circuitbreaker
 
 import (
+	"context"
 	"sync"
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/proto"
 )
 
 var (
-	defaultPodCache = &podCache{
-		currentHash: make(map[string]string),
+	defaultPodConfigCache = &podCache{
+		currentHash: make(map[string]sets.Set[string]),
 	}
-	//defaultHashCache = &hashCache{
-	//	store: make(map[string]sets.String),
-	//}
 )
 
 type podCache struct {
-	currentHash map[string]string
+	currentHash map[string]sets.Set[string]
 	mu          sync.RWMutex
 }
 
-func cacheKey(namespace, podName, cbName string) string {
-	return namespace + "/" + podName + "/" + cbName
+func cacheKey(namespace, podName string) string {
+	return namespace + "/" + podName
 }
 
-func (c *podCache) Get(key string) string {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	return c.currentHash[key]
+func (c *podCache) Has(namespace, podName, cbHash string) bool {
+	key := cacheKey(namespace, podName)
+	configs, ok := c.currentHash[key]
+	if !ok {
+		return false
+	}
+	return configs.Has(cbHash)
 }
 
-func (c *podCache) Update(key string, hash string) {
+func (c *podCache) Add(namespace, podName, cbHash string) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	c.currentHash[key] = hash
+	key := cacheKey(namespace, podName)
+	_, ok := c.currentHash[key]
+	if !ok {
+		c.currentHash[key] = sets.New[string](cbHash)
+		return
+	}
+	c.currentHash[key].Insert(cbHash)
 }
 
-func (c *podCache) Delete(key string) {
+func (c *podCache) Delete(namespace, podName string, cbHash ...string) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	delete(c.currentHash, key)
+	key := cacheKey(namespace, podName)
+	if len(cbHash) == 0 {
+		delete(c.currentHash, key)
+		return
+	}
+	_, ok := c.currentHash[key]
+	if !ok {
+		return
+	}
+	c.currentHash[key].Delete(cbHash...)
+}
+
+func newDeleteProcessor(c client.Client, ctx context.Context) *deleteProcessor {
+	processor := &deleteProcessor{
+		q:   workqueue.NewNamedDelayingQueue("delete-processor"),
+		ctx: ctx,
+		processFunc: func(item deleteItem) error {
+			po := &v1.Pod{}
+			if err := c.Get(ctx, types.NamespacedName{
+				Namespace: item.Namespace,
+				Name:      item.PodName,
+			}, po); err != nil {
+				return client.IgnoreNotFound(err)
+			}
+			if po.DeletionTimestamp != nil {
+				return nil
+			}
+			return deletePodConfig(ctx, &proto.CircuitBreaker{Name: item.ConfigName}, po.Status.PodIP)
+		},
+	}
+	go processor.processLoop()
+	return processor
+}
+
+type deleteProcessor struct {
+	q           workqueue.DelayingInterface
+	processFunc func(deleteItem) error
+	ctx         context.Context
+}
+
+func (d *deleteProcessor) Add(namespace, podName, config string) {
+	d.q.Add(deleteItem{Namespace: namespace, PodName: podName, ConfigName: config})
+}
+
+func (d *deleteProcessor) processLoop() {
+	go func() {
+		for {
+			item, shutdown := d.q.Get()
+			if shutdown {
+				break
+			}
+			val := item.(deleteItem)
+			if err := d.processFunc(val); err != nil {
+				d.q.AddAfter(item, 5*time.Second)
+			}
+			d.q.Done(item)
+		}
+	}()
+	<-d.ctx.Done()
+	d.q.ShutDown()
 }
 
-//
-//type hashCache struct {
-//	store map[string]sets.String
-//	mu    sync.RWMutex
-//}
-//
-//func (h *hashCache) Get(hash string) sets.String {
-//	h.mu.RLock()
-//	defer h.mu.RUnlock()
-//	val, ok := h.store[hash]
-//	if !ok {
-//		return sets.NewString()
-//	}
-//	return val.Clone()
-//}
-//
-//func (h *hashCache) Add(hash string, items ...string) {
-//	h.mu.Lock()
-//	defer h.mu.Unlock()
-//	val, ok := h.store[hash]
-//	if !ok {
-//		h.store[hash] = sets.NewString(items...)
-//	} else {
-//		val.Insert(items...)
-//	}
-//}
-//
-//func (h *hashCache) Replace(hash string, items ...string) {
-//	h.mu.Lock()
-//	defer h.mu.Unlock()
-//	h.store[hash] = sets.NewString(items...)
-//}
-//
-//func (h *hashCache) DeleteHash(hash string) {
-//	h.mu.Lock()
-//	defer h.mu.Unlock()
-//	delete(h.store, hash)
-//}
-//
-//func (h *hashCache) DeleteVal(hash string, items ...string) {
-//	h.mu.Lock()
-//	defer h.mu.Unlock()
-//	val, ok := h.store[hash]
-//	if ok {
-//		val.Delete(items...)
-//	}
-//}
+type deleteItem struct {
+	Namespace  string
+	PodName    string
+	ConfigName string
+}