Merge branch 'main' into group-client-reqs

chained/persist.go

@@ -136,9 +136,7 @@ func persistStats(statsFilePath string) {
 		time.Sleep(15 * time.Second)
 		statsMx.Lock()
 		dialers := make([]dialer.ProxyDialer, 0, len(statsTrackingDialers))
-		for _, d := range statsTrackingDialers {
-			dialers = append(dialers, d)
-		}
+		copy(dialers, statsTrackingDialers)
 		statsMx.Unlock()
 		doPersistStats(statsFilePath, dialers)
 	}

config/config.go

@@ -105,6 +105,9 @@ type options struct {
 	// dictate whether the fetcher will use dual fetching (from fronted and
 	// chained URLs) or not.
 	rt http.RoundTripper
+
+	// opName is the operation name to use for ops.Begin when fetching configs.
+	opName string
 }
 
 // pipeConfig creates a new config pipeline for reading a specified type of
@@ -122,7 +125,7 @@ func pipeConfig(opts *options) (stop func()) {
 
 	// lastCfg is accessed by both the current goroutine when dispatching
 	// saved or embedded configs, and in a separate goroutine for polling
-	// for remote configs.  There should never be mutual access by these
+	// for remote configs. There should never be mutual access by these
 	// goroutines, however, since the polling routine is started after the prior
 	// calls to dispatch return.
 	var (
@@ -140,7 +143,7 @@ func pipeConfig(opts *options) (stop func()) {
 		if reflect.DeepEqual(a, b) {
 			log.Debug("Config unchanged, ignoring")
 		} else {
-			log.Debug("Dispatching updated config")
+			log.Debugf("Dispatching updated config from %v", src)
 			opts.dispatch(cfg, src)
 			lastCfg = b
 		}
@@ -186,7 +189,7 @@ func pipeConfig(opts *options) (stop func()) {
 	// Now continually poll for new configs and pipe them back to the dispatch function.
 	if !opts.sticky {
 		fetcher := newHttpFetcher(opts.userConfig, opts.rt, opts.originURL)
-		go conf.configFetcher(stopCh,
+		go conf.configFetcher(opts.opName, stopCh,
 			func(cfg interface{}) {
 				dispatch(cfg, Fetched)
 			}, fetcher, opts.sleep,
@@ -308,9 +311,9 @@ func (conf *config) embedded(data []byte) (interface{}, error) {
 	return conf.unmarshaler(data)
 }
 
-func (conf *config) configFetcher(stopCh chan bool, dispatch func(interface{}), fetcher Fetcher, defaultSleep func() time.Duration, log golog.Logger) {
+func (conf *config) configFetcher(opName string, stopCh chan bool, dispatch func(interface{}), fetcher Fetcher, defaultSleep func() time.Duration, log golog.Logger) {
 	for {
-		sleepDuration := conf.fetchConfig(stopCh, dispatch, fetcher, log)
+		sleepDuration := conf.fetchConfig(opName, stopCh, dispatch, fetcher, log)
 		if sleepDuration == noSleep {
 			sleepDuration = defaultSleep()
 		}
@@ -325,8 +328,8 @@ func (conf *config) configFetcher(stopCh chan bool, dispatch func(interface{}),
 	}
 }
 
-func (conf *config) fetchConfig(stopCh chan bool, dispatch func(interface{}), fetcher Fetcher, log golog.Logger) time.Duration {
-	if bytes, sleepTime, err := fetcher.fetch(); err != nil {
+func (conf *config) fetchConfig(opName string, stopCh chan bool, dispatch func(interface{}), fetcher Fetcher, log golog.Logger) time.Duration {
+	if bytes, sleepTime, err := fetcher.fetch(opName); err != nil {
 		log.Errorf("Error fetching config: %v", err)
 		return sleepTime
 	} else if bytes == nil {

config/config_test.go

@@ -114,7 +114,7 @@ func TestProductionGlobal(t *testing.T) {
 
 	f := newHttpFetcher(newTestUserConfig(), &http.Transport{}, testURL)
 
-	cfgBytes, _, err := f.fetch()
+	cfgBytes, _, err := f.fetch("testOpName")
 	if !assert.NoError(t, err, "Error fetching global config from %s", testURL) {
 		return
 	}
@@ -185,7 +185,7 @@ func TestPollIntervals(t *testing.T) {
 		dispatch := func(cfg interface{}) {}
 
 		stopChan := make(chan bool)
-		go cfg.configFetcher(stopChan, dispatch, fetcher, func() time.Duration { return pollInterval }, log)
+		go cfg.configFetcher("testOpName", stopChan, dispatch, fetcher, func() time.Duration { return pollInterval }, log)
 		time.Sleep(waitTime)
 		close(stopChan)
 

config/fetcher.go

@@ -33,7 +33,7 @@ func ForceCountry(countryCode string) {
 
 // Fetcher is an interface for fetching config updates.
 type Fetcher interface {
-	fetch() ([]byte, time.Duration, error)
+	fetch(string) ([]byte, time.Duration, error)
 }
 
 // fetcher periodically fetches the latest cloud configuration.
@@ -66,8 +66,8 @@ func newHttpFetcher(conf common.UserConfig, rt http.RoundTripper, originURL stri
 	}
 }
 
-func (cf *fetcher) fetch() ([]byte, time.Duration, error) {
-	op := ops.Begin("fetch_config")
+func (cf *fetcher) fetch(opName string) ([]byte, time.Duration, error) {
+	op := ops.Begin(opName)
 	defer op.End()
 	result, sleep, err := cf.doFetch(context.Background(), op)
 	return result, sleep, op.FailIf(err)

config/fetcher_test.go

@@ -21,7 +21,7 @@ func TestFetcher(t *testing.T) {
 	rt := &http.Transport{}
 	configFetcher := newHttpFetcher(newTestUserConfig(), rt, common.GlobalURL)
 
-	bytes, _, err := configFetcher.fetch()
+	bytes, _, err := configFetcher.fetch("testOpName")
 	assert.Nil(t, err)
 	assert.True(t, len(bytes) > 200)
 }

config/initializer.go

@@ -97,6 +97,7 @@ func InitWithURLs(
 		},
 		sticky: isSticky(flags),
 		rt:     rt,
+		opName: "fetch_global",
 	}
 
 	stopGlobal := pipeConfig(globalOptions)