Repository: dragonflyoss/Dragonfly2
Pull request #1620 (2022-08-31 13:42:21)

feat: check whether scheduler is in the same cluster
Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- Check whether scheduler is in the same cluster.
<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.



Patch:
diff --git a/client/config/dynconfig_local.go b/client/config/dynconfig_local.go
index 70769453..0e6ff80f 100644
--- a/client/config/dynconfig_local.go
+++ b/client/config/dynconfig_local.go
@@ -59,22 +59,23 @@ func (d *dynconfigLocal) GetResolveSchedulerAddrs() ([]resolver.Address, error)
 		r := reachable.New(&reachable.Config{Address: addr})
 		if err := r.Check(); err != nil {
 			logger.Warnf("scheduler address %s is unreachable", addr)
-		} else {
-			if addrs[addr] {
-				continue
-			}
+			continue
+		}
 
-			host, _, err := net.SplitHostPort(addr)
-			if err != nil {
-				continue
-			}
+		if addrs[addr] {
+			continue
+		}
 
-			resolveAddrs = append(resolveAddrs, resolver.Address{
-				ServerName: host,
-				Addr:       addr,
-			})
-			addrs[addr] = true
+		host, _, err := net.SplitHostPort(addr)
+		if err != nil {
+			continue
 		}
+
+		resolveAddrs = append(resolveAddrs, resolver.Address{
+			ServerName: host,
+			Addr:       addr,
+		})
+		addrs[addr] = true
 	}
 
 	return resolveAddrs, nil
diff --git a/client/config/dynconfig_manager.go b/client/config/dynconfig_manager.go
index c97a8ec4..85b08942 100644
--- a/client/config/dynconfig_manager.go
+++ b/client/config/dynconfig_manager.go
@@ -74,25 +74,33 @@ func (d *dynconfigManager) GetResolveSchedulerAddrs() ([]resolver.Address, error
 	}
 
 	var (
-		addrs        = map[string]bool{}
-		resolveAddrs = []resolver.Address{}
+		addrs              = map[string]bool{}
+		resolveAddrs       = []resolver.Address{}
+		schedulerClusterID uint64
 	)
 	for _, scheduler := range schedulers {
+		// Check whether scheduler is in the same cluster.
+		if schedulerClusterID != 0 && schedulerClusterID != scheduler.SchedulerClusterId {
+			continue
+		}
+
 		addr := fmt.Sprintf("%s:%d", scheduler.GetIp(), scheduler.GetPort())
 		r := reachable.New(&reachable.Config{Address: addr})
 		if err := r.Check(); err != nil {
 			logger.Warnf("scheduler address %s is unreachable", addr)
-		} else {
-			if addrs[addr] {
-				continue
-			}
+			continue
+		}
 
-			resolveAddrs = append(resolveAddrs, resolver.Address{
-				ServerName: scheduler.GetIp(),
-				Addr:       addr,
-			})
-			addrs[addr] = true
+		if addrs[addr] {
+			continue
 		}
+
+		schedulerClusterID = scheduler.SchedulerClusterId
+		resolveAddrs = append(resolveAddrs, resolver.Address{
+			ServerName: scheduler.GetIp(),
+			Addr:       addr,
+		})
+		addrs[addr] = true
 	}
 
 	return resolveAddrs, nil
diff --git a/manager/config/config.go b/manager/config/config.go
index 4977fd75..d40257ab 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -266,6 +266,20 @@ type SecurityConfig struct {
 	// prefer: ServerHandshake supports tls and insecure (non-tls), ClientHandshake will only support tls
 	// default or empty: ServerHandshake supports tls and insecure (non-tls), ClientHandshake will only support insecure (non-tls)
 	TLSPolicy string `mapstructure:"tlsPolicy" yaml:"tlsPolicy"`
+
+	// CertSpec is the desired state of manager certificate.
+	CertSpec *CertSpec `mapstructure:"certSpec" yaml:"certSpec"`
+}
+
+type CertSpec struct {
+	// DNSNames is a list of dns names be set on the certificate.
+	DNSNames []string `mapstructure:"dnsNames" yaml:"dnsNames"`
+
+	// IPAddresses is a list of ip addresses be set on the certificate.
+	IPAddresses []string `mapstructure:"ipAddresses" yaml:"ipAddresses"`
+
+	// ValidityDuration is the validity duration  of certificate.
+	ValidityDuration time.Duration `mapstructure:"validityDuration" yaml:"validityDuration"`
 }
 
 // New config instance.
