Repository: go-kratos/kratos
Pull request #1231 (2021-07-23 06:19:45)

fix http resovler bug
之前的http resovler 一旦watch 返回error会导致整个流程return,现在改成先阻塞获取，再异步监听（永不退出）


Patch:
diff --git a/config/env/env.go b/config/env/env.go
index adf37c98..8190078d 100644
--- a/config/env/env.go
+++ b/config/env/env.go
@@ -16,7 +16,12 @@ func NewSource(prefixs ...string) config.Source {
 }
 
 func (e *env) Load() (kv []*config.KeyValue, err error) {
-	for _, envstr := range os.Environ() {
+	return e.load(os.Environ()), nil
+}
+
+func (e *env) load(envStrings []string) []*config.KeyValue {
+	var kv []*config.KeyValue
+	for _, envstr := range envStrings {
 		var k, v string
 		subs := strings.SplitN(envstr, "=", 2)
 		k = subs[0]
@@ -35,13 +40,18 @@ func (e *env) Load() (kv []*config.KeyValue, err error) {
 				k = k[1:]
 			}
 		}
-
+		if strings.Contains(k, "_") {
+			kv = append(kv, &config.KeyValue{
+				Key:   strings.Replace(k, "_", ".", -1),
+				Value: []byte(v),
+			})
+		}
 		kv = append(kv, &config.KeyValue{
 			Key:   k,
 			Value: []byte(v),
 		})
 	}
-	return
+	return kv
 }
 
 func (e *env) Watch() (config.Watcher, error) {
diff --git a/encoding/form/proto_decode.go b/encoding/form/proto_decode.go
index 0a817b30..94b12d9c 100644
--- a/encoding/form/proto_decode.go
+++ b/encoding/form/proto_decode.go
@@ -36,9 +36,10 @@ func populateFieldValues(v protoreflect.Message, fieldPath []string, values []st
 	var fd protoreflect.FieldDescriptor
 	for i, fieldName := range fieldPath {
 		fields := v.Descriptor().Fields()
-
-		if fd = fields.ByName(protoreflect.Name(fieldName)); fd == nil {
-			fd = fields.ByJSONName(fieldName)
+		if fd = getDescriptorByFieldAndName(fields, fieldName); fd == nil {
+			if len(fieldName) > 2 && strings.HasSuffix(fieldName, "[]") {
+				fd = getDescriptorByFieldAndName(fields, strings.TrimSuffix(fieldName, "[]"))
+			}
 			if fd == nil {
 				// ignore unexpected field.
 				return nil
@@ -72,6 +73,17 @@ func populateFieldValues(v protoreflect.Message, fieldPath []string, values []st
 	return populateField(fd, v, values[0])
 }
 
+func getDescriptorByFieldAndName(fields protoreflect.FieldDescriptors, fieldName string) protoreflect.FieldDescriptor {
+	var fd protoreflect.FieldDescriptor
+	if fd = fields.ByName(protoreflect.Name(fieldName)); fd == nil {
+		fd = fields.ByJSONName(fieldName)
+		if fd == nil {
+			return nil
+		}
+	}
+	return fd
+}
+
 func populateField(fd protoreflect.FieldDescriptor, v protoreflect.Message, value string) error {
 	val, err := parseField(fd, value)
 	if err != nil {
diff --git a/transport/http/resolver.go b/transport/http/resolver.go
index 77ed2699..a8737bf1 100644
--- a/transport/http/resolver.go
+++ b/transport/http/resolver.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/url"
 	"sync"
+	"time"
 
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/registry"
@@ -36,8 +37,9 @@ func parseTarget(endpoint string) (*Target, error) {
 }
 
 type resolver struct {
-	lock  sync.RWMutex
-	nodes []*registry.ServiceInstance
+	lock    sync.RWMutex
+	nodes   []*registry.ServiceInstance
+	updater Updater
 
 	target  *Target
 	watcher registry.Watcher
@@ -53,69 +55,73 @@ func newResolver(ctx context.Context, discovery registry.Discovery, target *Targ
 		target:  target,
 		watcher: watcher,
 		logger:  log.NewHelper(log.DefaultLogger),
+		updater: updater,
 	}
-	done := make(chan error, 1)
-	go func() {
-		for {
-			var executed bool
-			services, err := watcher.Next()
-			if err != nil {
-				r.logger.Errorf("http client watch service %v got unexpected error:=%v", target, err)
-				if block {
-					select {
-					case done <- err:
-					default:
-					}
-				}
-				return
-			}
-			var nodes []*registry.ServiceInstance
-			for _, in := range services {
-				_, endpoint, err := parseEndpoint(in.Endpoints)
+	if block {
+		done := make(chan error, 1)
+		go func() {
+			for {
+				services, err := watcher.Next()
 				if err != nil {
-					r.logger.Errorf("Failed to parse (%v) discovery endpoint: %v error %v", target, in.Endpoints, err)
-					continue
+					done <- err
+					return
 				}
-				if endpoint == "" {
-					continue
-				}
-				nodes = append(nodes, in)
-			}
-			if len(nodes) != 0 {
-				updater.Update(nodes)
-				r.lock.Lock()
-				r.nodes = nodes
-				r.lock.Unlock()
-				if block && !executed {
-					executed = true
+				r.update(services)
+				if len(r.nodes) > 0 {
 					done <- nil
+					return
 				}
-			} else {
-				r.logger.Warnf("[http resovler]Zero endpoint found,refused to write,ser: %s ins: %v", target.Endpoint, nodes)
 			}
-		}
-	}()
-	if block {
+		}()
 		select {
-		case e := <-done:
-			if e != nil {
+		case err := <-done:
+			if err != nil {
 				watcher.Stop()
+				return nil, err
 			}
-			return r, e
 		case <-ctx.Done():
 			r.logger.Errorf("http client watch service %v reaching context deadline!", target)
 			watcher.Stop()
 			return nil, ctx.Err()
 		}
 	}
+
+	go func() {
+		for {
+			services, err := watcher.Next()
+			if err != nil {
+				r.logger.Errorf("http client watch service %v got unexpected error:=%v", target, err)
+				time.Sleep(time.Second)
+				continue
+			}
+			r.update(services)
+		}
+	}()
+
 	return r, nil
 }
 
-func (r *resolver) fetch(ctx context.Context) []*registry.ServiceInstance {
-	r.lock.RLock()
-	nodes := r.nodes
-	r.lock.RUnlock()
-	return nodes
+func (r *resolver) update(services []*registry.ServiceInstance) {
+	var nodes []*registry.ServiceInstance
+	for _, in := range services {
+		_, endpoint, err := parseEndpoint(in.Endpoints)
+		if err != nil {
+			r.logger.Errorf("Failed to parse (%v) discovery endpoint: %v error %v", r.target, in.Endpoints, err)
+			continue
+		}
+		if endpoint == "" {
+			continue
+		}
+		nodes = append(nodes, in)
+	}
+	if len(nodes) != 0 {
+		r.updater.Update(nodes)
+		r.lock.Lock()
+		r.nodes = nodes
+		r.lock.Unlock()
+	} else {
+		r.logger.Warnf("[http resovler]Zero endpoint found,refused to write,ser: %s ins: %v", r.target.Endpoint, nodes)
+	}
 }
 
 func parseEndpoint(endpoints []string) (string, string, error) {


Test patch:
diff --git a/config/env/env_test.go b/config/env/env_test.go
index db5ec6d5..87f6def7 100644
--- a/config/env/env_test.go
+++ b/config/env/env_test.go
@@ -245,3 +245,67 @@ func TestEnvWithoutPrefix(t *testing.T) {
 		})
 	}
 }
+
+func Test_env_load(t *testing.T) {
+	type fields struct {
+		prefixs []string
+	}
+	type args struct {
+		envStrings []string
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   []*config.KeyValue
+	}{
+		{
+			name: "without prefixes",
+			fields: fields{
+				prefixs: nil,
+			},
+			args: args{
+				envStrings: []string{
+					"SERVICE_NAME=kratos_app",
+					"ADDR=192.168.0.1",
+					"AGE=20",
+				},
+			},
+			want: []*config.KeyValue{
+				{Key: "SERVICE_NAME", Value: []byte("kratos_app"), Format: ""},
+				{Key: "ADDR", Value: []byte("192.168.0.1"), Format: ""},
+				{Key: "AGE", Value: []byte("20"), Format: ""},
+			},
+		},
+
+		{
+			name: "with prefixes",
+			fields: fields{
+				prefixs: []string{"KRATOS_", "FOO"},
+			},
+			args: args{
+				envStrings: []string{
+					"KRATOS_SERVICE_NAME=kratos_app",
+					"KRATOS_ADDR=192.168.0.1",
+					"FOO_AGE=20",
+				},
+			},
+			want: []*config.KeyValue{
+				{Key: "SERVICE_NAME", Value: []byte("kratos_app"), Format: ""},
+				{Key: "ADDR", Value: []byte("192.168.0.1"), Format: ""},
+				{Key: "AGE", Value: []byte("20"), Format: ""},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := &env{
+				prefixs: tt.fields.prefixs,
+			}
+			got := e.load(tt.args.envStrings)
+			if !reflect.DeepEqual(tt.want, got) {
+				t.Errorf("env.load() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}