feat:add cache about networktopology (#2924)

Signed-off-by: huangmin <2107139596@qq.com>

pkg/cache/cache.go

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+//go:generate mockgen -destination cache_mock.go -source cache.go -package cache
 package cache
 
 import (

pkg/slices/slices.go

@@ -67,3 +67,21 @@ func Reverse[S ~[]T, T any](s S) {
 		s[i], s[j] = s[j], s[i]
 	}
 }
+
+// Complement removes duplicate elements of collections in first collection.
+func Complement[T comparable](a, b []T) []T {
+	var result []T
+
+	visited := make(map[T]struct{})
+	for _, v := range b {
+		visited[v] = struct{}{}
+	}
+
+	for _, v := range a {
+		if _, exists := visited[v]; !exists {
+			result = append(result, v)
+		}
+	}
+
+	return result
+}

pkg/slices/slices_test.go

@@ -178,3 +178,34 @@ func TestReverse(t *testing.T) {
 		})
 	}
 }
+
+func TestComplement(t *testing.T) {
+	tests := []struct {
+		name     string
+		source   []int
+		exclude  []int
+		expected []int
+	}{
+		{
+			name:     "slices with duplicates",
+			source:   []int{1, 2, 3},
+			exclude:  []int{1, 2, 4},
+			expected: []int{3},
+		},
+		{
+			name:     "slices with no duplicates",
+			source:   []int{1, 2, 3},
+			exclude:  []int{4, 5, 6},
+			expected: []int{1, 2, 3},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := Complement(tt.source, tt.exclude)
+			if !reflect.DeepEqual(result, tt.expected) {
+				t.Errorf("expected %v, but got %v", tt.expected, result)
+			}
+		})
+	}
+}

scheduler/config/config.go

@@ -342,6 +342,9 @@ type NetworkTopologyConfig struct {
 
 	// Probe is the configuration of probe.
 	Probe ProbeConfig `yaml:"probe" mapstructure:"probe"`
+
+	// Cache is the configuration of cache.
+	Cache CacheConfig `yaml:"cache" mapstructure:"cache"`
 }
 
 type ProbeConfig struct {
@@ -352,6 +355,14 @@ type ProbeConfig struct {
 	Count int `mapstructure:"count" yaml:"count"`
 }
 
+type CacheConfig struct {
+	// Interval is cache cleanup interval.
+	Interval time.Duration `yaml:"interval" mapstructure:"interval"`
+
+	// TTL is networkTopology cache items TLL.
+	TTL time.Duration `yaml:"ttl" mapstructure:"tll"`
+}
+
 type TrainerConfig struct {
 	// Enable trainer service.
 	Enable bool `yaml:"enable" mapstructure:"enable"`
@@ -458,6 +469,10 @@ func New() *Config {
 				QueueLength: DefaultProbeQueueLength,
 				Count:       DefaultProbeCount,
 			},
+			Cache: CacheConfig{
+				Interval: DefaultNetworkTopologyCacheInterval,
+				TTL:      DefaultNetworkTopologyCacheTLL,
+			},
 		},
 		Trainer: TrainerConfig{
 			Enable:        false,
@@ -644,6 +659,14 @@ func (cfg *Config) Validate() error {
 		return errors.New("probe requires parameter count")
 	}
 
+	if cfg.NetworkTopology.Cache.Interval <= 0 {
+		return errors.New("networkTopology requires parameter interval")
+	}
+
+	if cfg.NetworkTopology.Cache.TTL <= 0 {
+		return errors.New("networkTopology requires parameter ttl")
+	}
+
 	if cfg.Trainer.Enable {
 		if cfg.Trainer.Addr == "" {
 			return errors.New("trainer requires parameter addr")