add logs, add ut, fix breaker

Author: Eikykun

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

@@ -123,6 +123,7 @@ spec:
                             type: array
                         required:
                         - apiGroups
+                        - namespaces
                         - resources
                         - verbs
                         type: object

config/demo/circuitbreaker-demo.yaml

@@ -10,9 +10,9 @@ spec:
       interval: 1s
       limit: 20
     name: deletePod
-    properties:
-      sleepingWindowSize: 15m
-    recoverPolicy: SleepingWindow
+    recoverPolicy:
+      sleepingWindowSize: 10m
+      type: SleepingWindow
     resourceRules:
     - apiGroups:
       - ""
@@ -22,20 +22,24 @@ spec:
       - pods
       verbs:
       - delete
-    triggerPolicy: Normal
+    triggerPolicy: LimiterOnly
   - bucket:
       burst: 200
       interval: 1s
       limit: 20
     name: trafficOffLimit
     restRules:
     - method: POST
-      url: https://*.com/*/trafficOff
+      url: https://xxx.com/xxx/trafficOff
     triggerPolicy: LimiterOnly
+  selector:
+    matchLabels:
+      control-plane: kusionstack-operating
   trafficInterceptRules:
-  - contents:
+  - contentType: Regexp
+    contents:
     - .*(127.0.0.1).*
-    interceptType: White
+    interceptType: Whitelist
     methods:
     - POST
     - GET

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,omitempty"`
+	Namespaces []string `json:"namespaces"`
 }
 
 // RestRule defines the target rest resource of the limiting policy

pkg/manager/controllers/circuitbreaker/circuitbreaker_controller.go

@@ -110,8 +110,9 @@ func (r *CircuitBreakerReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 			klog.Errorf(msg)
 			reconcileErr = errors.Join(reconcileErr, stateErr)
 		} else {
+			klog.Infof("sync circuitbreaker %s to pod %s success, configHash=%s", cb.Name, utils.KeyFunc(&po), protoCb.ConfigHash)
 			currentHash = protoCb.ConfigHash
-			defaultPodConfigCache.Add(po.Namespace, state.PodName, cb.Name)
+			defaultPodConfigCache.Add(po.Namespace, po.Name, cb.Name)
 		}
 		status := &ctrlmeshv1alpha1.TargetStatus{
 			PodName:           po.Name,
@@ -289,6 +290,7 @@ func (r *CircuitBreakerReconciler) disableConfig(ctx context.Context, podIp stri
 	if resp != nil && resp.Msg != nil && !resp.Msg.Success {
 		return fmt.Errorf("fail to disable pod [%s, %s] circuit breaker config, %s", podIp, name, resp.Msg.Message)
 	}
+	klog.Infof("pod[ip=%s] circuitbreaker %s was disabled", podIp, name)
 	return nil
 }
 

pkg/manager/controllers/circuitbreaker/circuitbreaker_controller_test.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
+	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh"
 	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/constants"
 	ctrlmeshproto "github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/proto"
 	ctrlmeshv1alpha1 "github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/v1alpha1"
@@ -88,6 +89,9 @@ var circuitBreaker = &ctrlmeshv1alpha1.CircuitBreaker{
 						Verbs: []string{
 							"delete",
 						},
+						Namespaces: []string{
+							"*",
+						},
 					},
 				},
 				Bucket: ctrlmeshv1alpha1.Bucket{
@@ -142,6 +146,15 @@ func TestCircuitBreaker(t *testing.T) {
 	waitProcess()
 	g.Expect(c.Get(ctx, types.NamespacedName{Name: "testcb", Namespace: "default"}, cb)).Should(gomega.BeNil())
 	g.Expect(breakerManager.ValidateTrafficIntercept("aaa.aaa.aaa", "GET").Allowed).Should(gomega.BeFalse())
+	if cb.Labels == nil {
+		cb.Labels = map[string]string{}
+	}
+	cb.Labels[ctrlmesh.CtrlmeshCircuitBreakerDisableKey] = "true"
+	g.Expect(c.Update(ctx, cb)).Should(gomega.BeNil())
+	waitProcess()
+	g.Expect(c.Get(ctx, types.NamespacedName{Name: "testcb", Namespace: "default"}, cb)).Should(gomega.BeNil())
+	g.Expect(len(cb.Status.TargetStatus) == 0).Should(gomega.BeTrue())
+	g.Expect(len(cb.Finalizers) == 0).Should(gomega.BeTrue())
 	g.Expect(c.Delete(ctx, testBreaker)).Should(gomega.BeNil())
 	waitProcess()
 	g.Expect(c.Get(ctx, types.NamespacedName{Name: "testcb", Namespace: "default"}, cb)).Should(gomega.HaveOccurred())

pkg/manager/controllers/circuitbreaker/filter.go

@@ -19,6 +19,7 @@ package circuitbreaker
 import (
 	"sigs.k8s.io/controller-runtime/pkg/event"
 
+	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh"
 	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/utils/conv"
 	ctrlmeshv1alpha1 "github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/v1alpha1"
 )
@@ -43,6 +44,12 @@ func (b *BreakerPredicate) Update(e event.UpdateEvent) bool {
 	if newCB.DeletionTimestamp != nil || len(oldCB.Finalizers) != len(newCB.Finalizers) {
 		return true
 	}
+	if newCB.Labels != nil {
+		_, ok := newCB.Labels[ctrlmesh.CtrlmeshCircuitBreakerDisableKey]
+		if ok {
+			return true
+		}
+	}
 	oldProtoCB := conv.ConvertCircuitBreaker(oldCB)
 	newProtoCB := conv.ConvertCircuitBreaker(newCB)
 	return oldProtoCB.ConfigHash != newProtoCB.ConfigHash

pkg/proxy/circuitbreaker/manager.go

@@ -406,7 +406,7 @@ func (m *manager) doValidation(limitings []*ctrlmeshproto.RateLimiting, limiters
 					result.Reason = "OverLimit"
 					result.Message = fmt.Sprintf("the request is over limit by limiting rule: %s", limitings[idx].Name)
 					if limitings[idx].RecoverPolicy.Type == ctrlmeshproto.RateLimiting_RECOVER_POLICY_SLEEPING_WINDOW {
-						d, _ := time.ParseDuration(limitings[idx].Properties["sleepingWindowSize"])
+						d, _ := time.ParseDuration(limitings[idx].RecoverPolicy.SleepingWindowSize)
 						states[idx].triggerBreakerWithTimeWindow(d)
 						m.limiterStore.registerState(states[idx])
 					} else {