@@ -318,6 +332,11 @@ func New() *Config {
 		Security: &SecurityConfig{
 			AutoIssueCert: false,
 			TLSPolicy:     rpc.DefaultTLSPolicy,
+			CertSpec: &CertSpec{
+				IPAddresses:      DefaultCertIPAddresses,
+				DNSNames:         DefaultCertDNSNames,
+				ValidityDuration: DefaultCertValidityDuration,
+			},
 		},
 		Metrics: &MetricsConfig{
 			Enable:          false,
@@ -505,6 +524,22 @@ func (cfg *Config) Validate() error {
 		if !slices.Contains([]string{rpc.DefaultTLSPolicy, rpc.ForceTLSPolicy, rpc.PreferTLSPolicy}, cfg.Security.TLSPolicy) {
 			return errors.New("security requires parameter tlsPolicy")
 		}
+
+		if cfg.Security.CertSpec == nil {
+			return errors.New("security requires parameter certSpec")
+		}
+
+		if len(cfg.Security.CertSpec.IPAddresses) == 0 {
+			return errors.New("certSpec requires parameter ipAddresses")
+		}
+
+		if len(cfg.Security.CertSpec.DNSNames) == 0 {
+			return errors.New("certSpec requires parameter dnsNames")
+		}
+
+		if cfg.Security.CertSpec.ValidityDuration <= 0 {
+			return errors.New("certSpec requires parameter validityDuration")
+		}
 	}
 
 	if cfg.Metrics == nil {
diff --git a/manager/config/constants.go b/manager/config/constants.go
index 3b6ea169..b746bb9d 100644
--- a/manager/config/constants.go
+++ b/manager/config/constants.go
@@ -16,7 +16,11 @@
 
 package config
 
-import "time"
+import (
+	"time"
+
+	"d7y.io/dragonfly/v2/pkg/net/ip"
+)
 
 const (
 	// DatabaseTypeMysql is database type of mysql.
@@ -88,3 +92,14 @@ const (
 	// DefaultMetricsAddr is default address for metrics server.
 	DefaultMetricsAddr = ":8000"
 )
+
+var (
+	// DefaultCertIPAddresses is default manager ip addresses of certificate.
+	DefaultCertIPAddresses = []string{ip.IPv4}
+
+	// DefaultCertDNSNames is default manager dns names of certificate.
+	DefaultCertDNSNames = []string{"dragonfly-manager", "dragonfly-manager.dragonfly-system.svc"}
+
+	// DefaultCertValidityDuration is default manager validity duration of certificate.
+	DefaultCertValidityDuration = 10 * 365 * 24 * time.Hour
+)
diff --git a/manager/manager.go b/manager/manager.go
index f2b3c4f1..afbf2633 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -172,8 +172,13 @@ func New(cfg *config.Config, d dfpath.Dfpath) (*Server, error) {
 		// Manager GRPC server's tls varify must be false. If ClientCAs are required for client verification,
 		// the client cannot call the IssueCertificate api.
 		transportCredentials, err := rpc.NewServerCredentialsByCertify(cfg.Security.TLSPolicy, false, cert.Certificate, &certify.Certify{
-			CommonName:   ip.IPv4,
-			Issuer:       issuer.NewDragonflyManagerIssuer(&cert),
+			CommonName: ip.IPv4,
+			Issuer: issuer.NewDragonflyManagerIssuer(
+				&cert,
+				issuer.WithIPAddresses(cfg.Security.CertSpec.IPAddresses),
+				issuer.WithDNSNames(cfg.Security.CertSpec.DNSNames),
+				issuer.WithValidityDuration(cfg.Security.CertSpec.ValidityDuration),
+			),
 			RenewBefore:  time.Hour,
 			CertConfig:   nil,
 			IssueTimeout: 0,
diff --git a/pkg/issuer/dragonfly.go b/pkg/issuer/dragonfly.go
index c41b363e..63a852e4 100644
--- a/pkg/issuer/dragonfly.go
+++ b/pkg/issuer/dragonfly.go
@@ -38,6 +38,14 @@ import (
 	managerclient "d7y.io/dragonfly/v2/pkg/rpc/manager/client"
 )
 
+var (
+	// defaultSubjectOrganization is default organization of subject.
+	defaultSubjectOrganization = []string{"Dragonfly"}
+
+	// defaultValidityDuration is default validity duration of certificate.
+	defaultValidityDuration = 180 * 24 * int64(time.Hour.Seconds())
+)
+
 type dragonflyIssuer struct {
 	client managerclient.Client
 }
@@ -55,7 +63,7 @@ func (d *dragonflyIssuer) Issue(ctx context.Context, commonName string, certConf
 
 	request := &securityv1.CertificateRequest{
 		Csr:              string(csrPEM),
-		ValidityDuration: 180 * 24 * int64(time.Hour.Seconds()),
+		ValidityDuration: defaultValidityDuration,
 	}
 
 	response, err := d.client.IssueCertificate(ctx, request)
diff --git a/pkg/issuer/dragonfly_manager.go b/pkg/issuer/dragonfly_manager.go
index 5b652875..b0717817 100644
--- a/pkg/issuer/dragonfly_manager.go
+++ b/pkg/issuer/dragonfly_manager.go
@@ -25,35 +25,16 @@ import (
 	"crypto/x509/pkix"
 	"math/big"
 	"net"
-	"net/url"
 	"time"
 
 	"github.com/johanbrandhorst/certify"
-
-	"d7y.io/dragonfly/v2/pkg/net/ip"
-)
-
-var (
-	// defaultSubjectOrganization is default organization of subject.
-	defaultSubjectOrganization = []string{"Dragonfly"}
-
-	// defaultIPAddresses is default ip addresses of certificate.
-	defaultIPAddresses = []net.IP{net.ParseIP(ip.IPv4)}
-
-	// defaultDNSNames is default dns names of certificate.
-	defaultDNSNames = []string{"dragonfly-manager", "dragonfly-manager.dragonfly-system.svc"}
-
-	// defaultValidityDuration is default validity duration of certificate.
-	defaultValidityDuration = 10 * 365 * 24 * time.Hour
 )
 
 // GC provides issuer function.
 type dragonflyManagerIssuer struct {
 	tlsCACert        *tls.Certificate
 	dnsNames         []string
-	emailAddresses   []string
 	ipAddresses      []net.IP
-	uris             []*url.URL
 	validityDuration time.Duration
 }
 
@@ -67,24 +48,15 @@ func WithDNSNames(dnsNames []string) Option {
 	}
 }
 
-// WithEmailAddresses set the emailAddresses for dragonflyManagerIssuer.
-func WithEmailAddresses(emailAddrs []string) Option {
-	return func(i *dragonflyManagerIssuer) {
-		i.emailAddresses = emailAddrs
-	}
-}
-
 // WithIPAddresses set the ipAddresses for dragonflyManagerIssuer.
-func WithIPAddresses(ipAddrs []net.IP) Option {
+func WithIPAddresses(rawIPAddrs []string) Option {
 	return func(i *dragonflyManagerIssuer) {
-		i.ipAddresses = ipAddrs
-	}
-}
+		var ipAddrs []net.IP
+		for _, rawIPAddr := range rawIPAddrs {
+			ipAddrs = append(ipAddrs, net.ParseIP(rawIPAddr))
+		}
 
-// WithURIs set the uris for dragonflyManagerIssuer.
-func WithURIs(uris []*url.URL) Option {
-	return func(i *dragonflyManagerIssuer) {
-		i.uris = uris
+		i.ipAddresses = ipAddrs
 	}
 }
 
@@ -98,10 +70,7 @@ func WithValidityDuration(d time.Duration) Option {
 // NewDragonflyManagerIssuer returns a new certify.Issuer instence.
 func NewDragonflyManagerIssuer(tlsCACert *tls.Certificate, opts ...Option) certify.Issuer {
 	i := &dragonflyManagerIssuer{
-		tlsCACert:        tlsCACert,
-		dnsNames:         defaultDNSNames,
-		ipAddresses:      defaultIPAddresses,
-		validityDuration: defaultValidityDuration,
+		tlsCACert: tlsCACert,
 	}
 
 	for _, opt := range opts {
@@ -136,9 +105,8 @@ func (i *dragonflyManagerIssuer) Issue(ctx context.Context, commonName string, c
 			Organization: defaultSubjectOrganization,
 		},
 		DNSNames:              append(certConfig.SubjectAlternativeNames, i.dnsNames...),
-		EmailAddresses:        i.emailAddresses,
 		IPAddresses:           append(certConfig.IPSubjectAlternativeNames, i.ipAddresses...),
-		URIs:                  append(certConfig.URISubjectAlternativeNames, i.uris...),
+		URIs:                  certConfig.URISubjectAlternativeNames,
 		NotBefore:             now.Add(-10 * time.Minute).UTC(),
 		NotAfter:              now.Add(i.validityDuration).UTC(),
 		BasicConstraintsValid: true,
diff --git a/scheduler/config/dynconfig.go b/scheduler/config/dynconfig.go
index 6c292970..2d5b69f0 100644
--- a/scheduler/config/dynconfig.go
+++ b/scheduler/config/dynconfig.go
@@ -185,17 +185,18 @@ func (d *dynconfig) GetResolveSeedPeerAddrs() ([]resolver.Address, error) {
 		r := reachable.New(&reachable.Config{Address: addr})
 		if err := r.Check(); err != nil {
 			logger.Warnf("seed peer address %s is unreachable", addr)
-		} else {
-			if addrs[addr] {
-				continue
-			}
+			continue
+		}
 
-			resolveAddrs = append(resolveAddrs, resolver.Address{
-				ServerName: seedPeer.IP,
-				Addr:       addr,
-			})
-			addrs[addr] = true
+		if addrs[addr] {
+			continue
 		}
+
+		resolveAddrs = append(resolveAddrs, resolver.Address{
+			ServerName: seedPeer.IP,
+			Addr:       addr,
+		})
+		addrs[addr] = true
 	}
 
 	return resolveAddrs, nil


Test patch:
diff --git a/manager/config/config_test.go b/manager/config/config_test.go
index b25d7d9e..96ac0934 100644
--- a/manager/config/config_test.go
+++ b/manager/config/config_test.go
@@ -104,6 +104,11 @@ func TestManagerConfig_Load(t *testing.T) {
 			CACert:        "foo",
 			CAKey:         "bar",
 			TLSPolicy:     "force",
+			CertSpec: &CertSpec{
+				IPAddresses:      []string{"127.0.0.1"},
+				DNSNames:         []string{"foo"},
+				ValidityDuration: 1000,
+			},
 		},
 		Metrics: &MetricsConfig{
 			Enable:          true,
diff --git a/manager/config/testdata/manager.yaml b/manager/config/testdata/manager.yaml
index ecf50131..93ea33e5 100644
--- a/manager/config/testdata/manager.yaml
+++ b/manager/config/testdata/manager.yaml
@@ -62,6 +62,12 @@ security:
   caCert: foo
   caKey: bar
   tlsPolicy: force
+  certSpec:
+    ipAddresses:
+      - "127.0.0.1"
+    dnsNames:
+      - "foo"
+    validityDuration: 1000
 
 metrics:
   enable: true