store_test.go

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

     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  package load
    19  
    20  import (
    21  	"fmt"
    22  	"sort"
    23  	"sync"
    24  	"testing"
    25  
    26  	"github.com/google/go-cmp/cmp"
    27  	"github.com/google/go-cmp/cmp/cmpopts"
    28  )
    29  
    30  var (
    31  	dropCategories = []string{"drop_for_real", "drop_for_fun"}
    32  	localities     = []string{"locality-A", "locality-B"}
    33  	errTest        = fmt.Errorf("test error")
    34  )
    35  
    36  // rpcData wraps the rpc counts and load data to be pushed to the store.
    37  type rpcData struct {
    38  	start, success, failure int
    39  	serverData              map[string]float64 // Will be reported with successful RPCs.
    40  }
    41  
    42  // TestDrops spawns a bunch of goroutines which report drop data. After the
    43  // goroutines have exited, the test dumps the stats from the Store and makes
    44  // sure they are as expected.
    45  func TestDrops(t *testing.T) {
    46  	var (
    47  		drops = map[string]int{
    48  			dropCategories[0]: 30,
    49  			dropCategories[1]: 40,
    50  			"":                10,
    51  		}
    52  		wantStoreData = &Data{
    53  			TotalDrops: 80,
    54  			Drops: map[string]uint64{
    55  				dropCategories[0]: 30,
    56  				dropCategories[1]: 40,
    57  			},
    58  		}
    59  	)
    60  
    61  	ls := perClusterStore{}
    62  	var wg sync.WaitGroup
    63  	for category, count := range drops {
    64  		for i := 0; i < count; i++ {
    65  			wg.Add(1)
    66  			go func(c string) {
    67  				ls.CallDropped(c)
    68  				wg.Done()
    69  			}(category)
    70  		}
    71  	}
    72  	wg.Wait()
    73  
    74  	gotStoreData := ls.stats()
    75  	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval")); diff != "" {
    76  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
    77  	}
    78  }
    79  
    80  // TestLocalityStats spawns a bunch of goroutines which report rpc and load
    81  // data. After the goroutines have exited, the test dumps the stats from the
    82  // Store and makes sure they are as expected.
    83  func TestLocalityStats(t *testing.T) {
    84  	var (
    85  		localityData = map[string]rpcData{
    86  			localities[0]: {
    87  				start:      40,
    88  				success:    20,
    89  				failure:    10,
    90  				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
    91  			},
    92  			localities[1]: {
    93  				start:      80,
    94  				success:    40,
    95  				failure:    20,
    96  				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
    97  			},
    98  		}
    99  		wantStoreData = &Data{
   100  			LocalityStats: map[string]LocalityData{
   101  				localities[0]: {
   102  					RequestStats: RequestData{Succeeded: 20, Errored: 10, InProgress: 10},
   103  					LoadStats: map[string]ServerLoadData{
   104  						"net":  {Count: 20, Sum: 20},
   105  						"disk": {Count: 20, Sum: 40},
   106  						"cpu":  {Count: 20, Sum: 60},
   107  						"mem":  {Count: 20, Sum: 80},
   108  					},
   109  				},
   110  				localities[1]: {
   111  					RequestStats: RequestData{Succeeded: 40, Errored: 20, InProgress: 20},
   112  					LoadStats: map[string]ServerLoadData{
   113  						"net":  {Count: 40, Sum: 40},
   114  						"disk": {Count: 40, Sum: 80},
   115  						"cpu":  {Count: 40, Sum: 120},
   116  						"mem":  {Count: 40, Sum: 160},
   117  					},
   118  				},
   119  			},
   120  		}
   121  	)
   122  
   123  	ls := perClusterStore{}
   124  	var wg sync.WaitGroup
   125  	for locality, data := range localityData {
   126  		wg.Add(data.start)
   127  		for i := 0; i < data.start; i++ {
   128  			go func(l string) {
   129  				ls.CallStarted(l)
   130  				wg.Done()
   131  			}(locality)
   132  		}
   133  		// The calls to callStarted() need to happen before the other calls are
   134  		// made. Hence the wait here.
   135  		wg.Wait()
   136  
   137  		wg.Add(data.success)
   138  		for i := 0; i < data.success; i++ {
   139  			go func(l string, serverData map[string]float64) {
   140  				ls.CallFinished(l, nil)
   141  				for n, d := range serverData {
   142  					ls.CallServerLoad(l, n, d)
   143  				}
   144  				wg.Done()
   145  			}(locality, data.serverData)
   146  		}
   147  		wg.Add(data.failure)
   148  		for i := 0; i < data.failure; i++ {
   149  			go func(l string) {
   150  				ls.CallFinished(l, errTest)
   151  				wg.Done()
   152  			}(locality)
   153  		}
   154  		wg.Wait()
   155  	}
   156  
   157  	gotStoreData := ls.stats()
   158  	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval")); diff != "" {
   159  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   160  	}
   161  }
   162  
   163  func TestResetAfterStats(t *testing.T) {
   164  	// Push a bunch of drops, call stats and load stats, and leave inProgress to be non-zero.
   165  	// Dump the stats. Verify expected
   166  	// Push the same set of loads as before
   167  	// Now dump and verify the newly expected ones.
   168  	var (
   169  		drops = map[string]int{
   170  			dropCategories[0]: 30,
   171  			dropCategories[1]: 40,
   172  		}
   173  		localityData = map[string]rpcData{
   174  			localities[0]: {
   175  				start:      40,
   176  				success:    20,
   177  				failure:    10,
   178  				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
   179  			},
   180  			localities[1]: {
   181  				start:      80,
   182  				success:    40,
   183  				failure:    20,
   184  				serverData: map[string]float64{"net": 1, "disk": 2, "cpu": 3, "mem": 4},
   185  			},
   186  		}
   187  		wantStoreData = &Data{
   188  			TotalDrops: 70,
   189  			Drops: map[string]uint64{
   190  				dropCategories[0]: 30,
   191  				dropCategories[1]: 40,
   192  			},
   193  			LocalityStats: map[string]LocalityData{
   194  				localities[0]: {
   195  					RequestStats: RequestData{Succeeded: 20, Errored: 10, InProgress: 10},
   196  					LoadStats: map[string]ServerLoadData{
   197  						"net":  {Count: 20, Sum: 20},
   198  						"disk": {Count: 20, Sum: 40},
   199  						"cpu":  {Count: 20, Sum: 60},
   200  						"mem":  {Count: 20, Sum: 80},
   201  					},
   202  				},
   203  				localities[1]: {
   204  					RequestStats: RequestData{Succeeded: 40, Errored: 20, InProgress: 20},
   205  					LoadStats: map[string]ServerLoadData{
   206  						"net":  {Count: 40, Sum: 40},
   207  						"disk": {Count: 40, Sum: 80},
   208  						"cpu":  {Count: 40, Sum: 120},
   209  						"mem":  {Count: 40, Sum: 160},
   210  					},
   211  				},
   212  			},
   213  		}
   214  	)
   215  
   216  	reportLoad := func(ls *perClusterStore) {
   217  		for category, count := range drops {
   218  			for i := 0; i < count; i++ {
   219  				ls.CallDropped(category)
   220  			}
   221  		}
   222  		for locality, data := range localityData {
   223  			for i := 0; i < data.start; i++ {
   224  				ls.CallStarted(locality)
   225  			}
   226  			for i := 0; i < data.success; i++ {
   227  				ls.CallFinished(locality, nil)
   228  				for n, d := range data.serverData {
   229  					ls.CallServerLoad(locality, n, d)
   230  				}
   231  			}
   232  			for i := 0; i < data.failure; i++ {
   233  				ls.CallFinished(locality, errTest)
   234  			}
   235  		}
   236  	}
   237  
   238  	ls := perClusterStore{}
   239  	reportLoad(&ls)
   240  	gotStoreData := ls.stats()
   241  	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval")); diff != "" {
   242  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   243  	}
   244  
   245  	// The above call to stats() should have reset all load reports except the
   246  	// inProgress rpc count. We are now going to push the same load data into
   247  	// the store. So, we should expect to see twice the count for inProgress.
   248  	for _, l := range localities {
   249  		ls := wantStoreData.LocalityStats[l]
   250  		ls.RequestStats.InProgress *= 2
   251  		wantStoreData.LocalityStats[l] = ls
   252  	}
   253  	reportLoad(&ls)
   254  	gotStoreData = ls.stats()
   255  	if diff := cmp.Diff(wantStoreData, gotStoreData, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval")); diff != "" {
   256  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   257  	}
   258  }
   259  
   260  var sortDataSlice = cmp.Transformer("SortDataSlice", func(in []*Data) []*Data {
   261  	out := append([]*Data(nil), in...) // Copy input to avoid mutating it
   262  	sort.Slice(out,
   263  		func(i, j int) bool {
   264  			if out[i].Cluster < out[j].Cluster {
   265  				return true
   266  			}
   267  			if out[i].Cluster == out[j].Cluster {
   268  				return out[i].Service < out[j].Service
   269  			}
   270  			return false
   271  		},
   272  	)
   273  	return out
   274  })
   275  
   276  // Test all load are returned for the given clusters, and all clusters are
   277  // reported if no cluster is specified.
   278  func TestStoreStats(t *testing.T) {
   279  	var (
   280  		testClusters = []string{"c0", "c1", "c2"}
   281  		testServices = []string{"s0", "s1"}
   282  		testLocality = "test-locality"
   283  	)
   284  
   285  	store := NewStore()
   286  	for _, c := range testClusters {
   287  		for _, s := range testServices {
   288  			store.PerCluster(c, s).CallStarted(testLocality)
   289  			store.PerCluster(c, s).CallServerLoad(testLocality, "abc", 123)
   290  			store.PerCluster(c, s).CallDropped("dropped")
   291  			store.PerCluster(c, s).CallFinished(testLocality, nil)
   292  		}
   293  	}
   294  
   295  	wantC0 := []*Data{
   296  		{
   297  			Cluster: "c0", Service: "s0",
   298  			TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
   299  			LocalityStats: map[string]LocalityData{
   300  				"test-locality": {
   301  					RequestStats: RequestData{Succeeded: 1},
   302  					LoadStats:    map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
   303  				},
   304  			},
   305  		},
   306  		{
   307  			Cluster: "c0", Service: "s1",
   308  			TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
   309  			LocalityStats: map[string]LocalityData{
   310  				"test-locality": {
   311  					RequestStats: RequestData{Succeeded: 1},
   312  					LoadStats:    map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
   313  				},
   314  			},
   315  		},
   316  	}
   317  	// Call Stats with just "c0", this should return data for "c0", and not
   318  	// touch data for other clusters.
   319  	gotC0 := store.Stats([]string{"c0"})
   320  	if diff := cmp.Diff(wantC0, gotC0, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval"), sortDataSlice); diff != "" {
   321  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   322  	}
   323  
   324  	wantOther := []*Data{
   325  		{
   326  			Cluster: "c1", Service: "s0",
   327  			TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
   328  			LocalityStats: map[string]LocalityData{
   329  				"test-locality": {
   330  					RequestStats: RequestData{Succeeded: 1},
   331  					LoadStats:    map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
   332  				},
   333  			},
   334  		},
   335  		{
   336  			Cluster: "c1", Service: "s1",
   337  			TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
   338  			LocalityStats: map[string]LocalityData{
   339  				"test-locality": {
   340  					RequestStats: RequestData{Succeeded: 1},
   341  					LoadStats:    map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
   342  				},
   343  			},
   344  		},
   345  		{
   346  			Cluster: "c2", Service: "s0",
   347  			TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
   348  			LocalityStats: map[string]LocalityData{
   349  				"test-locality": {
   350  					RequestStats: RequestData{Succeeded: 1},
   351  					LoadStats:    map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
   352  				},
   353  			},
   354  		},
   355  		{
   356  			Cluster: "c2", Service: "s1",
   357  			TotalDrops: 1, Drops: map[string]uint64{"dropped": 1},
   358  			LocalityStats: map[string]LocalityData{
   359  				"test-locality": {
   360  					RequestStats: RequestData{Succeeded: 1},
   361  					LoadStats:    map[string]ServerLoadData{"abc": {Count: 1, Sum: 123}},
   362  				},
   363  			},
   364  		},
   365  	}
   366  	// Call Stats with empty slice, this should return data for all the
   367  	// remaining clusters, and not include c0 (because c0 data was cleared).
   368  	gotOther := store.Stats(nil)
   369  	if diff := cmp.Diff(wantOther, gotOther, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval"), sortDataSlice); diff != "" {
   370  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   371  	}
   372  }
   373  
   374  // Test the cases that if a cluster doesn't have load to report, its data is not
   375  // appended to the slice returned by Stats().
   376  func TestStoreStatsEmptyDataNotReported(t *testing.T) {
   377  	var (
   378  		testServices = []string{"s0", "s1"}
   379  		testLocality = "test-locality"
   380  	)
   381  
   382  	store := NewStore()
   383  	// "c0"'s RPCs all finish with success.
   384  	for _, s := range testServices {
   385  		store.PerCluster("c0", s).CallStarted(testLocality)
   386  		store.PerCluster("c0", s).CallFinished(testLocality, nil)
   387  	}
   388  	// "c1"'s RPCs never finish (always inprocess).
   389  	for _, s := range testServices {
   390  		store.PerCluster("c1", s).CallStarted(testLocality)
   391  	}
   392  
   393  	want0 := []*Data{
   394  		{
   395  			Cluster: "c0", Service: "s0",
   396  			LocalityStats: map[string]LocalityData{
   397  				"test-locality": {RequestStats: RequestData{Succeeded: 1}},
   398  			},
   399  		},
   400  		{
   401  			Cluster: "c0", Service: "s1",
   402  			LocalityStats: map[string]LocalityData{
   403  				"test-locality": {RequestStats: RequestData{Succeeded: 1}},
   404  			},
   405  		},
   406  		{
   407  			Cluster: "c1", Service: "s0",
   408  			LocalityStats: map[string]LocalityData{
   409  				"test-locality": {RequestStats: RequestData{InProgress: 1}},
   410  			},
   411  		},
   412  		{
   413  			Cluster: "c1", Service: "s1",
   414  			LocalityStats: map[string]LocalityData{
   415  				"test-locality": {RequestStats: RequestData{InProgress: 1}},
   416  			},
   417  		},
   418  	}
   419  	// Call Stats with empty slice, this should return data for all the
   420  	// clusters.
   421  	got0 := store.Stats(nil)
   422  	if diff := cmp.Diff(want0, got0, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval"), sortDataSlice); diff != "" {
   423  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   424  	}
   425  
   426  	want1 := []*Data{
   427  		{
   428  			Cluster: "c1", Service: "s0",
   429  			LocalityStats: map[string]LocalityData{
   430  				"test-locality": {RequestStats: RequestData{InProgress: 1}},
   431  			},
   432  		},
   433  		{
   434  			Cluster: "c1", Service: "s1",
   435  			LocalityStats: map[string]LocalityData{
   436  				"test-locality": {RequestStats: RequestData{InProgress: 1}},
   437  			},
   438  		},
   439  	}
   440  	// Call Stats with empty slice again, this should return data only for "c1",
   441  	// because "c0" data was cleared, but "c1" has in-progress RPCs.
   442  	got1 := store.Stats(nil)
   443  	if diff := cmp.Diff(want1, got1, cmpopts.EquateEmpty(), cmpopts.IgnoreFields(Data{}, "ReportInterval"), sortDataSlice); diff != "" {
   444  		t.Errorf("store.stats() returned unexpected diff (-want +got):\n%s", diff)
   445  	}
   446  }
   447
View as plain text