feat: resource add rdb for cache task (#3467)

Signed-off-by: Gaius <[email protected]>

scheduler/resource/resource.go

@@ -19,6 +19,7 @@
 package resource
 
 import (
+	"github.com/redis/go-redis/v9"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -63,6 +64,9 @@ type resource struct {
 	// Scheduler config.
 	config *config.Config
 
+	// Redis universal client interface.
+	rdb redis.UniversalClient
+
 	// TransportCredentials stores the Authenticator required to setup a client connection.
 	transportCredentials credentials.TransportCredentials
 }
@@ -78,6 +82,13 @@ func WithTransportCredentials(creds credentials.TransportCredentials) Option {
 	}
 }
 
+// WithRedisClient returns a Option which configures the redis client.
+func WithRedisClient(rdb redis.UniversalClient) Option {
+	return func(r *resource) {
+		r.rdb = rdb
+	}
+}
+
 // New returns Resource interface.
 func New(cfg *config.Config, gc gc.GC, dynconfig config.DynconfigInterface, options ...Option) (Resource, error) {
 	resource := &resource{config: cfg}

scheduler/scheduler.go

@@ -146,6 +146,9 @@ func New(ctx context.Context, cfg *config.Config, d dfpath.Dfpath) (*Server, err
 	}
 	s.announcer = announcer
 
+	// Initialize GC.
+	s.gc = gc.New(gc.WithLogger(logger.GCLogger))
+
 	// Initialize certify client.
 	var (
 		certifyClient              *certify.Certify
@@ -188,21 +191,16 @@ func New(ctx context.Context, cfg *config.Config, d dfpath.Dfpath) (*Server, err
 	}
 
 	// Initialize dynconfig client.
-	dynconfig, err := config.NewDynconfig(s.managerClient, filepath.Join(d.CacheDir(), dynconfig.CacheDirName), cfg, config.WithTransportCredentials(clientTransportCredentials))
-	if err != nil {
-		return nil, err
+	dynconfigOptions := []config.DynconfigOption{}
+	if clientTransportCredentials != nil {
+		dynconfigOptions = append(dynconfigOptions, config.WithTransportCredentials(clientTransportCredentials))
 	}
-	s.dynconfig = dynconfig
 
-	// Initialize GC.
-	s.gc = gc.New(gc.WithLogger(logger.GCLogger))
-
-	// Initialize resource.
-	resource, err := resource.New(cfg, s.gc, dynconfig, resource.WithTransportCredentials(clientTransportCredentials))
+	dynconfig, err := config.NewDynconfig(s.managerClient, filepath.Join(d.CacheDir(), dynconfig.CacheDirName), cfg, dynconfigOptions...)
 	if err != nil {
 		return nil, err
 	}
-	s.resource = resource
+	s.dynconfig = dynconfig
 
 	// Initialize redis client.
 	var rdb redis.UniversalClient
@@ -219,18 +217,33 @@ func New(ctx context.Context, cfg *config.Config, d dfpath.Dfpath) (*Server, err
 		}
 	}
 
+	// Initialize resource.
+	resourceOptions := []resource.Option{}
+	if clientTransportCredentials != nil {
+		resourceOptions = append(resourceOptions, resource.WithTransportCredentials(clientTransportCredentials))
+	}
+
+	if rdb != nil {
+		resourceOptions = append(resourceOptions, resource.WithRedisClient(rdb))
+	}
+
+	resource, err := resource.New(cfg, s.gc, dynconfig, resourceOptions...)
+	if err != nil {
+		return nil, err
+	}
+	s.resource = resource
+
 	// Initialize job service.
-	if cfg.Job.Enable && pkgredis.IsEnabled(cfg.Database.Redis.Addrs) {
+	if cfg.Job.Enable && rdb != nil {
 		s.job, err = job.New(cfg, resource)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	// Initialize options of evaluator.
+	// Initialize options of network topology options.
 	evaluatorNetworkTopologyOptions := []evaluator.NetworkTopologyOption{}
-	// Initialize network topology service.
-	if cfg.Scheduler.Algorithm == evaluator.NetworkTopologyAlgorithm {
+	if cfg.Scheduler.Algorithm == evaluator.NetworkTopologyAlgorithm && rdb != nil {
 		cache := cache.New(cfg.Scheduler.NetworkTopology.Cache.TTL, cfg.Scheduler.NetworkTopology.Cache.Interval)
 		s.networkTopology, err = networktopology.NewNetworkTopology(cfg.Scheduler.NetworkTopology, rdb, cache, resource, s.storage)
 		if err != nil {