1/* 2 * 3 * Copyright 2020 gRPC authors. 4 * 5 * Licensed under the Apache License, Version 2.0 (the "License"); 6 * you may not use this file except in compliance with the License. 7 * You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 * 17 */ 18 19package rls 20 21import ( 22 "bytes" 23 "encoding/json" 24 "fmt" 25 "time" 26 27 "github.com/golang/protobuf/jsonpb" 28 "github.com/golang/protobuf/ptypes" 29 durationpb "github.com/golang/protobuf/ptypes/duration" 30 "google.golang.org/grpc/balancer" 31 "google.golang.org/grpc/balancer/rls/internal/keys" 32 rlspb "google.golang.org/grpc/balancer/rls/internal/proto/grpc_lookup_v1" 33 "google.golang.org/grpc/internal/grpcutil" 34 "google.golang.org/grpc/resolver" 35 "google.golang.org/grpc/serviceconfig" 36) 37 38const ( 39 // This is max duration that we are willing to cache RLS responses. If the 40 // service config doesn't specify a value for max_age or if it specified a 41 // value greater that this, we will use this value instead. 42 maxMaxAge = 5 * time.Minute 43 // If lookup_service_timeout is not specified in the service config, we use 44 // a default of 10 seconds. 45 defaultLookupServiceTimeout = 10 * time.Second 46 // This is set to the targetNameField in the child policy config during 47 // service config validation. 48 dummyChildPolicyTarget = "target_name_to_be_filled_in_later" 49) 50 51// lbConfig contains the parsed and validated contents of the 52// loadBalancingConfig section of the service config. The RLS LB policy will 53// use this to directly access config data instead of ploughing through proto 54// fields. 55type lbConfig struct { 56 serviceconfig.LoadBalancingConfig 57 58 kbMap keys.BuilderMap 59 lookupService string 60 lookupServiceTimeout time.Duration 61 maxAge time.Duration 62 staleAge time.Duration 63 cacheSizeBytes int64 64 defaultTarget string 65 cpName string 66 cpTargetField string 67 cpConfig map[string]json.RawMessage 68} 69 70func (lbCfg *lbConfig) Equal(other *lbConfig) bool { 71 return lbCfg.kbMap.Equal(other.kbMap) && 72 lbCfg.lookupService == other.lookupService && 73 lbCfg.lookupServiceTimeout == other.lookupServiceTimeout && 74 lbCfg.maxAge == other.maxAge && 75 lbCfg.staleAge == other.staleAge && 76 lbCfg.cacheSizeBytes == other.cacheSizeBytes && 77 lbCfg.defaultTarget == other.defaultTarget && 78 lbCfg.cpName == other.cpName && 79 lbCfg.cpTargetField == other.cpTargetField && 80 cpConfigEqual(lbCfg.cpConfig, other.cpConfig) 81} 82 83func cpConfigEqual(am, bm map[string]json.RawMessage) bool { 84 if (bm == nil) != (am == nil) { 85 return false 86 } 87 if len(bm) != len(am) { 88 return false 89 } 90 91 for k, jsonA := range am { 92 jsonB, ok := bm[k] 93 if !ok { 94 return false 95 } 96 if !bytes.Equal(jsonA, jsonB) { 97 return false 98 } 99 } 100 return true 101} 102 103// This struct resembles the JSON respresentation of the loadBalancing config 104// and makes it easier to unmarshal. 105type lbConfigJSON struct { 106 RouteLookupConfig json.RawMessage 107 ChildPolicy []*loadBalancingConfig 108 ChildPolicyConfigTargetFieldName string 109} 110 111// loadBalancingConfig represents a single load balancing config, 112// stored in JSON format. 113// 114// TODO(easwars): This code seems to be repeated in a few places 115// (service_config.go and in the xds code as well). Refactor and re-use. 116type loadBalancingConfig struct { 117 Name string 118 Config json.RawMessage 119} 120 121// MarshalJSON returns a JSON encoding of l. 122func (l *loadBalancingConfig) MarshalJSON() ([]byte, error) { 123 return nil, fmt.Errorf("rls: loadBalancingConfig.MarshalJSON() is unimplemented") 124} 125 126// UnmarshalJSON parses the JSON-encoded byte slice in data and stores it in l. 127func (l *loadBalancingConfig) UnmarshalJSON(data []byte) error { 128 var cfg map[string]json.RawMessage 129 if err := json.Unmarshal(data, &cfg); err != nil { 130 return err 131 } 132 for name, config := range cfg { 133 l.Name = name 134 l.Config = config 135 } 136 return nil 137} 138 139// ParseConfig parses and validates the JSON representation of the service 140// config and returns the loadBalancingConfig to be used by the RLS LB policy. 141// 142// Helps implement the balancer.ConfigParser interface. 143// 144// The following validation checks are performed: 145// * routeLookupConfig: 146// ** grpc_keybuilders field: 147// - must have at least one entry 148// - must not have two entries with the same Name 149// - must not have any entry with a Name with the service field unset or 150// empty 151// - must not have any entries without a Name 152// - must not have a headers entry that has required_match set 153// - must not have two headers entries with the same key within one entry 154// ** lookup_service field: 155// - must be set and non-empty and must parse as a target URI 156// ** max_age field: 157// - if not specified or is greater than maxMaxAge, it will be reset to 158// maxMaxAge 159// ** stale_age field: 160// - if the value is greater than or equal to max_age, it is ignored 161// - if set, then max_age must also be set 162// ** valid_targets field: 163// - will be ignored 164// ** cache_size_bytes field: 165// - must be greater than zero 166// - TODO(easwars): Define a minimum value for this field, to be used when 167// left unspecified 168// * childPolicy field: 169// - must find a valid child policy with a valid config (the child policy must 170// be able to parse the provided config successfully when we pass it a dummy 171// target name in the target_field provided by the 172// childPolicyConfigTargetFieldName field) 173// * childPolicyConfigTargetFieldName field: 174// - must be set and non-empty 175func (*rlsBB) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, error) { 176 cfgJSON := &lbConfigJSON{} 177 if err := json.Unmarshal(c, cfgJSON); err != nil { 178 return nil, fmt.Errorf("rls: json unmarshal failed for service config {%+v}: %v", string(c), err) 179 } 180 181 m := jsonpb.Unmarshaler{AllowUnknownFields: true} 182 rlsProto := &rlspb.RouteLookupConfig{} 183 if err := m.Unmarshal(bytes.NewReader(cfgJSON.RouteLookupConfig), rlsProto); err != nil { 184 return nil, fmt.Errorf("rls: bad RouteLookupConfig proto {%+v}: %v", string(cfgJSON.RouteLookupConfig), err) 185 } 186 187 var childPolicy *loadBalancingConfig 188 for _, lbcfg := range cfgJSON.ChildPolicy { 189 if balancer.Get(lbcfg.Name) != nil { 190 childPolicy = lbcfg 191 break 192 } 193 } 194 195 kbMap, err := keys.MakeBuilderMap(rlsProto) 196 if err != nil { 197 return nil, err 198 } 199 200 lookupService := rlsProto.GetLookupService() 201 if lookupService == "" { 202 return nil, fmt.Errorf("rls: empty lookup_service in service config {%+v}", string(c)) 203 } 204 parsedTarget := grpcutil.ParseTarget(lookupService, false) 205 if parsedTarget.Scheme == "" { 206 parsedTarget.Scheme = resolver.GetDefaultScheme() 207 } 208 if resolver.Get(parsedTarget.Scheme) == nil { 209 return nil, fmt.Errorf("rls: invalid target URI in lookup_service {%s}", lookupService) 210 } 211 212 lookupServiceTimeout, err := convertDuration(rlsProto.GetLookupServiceTimeout()) 213 if err != nil { 214 return nil, fmt.Errorf("rls: failed to parse lookup_service_timeout in service config {%+v}: %v", string(c), err) 215 } 216 if lookupServiceTimeout == 0 { 217 lookupServiceTimeout = defaultLookupServiceTimeout 218 } 219 maxAge, err := convertDuration(rlsProto.GetMaxAge()) 220 if err != nil { 221 return nil, fmt.Errorf("rls: failed to parse max_age in service config {%+v}: %v", string(c), err) 222 } 223 staleAge, err := convertDuration(rlsProto.GetStaleAge()) 224 if err != nil { 225 return nil, fmt.Errorf("rls: failed to parse staleAge in service config {%+v}: %v", string(c), err) 226 } 227 if staleAge != 0 && maxAge == 0 { 228 return nil, fmt.Errorf("rls: stale_age is set, but max_age is not in service config {%+v}", string(c)) 229 } 230 if staleAge >= maxAge { 231 logger.Info("rls: stale_age {%v} is greater than max_age {%v}, ignoring it", staleAge, maxAge) 232 staleAge = 0 233 } 234 if maxAge == 0 || maxAge > maxMaxAge { 235 logger.Infof("rls: max_age in service config is %v, using %v", maxAge, maxMaxAge) 236 maxAge = maxMaxAge 237 } 238 cacheSizeBytes := rlsProto.GetCacheSizeBytes() 239 if cacheSizeBytes <= 0 { 240 return nil, fmt.Errorf("rls: cache_size_bytes must be greater than 0 in service config {%+v}", string(c)) 241 } 242 if childPolicy == nil { 243 return nil, fmt.Errorf("rls: childPolicy is invalid in service config {%+v}", string(c)) 244 } 245 if cfgJSON.ChildPolicyConfigTargetFieldName == "" { 246 return nil, fmt.Errorf("rls: childPolicyConfigTargetFieldName field is not set in service config {%+v}", string(c)) 247 } 248 // TODO(easwars): When we start instantiating the child policy from the 249 // parent RLS LB policy, we could make this function a method on the 250 // lbConfig object and share the code. We would be parsing the child policy 251 // config again during that time. The only difference betweeen now and then 252 // would be that we would be using real targetField name instead of the 253 // dummy. So, we could make the targetName field a parameter to this 254 // function during the refactor. 255 cpCfg, err := validateChildPolicyConfig(childPolicy, cfgJSON.ChildPolicyConfigTargetFieldName) 256 if err != nil { 257 return nil, err 258 } 259 260 return &lbConfig{ 261 kbMap: kbMap, 262 lookupService: lookupService, 263 lookupServiceTimeout: lookupServiceTimeout, 264 maxAge: maxAge, 265 staleAge: staleAge, 266 cacheSizeBytes: cacheSizeBytes, 267 defaultTarget: rlsProto.GetDefaultTarget(), 268 // TODO(easwars): Once we refactor validateChildPolicyConfig and make 269 // it a method on the lbConfig object, we could directly store the 270 // balancer.Builder and/or balancer.ConfigParser here instead of the 271 // Name. That would mean that we would have to create the lbConfig 272 // object here first before validating the childPolicy config, but 273 // that's a minor detail. 274 cpName: childPolicy.Name, 275 cpTargetField: cfgJSON.ChildPolicyConfigTargetFieldName, 276 cpConfig: cpCfg, 277 }, nil 278} 279 280// validateChildPolicyConfig validates the child policy config received in the 281// service config. This makes it possible for us to reject service configs 282// which contain invalid child policy configs which we know will fail for sure. 283// 284// It does the following: 285// * Unmarshals the provided child policy config into a map of string to 286// json.RawMessage. This allows us to add an entry to the map corresponding 287// to the targetFieldName that we received in the service config. 288// * Marshals the map back into JSON, finds the config parser associated with 289// the child policy and asks it to validate the config. 290// * If the validation succeeded, removes the dummy entry from the map and 291// returns it. If any of the above steps failed, it returns an error. 292func validateChildPolicyConfig(cp *loadBalancingConfig, cpTargetField string) (map[string]json.RawMessage, error) { 293 var childConfig map[string]json.RawMessage 294 if err := json.Unmarshal(cp.Config, &childConfig); err != nil { 295 return nil, fmt.Errorf("rls: json unmarshal failed for child policy config {%+v}: %v", cp.Config, err) 296 } 297 childConfig[cpTargetField], _ = json.Marshal(dummyChildPolicyTarget) 298 299 jsonCfg, err := json.Marshal(childConfig) 300 if err != nil { 301 return nil, fmt.Errorf("rls: json marshal failed for child policy config {%+v}: %v", childConfig, err) 302 } 303 builder := balancer.Get(cp.Name) 304 if builder == nil { 305 // This should never happen since we already made sure that the child 306 // policy name mentioned in the service config is a valid one. 307 return nil, fmt.Errorf("rls: balancer builder not found for child_policy %v", cp.Name) 308 } 309 parser, ok := builder.(balancer.ConfigParser) 310 if !ok { 311 return nil, fmt.Errorf("rls: balancer builder for child_policy does not implement balancer.ConfigParser: %v", cp.Name) 312 } 313 _, err = parser.ParseConfig(jsonCfg) 314 if err != nil { 315 return nil, fmt.Errorf("rls: childPolicy config validation failed: %v", err) 316 } 317 delete(childConfig, cpTargetField) 318 return childConfig, nil 319} 320 321func convertDuration(d *durationpb.Duration) (time.Duration, error) { 322 if d == nil { 323 return 0, nil 324 } 325 return ptypes.Duration(d) 326} 327