fix:fix circuit and fault store update

Author: spring-young

pkg/apis/ctrlmesh/constants/constants.go

@@ -63,7 +63,7 @@ const (
 	EnvEnableSim = "ENABLE_SIM"
 
 	EnvDisableCircuitBreaker         = "DISABLE_CIRCUIT_BREAKER"
-	EnvDisableFaultInjection         = "DISABLE_FAULT_INJECTION"
+	EnvEnableFaultInjection         = "ENABLE_FAULT_INJECTION"
 	EnvEnableApiServerCircuitBreaker = "ENABLE_API_SERVER_BREAKER"
 	EnvEnableRestCircuitBreaker      = "ENABLE_REST_BREAKER"
 	EnvEnableRestFaultInjection      = "ENABLE_REST_FAULT_INJECTION"
@@ -77,6 +77,7 @@ func AllProxySyncEnvKey() []string {
 		EnvIPTable,
 		EnvEnableWebHookProxy,
 		EnvDisableCircuitBreaker,
+		EnvEnableFaultInjection,
 		EnvEnableApiServerCircuitBreaker,
 		EnvEnableRestCircuitBreaker,
 		EnvEnableRestFaultInjection,

pkg/proxy/apiserver/handler.go

@@ -53,7 +53,7 @@ var (
 	enableIpTable       = os.Getenv(constants.EnvIPTable) == "true"
 
 	disableCircuitBreaker = os.Getenv(constants.EnvDisableCircuitBreaker) == "true"
-	disableFaultInjection = os.Getenv(constants.EnvDisableCircuitBreaker) == "true"
+	enableFaultInjection = os.Getenv(constants.EnvEnableFaultInjection) == "true"
 )
 
 type Proxy struct {
@@ -91,7 +91,7 @@ func NewProxy(opts *Options) (*Proxy, error) {
 	if opts.BreakerWrapperFunc != nil && !disableCircuitBreaker {
 		handler = opts.BreakerWrapperFunc(handler)
 	}
-	if opts.FaultInjectionWrapperFunc != nil && !disableFaultInjection {
+	if opts.FaultInjectionWrapperFunc != nil && enableFaultInjection {
 		handler = opts.FaultInjectionWrapperFunc(handler)
 	}
 	handler = genericfilters.WithWaitGroup(handler, opts.LongRunningFunc, opts.HandlerChainWaitGroup)

pkg/proxy/circuitbreaker/manager.go

@@ -79,6 +79,9 @@ func (m *manager) Sync(config *ctrlmeshproto.CircuitBreaker) (*ctrlmeshproto.Con
 				LimitingSnapshot: m.snapshot(config.Name),
 			}, nil
 		} else {
+			if ok{
+				m.unregisterRules(cb.Name)
+			}
 			m.breakerMap[config.Name] = config
 			m.registerRules(config)
 			var msg string
@@ -163,9 +166,6 @@ type ValidateResult struct {
 // RegisterRules register a circuit breaker to the local limiter store
 func (m *manager) registerRules(cb *ctrlmeshproto.CircuitBreaker) {
 	logger.Info("register rule", "circuit-breaker", cb.Name)
-	if _, ok := m.breakerMap[cb.Name]; ok {
-		m.unregisterRules(cb.Name)
-	}
 
 	for _, limiting := range cb.RateLimitings {
 		key := fmt.Sprintf("%s:%s", cb.Name, limiting.Name)

pkg/proxy/faultinjection/manager.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+
 	"math/rand"
 	"net/http"
 	"net/url"
@@ -30,9 +31,7 @@ import (
 
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
-	logf "sigs.k8s.io/controller-runtime/pkg/log"
-
-	// pkgfi "github.com/KusionStack/controller-mesh/circuitbreaker"
+	"k8s.io/klog/v2"
 
 	ctrlmeshproto "github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/proto"
 	"github.com/KusionStack/controller-mesh/pkg/utils"
@@ -41,7 +40,6 @@ import (
 const timeLayout = "15:04:05"
 
 var (
-	logger  = logf.Log.WithName("fault-injection-manager")
 	randNum = rand.New(rand.NewSource(time.Now().UnixNano()))
 )
 
@@ -89,6 +87,9 @@ func (m *manager) Sync(config *ctrlmeshproto.FaultInjection) (*ctrlmeshproto.Fau
 				Message: fmt.Sprintf("faultInjection spec hash not updated, hash %s", fi.ConfigHash),
 			}, nil
 		} else {
+			if ok{
+				m.unregisterRules(fi.Name)
+			}
 			m.faultInjectionMap[config.Name] = config
 			m.registerRules(config)
 			var msg string
@@ -117,16 +118,16 @@ func (m *manager) Sync(config *ctrlmeshproto.FaultInjection) (*ctrlmeshproto.Fau
 			}, nil
 		}
 	case ctrlmeshproto.FaultInjection_CHECK:
-		cb, ok := m.faultInjectionMap[config.Name]
+		fi, ok := m.faultInjectionMap[config.Name]
 		if !ok {
 			return &ctrlmeshproto.FaultInjectConfigResp{
 				Success: false,
-				Message: fmt.Sprintf("fault injection config %s not found", cb.Name),
+				Message: fmt.Sprintf("fault injection config %s not found", fi.Name),
 			}, nil
-		} else if config.ConfigHash != cb.ConfigHash {
+		} else if config.ConfigHash != fi.ConfigHash {
 			return &ctrlmeshproto.FaultInjectConfigResp{
 				Success: false,
-				Message: fmt.Sprintf("unequal fault injection %s hash, old %s, new %s", cb.Name, cb.ConfigHash, config.ConfigHash),
+				Message: fmt.Sprintf("unequal fault injection %s hash, old %s, new %s", fi.Name, fi.ConfigHash, config.ConfigHash),
 			}, nil
 		}
 		return &ctrlmeshproto.FaultInjectConfigResp{
@@ -150,17 +151,14 @@ func (m *manager) HandlerWrapper() func(http.Handler) http.Handler {
 
 func (m *manager) FaultInjectionRest(URL string, method string) (result *FaultInjectionResult) {
 	now := time.Now()
-	defer func() {
-		logger.Info("validate rest", "URL", URL, "method", method, "result", result, "cost time", time.Since(now).String())
-	}()
-
 	urls := generateWildcardUrls(URL, method)
 	for _, url := range urls {
 		faultInjections, states := m.faultInjectionStore.byIndex(IndexRest, url)
 		if len(faultInjections) == 0 {
 			continue
 		}
 		result = m.doFaultInjection(faultInjections, states)
+		klog.Infof("validate rest, URL: %s, method:%s, result: %v, cost time: %v ", URL, method, result, time.Since(now).String())
 		return result
 	}
 	result = &FaultInjectionResult{Abort: false, Reason: "No rule match"}
@@ -173,7 +171,7 @@ func generateWildcardUrls(URL string, method string) []string {
 	URL = strings.TrimSuffix(URL, "/")
 	u, err := url.Parse(URL)
 	if err != nil {
-		logger.Error(err, "failed to url", "URL", URL, "method", method)
+		klog.Errorf("failed to url, URL: %s, method: %s,err: %v", URL, method, err)
 		return result
 	}
 	if len(u.Path) > 0 {
@@ -189,16 +187,14 @@ func generateWildcardUrls(URL string, method string) []string {
 
 func (m *manager) FaultInjectionResource(namespace, apiGroup, resource, verb string) (result *FaultInjectionResult) {
 	now := time.Now()
-	defer func() {
-		logger.Info("validate resource", "namespace", namespace, "apiGroup", apiGroup, "resource", resource, "verb", verb, "result", result, "cost time", time.Since(now).String())
-	}()
 	seeds := generateWildcardSeeds(namespace, apiGroup, resource, verb)
 	for _, seed := range seeds {
 		faultInjections, states := m.faultInjectionStore.byIndex(IndexResource, seed)
 		if len(faultInjections) == 0 {
 			continue
 		}
 		result = m.doFaultInjection(faultInjections, states)
+		klog.Infof("validate resource, namespace: %s, apiGroup: %s, resource: %s, verb: %s, result: %v, cost time: %v, ", namespace, apiGroup, resource, verb, result, time.Since(now).String())
 		return result
 	}
 	result = &FaultInjectionResult{Abort: false, Reason: "No rule match"}
@@ -253,8 +249,12 @@ func withFaultInjection(injector FaultInjector, handler http.Handler) http.Handl
 			if apiErr.Code != http.StatusOK {
 				w.Header().Set("Content-Type", "application/json")
 				w.WriteHeader(int(apiErr.Code))
-				json.NewEncoder(w).Encode(apiErr)
-				logger.Info("faultinjection rule", fmt.Sprintf("fault injection, %s, %s,%d", result.Reason, result.Message, result.ErrCode))
+				if err := json.NewEncoder(w).Encode(apiErr); err != nil {
+					// Error encoding the JSON response, at this point the headers are already written.
+					klog.Errorf("failed to write api error response: %v", err)
+					return
+				}
+				klog.Infof("faultinjection rule: %s", fmt.Sprintf("fault injection, %s, %s,%d", apiErr.Reason, apiErr.Message, apiErr.Code))
 				return
 			}
 		}
@@ -278,7 +278,7 @@ func (m *manager) doFaultInjection(faultInjections []*ctrlmeshproto.HTTPFaultInj
 			if isInpercentRange(faultInjections[idx].Delay.Percent) {
 				delay := faultInjections[idx].Delay.GetFixedDelay()
 				delayDuration := delay.AsDuration()
-				logger.Info("Delaying time ", "for", delayDuration)
+				klog.Infof("Delaying time: %v ", delayDuration)
 				time.Sleep(delayDuration)
 			}
 		}
@@ -398,11 +398,7 @@ func isEffectiveTimeRange(timeRange *ctrlmeshproto.EffectiveTimeRange) bool {
 
 // RegisterRules register a fault injection to the local store
 func (m *manager) registerRules(fi *ctrlmeshproto.FaultInjection) {
-	logger.Info("register rule", "faultInjection", fi.Name)
-	if _, ok := m.faultInjectionMap[fi.Name]; ok {
-		m.unregisterRules(fi.Name)
-	}
-
+	klog.Infof("register rule, faultInjection: %s", fi.Name)
 	for _, faultInjection := range fi.HttpFaultInjections {
 		key := fmt.Sprintf("%s:%s", fi.Name, faultInjection.Name)
 		m.faultInjectionStore.createOrUpdateRule(
@@ -413,7 +409,7 @@ func (m *manager) registerRules(fi *ctrlmeshproto.FaultInjection) {
 
 // UnregisterRules unregister a fault injection to the local store
 func (m *manager) unregisterRules(fiName string) {
-	logger.Info("unregister rule", "faultInjection", fiName)
+	klog.Infof("unregister rule, faultInjection: %s", fiName)
 	fi, ok := m.faultInjectionMap[fiName]
 	if !ok {
 		return

pkg/proxy/http/http_proxy.go

@@ -37,8 +37,10 @@ import (
 )
 
 var (
-	enableRestBreaker = os.Getenv(constants.EnvEnableRestCircuitBreaker) == "true"
-	logger            = logf.Log.WithName("http-proxy")
+	enableRestBreaker        = os.Getenv(constants.EnvEnableRestCircuitBreaker) == "true"
+	enableRestFaultInjection = os.Getenv(constants.EnvEnableRestFaultInjection) == "true"
+
+	logger = logf.Log.WithName("http-proxy")
 )
 
 type ITProxy interface {
@@ -86,16 +88,20 @@ func (t *tproxy) handleHTTP(resp http.ResponseWriter, req *http.Request) {
 	}
 	klog.Infof("handel http request, url: %s ", realEndPointUrl.String())
 	// faultinjection
-	result := t.FaultInjector.FaultInjectionRest(req.Header.Get(meshhttp.HeaderMeshRealEndpoint), req.Method)
-	if result.Abort {
-		apiErr := utils.HttpToAPIError(int(result.ErrCode), req.Method, result.Message)
-		resp.Header().Set("Content-Type", "application/json")
-		resp.WriteHeader(int(apiErr.Code))
-		if err := json.NewEncoder(resp).Encode(apiErr); err != nil {
-			http.Error(resp, fmt.Sprintf("fail to inject fault %v", err), http.StatusInternalServerError)
+	if enableRestFaultInjection {
+		result := t.FaultInjector.FaultInjectionRest(req.Header.Get(meshhttp.HeaderMeshRealEndpoint), req.Method)
+		if result.Abort {
+			apiErr := utils.HttpToAPIError(int(result.ErrCode), req.Method, result.Message)
+			if apiErr.Code != http.StatusOK {
+				resp.Header().Set("Content-Type", "application/json")
+				resp.WriteHeader(int(apiErr.Code))
+				if err := json.NewEncoder(resp).Encode(apiErr); err != nil {
+					http.Error(resp, fmt.Sprintf("fail to inject fault %v", err), http.StatusInternalServerError)
+				}
+				klog.Infof("faultInjection rule, rule: %s", fmt.Sprintf("fault injection, %s, %s,%d", result.Reason, result.Message, result.ErrCode))
+				return
+			}
 		}
-		klog.Infof("faultInjection rule, rule: %s", fmt.Sprintf("fault injection, %s, %s,%d", result.Reason, result.Message, result.ErrCode))
-		return
 	}
 
 	// circuitbreaker