client_new.go

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

     1  /*
     2   *
     3   * Copyright 2022 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  
    19  package xdsclient
    20  
    21  import (
    22  	"bytes"
    23  	"context"
    24  	"encoding/json"
    25  	"fmt"
    26  	"sync"
    27  	"sync/atomic"
    28  	"time"
    29  
    30  	"google.golang.org/grpc/internal"
    31  	"google.golang.org/grpc/internal/cache"
    32  	"google.golang.org/grpc/internal/grpcsync"
    33  	"google.golang.org/grpc/internal/xds/bootstrap"
    34  	"google.golang.org/grpc/xds/internal/xdsclient/xdsresource"
    35  )
    36  
    37  // New returns a new xDS client configured by the bootstrap file specified in env
    38  // variable GRPC_XDS_BOOTSTRAP or GRPC_XDS_BOOTSTRAP_CONFIG.
    39  //
    40  // The returned client is a reference counted singleton instance. This function
    41  // creates a new client only when one doesn't already exist.
    42  //
    43  // The second return value represents a close function which releases the
    44  // caller's reference on the returned client.  The caller is expected to invoke
    45  // it once they are done using the client. The underlying client will be closed
    46  // only when all references are released, and it is safe for the caller to
    47  // invoke this close function multiple times.
    48  func New() (XDSClient, func(), error) {
    49  	return newRefCountedWithConfig(nil)
    50  }
    51  
    52  // NewWithConfig returns a new xDS client configured by the given config.
    53  //
    54  // The second return value represents a close function which releases the
    55  // caller's reference on the returned client.  The caller is expected to invoke
    56  // it once they are done using the client. The underlying client will be closed
    57  // only when all references are released, and it is safe for the caller to
    58  // invoke this close function multiple times.
    59  //
    60  // # Internal/Testing Only
    61  //
    62  // This function should ONLY be used for internal (c2p resolver) and/or testing
    63  // purposese. DO NOT use this elsewhere. Use New() instead.
    64  func NewWithConfig(config *bootstrap.Config) (XDSClient, func(), error) {
    65  	return newRefCountedWithConfig(config)
    66  }
    67  
    68  // newWithConfig returns a new xdsClient with the given config.
    69  func newWithConfig(config *bootstrap.Config, watchExpiryTimeout time.Duration, idleAuthorityDeleteTimeout time.Duration) (*clientImpl, error) {
    70  	ctx, cancel := context.WithCancel(context.Background())
    71  	c := &clientImpl{
    72  		done:               grpcsync.NewEvent(),
    73  		config:             config,
    74  		watchExpiryTimeout: watchExpiryTimeout,
    75  		serializer:         grpcsync.NewCallbackSerializer(ctx),
    76  		serializerClose:    cancel,
    77  		resourceTypes:      newResourceTypeRegistry(),
    78  		authorities:        make(map[string]*authority),
    79  		idleAuthorities:    cache.NewTimeoutCache(idleAuthorityDeleteTimeout),
    80  	}
    81  
    82  	c.logger = prefixLogger(c)
    83  	c.logger.Infof("Created client to xDS management server: %s", config.XDSServer)
    84  	return c, nil
    85  }
    86  
    87  // NewWithConfigForTesting returns an xDS client for the specified bootstrap
    88  // config, separate from the global singleton.
    89  //
    90  // The second return value represents a close function which the caller is
    91  // expected to invoke once they are done using the client.  It is safe for the
    92  // caller to invoke this close function multiple times.
    93  //
    94  // # Testing Only
    95  //
    96  // This function should ONLY be used for testing purposes.
    97  // TODO(easwars): Document the new close func.
    98  func NewWithConfigForTesting(config *bootstrap.Config, watchExpiryTimeout, authorityIdleTimeout time.Duration) (XDSClient, func(), error) {
    99  	cl, err := newWithConfig(config, watchExpiryTimeout, authorityIdleTimeout)
   100  	if err != nil {
   101  		return nil, nil, err
   102  	}
   103  	return cl, grpcsync.OnceFunc(cl.close), nil
   104  }
   105  
   106  func init() {
   107  	internal.TriggerXDSResourceNameNotFoundClient = triggerXDSResourceNameNotFoundClient
   108  }
   109  
   110  var singletonClientForTesting = atomic.Pointer[clientRefCounted]{}
   111  
   112  func triggerXDSResourceNameNotFoundClient(resourceType, resourceName string) error {
   113  	c := singletonClientForTesting.Load()
   114  	return internal.TriggerXDSResourceNameNotFoundForTesting.(func(func(xdsresource.Type, string) error, string, string) error)(c.clientImpl.triggerResourceNotFoundForTesting, resourceType, resourceName)
   115  }
   116  
   117  // NewWithBootstrapContentsForTesting returns an xDS client for this config,
   118  // separate from the global singleton.
   119  //
   120  // The second return value represents a close function which the caller is
   121  // expected to invoke once they are done using the client.  It is safe for the
   122  // caller to invoke this close function multiple times.
   123  //
   124  // # Testing Only
   125  //
   126  // This function should ONLY be used for testing purposes.
   127  func NewWithBootstrapContentsForTesting(contents []byte) (XDSClient, func(), error) {
   128  	// Normalize the contents
   129  	buf := bytes.Buffer{}
   130  	err := json.Indent(&buf, contents, "", "")
   131  	if err != nil {
   132  		return nil, nil, fmt.Errorf("xds: error normalizing JSON: %v", err)
   133  	}
   134  	contents = bytes.TrimSpace(buf.Bytes())
   135  
   136  	c, err := getOrMakeClientForTesting(contents)
   137  	if err != nil {
   138  		return nil, nil, err
   139  	}
   140  	singletonClientForTesting.Store(c)
   141  	return c, grpcsync.OnceFunc(func() {
   142  		clientsMu.Lock()
   143  		defer clientsMu.Unlock()
   144  		if c.decrRef() == 0 {
   145  			c.close()
   146  			delete(clients, string(contents))
   147  			singletonClientForTesting.Store(nil)
   148  		}
   149  	}), nil
   150  }
   151  
   152  // getOrMakeClientForTesting creates a new reference counted client (separate
   153  // from the global singleton) for the given config, or returns an existing one.
   154  // It takes care of incrementing the reference count for the returned client,
   155  // and leaves the caller responsible for decrementing the reference count once
   156  // the client is no longer needed.
   157  func getOrMakeClientForTesting(config []byte) (*clientRefCounted, error) {
   158  	clientsMu.Lock()
   159  	defer clientsMu.Unlock()
   160  
   161  	if c := clients[string(config)]; c != nil {
   162  		c.incrRef()
   163  		return c, nil
   164  	}
   165  
   166  	bcfg, err := bootstrap.NewConfigFromContents(config)
   167  	if err != nil {
   168  		return nil, fmt.Errorf("bootstrap config %s: %v", string(config), err)
   169  	}
   170  	cImpl, err := newWithConfig(bcfg, defaultWatchExpiryTimeout, defaultIdleAuthorityDeleteTimeout)
   171  	if err != nil {
   172  		return nil, fmt.Errorf("creating xDS client: %v", err)
   173  	}
   174  	c := &clientRefCounted{clientImpl: cImpl, refCount: 1}
   175  	clients[string(config)] = c
   176  	return c, nil
   177  }
   178  
   179  var (
   180  	clients   = map[string]*clientRefCounted{}
   181  	clientsMu sync.Mutex
   182  )
   183
View as plain text