[proxy]: Fix data race warning && Add service registry document

doc/controllers.md

@@ -1,8 +1,8 @@
-
 # Controllers
 
 - [Controllers](#controllers)
   - [System Controllers](#system-controllers)
+    - [ServiceRegistry](#serviceregistry)
     - [TrafficController](#trafficcontroller)
     - [RawConfigTrafficController](#rawconfigtrafficcontroller)
       - [HTTPServer](#httpserver)
@@ -46,6 +46,18 @@ The two categories are conceptual, which means they are not strict distinctions.
 
 For now, all system controllers can not be configured. It may gain this capability if necessary in the future.
 
+### ServiceRegistry
+
+We use system controller `ServiceRegistry` to be the service hub for all service registries. Current the drivers are
+
+- [ConsulServiceRegistry](#consulserviceregistry)
+- [EtcdServiceRegistry](#etcdserviceregistry)
+- [EurekaServiceRegistry](#eurekaserviceregistry)
+- [ZookeeperServiceRegistry](#zookeeperserviceregistry)
+- [NacosServiceRegistry](#nacosserviceregistry)
+
+The drivers need to offer notifying change periodically, and operations to the external service registry.
+
 ### TrafficController
 
 TrafficController handles the lifecycle of HTTPServer and HTTPPipeline and their relationship. It manages the resource in a namespaced way. HTTPServer accepts incoming traffic and routes it to HTTPPipelines in the same namespace. Most other controllers could handle traffic by leverage the ability of TrafficController..

pkg/filter/proxy/pool.go

@@ -129,25 +129,23 @@ func (p *pool) status() *PoolStatus {
 }
 
 func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader) string {
-	// FIXME: When the request matched mirror handler, there will be
-	// data race warning. Because after proxy handled the request,
-	// the HTTPPipeline and mux of HTTPServer will call ctx.AddTag() too.
-	// Even the the situation is not a bug, but the cautious data race
-	// detector of golang will warn it.
-	// Add a method ctx.AddTagWithLock()?
 	addTag := func(subPrefix, msg string) {
 		tag := stringtool.Cat(p.tagPrefix, "#", subPrefix, ": ", msg)
 		ctx.Lock()
 		ctx.AddTag(tag)
 		ctx.Unlock()
 	}
 
-	w := ctx.Response()
+	setStatusCode := func(code int) {
+		ctx.Lock()
+		ctx.Response().SetStatusCode(code)
+		ctx.Unlock()
+	}
 
 	server, err := p.servers.next(ctx)
 	if err != nil {
 		addTag("serverErr", err.Error())
-		w.SetStatusCode(http.StatusServiceUnavailable)
+		setStatusCode(http.StatusServiceUnavailable)
 		return resultInternalError
 	}
 	addTag("addr", server.URL)
@@ -157,7 +155,7 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader) string {
 		msg := stringtool.Cat("prepare request failed: ", err.Error())
 		logger.Errorf("BUG: %s", msg)
 		addTag("bug", msg)
-		w.SetStatusCode(http.StatusInternalServerError)
+		setStatusCode(http.StatusInternalServerError)
 		return resultInternalError
 	}
 
@@ -174,20 +172,22 @@ func (p *pool) handle(ctx context.HTTPContext, reqBody io.Reader) string {
 			return resultClientError
 		}
 
-		w.SetStatusCode(http.StatusServiceUnavailable)
+		setStatusCode(http.StatusServiceUnavailable)
 		return resultServerError
 	}
 
 	addTag("code", strconv.Itoa(resp.StatusCode))
 
 	ctx.Lock()
 	defer ctx.Unlock()
+	// NOTE: The code below can't use addTag and setStatusCode in case of deadlock.
+
 	respBody := p.statRequestResponse(ctx, req, resp, span)
 
 	if p.writeResponse {
-		w.SetStatusCode(resp.StatusCode)
-		w.Header().AddFromStd(resp.Header)
-		w.SetBody(respBody)
+		ctx.Response().SetStatusCode(resp.StatusCode)
+		ctx.Response().Header().AddFromStd(resp.Header)
+		ctx.Response().SetBody(respBody)
 
 		return ""
 	}

pkg/filter/proxy/proxy.go

@@ -279,14 +279,7 @@ func (b *Proxy) handle(ctx context.HTTPContext) (result string) {
 
 		wg := &sync.WaitGroup{}
 		wg.Add(1)
-		defer func() {
-			if result == "" {
-				// NOTE: Waiting for mirrorPool finishing
-				// only if mainPool/candidatePool handled
-				// with normal result.
-				wg.Wait()
-			}
-		}()
+		defer wg.Wait()
 
 		go func() {
 			defer wg.Done()