service config: default service config

clientconn.go

@@ -86,6 +86,9 @@ var (
 	// errCredentialsConflict indicates that grpc.WithTransportCredentials()
 	// and grpc.WithInsecure() are both called for a connection.
 	errCredentialsConflict = errors.New("grpc: transport credentials are set for an insecure connection (grpc.WithTransportCredentials() and grpc.WithInsecure() are both called)")
+	// errInvalidDefaultServiceConfig indicates that grpc.WithDefaultServiceConfig(string) provides
+	// an invalid service config string.
+	errInvalidDefaultServiceConfig = errors.New("grpc: the provided default service config is invalid")
 )
 
 const (
@@ -173,6 +176,13 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 		}
 	}
 
+	if cc.dopts.defaultServiceConfigRawJSON != nil {
+		sc, err := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON)
+		if err != nil {
+			return nil, errInvalidDefaultServiceConfig
+		}
+		cc.dopts.defaultServiceConfig = sc
+	}
 	cc.mkp = cc.dopts.copts.KeepaliveParams
 
 	if cc.dopts.copts.Dialer == nil {
@@ -383,7 +393,6 @@ type ClientConn struct {
 	mu              sync.RWMutex
 	resolverWrapper *ccResolverWrapper
 	sc              ServiceConfig
-	scRaw           string
 	conns           map[*addrConn]struct{}
 	// Keepalive parameter can be updated if a GoAway is received.
 	mkp             keepalive.ClientParameters
@@ -430,7 +439,8 @@ func (cc *ClientConn) scWatcher() {
 			// TODO: load balance policy runtime change is ignored.
 			// We may revisit this decision in the future.
 			cc.sc = sc
-			cc.scRaw = ""
+			s := ""
+			cc.sc.rawJSONString = &s
 			cc.mu.Unlock()
 		case <-cc.ctx.Done():
 			return
@@ -457,6 +467,24 @@ func (cc *ClientConn) waitForResolvedAddrs(ctx context.Context) error {
 	}
 }
 
+// gRPC should resort to default service config when:
+// * resolver service config is disabled
+// * or, resolver does not return a service config or returns an invalid one.
+func (cc *ClientConn) fallbackToDefaultServiceConfig(sc string) bool {
+	if cc.dopts.disableServiceConfig {
+		return true
+	}
+	// The logic below is temporary, will be removed once we change the resolver.State ServiceConfig field type.
+	// Right now, we assume that empty service config string means resolver does not return a config.
+	if sc == "" {
+		return true
+	}
+	// TODO: the logic below is temporary. Once we finish the logic to validate service config
+	// in resolver, we will replace the logic below.
+	_, err := parseServiceConfig(sc)
+	return err != nil
+}
+
 func (cc *ClientConn) updateResolverState(s resolver.State) error {
 	cc.mu.Lock()
 	defer cc.mu.Unlock()
@@ -467,26 +495,18 @@ func (cc *ClientConn) updateResolverState(s resolver.State) error {
 		return nil
 	}
 
-	if !cc.dopts.disableServiceConfig && cc.scRaw != s.ServiceConfig {
-		// New service config; apply it.
+	if cc.fallbackToDefaultServiceConfig(s.ServiceConfig) {
+		if cc.dopts.defaultServiceConfig != nil && cc.sc.rawJSONString == nil {
+			cc.applyServiceConfig(cc.dopts.defaultServiceConfig)
+		}
+	} else {
+		// TODO: the parsing logic below will be moved inside resolver.
 		sc, err := parseServiceConfig(s.ServiceConfig)
 		if err != nil {
-			fmt.Println("error parsing config: ", err)
 			return err
 		}
-		cc.scRaw = s.ServiceConfig
-		cc.sc = sc
-
-		if cc.sc.retryThrottling != nil {
-			newThrottler := &retryThrottler{
-				tokens: cc.sc.retryThrottling.MaxTokens,
-				max:    cc.sc.retryThrottling.MaxTokens,
-				thresh: cc.sc.retryThrottling.MaxTokens / 2,
-				ratio:  cc.sc.retryThrottling.TokenRatio,
-			}
-			cc.retryThrottler.Store(newThrottler)
-		} else {
-			cc.retryThrottler.Store((*retryThrottler)(nil))
+		if cc.sc.rawJSONString == nil || *cc.sc.rawJSONString != s.ServiceConfig {
+			cc.applyServiceConfig(sc)
 		}
 	}
 
@@ -748,6 +768,28 @@ func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method st
 	return t, done, nil
 }
 
+func (cc *ClientConn) applyServiceConfig(sc *ServiceConfig) error {
+	if sc == nil {
+		// should never reach here.
+		return fmt.Errorf("got nil pointer for service config")
+	}
+	cc.sc = *sc
+
+	if cc.sc.retryThrottling != nil {
+		newThrottler := &retryThrottler{
+			tokens: cc.sc.retryThrottling.MaxTokens,
+			max:    cc.sc.retryThrottling.MaxTokens,
+			thresh: cc.sc.retryThrottling.MaxTokens / 2,
+			ratio:  cc.sc.retryThrottling.TokenRatio,
+		}
+		cc.retryThrottler.Store(newThrottler)
+	} else {
+		cc.retryThrottler.Store((*retryThrottler)(nil))
+	}
+
+	return nil
+}
+
 func (cc *ClientConn) resolveNow(o resolver.ResolveNowOption) {
 	cc.mu.RLock()
 	r := cc.resolverWrapper

clientconn_test.go

@@ -1232,3 +1232,90 @@ func (s) TestUpdateAddresses_RetryFromFirstAddr(t *testing.T) {
 		t.Fatal("timed out waiting for any server to be contacted after tryUpdateAddrs")
 	}
 }
+
+func (s) TestDefaultServiceConfig(t *testing.T) {
+	r, cleanup := manual.GenerateAndRegisterManualResolver()
+	defer cleanup()
+	addr := r.Scheme() + ":///non.existent"
+	js := `{
+    "methodConfig": [
+        {
+            "name": [
+                {
+                    "service": "foo",
+                    "method": "bar"
+                }
+            ],
+            "waitForReady": true
+        }
+    ]
+}`
+	testInvalidDefaultServiceConfig(t)
+	testDefaultServiceConfigWhenResolverServiceConfigDisabled(t, r, addr, js)
+	testDefaultServiceConfigWhenResolverDoesNotReturnServiceConfig(t, r, addr, js)
+	testDefaultServiceConfigWhenResolverReturnInvalidServiceConfig(t, r, addr, js)
+}
+
+func verifyWaitForReadyEqualsTrue(cc *ClientConn) bool {
+	var i int
+	for i = 0; i < 10; i++ {
+		mc := cc.GetMethodConfig("/foo/bar")
+		if mc.WaitForReady != nil && *mc.WaitForReady == true {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	return i != 10
+}
+
+func testInvalidDefaultServiceConfig(t *testing.T) {
+	_, err := Dial("fake.com", WithInsecure(), WithDefaultServiceConfig(""))
+	if err != errInvalidDefaultServiceConfig {
+		t.Fatalf("Dial got err: %v, want: %v", err, errInvalidDefaultServiceConfig)
+	}
+}
+
+func testDefaultServiceConfigWhenResolverServiceConfigDisabled(t *testing.T, r resolver.Resolver, addr string, js string) {
+	cc, err := Dial(addr, WithInsecure(), WithDisableServiceConfig(), WithDefaultServiceConfig(js))
+	if err != nil {
+		t.Fatalf("Dial(%s, _) = _, %v, want _, <nil>", addr, err)
+	}
+	defer cc.Close()
+	// Resolver service config gets ignored since resolver service config is disabled.
+	r.(*manual.Resolver).UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: addr}},
+		ServiceConfig: "{}",
+	})
+	if !verifyWaitForReadyEqualsTrue(cc) {
+		t.Fatal("default service config failed to be applied after 1s")
+	}
+}
+
+func testDefaultServiceConfigWhenResolverDoesNotReturnServiceConfig(t *testing.T, r resolver.Resolver, addr string, js string) {
+	cc, err := Dial(addr, WithInsecure(), WithDefaultServiceConfig(js))
+	if err != nil {
+		t.Fatalf("Dial(%s, _) = _, %v, want _, <nil>", addr, err)
+	}
+	defer cc.Close()
+	r.(*manual.Resolver).UpdateState(resolver.State{
+		Addresses: []resolver.Address{{Addr: addr}},
+	})
+	if !verifyWaitForReadyEqualsTrue(cc) {
+		t.Fatal("default service config failed to be applied after 1s")
+	}
+}
+
+func testDefaultServiceConfigWhenResolverReturnInvalidServiceConfig(t *testing.T, r resolver.Resolver, addr string, js string) {
+	cc, err := Dial(addr, WithInsecure(), WithDefaultServiceConfig(js))
+	if err != nil {
+		t.Fatalf("Dial(%s, _) = _, %v, want _, <nil>", addr, err)
+	}
+	defer cc.Close()
+	r.(*manual.Resolver).UpdateState(resolver.State{
+		Addresses:     []resolver.Address{{Addr: addr}},
+		ServiceConfig: "{something wrong,}",
+	})
+	if !verifyWaitForReadyEqualsTrue(cc) {
+		t.Fatal("default service config failed to be applied after 1s")
+	}
+}

dialoptions.go

@@ -55,14 +55,16 @@ type dialOptions struct {
 	// balancer, and also by WithBalancerName dial option.
 	balancerBuilder balancer.Builder
 	// This is to support grpclb.
-	resolverBuilder      resolver.Builder
-	reqHandshake         envconfig.RequireHandshakeSetting
-	channelzParentID     int64
-	disableServiceConfig bool
-	disableRetry         bool
-	disableHealthCheck   bool
-	healthCheckFunc      internal.HealthChecker
-	minConnectTimeout    func() time.Duration
+	resolverBuilder             resolver.Builder
+	reqHandshake                envconfig.RequireHandshakeSetting
+	channelzParentID            int64
+	disableServiceConfig        bool
+	disableRetry                bool
+	disableHealthCheck          bool
+	healthCheckFunc             internal.HealthChecker
+	minConnectTimeout           func() time.Duration
+	defaultServiceConfig        *ServiceConfig
+	defaultServiceConfigRawJSON *string
 }
 
 // DialOption configures how we set up the connection.
@@ -447,6 +449,18 @@ func WithDisableServiceConfig() DialOption {
 	})
 }
 
