feat: add E2E tests for cases that peers going offline

Author: BruceAko

deploy/docker-compose/template/scheduler.template.yaml

@@ -61,7 +61,7 @@ scheduler:
     # then the task will also be reclaimed.
     taskGCInterval: 30m
     # hostGCInterval is the interval of host gc.
-    hostGCInterval: 6h
+    hostGCInterval: 5m
     # hostTTL is time to live of host. If host announces message to scheduler,
     # then HostTTl will be reset.
     hostTTL: 1h

pkg/rpc/scheduler/client/client_v1.go

@@ -90,7 +90,7 @@ func GetV1(ctx context.Context, dynconfig config.Dynconfig, opts ...grpc.DialOpt
 	}, nil
 }
 
-// GetV1ByAddr returns v2 version of the scheduler client by address.
+// GetV1ByAddr returns v1 version of the scheduler client by address.
 func GetV1ByAddr(ctx context.Context, target string, opts ...grpc.DialOption) (V1, error) {
 	conn, err := grpc.DialContext(
 		ctx,

pkg/rpc/scheduler/client/client_v2.go

@@ -21,6 +21,7 @@ package client
 import (
 	"context"
 	"math"
+	"sync"
 
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
@@ -30,6 +31,7 @@ import (
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/balancer"
+	"google.golang.org/protobuf/types/known/emptypb"
 
 	commonv2 "d7y.io/api/v2/pkg/apis/common/v2"
 	schedulerv2 "d7y.io/api/v2/pkg/apis/scheduler/v2"
@@ -145,6 +147,9 @@ type V2 interface {
 	// AnnounceHost announces host to scheduler.
 	AnnounceHost(context.Context, *schedulerv2.AnnounceHostRequest, ...grpc.CallOption) error
 
+	// ListHosts lists hosts in scheduler.
+	ListHosts(ctx context.Context, opts ...grpc.CallOption) (*schedulerv2.ListHostsResponse, error)
+
 	// DeleteHost releases host in scheduler.
 	DeleteHost(context.Context, *schedulerv2.DeleteHostRequest, ...grpc.CallOption) error
 
@@ -253,6 +258,56 @@ func (v *v2) AnnounceHost(ctx context.Context, req *schedulerv2.AnnounceHostRequ
 	return eg.Wait()
 }
 
+// ListHosts lists host in all schedulers.
+func (v *v2) ListHosts(ctx context.Context, opts ...grpc.CallOption) (*schedulerv2.ListHostsResponse, error) {
+	ctx, cancel := context.WithTimeout(ctx, contextTimeout)
+	defer cancel()
+
+	circle, err := v.GetCircle()
+	if err != nil {
+		return nil, err
+	}
+
+	responses := make([]*schedulerv2.ListHostsResponse, len(circle))
+	mu := sync.Mutex{}
+
+	eg, _ := errgroup.WithContext(ctx)
+	for _, virtualTaskID := range circle {
+		virtualTaskID := virtualTaskID
+		eg.Go(func() error {
+			response, err := v.SchedulerClient.ListHosts(
+				context.WithValue(ctx, pkgbalancer.ContextKey, virtualTaskID),
+				new(emptypb.Empty),
+				opts...,
+			)
+			if err != nil {
+				return err
+			}
+
+			mu.Lock()
+			responses = append(responses, response)
+			mu.Unlock()
+
+			return nil
+		})
+	}
+
+	idSet := map[string]struct{}{}
+	var hosts []*commonv2.Host
+	for _, response := range responses {
+		for _, host := range response.Hosts {
+			if _, exist := idSet[host.Id]; !exist {
+				idSet[host.Id] = struct{}{}
+				hosts = append(hosts, host)
+			}
+		}
+	}
+
+	return &schedulerv2.ListHostsResponse{
+		Hosts: hosts,
+	}, eg.Wait()
+}
+
 // DeleteHost releases host in all schedulers.
 func (v *v2) DeleteHost(ctx context.Context, req *schedulerv2.DeleteHostRequest, opts ...grpc.CallOption) error {
 	ctx, cancel := context.WithTimeout(ctx, contextTimeout)

pkg/rpc/scheduler/client/mocks/client_v2_mock.go

@@ -224,6 +224,20 @@ var (
 	}, []string{"os", "platform", "platform_family", "platform_version",
 		"kernel_version", "git_version", "git_commit", "go_version", "build_platform"})
 
+	ListHostsCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "list_hosts_total",
+		Help:      "Counter of the number of the list hosts.",
+	})
+
+	ListHostsCountFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "list_hosts_failure_total",
+		Help:      "Counter of the number of failed of the list hosts.",
+	})
+
 	LeaveHostCount = promauto.NewCounter(prometheus.CounterOpts{
 		Namespace: types.MetricsNamespace,
 		Subsystem: types.SchedulerMetricsName,

scheduler/metrics/metrics.go

@@ -138,10 +138,18 @@ func (s *schedulerServerV2) AnnounceHost(ctx context.Context, req *schedulerv2.A
 	return new(emptypb.Empty), nil
 }
 
-// TODO Implement the following methods.
 // ListHosts lists hosts in scheduler.
 func (s *schedulerServerV2) ListHosts(ctx context.Context, _ *emptypb.Empty) (*schedulerv2.ListHostsResponse, error) {
-	return nil, nil
+	// Collect ListHostsCount metrics.
+	metrics.ListHostsCount.Inc()
+	resp, err := s.service.ListHosts(ctx)
+	if err != nil {
+		// Collect ListHostsFailureCount metrics.
+		metrics.ListHostsCountFailureCount.Inc()
+		return nil, err
+	}
+
+	return resp, nil
 }
 
 // DeleteHost releases host in scheduler.

scheduler/rpcserver/scheduler_server_v2.go

@@ -693,6 +693,84 @@ func (v *V2) AnnounceHost(ctx context.Context, req *schedulerv2.AnnounceHostRequ
 	return nil
 }
 
+// ListHosts lists hosts in scheduler.
+func (v *V2) ListHosts(ctx context.Context) (*schedulerv2.ListHostsResponse, error) {
+	hosts := v.resource.HostManager().LoadAll()
+
+	resHosts := make([]*commonv2.Host, len(hosts))
+	for i, host := range hosts {
+		resHosts[i] = &commonv2.Host{
+			Id:              host.ID,
+			Type:            uint32(host.Type),
+			Hostname:        host.Hostname,
+			Ip:              host.IP,
+			Port:            host.Port,
+			DownloadPort:    host.DownloadPort,
+			Os:              host.OS,
+			Platform:        host.Platform,
+			PlatformFamily:  host.PlatformFamily,
+			PlatformVersion: host.PlatformVersion,
+			KernelVersion:   host.KernelVersion,
+			Cpu: &commonv2.CPU{
+				LogicalCount:   host.CPU.LogicalCount,
+				PhysicalCount:  host.CPU.PhysicalCount,
+				Percent:        host.CPU.Percent,
+				ProcessPercent: host.CPU.ProcessPercent,
+				Times: &commonv2.CPUTimes{
+					User:      host.CPU.Times.User,
+					System:    host.CPU.Times.System,
+					Idle:      host.CPU.Times.Idle,
+					Nice:      host.CPU.Times.Nice,
+					Iowait:    host.CPU.Times.Iowait,
+					Irq:       host.CPU.Times.Irq,
+					Softirq:   host.CPU.Times.Softirq,
+					Steal:     host.CPU.Times.Steal,
+					Guest:     host.CPU.Times.Guest,
+					GuestNice: host.CPU.Times.GuestNice,
+				},
+			},
+			Memory: &commonv2.Memory{
+				Total:              host.Memory.Total,
+				Available:          host.Memory.Available,
+				Used:               host.Memory.Used,
+				UsedPercent:        host.Memory.UsedPercent,
+				ProcessUsedPercent: host.Memory.ProcessUsedPercent,
+				Free:               host.Memory.Free,
+			},
+			Network: &commonv2.Network{
+				TcpConnectionCount:       host.Network.TCPConnectionCount,
+				UploadTcpConnectionCount: host.Network.UploadTCPConnectionCount,
+				Location:                 &host.Network.Location,
+				Idc:                      &host.Network.IDC,
+			},
+			Disk: &commonv2.Disk{
+				Total:             host.Disk.Total,
+				Free:              host.Disk.Free,
+				Used:              host.Disk.Used,
+				UsedPercent:       host.Disk.UsedPercent,
+				InodesTotal:       host.Disk.InodesTotal,
+				InodesUsed:        host.Disk.InodesUsed,
+				InodesFree:        host.Disk.InodesFree,
+				InodesUsedPercent: host.Disk.InodesUsedPercent,
+			},
+			Build: &commonv2.Build{
+				GitVersion: host.Build.GitVersion,
+				GitCommit:  &host.Build.GitCommit,
+				GoVersion:  &host.Build.GoVersion,
+				Platform:   &host.Build.Platform,
+			},
+			SchedulerClusterId: host.SchedulerClusterID,
+			DisableShared:      host.DisableShared,
+		}
+	}
+
+	resp := &schedulerv2.ListHostsResponse{
+		Hosts: resHosts,
+	}
+
+	return resp, nil
+}
+
 // DeleteHost releases host in scheduler.
 func (v *V2) DeleteHost(ctx context.Context, req *schedulerv2.DeleteHostRequest) error {
 	log := logger.WithHostID(req.GetHostId())

scheduler/service/service_v2.go

@@ -0,0 +1,48 @@
+/*
+ *     Copyright 2024 The Dragonfly Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package e2e
+
+import (
+	"context"
+	"fmt"
+
+	. "github.com/onsi/ginkgo/v2" //nolint
+	. "github.com/onsi/gomega"    //nolint
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	schedulerclient "d7y.io/dragonfly/v2/pkg/rpc/scheduler/client"
+	"d7y.io/dragonfly/v2/test/e2e/v2/util"
+)
+
+var _ = Describe("Clients go offline normally and abnormally", func() {
+	Context("scheduler clears peer metadata", func() {
+		It("number of hosts should be ok", Label("host", "leave"), func() {
+			hostCount := util.Servers[util.SeedClientServerName].Replicas + util.Servers[util.ClientServerName].Replicas + 2
+
+			grpcCredentials := insecure.NewCredentials()
+			schedulerClient, err := schedulerclient.GetV2ByAddr(context.Background(), ":8002", grpc.WithTransportCredentials(grpcCredentials))
+			Expect(err).NotTo(HaveOccurred())
+
+			response, err := schedulerClient.ListHosts(context.Background())
+			Expect(err).NotTo(HaveOccurred())
+			fmt.Println(response)
+
+			Expect(response.Hosts).To(Equal(hostCount))
+		})
+	})
+})

test/e2e/v2/leave_host_test.go

@@ -96,6 +96,7 @@ seedClient:
 
 client:
   enable: true
+  replicas: 3
   image:
     repository: dragonflyoss/client
     tag: latest

test/testdata/charts/config-v2.yaml

@@ -12,6 +12,9 @@ nodes:
       - containerPort: 4003
         hostPort: 4003
         protocol: TCP
+      - containerPort: 8002
+        hostPort: 8002
+        protocol: TCP
     extraMounts:
       - hostPath: ./test/testdata/containerd/config-v2.toml
         containerPath: /etc/containerd/config.toml