authority.go

Documentation: google.golang.org/grpc/xds/internal/xdsclient

     1  /*
     2   *
     3   * Copyright 2021 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  package xdsclient
    19  
    20  import (
    21  	"context"
    22  	"errors"
    23  	"fmt"
    24  	"strings"
    25  	"sync"
    26  	"time"
    27  
    28  	"google.golang.org/grpc/internal/grpclog"
    29  	"google.golang.org/grpc/internal/grpcsync"
    30  	"google.golang.org/grpc/internal/xds/bootstrap"
    31  	"google.golang.org/grpc/xds/internal/xdsclient/load"
    32  	"google.golang.org/grpc/xds/internal/xdsclient/transport"
    33  	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
    34  	"google.golang.org/protobuf/types/known/anypb"
    35  )
    36  
    37  type watchState int
    38  
    39  const (
    40  	watchStateStarted   watchState = iota // Watch started, request not yet set.
    41  	watchStateRequested                   // Request sent for resource being watched.
    42  	watchStateReceived                    // Response received for resource being watched.
    43  	watchStateTimeout                     // Watch timer expired, no response.
    44  	watchStateCanceled                    // Watch cancelled.
    45  )
    46  
    47  type resourceState struct {
    48  	watchers        map[xdsresource.ResourceWatcher]bool // Set of watchers for this resource
    49  	cache           xdsresource.ResourceData             // Most recent ACKed update for this resource
    50  	md              xdsresource.UpdateMetadata           // Metadata for the most recent update
    51  	deletionIgnored bool                                 // True if resource deletion was ignored for a prior update
    52  
    53  	// Common watch state for all watchers of this resource.
    54  	wTimer *time.Timer // Expiry timer
    55  	wState watchState  // State of the watch
    56  }
    57  
    58  // authority wraps all state associated with a single management server. It
    59  // contains the transport used to communicate with the management server and a
    60  // cache of resource state for resources requested from the management server.
    61  //
    62  // Bootstrap configuration could contain multiple entries in the authorities map
    63  // that share the same server config (server address and credentials to use). We
    64  // share the same authority instance amongst these entries, and the reference
    65  // counting is taken care of by the `clientImpl` type.
    66  type authority struct {
    67  	serverCfg          *bootstrap.ServerConfig       // Server config for this authority
    68  	bootstrapCfg       *bootstrap.Config             // Full bootstrap configuration
    69  	refCount           int                           // Reference count of watches referring to this authority
    70  	serializer         *grpcsync.CallbackSerializer  // Callback serializer for invoking watch callbacks
    71  	resourceTypeGetter func(string) xdsresource.Type // ResourceType registry lookup
    72  	transport          *transport.Transport          // Underlying xDS transport to the management server
    73  	watchExpiryTimeout time.Duration                 // Resource watch expiry timeout
    74  	logger             *grpclog.PrefixLogger
    75  
    76  	// A two level map containing the state of all the resources being watched.
    77  	//
    78  	// The first level map key is the ResourceType (Listener, Route etc). This
    79  	// allows us to have a single map for all resources instead of having per
    80  	// resource-type maps.
    81  	//
    82  	// The second level map key is the resource name, with the value being the
    83  	// actual state of the resource.
    84  	resourcesMu sync.Mutex
    85  	resources   map[xdsresource.Type]map[string]*resourceState
    86  	closed      bool
    87  }
    88  
    89  // authorityArgs is a convenience struct to wrap arguments required to create a
    90  // new authority. All fields here correspond directly to appropriate fields
    91  // stored in the authority struct.
    92  type authorityArgs struct {
    93  	// The reason for passing server config and bootstrap config separately
    94  	// (although the former is part of the latter) is because authorities in the
    95  	// bootstrap config might contain an empty server config, and in this case,
    96  	// the top-level server config is to be used.
    97  	serverCfg          *bootstrap.ServerConfig
    98  	bootstrapCfg       *bootstrap.Config
    99  	serializer         *grpcsync.CallbackSerializer
   100  	resourceTypeGetter func(string) xdsresource.Type
   101  	watchExpiryTimeout time.Duration
   102  	logger             *grpclog.PrefixLogger
   103  }
   104  
   105  func newAuthority(args authorityArgs) (*authority, error) {
   106  	ret := &authority{
   107  		serverCfg:          args.serverCfg,
   108  		bootstrapCfg:       args.bootstrapCfg,
   109  		serializer:         args.serializer,
   110  		resourceTypeGetter: args.resourceTypeGetter,
   111  		watchExpiryTimeout: args.watchExpiryTimeout,
   112  		logger:             args.logger,
   113  		resources:          make(map[xdsresource.Type]map[string]*resourceState),
   114  	}
   115  
   116  	tr, err := transport.New(transport.Options{
   117  		ServerCfg:      *args.serverCfg,
   118  		OnRecvHandler:  ret.handleResourceUpdate,
   119  		OnErrorHandler: ret.newConnectionError,
   120  		OnSendHandler:  ret.transportOnSendHandler,
   121  		Logger:         args.logger,
   122  		NodeProto:      args.bootstrapCfg.NodeProto,
   123  	})
   124  	if err != nil {
   125  		return nil, fmt.Errorf("creating new transport to %q: %v", args.serverCfg, err)
   126  	}
   127  	ret.transport = tr
   128  	return ret, nil
   129  }
   130  
   131  // transportOnSendHandler is called by the underlying transport when it sends a
   132  // resource request successfully. Timers are activated for resources waiting for
   133  // a response.
   134  func (a *authority) transportOnSendHandler(u *transport.ResourceSendInfo) {
   135  	rType := a.resourceTypeGetter(u.URL)
   136  	// Resource type not found is not expected under normal circumstances, since
   137  	// the resource type url passed to the transport is determined by the authority.
   138  	if rType == nil {
   139  		a.logger.Warningf("Unknown resource type url: %s.", u.URL)
   140  		return
   141  	}
   142  	a.resourcesMu.Lock()
   143  	defer a.resourcesMu.Unlock()
   144  	a.startWatchTimersLocked(rType, u.ResourceNames)
   145  }
   146  
   147  func (a *authority) handleResourceUpdate(resourceUpdate transport.ResourceUpdate) error {
   148  	rType := a.resourceTypeGetter(resourceUpdate.URL)
   149  	if rType == nil {
   150  		return xdsresource.NewErrorf(xdsresource.ErrorTypeResourceTypeUnsupported, "Resource URL %v unknown in response from server", resourceUpdate.URL)
   151  	}
   152  
   153  	opts := &xdsresource.DecodeOptions{
   154  		BootstrapConfig: a.bootstrapCfg,
   155  		ServerConfig:    a.serverCfg,
   156  	}
   157  	updates, md, err := decodeAllResources(opts, rType, resourceUpdate)
   158  	a.updateResourceStateAndScheduleCallbacks(rType, updates, md)
   159  	return err
   160  }
   161  
   162  func (a *authority) updateResourceStateAndScheduleCallbacks(rType xdsresource.Type, updates map[string]resourceDataErrTuple, md xdsresource.UpdateMetadata) {
   163  	a.resourcesMu.Lock()
   164  	defer a.resourcesMu.Unlock()
   165  
   166  	resourceStates := a.resources[rType]
   167  	for name, uErr := range updates {
   168  		if state, ok := resourceStates[name]; ok {
   169  			// Cancel the expiry timer associated with the resource once a
   170  			// response is received, irrespective of whether the update is a
   171  			// good one or not.
   172  			//
   173  			// We check for watch states `started` and `requested` here to
   174  			// accommodate for a race which can happen in the following
   175  			// scenario:
   176  			// - When a watch is registered, it is possible that the ADS stream
   177  			//   is not yet created. In this case, the request for the resource
   178  			//   is not sent out immediately. An entry in the `resourceStates`
   179  			//   map is created with a watch state of `started`.
   180  			// - Once the stream is created, it is possible that the management
   181  			//   server might respond with the requested resource before we send
   182  			//   out request for the same. If we don't check for `started` here,
   183  			//   and move the state to `received`, we will end up starting the
   184  			//   timer when the request gets sent out. And since the management
   185  			//   server already sent us the resource, there is a good chance
   186  			//   that it will not send it again. This would eventually lead to
   187  			//   the timer firing, even though we have the resource in the
   188  			//   cache.
   189  			if state.wState == watchStateStarted || state.wState == watchStateRequested {
   190  				// It is OK to ignore the return value from Stop() here because
   191  				// if the timer has already fired, it means that the timer watch
   192  				// expiry callback is blocked on the same lock that we currently
   193  				// hold. Since we move the state to `received` here, the timer
   194  				// callback will be a no-op.
   195  				if state.wTimer != nil {
   196  					state.wTimer.Stop()
   197  				}
   198  				state.wState = watchStateReceived
   199  			}
   200  
   201  			if uErr.err != nil {
   202  				// On error, keep previous version of the resource. But update
   203  				// status and error.
   204  				state.md.ErrState = md.ErrState
   205  				state.md.Status = md.Status
   206  				for watcher := range state.watchers {
   207  					watcher := watcher
   208  					err := uErr.err
   209  					a.serializer.Schedule(func(context.Context) { watcher.OnError(err) })
   210  				}
   211  				continue
   212  			}
   213  
   214  			if state.deletionIgnored {
   215  				state.deletionIgnored = false
   216  				a.logger.Infof("A valid update was received for resource %q of type %q after previously ignoring a deletion", name, rType.TypeName())
   217  			}
   218  			// Notify watchers only if this is a first time update or it is different
   219  			// from the one currently cached.
   220  			if state.cache == nil || !state.cache.Equal(uErr.resource) {
   221  				for watcher := range state.watchers {
   222  					watcher := watcher
   223  					resource := uErr.resource
   224  					a.serializer.Schedule(func(context.Context) { watcher.OnUpdate(resource) })
   225  				}
   226  			}
   227  			// Sync cache.
   228  			a.logger.Debugf("Resource type %q with name %q added to cache", rType.TypeName(), name)
   229  			state.cache = uErr.resource
   230  			// Set status to ACK, and clear error state. The metadata might be a
   231  			// NACK metadata because some other resources in the same response
   232  			// are invalid.
   233  			state.md = md
   234  			state.md.ErrState = nil
   235  			state.md.Status = xdsresource.ServiceStatusACKed
   236  			if md.ErrState != nil {
   237  				state.md.Version = md.ErrState.Version
   238  			}
   239  		}
   240  	}
   241  
   242  	// If this resource type requires that all resources be present in every
   243  	// SotW response from the server, a response that does not include a
   244  	// previously seen resource will be interpreted as a deletion of that
   245  	// resource unless ignore_resource_deletion option was set in the server
   246  	// config.
   247  	if !rType.AllResourcesRequiredInSotW() {
   248  		return
   249  	}
   250  	for name, state := range resourceStates {
   251  		if state.cache == nil {
   252  			// If the resource state does not contain a cached update, which can
   253  			// happen when:
   254  			// - resource was newly requested but has not yet been received, or,
   255  			// - resource was removed as part of a previous update,
   256  			// we don't want to generate an error for the watchers.
   257  			//
   258  			// For the first of the above two conditions, this ADS response may
   259  			// be in reaction to an earlier request that did not yet request the
   260  			// new resource, so its absence from the response does not
   261  			// necessarily indicate that the resource does not exist. For that
   262  			// case, we rely on the request timeout instead.
   263  			//
   264  			// For the second of the above two conditions, we already generated
   265  			// an error when we received the first response which removed this
   266  			// resource. So, there is no need to generate another one.
   267  			continue
   268  		}
   269  		if _, ok := updates[name]; !ok {
   270  			// The metadata status is set to "ServiceStatusNotExist" if a
   271  			// previous update deleted this resource, in which case we do not
   272  			// want to repeatedly call the watch callbacks with a
   273  			// "resource-not-found" error.
   274  			if state.md.Status == xdsresource.ServiceStatusNotExist {
   275  				continue
   276  			}
   277  			// Per A53, resource deletions are ignored if the `ignore_resource_deletion`
   278  			// server feature is enabled through the bootstrap configuration. If the
   279  			// resource deletion is to be ignored, the resource is not removed from
   280  			// the cache and the corresponding OnResourceDoesNotExist() callback is
   281  			// not invoked on the watchers.
   282  			if a.serverCfg.IgnoreResourceDeletion {
   283  				if !state.deletionIgnored {
   284  					state.deletionIgnored = true
   285  					a.logger.Warningf("Ignoring resource deletion for resource %q of type %q", name, rType.TypeName())
   286  				}
   287  				continue
   288  			}
   289  			// If resource exists in cache, but not in the new update, delete
   290  			// the resource from cache, and also send a resource not found error
   291  			// to indicate resource removed. Metadata for the resource is still
   292  			// maintained, as this is required by CSDS.
   293  			state.cache = nil
   294  			state.md = xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusNotExist}
   295  			for watcher := range state.watchers {
   296  				watcher := watcher
   297  				a.serializer.Schedule(func(context.Context) { watcher.OnResourceDoesNotExist() })
   298  			}
   299  		}
   300  	}
   301  }
   302  
   303  type resourceDataErrTuple struct {
   304  	resource xdsresource.ResourceData
   305  	err      error
   306  }
   307  
   308  func decodeAllResources(opts *xdsresource.DecodeOptions, rType xdsresource.Type, update transport.ResourceUpdate) (map[string]resourceDataErrTuple, xdsresource.UpdateMetadata, error) {
   309  	timestamp := time.Now()
   310  	md := xdsresource.UpdateMetadata{
   311  		Version:   update.Version,
   312  		Timestamp: timestamp,
   313  	}
   314  
   315  	topLevelErrors := make([]error, 0)           // Tracks deserialization errors, where we don't have a resource name.
   316  	perResourceErrors := make(map[string]error)  // Tracks resource validation errors, where we have a resource name.
   317  	ret := make(map[string]resourceDataErrTuple) // Return result, a map from resource name to either resource data or error.
   318  	for _, r := range update.Resources {
   319  		result, err := rType.Decode(opts, r)
   320  
   321  		// Name field of the result is left unpopulated only when resource
   322  		// deserialization fails.
   323  		name := ""
   324  		if result != nil {
   325  			name = xdsresource.ParseName(result.Name).String()
   326  		}
   327  		if err == nil {
   328  			ret[name] = resourceDataErrTuple{resource: result.Resource}
   329  			continue
   330  		}
   331  		if name == "" {
   332  			topLevelErrors = append(topLevelErrors, err)
   333  			continue
   334  		}
   335  		perResourceErrors[name] = err
   336  		// Add place holder in the map so we know this resource name was in
   337  		// the response.
   338  		ret[name] = resourceDataErrTuple{err: err}
   339  	}
   340  
   341  	if len(topLevelErrors) == 0 && len(perResourceErrors) == 0 {
   342  		md.Status = xdsresource.ServiceStatusACKed
   343  		return ret, md, nil
   344  	}
   345  
   346  	md.Status = xdsresource.ServiceStatusNACKed
   347  	errRet := combineErrors(rType.TypeName(), topLevelErrors, perResourceErrors)
   348  	md.ErrState = &xdsresource.UpdateErrorMetadata{
   349  		Version:   update.Version,
   350  		Err:       errRet,
   351  		Timestamp: timestamp,
   352  	}
   353  	return ret, md, errRet
   354  }
   355  
   356  // startWatchTimersLocked is invoked upon transport.OnSend() callback with resources
   357  // requested on the underlying ADS stream. This satisfies the conditions to start
   358  // watch timers per A57 [https://github.com/grpc/proposal/blob/master/A57-xds-client-failure-mode-behavior.md#handling-resources-that-do-not-exist]
   359  //
   360  // Caller must hold a.resourcesMu.
   361  func (a *authority) startWatchTimersLocked(rType xdsresource.Type, resourceNames []string) {
   362  	resourceStates := a.resources[rType]
   363  	for _, resourceName := range resourceNames {
   364  		if state, ok := resourceStates[resourceName]; ok {
   365  			if state.wState != watchStateStarted {
   366  				continue
   367  			}
   368  			state.wTimer = time.AfterFunc(a.watchExpiryTimeout, func() {
   369  				a.handleWatchTimerExpiry(rType, resourceName, state)
   370  			})
   371  			state.wState = watchStateRequested
   372  		}
   373  	}
   374  }
   375  
   376  // stopWatchTimersLocked is invoked upon connection errors to stops watch timers
   377  // for resources that have been requested, but not yet responded to by the management
   378  // server.
   379  //
   380  // Caller must hold a.resourcesMu.
   381  func (a *authority) stopWatchTimersLocked() {
   382  	for _, rType := range a.resources {
   383  		for resourceName, state := range rType {
   384  			if state.wState != watchStateRequested {
   385  				continue
   386  			}
   387  			if !state.wTimer.Stop() {
   388  				// If the timer has already fired, it means that the timer watch expiry
   389  				// callback is blocked on the same lock that we currently hold. Don't change
   390  				// the watch state and instead let the watch expiry callback handle it.
   391  				a.logger.Warningf("Watch timer for resource %v already fired. Ignoring here.", resourceName)
   392  				continue
   393  			}
   394  			state.wTimer = nil
   395  			state.wState = watchStateStarted
   396  		}
   397  	}
   398  }
   399  
   400  // newConnectionError is called by the underlying transport when it receives a
   401  // connection error. The error will be forwarded to all the resource watchers.
   402  func (a *authority) newConnectionError(err error) {
   403  	a.resourcesMu.Lock()
   404  	defer a.resourcesMu.Unlock()
   405  
   406  	a.stopWatchTimersLocked()
   407  
   408  	// We do not consider it an error if the ADS stream was closed after having received
   409  	// a response on the stream. This is because there are legitimate reasons why the server
   410  	// may need to close the stream during normal operations, such as needing to rebalance
   411  	// load or the underlying connection hitting its max connection age limit.
   412  	// See gRFC A57 for more details.
   413  	if xdsresource.ErrType(err) == xdsresource.ErrTypeStreamFailedAfterRecv {
   414  		a.logger.Warningf("Watchers not notified since ADS stream failed after having received at least one response: %v", err)
   415  		return
   416  	}
   417  
   418  	for _, rType := range a.resources {
   419  		for _, state := range rType {
   420  			// Propagate the connection error from the transport layer to all watchers.
   421  			for watcher := range state.watchers {
   422  				watcher := watcher
   423  				a.serializer.Schedule(func(context.Context) {
   424  					watcher.OnError(xdsresource.NewErrorf(xdsresource.ErrorTypeConnection, "xds: error received from xDS stream: %v", err))
   425  				})
   426  			}
   427  		}
   428  	}
   429  }
   430  
   431  // Increments the reference count. Caller must hold parent's authorityMu.
   432  func (a *authority) refLocked() {
   433  	a.refCount++
   434  }
   435  
   436  // Decrements the reference count. Caller must hold parent's authorityMu.
   437  func (a *authority) unrefLocked() int {
   438  	a.refCount--
   439  	return a.refCount
   440  }
   441  
   442  func (a *authority) close() {
   443  	a.transport.Close()
   444  
   445  	a.resourcesMu.Lock()
   446  	a.closed = true
   447  	a.resourcesMu.Unlock()
   448  }
   449  
   450  func (a *authority) watchResource(rType xdsresource.Type, resourceName string, watcher xdsresource.ResourceWatcher) func() {
   451  	a.logger.Debugf("New watch for type %q, resource name %q", rType.TypeName(), resourceName)
   452  	a.resourcesMu.Lock()
   453  	defer a.resourcesMu.Unlock()
   454  
   455  	// Lookup the ResourceType specific resources from the top-level map. If
   456  	// there is no entry for this ResourceType, create one.
   457  	resources := a.resources[rType]
   458  	if resources == nil {
   459  		resources = make(map[string]*resourceState)
   460  		a.resources[rType] = resources
   461  	}
   462  
   463  	// Lookup the resourceState for the particular resource that the watch is
   464  	// being registered for. If this is the first watch for this resource,
   465  	// instruct the transport layer to send a DiscoveryRequest for the same.
   466  	state := resources[resourceName]
   467  	if state == nil {
   468  		a.logger.Debugf("First watch for type %q, resource name %q", rType.TypeName(), resourceName)
   469  		state = &resourceState{
   470  			watchers: make(map[xdsresource.ResourceWatcher]bool),
   471  			md:       xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusRequested},
   472  			wState:   watchStateStarted,
   473  		}
   474  		resources[resourceName] = state
   475  		a.sendDiscoveryRequestLocked(rType, resources)
   476  	}
   477  	// Always add the new watcher to the set of watchers.
   478  	state.watchers[watcher] = true
   479  
   480  	// If we have a cached copy of the resource, notify the new watcher.
   481  	if state.cache != nil {
   482  		if a.logger.V(2) {
   483  			a.logger.Infof("Resource type %q with resource name %q found in cache: %s", rType.TypeName(), resourceName, state.cache.ToJSON())
   484  		}
   485  		resource := state.cache
   486  		a.serializer.Schedule(func(context.Context) { watcher.OnUpdate(resource) })
   487  	}
   488  
   489  	return func() {
   490  		a.resourcesMu.Lock()
   491  		defer a.resourcesMu.Unlock()
   492  
   493  		// We already have a reference to the resourceState for this particular
   494  		// resource. Avoid indexing into the two-level map to figure this out.
   495  
   496  		// Delete this particular watcher from the list of watchers, so that its
   497  		// callback will not be invoked in the future.
   498  		state.wState = watchStateCanceled
   499  		delete(state.watchers, watcher)
   500  		if len(state.watchers) > 0 {
   501  			return
   502  		}
   503  
   504  		// There are no more watchers for this resource, delete the state
   505  		// associated with it, and instruct the transport to send a request
   506  		// which does not include this resource name.
   507  		a.logger.Debugf("Removing last watch for type %q, resource name %q", rType.TypeName(), resourceName)
   508  		delete(resources, resourceName)
   509  		a.sendDiscoveryRequestLocked(rType, resources)
   510  	}
   511  }
   512  
   513  func (a *authority) handleWatchTimerExpiry(rType xdsresource.Type, resourceName string, state *resourceState) {
   514  	a.resourcesMu.Lock()
   515  	defer a.resourcesMu.Unlock()
   516  
   517  	if a.closed {
   518  		return
   519  	}
   520  	a.logger.Warningf("Watch for resource %q of type %s timed out", resourceName, rType.TypeName())
   521  
   522  	switch state.wState {
   523  	case watchStateRequested:
   524  		// This is the only state where we need to handle the timer expiry by
   525  		// invoking appropriate watch callbacks. This is handled outside the switch.
   526  	case watchStateCanceled:
   527  		return
   528  	default:
   529  		a.logger.Warningf("Unexpected watch state %q for resource %q.", state.wState, resourceName)
   530  		return
   531  	}
   532  
   533  	state.wState = watchStateTimeout
   534  	// With the watch timer firing, it is safe to assume that the resource does
   535  	// not exist on the management server.
   536  	state.cache = nil
   537  	state.md = xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusNotExist}
   538  	for watcher := range state.watchers {
   539  		watcher := watcher
   540  		a.serializer.Schedule(func(context.Context) { watcher.OnResourceDoesNotExist() })
   541  	}
   542  }
   543  
   544  func (a *authority) triggerResourceNotFoundForTesting(rType xdsresource.Type, resourceName string) {
   545  	a.resourcesMu.Lock()
   546  	defer a.resourcesMu.Unlock()
   547  
   548  	if a.closed {
   549  		return
   550  	}
   551  	resourceStates := a.resources[rType]
   552  	state, ok := resourceStates[resourceName]
   553  	if !ok {
   554  		return
   555  	}
   556  	// if watchStateTimeout already triggered resource not found above from
   557  	// normal watch expiry.
   558  	if state.wState == watchStateCanceled || state.wState == watchStateTimeout {
   559  		return
   560  	}
   561  	state.wState = watchStateTimeout
   562  	state.cache = nil
   563  	state.md = xdsresource.UpdateMetadata{Status: xdsresource.ServiceStatusNotExist}
   564  	for watcher := range state.watchers {
   565  		watcher := watcher
   566  		a.serializer.Schedule(func(context.Context) { watcher.OnResourceDoesNotExist() })
   567  	}
   568  }
   569  
   570  // sendDiscoveryRequestLocked sends a discovery request for the specified
   571  // resource type and resource names. Even though this method does not directly
   572  // access the resource cache, it is important that `resourcesMu` be beld when
   573  // calling this method to ensure that a consistent snapshot of resource names is
   574  // being requested.
   575  func (a *authority) sendDiscoveryRequestLocked(rType xdsresource.Type, resources map[string]*resourceState) {
   576  	resourcesToRequest := make([]string, len(resources))
   577  	i := 0
   578  	for name := range resources {
   579  		resourcesToRequest[i] = name
   580  		i++
   581  	}
   582  	a.transport.SendRequest(rType.TypeURL(), resourcesToRequest)
   583  }
   584  
   585  func (a *authority) reportLoad() (*load.Store, func()) {
   586  	return a.transport.ReportLoad()
   587  }
   588  
   589  func (a *authority) dumpResources() map[string]map[string]xdsresource.UpdateWithMD {
   590  	a.resourcesMu.Lock()
   591  	defer a.resourcesMu.Unlock()
   592  
   593  	dump := make(map[string]map[string]xdsresource.UpdateWithMD)
   594  	for rType, resourceStates := range a.resources {
   595  		states := make(map[string]xdsresource.UpdateWithMD)
   596  		for name, state := range resourceStates {
   597  			var raw *anypb.Any
   598  			if state.cache != nil {
   599  				raw = state.cache.Raw()
   600  			}
   601  			states[name] = xdsresource.UpdateWithMD{
   602  				MD:  state.md,
   603  				Raw: raw,
   604  			}
   605  		}
   606  		dump[rType.TypeURL()] = states
   607  	}
   608  	return dump
   609  }
   610  
   611  func combineErrors(rType string, topLevelErrors []error, perResourceErrors map[string]error) error {
   612  	var errStrB strings.Builder
   613  	errStrB.WriteString(fmt.Sprintf("error parsing %q response: ", rType))
   614  	if len(topLevelErrors) > 0 {
   615  		errStrB.WriteString("top level errors: ")
   616  		for i, err := range topLevelErrors {
   617  			if i != 0 {
   618  				errStrB.WriteString(";\n")
   619  			}
   620  			errStrB.WriteString(err.Error())
   621  		}
   622  	}
   623  	if len(perResourceErrors) > 0 {
   624  		var i int
   625  		for name, err := range perResourceErrors {
   626  			if i != 0 {
   627  				errStrB.WriteString(";\n")
   628  			}
   629  			i++
   630  			errStrB.WriteString(fmt.Sprintf("resource %q: %v", name, err.Error()))
   631  		}
   632  	}
   633  	return errors.New(errStrB.String())
   634  }
   635
View as plain text