Repository: dragonflyoss/Dragonfly2
Pull request #676 (2021-09-26 13:43:59)

Add manager metrics and logger
<!--- Provide a general summary of your changes in the Title above -->

## Description
- Add manager middlewares.
- Add manager grpc metrics.
- Add mysql logger.
- Add redis logger.
<!--- 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! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#676](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (13e8c68) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/d668819712d81ddc4bd7a2ba0aa5faa08a025b85?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (d668819) will **decrease** coverage by `0.12%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676/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/676?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main     #676      +/-   ##
==========================================
- Coverage   38.61%   38.48%   -0.13%     
==========================================
  Files          89       89              
  Lines        6982     6987       +5     
==========================================
- Hits         2696     2689       -7     
- Misses       3904     3916      +12     
  Partials      382      382              
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| unittests | `38.48% <0.00%> (-0.13%)` | :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/676?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [client/daemon/storage/local\_storage.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9zdG9yYWdlL2xvY2FsX3N0b3JhZ2UuZ28=) | `30.59% <0.00%> (ø)` | |
| [manager/config/config.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-bWFuYWdlci9jb25maWcvY29uZmlnLmdv) | `0.00% <0.00%> (ø)` | |
| [client/daemon/peer/peertask\_stream.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3N0cmVhbS5nbw==) | `55.66% <0.00%> (-1.62%)` | :arrow_down: |
| [client/daemon/peer/peertask\_base.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2Jhc2UuZ28=) | `33.67% <0.00%> (-0.62%)` | :arrow_down: |
| [client/daemon/peer/peertask\_file.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/676/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2ZpbGUuZ28=) | `54.77% <0.00%> (+0.35%)` | :arrow_up: |