+// WithDefaultServiceConfig returns a DialOption that configures the default service config, which
+// will be used in cases where:
+// 1. WithDisableServiceConfig is called.
+// 2. Resolver does not return service config or if the resolver gets and invalid config.
+//
+// This API is EXPERIMENTAL.
+func WithDefaultServiceConfig(s string) DialOption {
+	return newFuncDialOption(func(o *dialOptions) {
+		o.defaultServiceConfigRawJSON = &s
+	})
+}
+
 // WithDisableRetry returns a DialOption that disables retries, even if the
 // service config enables them.  This does not impact transparent retries, which
 // will happen automatically if no data is written to the wire or if the RPC is

resolver_conn_wrapper.go

@@ -118,7 +118,7 @@ func (ccr *ccResolverWrapper) UpdateState(s resolver.State) {
 	ccr.curState = s
 }
 
-// NewAddress is called by the resolver implemenetion to send addresses to gRPC.
+// NewAddress is called by the resolver implementation to send addresses to gRPC.
 func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
 	if ccr.isDone() {
 		return
@@ -131,7 +131,7 @@ func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
 	ccr.cc.updateResolverState(ccr.curState)
 }
 
-// NewServiceConfig is called by the resolver implemenetion to send service
+// NewServiceConfig is called by the resolver implementation to send service
 // configs to gRPC.
 func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
 	if ccr.isDone() {

service_config.go

@@ -99,6 +99,10 @@ type ServiceConfig struct {
 	// healthCheckConfig must be set as one of the requirement to enable LB channel
 	// health check.
 	healthCheckConfig *healthCheckConfig
+	// rawJSONString stores the pointer to the original service config json string that get parsed into
+	// this service config struct. Null pointer means this is a service config struct just defined,
+	// but no json string has been parsed and initialize the struct.
+	rawJSONString *string
 }
 
 // healthCheckConfig defines the go-native version of the LB channel health check config.
@@ -238,24 +242,22 @@ type jsonSC struct {
 	HealthCheckConfig   *healthCheckConfig
 }
 
-func parseServiceConfig(js string) (ServiceConfig, error) {
-	if len(js) == 0 {
-		return ServiceConfig{}, fmt.Errorf("no JSON service config provided")
-	}
+func parseServiceConfig(js string) (*ServiceConfig, error) {
 	var rsc jsonSC
 	err := json.Unmarshal([]byte(js), &rsc)
 	if err != nil {
 		grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
-		return ServiceConfig{}, err
+		return nil, err
 	}
 	sc := ServiceConfig{
 		LB:                rsc.LoadBalancingPolicy,
 		Methods:           make(map[string]MethodConfig),
 		retryThrottling:   rsc.RetryThrottling,
 		healthCheckConfig: rsc.HealthCheckConfig,
+		rawJSONString:     &js,
 	}
 	if rsc.MethodConfig == nil {
-		return sc, nil
+		return &sc, nil
 	}
 
 	for _, m := range *rsc.MethodConfig {
@@ -265,7 +267,7 @@ func parseServiceConfig(js string) (ServiceConfig, error) {
 		d, err := parseDuration(m.Timeout)
 		if err != nil {
 			grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
-			return ServiceConfig{}, err
+			return nil, err
 		}
 
 		mc := MethodConfig{
@@ -274,7 +276,7 @@ func parseServiceConfig(js string) (ServiceConfig, error) {
 		}
 		if mc.retryPolicy, err = convertRetryPolicy(m.RetryPolicy); err != nil {
 			grpclog.Warningf("grpc: parseServiceConfig error unmarshaling %s due to %v", js, err)
-			return ServiceConfig{}, err
+			return nil, err
 		}
 		if m.MaxRequestMessageBytes != nil {
 			if *m.MaxRequestMessageBytes > int64(maxInt) {
@@ -305,7 +307,7 @@ func parseServiceConfig(js string) (ServiceConfig, error) {
 			sc.retryThrottling = nil
 		}
 	}
-	return sc, nil
+	return &sc, nil
 }
 
 func convertRetryPolicy(jrp *jsonRetryPolicy) (p *retryPolicy, err error) {