Repository: dragonflyoss/Dragonfly2
Pull request #1577 (2022-08-18 10:38:35)

feat: manager implements issue certificate grpc
Signed-off-by: Jim Ma <majinjing3@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description

<!--- 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)
- [ ] 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.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1577](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (30d01e2) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/22bfa879da935e8c678cf9a32cc1e64ad9a2fb2b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (22bfa87) will **decrease** coverage by `2.37%`.
> The diff coverage is `53.73%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1577      +/-   ##
==========================================
- Coverage   44.66%   42.29%   -2.38%     
==========================================
  Files         116      118       +2     
  Lines       10568    11274     +706     
==========================================
+ Hits         4720     4768      +48     
- Misses       5362     6010     +648     
- Partials      486      496      +10     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `42.29% <53.73%> (-2.38%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [manager/rpcserver/rpcserver.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-bWFuYWdlci9ycGNzZXJ2ZXIvcnBjc2VydmVyLmdv) | `2.27% <42.10%> (ø)` | |
| [manager/rpcserver/cert.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-bWFuYWdlci9ycGNzZXJ2ZXIvY2VydC5nbw==) | `58.33% <58.33%> (ø)` | |
| [...ent/daemon/peer/peertask\_piecetask\_synchronizer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1577/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19zeW5jaHJvbml6ZXIuZ28=) | `48.94% <0.00%> (+2.10%)` | :arrow_up: |


Patch:
diff --git a/client/config/dynconfig_manager.go b/client/config/dynconfig_manager.go
index fa4e129a..067efc1e 100644
--- a/client/config/dynconfig_manager.go
+++ b/client/config/dynconfig_manager.go
@@ -128,7 +128,15 @@ func (d *dynconfigManager) Get() (*DynconfigData, error) {
 
 // Refresh refreshes dynconfig in cache.
 func (d *dynconfigManager) Refresh() error {
-	return d.Dynconfig.Refresh()
+	if err := d.Dynconfig.Refresh(); err != nil {
+		return err
+	}
+
+	if err := d.Notify(); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // Register allows an instance to register itself to listen/observe events.
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 81510db5..0e59e338 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -507,16 +507,15 @@ func (cd *clientDaemon) Serve() error {
 			return err
 		}
 
-		g.Go(func() error {
-			logger.Info("keepalive to manager")
+		go func() {
+			logger.Info("start keepalive to manager")
 			cd.managerClient.KeepAlive(cd.Option.Scheduler.Manager.SeedPeer.KeepAlive.Interval, &managerv1.KeepAliveRequest{
 				SourceType: managerv1.SourceType_SEED_PEER_SOURCE,
 				HostName:   cd.Option.Host.Hostname,
 				Ip:         cd.Option.Host.AdvertiseIP,
 				ClusterId:  uint64(cd.Option.Scheduler.Manager.SeedPeer.ClusterID),
 			})
-			return err
-		})
+		}()
 	}
 
 	if cd.Option.AliveTime.Duration > 0 {
diff --git a/client/daemon/peer/peertask_piecetask_poller.go b/client/daemon/peer/peertask_piecetask_poller.go
index 3e69064f..430325e8 100644
--- a/client/daemon/peer/peertask_piecetask_poller.go
+++ b/client/daemon/peer/peertask_piecetask_poller.go
@@ -167,10 +167,11 @@ func (poller *pieceTaskPoller) getPieceTasksByPeer(
 		ctx, cancel := context.WithTimeout(ptc.ctx, 4*time.Second)
 		defer cancel()
 
-		client, err := dfdaemonclient.GetClient(dfnet.NetAddr{
+		netAddr := &dfnet.NetAddr{
 			Type: dfnet.TCP,
 			Addr: fmt.Sprintf("%s:%d", peer.Ip, peer.RpcPort),
-		}.GetEndpoint())
+		}
+		client, err := dfdaemonclient.GetClient(netAddr.String())
 		if err != nil {
 			ptc.Errorf("get dfdaemon client error: %s", err)
 			span.RecordError(err)
diff --git a/client/daemon/peer/peertask_piecetask_synchronizer.go b/client/daemon/peer/peertask_piecetask_synchronizer.go
index 2c9f48ba..d9d155cb 100644
--- a/client/daemon/peer/peertask_piecetask_synchronizer.go
+++ b/client/daemon/peer/peertask_piecetask_synchronizer.go
@@ -162,10 +162,11 @@ func (s *pieceTaskSyncManager) newPieceTaskSynchronizer(
 		delete(s.workers, dstPeer.PeerId)
 	}
 
-	client, err := dfdaemonclient.GetClient(dfnet.NetAddr{
+	netAddr := &dfnet.NetAddr{
 		Type: dfnet.TCP,
 		Addr: fmt.Sprintf("%s:%d", dstPeer.Ip, dstPeer.RpcPort),
-	}.GetEndpoint())
+	}
+	client, err := dfdaemonclient.GetClient(netAddr.String())
 	if err != nil {
 		s.peerTaskConductor.Errorf("get dfdaemon client error: %s, dest peer: %s", err, dstPeer.PeerId)
 		return err
diff --git a/cmd/dfcache/cmd/root.go b/cmd/dfcache/cmd/root.go
index b3ca784b..e89c9d8a 100644
--- a/cmd/dfcache/cmd/root.go
+++ b/cmd/dfcache/cmd/root.go
@@ -172,7 +172,8 @@ func runDfcacheSubcmd(cmdName string, args []string) error {
 
 // checkDaemon checks if daemon is running
 func checkDaemon(daemonSockPath string) (client.Client, error) {
-	dfdaemonClient, err := client.GetClient(dfnet.NetAddr{Type: dfnet.UNIX, Addr: daemonSockPath}.GetEndpoint())
+	netAddr := &dfnet.NetAddr{Type: dfnet.UNIX, Addr: daemonSockPath}
+	dfdaemonClient, err := client.GetClient(netAddr.String())
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/dfget/cmd/daemon.go b/cmd/dfget/cmd/daemon.go
index f67616ca..70b78e49 100644
--- a/cmd/dfget/cmd/daemon.go
+++ b/cmd/dfget/cmd/daemon.go
@@ -129,7 +129,8 @@ func initDaemonDfpath(cfg *config.DaemonOption) (dfpath.Dfpath, error) {
 
 func runDaemon(d dfpath.Dfpath) error {
 	logger.Infof("Version:\n%s", version.Version())
-	daemonClient, err := client.GetClient(dfnet.NetAddr{Type: dfnet.UNIX, Addr: d.DaemonSockPath()}.GetEndpoint())
+	netAddr := &dfnet.NetAddr{Type: dfnet.UNIX, Addr: d.DaemonSockPath()}
+	daemonClient, err := client.GetClient(netAddr.String())
 	if err != nil {
 		return err
 	}
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index 84f319b8..9257bdc2 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -244,7 +244,8 @@ func runDfget(dfgetLockPath, daemonSockPath string) error {
 
 // checkAndSpawnDaemon do checking at three checkpoints
 func checkAndSpawnDaemon(dfgetLockPath, daemonSockPath string) (client.Client, error) {
-	dfdaemonClient, err := client.GetClient(dfnet.NetAddr{Type: dfnet.UNIX, Addr: daemonSockPath}.GetEndpoint())
+	netAddr := &dfnet.NetAddr{Type: dfnet.UNIX, Addr: daemonSockPath}
+	dfdaemonClient, err := client.GetClient(netAddr.String())
 	if err != nil {
 		return nil, err
 	}
diff --git a/go.mod b/go.mod
index dab58e79..4d2f6052 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,7 @@ module d7y.io/dragonfly/v2
 go 1.18
 
 require (
-	d7y.io/api v1.0.9
+	d7y.io/api v1.1.0
 	github.com/RichardKnop/machinery v1.10.6
 	github.com/Showmax/go-fqdn v1.0.0
 	github.com/VividCortex/mysqlerr v1.0.0
diff --git a/go.sum b/go.sum
index c3bff36a..ec2dec08 100644
--- a/go.sum
+++ b/go.sum
@@ -71,6 +71,8 @@ cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3f
 cloud.google.com/go/storage v1.22.1/go.mod h1:S8N1cAStu7BOeFfE8KAQzmyyLkK8p/vmRq6kuBTW58Y=
 d7y.io/api v1.0.9 h1:KrKk1ryT771msyBxRXEX9B7abzh6h3yHyVPuSeu3Ypo=
 d7y.io/api v1.0.9/go.mod h1:GFnWPZFe4DUW70aOQikRZF0pvXpbUwAsGSCAZFFitPo=
+d7y.io/api v1.1.0 h1:+PkkqokEe+82Fjimgcrq0aVp6irGOGyYRXd9ZZ12aKk=
+d7y.io/api v1.1.0/go.mod h1:GFnWPZFe4DUW70aOQikRZF0pvXpbUwAsGSCAZFFitPo=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20201218220906-28db891af037/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v0.19.0/go.mod h1:h6H6c8enJmmocHUbLiiGY6sx7f9i+X3m1CHdd5c6Rdw=
diff --git a/manager/manager.go b/manager/manager.go
index 328fb004..1b8d5038 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -154,7 +154,11 @@ func New(cfg *config.Config, d dfpath.Dfpath) (*Server, error) {
 	}
 
 	// Initialize GRPC server
-	grpcServer := rpcserver.New(cfg, db, cache, searcher, objectStorage, cfg.ObjectStorage)
+	grpcServer, err := rpcserver.New(cfg, db, cache, searcher, objectStorage, cfg.ObjectStorage, "", "")
+	if err != nil {
+		return nil, err
+	}
+
 	s.grpcServer = grpcServer
 
 	// Initialize prometheus
diff --git a/manager/rpcserver/cert.go b/manager/rpcserver/cert.go
new file mode 100644
index 00000000..554121b1
--- /dev/null
+++ b/manager/rpcserver/cert.go
@@ -0,0 +1,116 @@
+/*
+ *     Copyright 2022 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 rpcserver
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/x509"
+	"encoding/pem"
+	"fmt"
+	"math/big"
+	"net"
+	"time"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/status"
+
+	securityv1 "d7y.io/api/pkg/apis/security/v1"
+)
+
+func (s *Server) IssueCertificate(ctx context.Context, request *securityv1.CertificateRequest) (*securityv1.CertificateResponse, error) {
+	if s.ca == nil {
+		return nil, status.Errorf(codes.Unavailable, "ca is missing for this manager instance")
+	}
+
+	var (
+		ip  string
+		err error
+	)
+
+	p, ok := peer.FromContext(ctx)
+	if !ok {
+		return nil, status.Errorf(codes.InvalidArgument, "invalid grpc peer info")
+	}
+
+	if addr, ok := p.Addr.(*net.TCPAddr); ok {
+		ip = addr.IP.String()
+	} else {
+		ip, _, err = net.SplitHostPort(p.Addr.String())
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// decode csr pem
+	block, _ := pem.Decode([]byte(request.Csr))
+	if block == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "invalid csr format")
+	}
+
+	// parse csr
+	csr, err := x509.ParseCertificateRequest(block.Bytes)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "invalid csr format: %s", err.Error())
+	}
+
+	// check csr signature
+	// TODO check csr common name and so on
+	if err = csr.CheckSignature(); err != nil {
+		return nil, err
+	}
+
+	serial, err := rand.Int(rand.Reader, (&big.Int{}).Exp(big.NewInt(2), big.NewInt(159), nil))
+	if err != nil {
+		return nil, err
+	}
+
+	// check certificate duration
+	now := time.Now()
+	duration := time.Duration(request.ValidityDuration) * time.Second
+	if duration == 0 {
+		duration = time.Hour
+	}
+
+	template := x509.Certificate{
+		SerialNumber:          serial,
+		Subject:               csr.Subject,
+		IPAddresses:           []net.IP{net.ParseIP(ip)}, // only valid for peer ip
+		NotBefore:             now.Add(-10 * time.Minute).UTC(),
+		NotAfter:              now.Add(duration).UTC(),
+		BasicConstraintsValid: true,
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageDataEncipherment | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+	}
+
+	certificate, err := x509.CreateCertificate(rand.Reader, &template, s.ca.Leaf, csr.PublicKey, s.ca.PrivateKey)
+	if err != nil {
+		return nil, fmt.Errorf("failed to generate certificate, error: %s", err)
+	}
+
+	// encode into PEM format
+	var certPEM bytes.Buffer
+	if err = pem.Encode(&certPEM, &pem.Block{Type: "CERTIFICATE", Bytes: certificate}); err != nil {
+		return nil, err
+	}
+
+	return &securityv1.CertificateResponse{
+		CertificateChain: append([]string{certPEM.String()}, s.certChain...),
+	}, nil
+}
diff --git a/manager/rpcserver/rpcserver.go b/manager/rpcserver/rpcserver.go
index 9d8964ca..e851ee4a 100644
--- a/manager/rpcserver/rpcserver.go
+++ b/manager/rpcserver/rpcserver.go
@@ -17,7 +17,11 @@
 package rpcserver
 
 import (
+	"bytes"
 	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/pem"
 	"errors"
 	"io"
 	"time"
@@ -28,8 +32,8 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	emptypb "google.golang.org/protobuf/types/known/emptypb"
-	timestamppb "google.golang.org/protobuf/types/known/timestamppb"
+	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
 	"gorm.io/gorm"
 
 	managerv1 "d7y.io/api/pkg/apis/manager/v1"
@@ -62,14 +66,29 @@ type Server struct {
 	objectStorage objectstorage.ObjectStorage
 	// Object storage configuration.
 	objectStorageConfig *config.ObjectStorageConfig
+	// ca certificate to sign certificates
+	ca *tls.Certificate
+	// ca certificate chain
+	certChain []string
 }
 
 // New returns a new manager server from the given options.
 func New(
 	cfg *config.Config, database *database.Database, cache *cache.Cache, searcher searcher.Searcher,
-	objectStorage objectstorage.ObjectStorage, objectStorageConfig *config.ObjectStorageConfig, opts ...grpc.ServerOption,
-) *grpc.Server {
-	return managerserver.New(&Server{
+	objectStorage objectstorage.ObjectStorage, objectStorageConfig *config.ObjectStorageConfig,
+	caCertPEM string, caPrivateKeyPEM string, opts ...grpc.ServerOption,
+) (*grpc.Server, error) {
+	server, err := newServer(cfg, database, cache, searcher, objectStorage, objectStorageConfig, caCertPEM, caPrivateKeyPEM)
+	if err != nil {
+		return nil, err
+	}
+	return managerserver.New(server, opts...), nil
+}
+
+func newServer(cfg *config.Config, database *database.Database, cache *cache.Cache, searcher searcher.Searcher,
+	objectStorage objectstorage.ObjectStorage, objectStorageConfig *config.ObjectStorageConfig,
+	caCertPEM string, caPrivateKeyPEM string) (*Server, error) {
+	s := &Server{
 		config:              cfg,
 		db:                  database.DB,
 		rdb:                 database.RDB,
@@ -77,7 +96,31 @@ func New(
 		searcher:            searcher,
 		objectStorage:       objectStorage,
 		objectStorageConfig: objectStorageConfig,
-	}, opts...)
+	}
+
+	if caCertPEM != "" || caPrivateKeyPEM != "" {
+		ca, err := tls.X509KeyPair([]byte(caCertPEM), []byte(caPrivateKeyPEM))
+		if err != nil {
+			return nil, err
+		}
+		// load x509 cert
+		ca.Leaf, err = x509.ParseCertificate(ca.Certificate[0])
+		if err != nil {
+			return nil, err
+		}
+
+		s.ca = &ca
+
+		// parse all cert
+		for _, cert := range ca.Certificate {
+			var certChainPEM bytes.Buffer
+			if err = pem.Encode(&certChainPEM, &pem.Block{Type: "CERTIFICATE", Bytes: cert}); err != nil {
+				return nil, err
+			}
+			s.certChain = append(s.certChain, certChainPEM.String())
+		}
+	}
+	return s, nil
 }
 
 // Get SeedPeer and SeedPeer cluster configuration.
diff --git a/pkg/dfnet/dfnet.go b/pkg/dfnet/dfnet.go
index 7a72b0b6..5506d0ba 100644
--- a/pkg/dfnet/dfnet.go
+++ b/pkg/dfnet/dfnet.go
@@ -19,6 +19,7 @@ package dfnet
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 
 	"gopkg.in/yaml.v3"
 )
@@ -26,36 +27,39 @@ import (
 type NetworkType string
 
 const (
-	TCP   NetworkType = "tcp"
-	UNIX  NetworkType = "unix"
-	VSOCK NetworkType = "vsock"
+	// TCP represents protocol of tcp.
+	TCP NetworkType = "tcp"
+
+	// TCP represents protocol of unix.
+	UNIX NetworkType = "unix"
 
-	TCPEndpointPrefix   string = "dns:///"
-	UnixEndpointPrefix  string = "unix://"
-	VsockEndpointPrefix string = "vsock://"
+	// TCP represents protocol of vsock.
+	VSOCK NetworkType = "vsock"
 )
 
+// NetAddr is the definition structure of grpc address,
+// refer to https://github.com/grpc/grpc/blob/master/doc/naming.md.
 type NetAddr struct {
+	// Type is the type of network.
 	Type NetworkType `mapstructure:"type" yaml:"type"`
-	// see https://github.com/grpc/grpc/blob/master/doc/naming.md
+
+	// Addr is the address of network.
 	Addr string `mapstructure:"addr" yaml:"addr"`
 }
 
-func (n NetAddr) GetEndpoint() string {
+// String returns the endpoint of network address.
+func (n *NetAddr) String() string {
 	switch n.Type {
 	case UNIX:
-		return UnixEndpointPrefix + n.Addr
+		return fmt.Sprintf("unix://%s", n.Addr)
 	case VSOCK:
-		return VsockEndpointPrefix + n.Addr
+		return fmt.Sprintf("vsock://%s", n.Addr)
 	default:
-		return TCPEndpointPrefix + n.Addr
+		return fmt.Sprintf("dns:///%s", n.Addr)
 	}
 }
 
-func (n NetAddr) String() string {
-	return n.GetEndpoint()
-}
-
+// UnmarshalJSON parses the JSON-encoded data and stores the result in NetAddr.
 func (n *NetAddr) UnmarshalJSON(b []byte) error {
 	var v any
 	if err := json.Unmarshal(b, &v); err != nil {
@@ -77,6 +81,7 @@ func (n *NetAddr) UnmarshalJSON(b []byte) error {
 	}
 }
 
+// UnmarshalYAML parses the YAML-encoded data and stores the result in NetAddr.
 func (n *NetAddr) UnmarshalYAML(node *yaml.Node) error {
 	switch node.Kind {
 	case yaml.ScalarNode:
@@ -84,6 +89,7 @@ func (n *NetAddr) UnmarshalYAML(node *yaml.Node) error {
 		if err := node.Decode(&addr); err != nil {
 			return err
 		}
+
 		n.Type = TCP
 		n.Addr = addr
 		return nil
@@ -97,9 +103,11 @@ func (n *NetAddr) UnmarshalYAML(node *yaml.Node) error {
 			if err := node.Content[i].Decode(&key); err != nil {
 				return err
 			}
+
 			if err := node.Content[i+1].Decode(&value); err != nil {
 				return err
 			}
+
 			m[key] = value
 		}
 
@@ -111,35 +119,25 @@ func (n *NetAddr) UnmarshalYAML(node *yaml.Node) error {
 		if err := n.unmarshal(yaml.Unmarshal, b); err != nil {
 			return err
 		}
+
 		return nil
 	default:
 		return errors.New("invalid net addr")
 	}
 }
 
+// unmarshal parses the encoded data and stores the result.
 func (n *NetAddr) unmarshal(unmarshal func(in []byte, out any) (err error), b []byte) error {
-	nt := struct {
+	netAddr := struct {
 		Type NetworkType `json:"type" yaml:"type"`
-		Addr string      `json:"addr" yaml:"addr"` // see https://github.com/grpc/grpc/blob/master/doc/naming.md
+		Addr string      `json:"addr" yaml:"addr"`
 	}{}
 
-	if err := unmarshal(b, &nt); err != nil {
+	if err := unmarshal(b, &netAddr); err != nil {
 		return err
 	}
 
-	n.Type = nt.Type
-	n.Addr = nt.Addr
-
+	n.Type = netAddr.Type
+	n.Addr = netAddr.Addr
 	return nil
 }
-
-func Convert2NetAddr(addrs []string) []NetAddr {
-	netAddrs := make([]NetAddr, 0, len(addrs))
-	for i := range addrs {
-		netAddrs = append(netAddrs, NetAddr{
-			Type: TCP,
-			Addr: addrs[i],
-		})
-	}
-	return netAddrs
-}
diff --git a/pkg/rpc/dfdaemon/client/client.go b/pkg/rpc/dfdaemon/client/client.go
index 4589db53..88ce3e07 100644
--- a/pkg/rpc/dfdaemon/client/client.go
+++ b/pkg/rpc/dfdaemon/client/client.go
@@ -62,6 +62,7 @@ func GetClient(target string, opts ...grpc.DialOption) (Client, error) {
 		append([]grpc.DialOption{
 			grpc.WithTransportCredentials(insecure.NewCredentials()),
 			grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(
+				rpc.ConvertErrorUnaryClientInterceptor,
 				otelgrpc.UnaryClientInterceptor(),
 				grpc_prometheus.UnaryClientInterceptor,
 				grpc_zap.UnaryClientInterceptor(logger.GrpcLogger.Desugar()),
@@ -72,6 +73,7 @@ func GetClient(target string, opts ...grpc.DialOption) (Client, error) {
 				),
 			)),
 			grpc.WithStreamInterceptor(grpc_middleware.ChainStreamClient(
+				rpc.ConvertErrorStreamClientInterceptor,
 				otelgrpc.StreamClientInterceptor(),
 				grpc_prometheus.StreamClientInterceptor,
 				grpc_zap.StreamClientInterceptor(logger.GrpcLogger.Desugar()),
diff --git a/pkg/rpc/manager/server/server.go b/pkg/rpc/manager/server/server.go
index 25990d52..ba27f8bf 100644
--- a/pkg/rpc/manager/server/server.go
+++ b/pkg/rpc/manager/server/server.go
@@ -29,6 +29,7 @@ import (
 	healthpb "google.golang.org/grpc/health/grpc_health_v1"
 
 	managerv1 "d7y.io/api/pkg/apis/manager/v1"
+	securityv1 "d7y.io/api/pkg/apis/security/v1"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc"
@@ -67,6 +68,7 @@ func New(svr managerv1.ManagerServer, opts ...grpc.ServerOption) *grpc.Server {
 
 	// Register servers on grpc server.
 	managerv1.RegisterManagerServer(grpcServer, svr)
+	securityv1.RegisterCertificateServiceServer(grpcServer, svr.(securityv1.CertificateServiceServer))
 
 	// Register health on grpc server.
 	healthpb.RegisterHealthServer(grpcServer, health.NewServer())
diff --git a/pkg/rpc/vsock.go b/pkg/rpc/vsock.go
index 61beb6d2..f7395855 100644
--- a/pkg/rpc/vsock.go
+++ b/pkg/rpc/vsock.go
@@ -55,5 +55,5 @@ func VsockDialer(ctx context.Context, address string) (net.Conn, error) {
 
 // IsVsock returns whether the address is vsock.
 func IsVsock(target string) bool {
-	return strings.HasPrefix(target, dfnet.VsockEndpointPrefix)
+	return strings.HasPrefix(target, string(dfnet.VSOCK))
 }
diff --git a/scheduler/config/dynconfig.go b/scheduler/config/dynconfig.go
index ff3ce08d..5533fc29 100644
--- a/scheduler/config/dynconfig.go
+++ b/scheduler/config/dynconfig.go
@@ -259,7 +259,15 @@ func (d *dynconfig) Get() (*DynconfigData, error) {
 
 // Refresh refreshes dynconfig in cache.
 func (d *dynconfig) Refresh() error {
-	return d.Dynconfig.Refresh()
+	if err := d.Dynconfig.Refresh(); err != nil {
+		return err
+	}
+
+	if err := d.Notify(); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // Register allows an instance to register itself to listen/observe events.


Test patch:
diff --git a/client/daemon/rpcserver/rpcserver_test.go b/client/daemon/rpcserver/rpcserver_test.go
index d701d1d2..98bcb919 100644
--- a/client/daemon/rpcserver/rpcserver_test.go
+++ b/client/daemon/rpcserver/rpcserver_test.go
@@ -562,10 +562,11 @@ func setupPeerServerAndClient(t *testing.T, srv *server, assert *testifyassert.A
 		}
 	}()
 
-	client, err := dfdaemonclient.GetClient(dfnet.NetAddr{
+	netAddr := &dfnet.NetAddr{
 		Type: dfnet.TCP,
 		Addr: fmt.Sprintf(":%d", port),
-	}.GetEndpoint())
+	}
+	client, err := dfdaemonclient.GetClient(netAddr.String())
 	assert.Nil(err, "grpc dial should be ok")
 	return client
 }
diff --git a/manager/rpcserver/cert_test.go b/manager/rpcserver/cert_test.go
new file mode 100644
index 00000000..33f8e24b
--- /dev/null
+++ b/manager/rpcserver/cert_test.go
@@ -0,0 +1,174 @@
+/*
+ *     Copyright 2022 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 rpcserver
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+	testifyassert "github.com/stretchr/testify/assert"
+	testifyrequire "github.com/stretchr/testify/require"
+	"google.golang.org/grpc/peer"
+	"gorm.io/gorm"
+
+	securityv1 "d7y.io/api/pkg/apis/security/v1"
+
+	"d7y.io/dragonfly/v2/manager/database"
+)
+
+func TestIssueCertificate(t *testing.T) {
+	assert := testifyassert.New(t)
+	require := testifyrequire.New(t)
+
+	caCert, caKey := genCA()
+
+	testCases := []struct {
+		name   string
+		peerIP string
+	}{
+		{
+			name:   "ipv4",
+			peerIP: "1.1.1.1",
+		},
+		{
+			name:   "ipv6",
+			peerIP: "1::1",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			template := &x509.CertificateRequest{
+				Subject: pkix.Name{
+					Country:            []string{"China"},
+					Organization:       []string{"Dragonfly"},
+					OrganizationalUnit: []string{"Development"},
+				},
+			}
+
+			pk, err := rsa.GenerateKey(rand.Reader, 4096)
+			require.Nilf(err, "GenerateKey should be ok")
+
+			csr, err := x509.CreateCertificateRequest(rand.Reader, template, pk)
+			require.Nilf(err, "CreateCertificateRequest should be ok")
+
+			var csrPEM bytes.Buffer
+			err = pem.Encode(&csrPEM, &pem.Block{Type: "CERTIFICATE REQUEST", Bytes: csr})
+			require.Nilf(err, "pem.Encode should be ok")
+
+			server, err := newServer(nil,
+				&database.Database{
+					DB:  &gorm.DB{},
+					RDB: &redis.Client{},
+				},
+				nil, nil, nil, nil, caCert, caKey)
+
+			require.Nilf(err, "newServer should be ok")
+
+			ctx := peer.NewContext(
+				context.Background(),
+				&peer.Peer{
+					Addr: &net.TCPAddr{
+						IP:   net.ParseIP(tc.peerIP),
+						Port: 65008,
+					},
+				})
+
+			response, err := server.IssueCertificate(
+				ctx,
+				&securityv1.CertificateRequest{
+					Csr:              csrPEM.String(),
+					ValidityDuration: 0,
+				})
+
+			assert.Nilf(err, "IssueCertificate should be ok")
+			assert.NotNilf(response, "IssueCertificate should not be nil")
+			assert.Equal(len(response.CertificateChain), len(server.certChain)+1)
+
+			cert := readCert(response.CertificateChain[0])
+			assert.Equal(len(cert.IPAddresses), 1)
+			assert.True(cert.IPAddresses[0].Equal(net.ParseIP(tc.peerIP)))
+
+			assert.Equal(cert.KeyUsage, x509.KeyUsageDigitalSignature|x509.KeyUsageDataEncipherment|x509.KeyUsageKeyEncipherment)
+			assert.Equal(cert.ExtKeyUsage, []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth})
+		})
+	}
+}
+
+func genCA() (cert, key string) {
+	pk, err := rsa.GenerateKey(rand.Reader, 4096)
+	if err != nil {
+		panic(err)
+	}
+
+	ca := &x509.Certificate{
+		SerialNumber: big.NewInt(2022),
+		Subject: pkix.Name{
+			Country:            []string{"China"},
+			Organization:       []string{"Dragonfly"},
+			OrganizationalUnit: []string{"Development"},
+			Locality:           []string{"Hangzhou"},
+			Province:           []string{"Zhejiang"},
+		},
+		NotBefore:             time.Now().Add(-10 * time.Minute).UTC(),
+		NotAfter:              time.Now().Add(time.Hour).UTC(),
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
+	}
+
+	caBytes, err := x509.CreateCertificate(rand.Reader, ca, ca, &pk.PublicKey, pk)
+	if err != nil {
+		panic(err)
+	}
+
+	var (
+		caCertPEM       bytes.Buffer
+		caPrivateKeyPEM bytes.Buffer
+	)
+
+	if err = pem.Encode(&caCertPEM, &pem.Block{Type: "CERTIFICATE", Bytes: caBytes}); err != nil {
+		panic(err)
+	}
+
+	if err = pem.Encode(&caPrivateKeyPEM, &pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(pk)}); err != nil {
+		panic(err)
+	}
+
+	key, cert = caPrivateKeyPEM.String(), caCertPEM.String()
+	return
+}
+
+func readCert(certPEM string) *x509.Certificate {
+	p, _ := pem.Decode([]byte(certPEM))
+	cert, err := x509.ParseCertificate(p.Bytes)
+	if err != nil {
+		panic(err)
+	}
+	return cert
+}