Patch:
diff --git a/client/daemon/storage/local_storage.go b/client/daemon/storage/local_storage.go
index 9bd4b9fa..b9a11cb1 100644
--- a/client/daemon/storage/local_storage.go
+++ b/client/daemon/storage/local_storage.go
@@ -166,7 +166,7 @@ func (t *localTaskStore) ReadPiece(ctx context.Context, req *ReadPieceRequest) (
 
 	if _, err = file.Seek(req.Range.Start, io.SeekStart); err != nil {
 		file.Close()
-		t.Errorf("file seek filed: %v", err)
+		t.Errorf("file seek failed: %v", err)
 		return nil, nil, err
 	}
 	// who call ReadPiece, who close the io.ReadCloser
@@ -181,7 +181,7 @@ func (t *localTaskStore) ReadAllPieces(ctx context.Context, req *PeerTaskMetaDat
 	}
 	if _, err = file.Seek(0, io.SeekStart); err != nil {
 		file.Close()
-		t.Errorf("file seek filed: %v", err)
+		t.Errorf("file seek failed: %v", err)
 		return nil, err
 	}
 	// who call ReadPiece, who close the io.ReadCloser
diff --git a/go.mod b/go.mod
index 8628202e..41efc2fa 100644
--- a/go.mod
+++ b/go.mod
@@ -35,6 +35,8 @@ require (
 	github.com/google/go-querystring v1.1.0 // indirect
 	github.com/google/uuid v1.2.0
 	github.com/gorilla/mux v1.8.0
+	github.com/grpc-ecosystem/go-grpc-middleware v1.0.0
+	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/jackc/pgproto3/v2 v2.1.1 // indirect
 	github.com/jarcoal/httpmock v1.0.8
 	github.com/klauspost/compress v1.13.1 // indirect
@@ -50,7 +52,7 @@ require (
 	github.com/pelletier/go-toml v1.8.1 // indirect
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.7.1 // indirect
+	github.com/prometheus/client_golang v1.7.1
 	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/schollz/progressbar/v3 v3.8.2
 	github.com/serialx/hashring v0.0.0-20200727003509-22c0c7ab6b1b
@@ -88,4 +90,5 @@ require (
 	gorm.io/plugin/soft_delete v1.0.2
 	k8s.io/apimachinery v0.20.6
 	k8s.io/client-go v11.0.0+incompatible
+	moul.io/zapgorm2 v1.1.0
 )
diff --git a/go.sum b/go.sum
index d3eb7f1d..a424a2d9 100644
--- a/go.sum
+++ b/go.sum
@@ -300,6 +300,7 @@ github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4
 github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang-jwt/jwt/v4 v4.0.0 h1:RAqyYixv1p7uEnocuy8P1nru5wprCh/MH2BIlW5z5/o=
 github.com/golang-jwt/jwt/v4 v4.0.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
@@ -400,7 +401,9 @@ github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+
 github.com/gorilla/sessions v1.2.0 h1:S7P+1Hm5V/AT9cjEcUD5uDaQSX0OE577aCXgoaKpYbQ=
 github.com/gorilla/sessions v1.2.0/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:Iju5GlWwrvL6UBg4zJJt3btmonfrMlCDdsejg4CZE7c=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
+github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
 github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t/GYV9WGTSLVdBhOQFDY=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
@@ -879,7 +882,6 @@ go.opentelemetry.io/otel/trace v0.20.0/go.mod h1:6GjCW8zgDjwGHGa6GkyeB8+/5vjT16g
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
-go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -1395,6 +1397,8 @@ k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.4.0 h1:7+X0fUguPyrKEC4WjH8iGDg3laWgMo5tMnRTIGTTxGQ=
 k8s.io/klog/v2 v2.4.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
 k8s.io/kube-openapi v0.0.0-20201113171705-d219536bb9fd/go.mod h1:WOJ3KddDSol4tAGcJo0Tvi+dK12EcqSLqcWsryKMpfM=
+moul.io/zapgorm2 v1.1.0 h1:qwAlMBYf+qJkJ7PAzJl4oCe6eS6QGiKAXUPeis0+RBE=
+moul.io/zapgorm2 v1.1.0/go.mod h1:emRfKjNqSzVj5lcgasBdovIXY1jSOwFz2GQZn1Rddks=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4=
 rsc.io/quote/v3 v3.1.0/go.mod h1:yEA65RcK8LyAZtP9Kv3t0HmxON59tX3rD+tICJqUlj0=
diff --git a/manager/config/config.go b/manager/config/config.go
index 12243ea1..e4ba185c 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -35,6 +35,7 @@ type ServerConfig struct {
 	PublicPath string           `yaml:"publicPath" mapstructure:"publicPath"`
 	GRPC       *TCPListenConfig `yaml:"grpc" mapstructure:"grpc"`
 	REST       *RestConfig      `yaml:"rest" mapstructure:"rest"`
+	Metric     *RestConfig      `yaml:"metric" mapstructure:"metric"`
 }
 
 type DatabaseConfig struct {
@@ -105,6 +106,9 @@ func New() *Config {
 			REST: &RestConfig{
 				Addr: ":8080",
 			},
+			Metric: &RestConfig{
+				Addr: ":8000",
+			},
 		},
 		Database: &DatabaseConfig{
 			Redis: &RedisConfig{
@@ -174,11 +178,15 @@ func (cfg *Config) Validate() error {
 
 	if cfg.Server != nil {
 		if cfg.Server.GRPC == nil {
-			return errors.New("empty grpc config is not specified")
+			return errors.New("empty grpc server config is not specified")
 		}
 
 		if cfg.Server.REST == nil {
-			return errors.New("empty rest config is not specified")
+			return errors.New("empty rest server config is not specified")
+		}
+
+		if cfg.Server.Metric == nil {
+			return errors.New("empty metric server config is not specified")
 		}
 	}
 
diff --git a/manager/database/database.go b/manager/database/database.go
index d64a4624..cd967e2b 100644
--- a/manager/database/database.go
+++ b/manager/database/database.go
@@ -20,12 +20,15 @@ import (
 	"context"
 	"fmt"
 
-	"d7y.io/dragonfly/v2/manager/config"
-	"d7y.io/dragonfly/v2/manager/model"
 	"github.com/go-redis/redis/v8"
 	"gorm.io/driver/mysql"
 	"gorm.io/gorm"
 	"gorm.io/gorm/schema"
+	"moul.io/zapgorm2"
+
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/manager/config"
+	"d7y.io/dragonfly/v2/manager/model"
 )
 
 type Database struct {
@@ -51,6 +54,8 @@ func New(cfg *config.Config) (*Database, error) {
 }
 
 func NewRedis(cfg *config.RedisConfig) (*redis.Client, error) {
+	redis.SetLogger(&redisLogger{})
+
 	client := redis.NewClient(&redis.Options{
 		Addr:     fmt.Sprintf("%s:%d", cfg.Host, cfg.Port),
 		Password: cfg.Password,
@@ -65,6 +70,8 @@ func NewRedis(cfg *config.RedisConfig) (*redis.Client, error) {
 }
 
 func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
+	logger := zapgorm2.New(logger.CoreLogger.Desugar())
+
 	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=True&loc=Local&interpolateParams=true", cfg.User, cfg.Password, cfg.Host, cfg.Port, cfg.DBName)
 	dialector := mysql.Open(dsn)
 	db, err := gorm.Open(dialector, &gorm.Config{
@@ -72,6 +79,7 @@ func newMyqsl(cfg *config.MysqlConfig) (*gorm.DB, error) {
 			SingularTable: true,
 		},
 		DisableForeignKeyConstraintWhenMigrating: true,
+		Logger:                                   logger,
 	})
 	if err != nil {
 		return nil, err
diff --git a/manager/database/redis_logger.go b/manager/database/redis_logger.go
new file mode 100644
index 00000000..24100f82
--- /dev/null
+++ b/manager/database/redis_logger.go
@@ -0,0 +1,30 @@
+/*
+ *     Copyright 2020 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 database
+
+import (
+	"context"
+	"fmt"
+
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+)
+
+type redisLogger struct{}
+
+func (rl *redisLogger) Printf(ctx context.Context, format string, v ...interface{}) {
+	logger.CoreLogger.Desugar().Info(fmt.Sprintf(format, v...))
+}
diff --git a/manager/manager.go b/manager/manager.go
index 8a9b9325..b902dc8b 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -26,13 +26,13 @@ import (
 	"d7y.io/dragonfly/v2/manager/config"
 	"d7y.io/dragonfly/v2/manager/database"
 	"d7y.io/dragonfly/v2/manager/job"
+	"d7y.io/dragonfly/v2/manager/metric"
 	"d7y.io/dragonfly/v2/manager/permission/rbac"
-	"d7y.io/dragonfly/v2/manager/proxy"
 	"d7y.io/dragonfly/v2/manager/router"
 	"d7y.io/dragonfly/v2/manager/searcher"
 	"d7y.io/dragonfly/v2/manager/service"
 	"d7y.io/dragonfly/v2/pkg/rpc"
-	"d7y.io/dragonfly/v2/pkg/rpc/manager"
+	grpc_manager_server "d7y.io/dragonfly/v2/pkg/rpc/manager/server"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
 )
@@ -51,8 +51,8 @@ type Server struct {
 	// REST server
 	restServer *http.Server
 
-	// Proxy server
-	proxyServer proxy.Proxy
+	// Metric server
+	metricServer *http.Server
 }
 
 func New(cfg *config.Config) (*Server, error) {
@@ -83,9 +83,6 @@ func New(cfg *config.Config) (*Server, error) {
 		return nil, err
 	}
 
-	// Initialize Proxy server
-	proxyServer := proxy.New(cfg.Database.Redis)
-
 	// Initialize REST server
 	restService := service.NewREST(db, cache, job, enforcer)
 	router, err := router.Init(cfg, restService, enforcer)
@@ -109,27 +106,20 @@ func New(cfg *config.Config) (*Server, error) {
 	if cfg.Options.Telemetry.Jaeger != "" {
 		opts = append(opts, grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()), grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()))
 	}
+	grpcServer := grpc_manager_server.New(grpcService, opts...)
 
-	grpcServer := grpc.NewServer(opts...)
-	manager.RegisterManagerServer(grpcServer, grpcService)
+	// Initialize prometheus
+	metricServer := metric.New(cfg.Server.Metric, grpcServer)
 
 	return &Server{
-		config:      cfg,
-		grpcServer:  grpcServer,
-		restServer:  restServer,
-		proxyServer: proxyServer,
+		config:       cfg,
+		grpcServer:   grpcServer,
+		restServer:   restServer,
+		metricServer: metricServer,
 	}, nil
 }
 
 func (s *Server) Serve() error {
-	// Started Proxy server
-	go func() {
-		logger.Info("started proxy")
-		if err := s.proxyServer.Serve(); err != nil {
-			logger.Fatalf("proxy server closed unexpect: %+v", err)
-		}
-	}()
-
 	// Started REST server
 	go func() {
 		logger.Infof("started rest server at %s", s.restServer.Addr)
@@ -141,6 +131,17 @@ func (s *Server) Serve() error {
 		}
 	}()
 
+	// Started metric server
+	go func() {
+		logger.Infof("started metric server at %s", s.metricServer.Addr)
+		if err := s.metricServer.ListenAndServe(); err != nil {
+			if err == http.ErrServerClosed {
+				return
+			}
+			logger.Fatalf("metric server closed unexpect: %+v", err)
+		}
+	}()
+
 	// Generate GRPC listener
 	lis, _, err := rpc.ListenWithPortRange(s.config.Server.GRPC.Listen, s.config.Server.GRPC.PortRange.Start, s.config.Server.GRPC.PortRange.End)
 	if err != nil {
@@ -165,9 +166,11 @@ func (s *Server) Stop() {
 	}
 	logger.Info("rest server closed under request")
 
-	// Stop Proxy server
-	s.proxyServer.Stop()
-	logger.Info("proxy server closed under request")
+	// Stop metric server
+	if err := s.metricServer.Shutdown(context.Background()); err != nil {
+		logger.Errorf("metric server failed to stop: %+v", err)
+	}
+	logger.Info("metric server closed under request")
 
 	// Stop GRPC server
 	stopped := make(chan struct{})
diff --git a/manager/metric/metric.go b/manager/metric/metric.go
new file mode 100644
index 00000000..1944c276
--- /dev/null
+++ b/manager/metric/metric.go
@@ -0,0 +1,38 @@
+/*
+ *     Copyright 2020 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 metric
+
+import (
+	"net/http"
+
+	"d7y.io/dragonfly/v2/manager/config"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"google.golang.org/grpc"
+)
+
+func New(cfg *config.RestConfig, grpcServer *grpc.Server) *http.Server {
+	grpc_prometheus.Register(grpcServer)
+
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+
+	return &http.Server{
+		Addr:    cfg.Addr,
+		Handler: mux,
+	}
+}
diff --git a/manager/proxy/proxy.go b/manager/proxy/proxy.go
deleted file mode 100644
index 8b64a836..00000000
--- a/manager/proxy/proxy.go
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- *     Copyright 2020 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 proxy
-
-import (
-	"fmt"
-	"io"
-	"net"
-	"sync"
-
-	logger "d7y.io/dragonfly/v2/internal/dflog"
-	"d7y.io/dragonfly/v2/manager/config"
-)
-
-type Proxy interface {
-	Serve() error
-	Stop()
-}
-
-type proxy struct {
-	from string
-	to   string
-	done chan struct{}
-}
-
-func New(cfg *config.RedisConfig) Proxy {
-	return &proxy{
-		from: fmt.Sprintf(":%d", cfg.Port),
-		to:   fmt.Sprintf("%s:%d", cfg.Host, cfg.Port),
-		done: make(chan struct{}),
-	}
-}
-
-func (p *proxy) Serve() error {
-	listener, err := net.Listen("tcp", p.from)
-	if err != nil {
-		return err
-	}
-	defer listener.Close()
-
-	for {
-		select {
-		case <-p.done:
-			return nil
-		default:
-			conn, err := listener.Accept()
-			if err != nil {
-				logger.Errorf("error accepting conn %v", err)
-			} else {
-				go p.handleConn(conn)
-			}
-		}
-	}
-}
-
-func (p *proxy) Stop() {
-	if p.done == nil {
-		return
-	}
-	close(p.done)
-	p.done = nil
-}
-
-func (p *proxy) handleConn(conn net.Conn) {
-	logger.Infof("handling", conn)
-	defer logger.Infof("done handling", conn)
-	defer conn.Close()
-	rConn, err := net.Dial("tcp", p.to)
-	if err != nil {
-		logger.Errorf("error dialing remote host %v", err)
-		return
-	}
-	defer rConn.Close()
-
-	wg := &sync.WaitGroup{}
-	wg.Add(2)
-	go p.copy(rConn, conn, wg)
-	go p.copy(conn, rConn, wg)
-	wg.Wait()
-}
-
-func (p *proxy) copy(from, to net.Conn, wg *sync.WaitGroup) {
-	defer wg.Done()
-	select {
-	case <-p.done:
-		return
-	default:
-		if _, err := io.Copy(to, from); err != nil {
-			logger.Errorf("error copy %v", err)
-			p.Stop()
-			return
-		}
-	}
-}
diff --git a/manager/router/router.go b/manager/router/router.go
index 6841e911..36855407 100644
--- a/manager/router/router.go
+++ b/manager/router/router.go
@@ -26,6 +26,7 @@ import (
 	"d7y.io/dragonfly/v2/manager/handlers"
 	"d7y.io/dragonfly/v2/manager/middlewares"
 	"d7y.io/dragonfly/v2/manager/service"
+
 	"github.com/casbin/casbin/v2"
 	"github.com/gin-contrib/cors"
 	"github.com/gin-contrib/static"
@@ -59,7 +60,7 @@ func Init(cfg *config.Config, service service.REST, enforcer *casbin.Enforcer) (
 	r := gin.New()
 	h := handlers.New(service)
 
-	// Prometheus
+	// Prometheus metrics
 	p := ginprometheus.NewPrometheus(PrometheusSubsystemName)
 	p.Use(r)
 
diff --git a/manager/service/service_grpc.go b/manager/service/service_grpc.go
index e1c7d5d0..8b2e7d46 100644
--- a/manager/service/service_grpc.go
+++ b/manager/service/service_grpc.go
@@ -53,10 +53,6 @@ func NewGRPC(database *database.Database, cache *cache.Cache, searcher searcher.
 }
 
 func (s *GRPC) GetCDN(ctx context.Context, req *manager.GetCDNRequest) (*manager.CDN, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	var pbCDN manager.CDN
 	cacheKey := cache.MakeCDNCacheKey(req.HostName, uint(req.CdnClusterId))
 
@@ -119,10 +115,6 @@ func (s *GRPC) GetCDN(ctx context.Context, req *manager.GetCDNRequest) (*manager
 }
 
 func (s *GRPC) createCDN(ctx context.Context, req *manager.UpdateCDNRequest) (*manager.CDN, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	cdn := model.CDN{
 		HostName:     req.HostName,
 		IDC:          req.Idc,
@@ -150,10 +142,6 @@ func (s *GRPC) createCDN(ctx context.Context, req *manager.UpdateCDNRequest) (*m
 }
 
 func (s *GRPC) UpdateCDN(ctx context.Context, req *manager.UpdateCDNRequest) (*manager.CDN, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	cdn := model.CDN{}
 	if err := s.db.First(&cdn, model.CDN{
 		HostName:     req.HostName,
@@ -196,10 +184,6 @@ func (s *GRPC) UpdateCDN(ctx context.Context, req *manager.UpdateCDNRequest) (*m
 }
 
 func (s *GRPC) GetScheduler(ctx context.Context, req *manager.GetSchedulerRequest) (*manager.Scheduler, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	var pbScheduler manager.Scheduler
 	cacheKey := cache.MakeSchedulerCacheKey(req.HostName, uint(req.SchedulerClusterId))
 
@@ -294,10 +278,6 @@ func (s *GRPC) GetScheduler(ctx context.Context, req *manager.GetSchedulerReques
 }
 
 func (s *GRPC) createScheduler(ctx context.Context, req *manager.UpdateSchedulerRequest) (*manager.Scheduler, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	var netConfig model.JSONMap
 	if len(req.NetConfig) > 0 {
 		if err := netConfig.UnmarshalJSON(req.NetConfig); err != nil {
@@ -335,10 +315,6 @@ func (s *GRPC) createScheduler(ctx context.Context, req *manager.UpdateScheduler
 }
 
 func (s *GRPC) UpdateScheduler(ctx context.Context, req *manager.UpdateSchedulerRequest) (*manager.Scheduler, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	scheduler := model.Scheduler{}
 	if err := s.db.First(&scheduler, model.Scheduler{
 		HostName:           req.HostName,
@@ -391,10 +367,6 @@ func (s *GRPC) UpdateScheduler(ctx context.Context, req *manager.UpdateScheduler
 }
 
 func (s *GRPC) ListSchedulers(ctx context.Context, req *manager.ListSchedulersRequest) (*manager.ListSchedulersResponse, error) {
-	if err := req.Validate(); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	var pbListSchedulersResponse manager.ListSchedulersResponse
 	cacheKey := cache.MakeSchedulersCacheKey(req.HostName)
 
@@ -467,10 +439,6 @@ func (s *GRPC) KeepAlive(m manager.Manager_KeepAliveServer) error {
 		logger.Errorf("keepalive failed for the first time: %v", err)
 		return status.Error(codes.Unknown, err.Error())
 	}
-	if err := req.Validate(); err != nil {
-		return status.Error(codes.InvalidArgument, err.Error())
-	}
-
 	hostName := req.HostName
 	sourceType := req.SourceType
 	clusterID := uint(req.ClusterId)
diff --git a/pkg/rpc/manager/client/client.go b/pkg/rpc/manager/client/client.go
index 4ff56e91..4ed96dbf 100644
--- a/pkg/rpc/manager/client/client.go
+++ b/pkg/rpc/manager/client/client.go
@@ -20,7 +20,12 @@ import (
 	"context"
 	"time"
 
+	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
+
+	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
+	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/backoff"
 )
@@ -59,6 +64,10 @@ func New(target string) (Client, error) {
 				MaxDelay:   backoffMaxDelay,
 			},
 		}),
+		grpc.WithStreamInterceptor(grpc_middleware.ChainStreamClient(
+			grpc_prometheus.StreamClientInterceptor,
+			grpc_zap.StreamClientInterceptor(logger.GrpcLogger.Desugar()),
+		)),
 	)
 	if err != nil {
 		return nil, err
diff --git a/pkg/rpc/manager/manager_grpc.pb.go b/pkg/rpc/manager/manager_grpc.pb.go
index 0f9f3d52..a1bd1f39 100644
--- a/pkg/rpc/manager/manager_grpc.pb.go
+++ b/pkg/rpc/manager/manager_grpc.pb.go
@@ -4,6 +4,7 @@ package manager
 
 import (
 	context "context"
+
 	grpc "google.golang.org/grpc"
 	codes "google.golang.org/grpc/codes"
 	status "google.golang.org/grpc/status"
diff --git a/pkg/rpc/manager/server/server.go b/pkg/rpc/manager/server/server.go
new file mode 100644
index 00000000..b0072fe8
--- /dev/null
+++ b/pkg/rpc/manager/server/server.go
@@ -0,0 +1,51 @@
+/*
+ *     Copyright 2020 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 server
+
+import (
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/pkg/rpc/manager"
+
+	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
+	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
+	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
+	grpc_validator "github.com/grpc-ecosystem/go-grpc-middleware/validator"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"google.golang.org/grpc"
+)
+
+func New(managerServer manager.ManagerServer, opts ...grpc.ServerOption) *grpc.Server {
+	defaultOptions := []grpc.ServerOption{
+		grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(
+			grpc_validator.StreamServerInterceptor(),
+			grpc_recovery.StreamServerInterceptor(),
+			grpc_prometheus.StreamServerInterceptor,
+			grpc_zap.StreamServerInterceptor(logger.GrpcLogger.Desugar()),
+		)),
+		grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
+			grpc_validator.UnaryServerInterceptor(),
+			grpc_recovery.UnaryServerInterceptor(),
+			grpc_prometheus.UnaryServerInterceptor,
+			grpc_zap.UnaryServerInterceptor(logger.GrpcLogger.Desugar()),
+		)),
+	}
+
+	grpcServer := grpc.NewServer(append(defaultOptions, opts...)...)
+	manager.RegisterManagerServer(grpcServer, managerServer)
+
+	return grpcServer
+}


Test patch:
diff --git a/manager/config/config_test.go b/manager/config/config_test.go
index ea365838..0fff025b 100644
--- a/manager/config/config_test.go
+++ b/manager/config/config_test.go
@@ -41,6 +41,9 @@ func TestManagerConfig_Load(t *testing.T) {
 			REST: &RestConfig{
 				Addr: ":8080",
 			},
+			Metric: &RestConfig{
+				Addr: ":8000",
+			},
 		},
 		Database: &DatabaseConfig{
 			Mysql: &MysqlConfig{
diff --git a/manager/config/testdata/manager.yaml b/manager/config/testdata/manager.yaml
index 18b32729..5bb98939 100644
--- a/manager/config/testdata/manager.yaml
+++ b/manager/config/testdata/manager.yaml
@@ -7,6 +7,8 @@ server:
       end: 65003
   rest:
     addr: :8080
+  metric:
+    addr: :8000
 
 database:
   mysql: