feat: add grpc metric and refactor grpc server (#686)

* feat: add grpc metric and refactor grpc server

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

cdn/cdn.go

@@ -20,9 +20,12 @@ package cdn
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"runtime"
+	"time"
 
 	"d7y.io/dragonfly/v2/cdn/config"
+	"d7y.io/dragonfly/v2/cdn/metrics"
 	"d7y.io/dragonfly/v2/cdn/plugins"
 	"d7y.io/dragonfly/v2/cdn/rpcserver"
 	"d7y.io/dragonfly/v2/cdn/supervisor/cdn"
@@ -32,7 +35,6 @@ import (
 	"d7y.io/dragonfly/v2/cdn/supervisor/task"
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	"d7y.io/dragonfly/v2/pkg/rpc"
-	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/server"
 	"d7y.io/dragonfly/v2/pkg/rpc/manager"
 	managerclient "d7y.io/dragonfly/v2/pkg/rpc/manager/client"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
@@ -41,9 +43,21 @@ import (
 	"google.golang.org/grpc"
 )
 
+const (
+	gracefulStopTimeout = 10 * time.Second
+)
+
 type Server struct {
-	config        *config.Config
-	seedServer    server.SeederServer
+	// Server configuration
+	config *config.Config
+
+	// GRPC server
+	grpcServer *grpc.Server
+
+	// Metrics server
+	metricsServer *http.Server
+
+	// Manager client
 	managerClient managerclient.Client
 }
 
@@ -86,11 +100,20 @@ func New(cfg *config.Config) (*Server, error) {
 	storageMgr.Initialize(taskMgr)
 
 	// Initialize storage manager
-	cdnSeedServer, err := rpcserver.NewCdnSeedServer(cfg, taskMgr)
+	var opts []grpc.ServerOption
+	if s.config.Options.Telemetry.Jaeger != "" {
+		opts = append(opts, grpc.ChainUnaryInterceptor(otelgrpc.UnaryServerInterceptor()), grpc.ChainStreamInterceptor(otelgrpc.StreamServerInterceptor()))
+	}
+	grpcServer, err := rpcserver.New(cfg, taskMgr, opts...)
 	if err != nil {
 		return nil, errors.Wrap(err, "create seedServer")
 	}
-	s.seedServer = cdnSeedServer
+	s.grpcServer = grpcServer
+
+	// Initialize prometheus
+	if cfg.Metrics != nil {
+		s.metricsServer = metrics.New(cfg.Metrics, grpcServer)
+	}
 
 	// Initialize manager client
 	if cfg.Manager.Addr != "" {
@@ -118,21 +141,27 @@ func New(cfg *config.Config) (*Server, error) {
 	return s, nil
 }
 
-func (s *Server) Serve() (err error) {
-	defer func() {
-		if rec := recover(); rec != nil {
-			err = errors.New(fmt.Sprintf("%v", rec))
-		}
-	}()
-
+func (s *Server) Serve() error {
 	// Start GC
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	err = gc.StartGC(ctx)
-	if err != nil {
+	if err := gc.StartGC(ctx); err != nil {
 		return err
 	}
 
+	// Started metrics server
+	if s.metricsServer != nil {
+		go func() {
+			logger.Infof("started metrics server at %s", s.metricsServer.Addr)
+			if err := s.metricsServer.ListenAndServe(); err != nil {
+				if err == http.ErrServerClosed {
+					return
+				}
+				logger.Fatalf("metrics server closed unexpect: %+v", err)
+			}
+		}()
+	}
+
 	// Serve Keepalive
 	if s.managerClient != nil {
 		go func() {
@@ -145,24 +174,51 @@ func (s *Server) Serve() (err error) {
 		}()
 	}
 
-	// Serve GRPC
-	var opts []grpc.ServerOption
-	if s.config.Options.Telemetry.Jaeger != "" {
-		opts = append(opts, grpc.ChainUnaryInterceptor(otelgrpc.UnaryServerInterceptor()), grpc.ChainStreamInterceptor(otelgrpc.StreamServerInterceptor()))
-	}
-	err = rpc.StartTCPServer(s.config.ListenPort, s.config.ListenPort, s.seedServer, opts...)
+	// Generate GRPC listener
+	lis, _, err := rpc.ListenWithPortRange(iputils.HostIP, s.config.ListenPort, s.config.ListenPort)
 	if err != nil {
-		return errors.Wrap(err, "start tcp server")
+		logger.Fatalf("net listener failed to start: %+v", err)
+	}
+	defer lis.Close()
+
+	// Started GRPC server
+	logger.Infof("started grpc server at %s://%s", lis.Addr().Network(), lis.Addr().String())
+	if err := s.grpcServer.Serve(lis); err != nil {
+		logger.Errorf("stoped grpc server: %+v", err)
+		return err
 	}
+
 	return nil
 }
 
 func (s *Server) Stop() {
+	// Stop manager client
 	if s.managerClient != nil {
 		s.managerClient.Close()
 		logger.Info("manager client closed")
 	}
 
-	rpc.StopServer()
-	logger.Info("grpc server closed under request")
+	// Stop metrics server
+	if s.metricsServer != nil {
+		if err := s.metricsServer.Shutdown(context.Background()); err != nil {
+			logger.Errorf("metrics server failed to stop: %+v", err)
+		}
+		logger.Info("metrics server closed under request")
+	}
+
+	// Stop GRPC server
+	stopped := make(chan struct{})
+	go func() {
+		s.grpcServer.GracefulStop()
+		logger.Info("grpc server closed under request")
+		close(stopped)
+	}()
+
+	t := time.NewTimer(gracefulStopTimeout)
+	select {
+	case <-t.C:
+		s.grpcServer.Stop()
+	case <-stopped:
+		t.Stop()
+	}
 }

cdn/config/config.go

@@ -189,6 +189,9 @@ type BaseProperties struct {
 
 	// Host configuration
 	Host HostConfig `yaml:"host" mapstructure:"host"`
+
+	// Metrics configuration
+	Metrics *RestConfig `yaml:"metrics" mapstructure:"metrics"`
 }
 
 type ManagerConfig struct {
@@ -214,3 +217,7 @@ type HostConfig struct {
 	// IDC for scheduler
 	IDC string `mapstructure:"idc" yaml:"idc"`
 }
+
+type RestConfig struct {
+	Addr string `yaml:"addr" mapstructure:"addr"`
+}

cdn/metrics/metrics.go

@@ -0,0 +1,72 @@
+/*
+ *     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 metrics
+
+import (
+	"net/http"
+
+	"d7y.io/dragonfly/v2/cdn/config"
+	"d7y.io/dragonfly/v2/internal/constants"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"google.golang.org/grpc"
+)
+
+// Variables declared for metrics.
+var (
+	DownloadCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: constants.MetricsNamespace,
+		Subsystem: constants.CDNMetricsName,
+		Name:      "download_total",
+		Help:      "Counter of the number of the downloading.",
+	})
+
+	DownloadFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: constants.MetricsNamespace,
+		Subsystem: constants.CDNMetricsName,
+		Name:      "download_failure_total",
+		Help:      "Counter of the number of failed of the downloading.",
+	})
+
+	DownloadTraffic = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: constants.MetricsNamespace,
+		Subsystem: constants.CDNMetricsName,
+		Name:      "download_traffic",
+		Help:      "Counter of the number of download traffic.",
+	})
+
+	ConcurrentDownloadGauge = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: constants.MetricsNamespace,
+		Subsystem: constants.CDNMetricsName,
+		Name:      "concurrent_download_total",
+		Help:      "Gauger of the number of concurrent of the downloading.",
+	})
+)
+
+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,
+	}
+}

cdn/rpcserver/rpcserver.go

@@ -32,45 +32,34 @@ import (
 	"d7y.io/dragonfly/v2/internal/idgen"
 	"d7y.io/dragonfly/v2/pkg/rpc/base"
 	"d7y.io/dragonfly/v2/pkg/rpc/cdnsystem"
+	cdnserver "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/server"
 	"d7y.io/dragonfly/v2/pkg/util/digestutils"
 	"d7y.io/dragonfly/v2/pkg/util/net/iputils"
 	"d7y.io/dragonfly/v2/pkg/util/net/urlutils"
 	"d7y.io/dragonfly/v2/pkg/util/stringutils"
 	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc"
 )
 
-var tracer trace.Tracer
+var tracer = otel.Tracer("cdn-server")
 
-func init() {
-	tracer = otel.Tracer("cdn-server")
-}
-
-type options struct {
-	tracer trace.Tracer
-}
-
-type Option func(*options)
-
-func WithTracer(tracer trace.Tracer) Option {
-	return func(o *options) {
-		o.tracer = tracer
-	}
-}
-
-// CdnSeedServer is used to implement cdnsystem.SeederServer.
-type CdnSeedServer struct {
+type server struct {
+	*grpc.Server
 	taskMgr supervisor.SeedTaskMgr
 	cfg     *config.Config
 }
 
-// NewCdnSeedServer returns a new Manager Object.
-func NewCdnSeedServer(cfg *config.Config, taskMgr supervisor.SeedTaskMgr) (*CdnSeedServer, error) {
-	return &CdnSeedServer{
+// New returns a new Manager Object.
+func New(cfg *config.Config, taskMgr supervisor.SeedTaskMgr, opts ...grpc.ServerOption) (*grpc.Server, error) {
+	svr := &server{
 		taskMgr: taskMgr,
 		cfg:     cfg,
-	}, nil
+	}
+
+	svr.Server = cdnserver.New(svr, opts...)
+	return svr.Server, nil
 }
 
 func constructRegisterRequest(req *cdnsystem.SeedRequest) (*types.TaskRegisterRequest, error) {
@@ -114,7 +103,7 @@ func checkSeedRequestParams(req *cdnsystem.SeedRequest) error {
 	return nil
 }
 
-func (css *CdnSeedServer) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRequest, psc chan<- *cdnsystem.PieceSeed) (err error) {
+func (css *server) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRequest, psc chan<- *cdnsystem.PieceSeed) (err error) {
 	var span trace.Span
 	ctx, span = tracer.Start(ctx, config.SpanObtainSeeds, trace.WithSpanKind(trace.SpanKindServer))
 	defer span.End()
@@ -185,7 +174,7 @@ func (css *CdnSeedServer) ObtainSeeds(ctx context.Context, req *cdnsystem.SeedRe
 	return nil
 }
 
-func (css *CdnSeedServer) GetPieceTasks(ctx context.Context, req *base.PieceTaskRequest) (piecePacket *base.PiecePacket, err error) {
+func (css *server) GetPieceTasks(ctx context.Context, req *base.PieceTaskRequest) (piecePacket *base.PiecePacket, err error) {
 	var span trace.Span
 	ctx, span = tracer.Start(ctx, config.SpanGetPieceTasks, trace.WithSpanKind(trace.SpanKindServer))
 	defer span.End()

cdn/rpcserver/rpcserver_test.go

@@ -48,7 +48,7 @@ func TestCdnSeedServer_GetPieceTasks(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			css := &CdnSeedServer{
+			css := &server{
 				taskMgr: tt.fields.taskMgr,
 				cfg:     tt.fields.cfg,
 			}
@@ -84,7 +84,7 @@ func TestCdnSeedServer_ObtainSeeds(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			css := &CdnSeedServer{
+			css := &server{
 				taskMgr: tt.fields.taskMgr,
 				cfg:     tt.fields.cfg,
 			}
@@ -103,14 +103,14 @@ func TestNewCdnSeedServer(t *testing.T) {
 	tests := []struct {
 		name    string
 		args    args
-		want    *CdnSeedServer
+		want    *server
 		wantErr bool
 	}{
 		// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := NewCdnSeedServer(tt.args.cfg, tt.args.taskMgr)
+			got, err := New(tt.args.cfg, tt.args.taskMgr)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("NewCdnSeedServer() error = %v, wantErr %v", err, tt.wantErr)
 				return

cdn/supervisor/cdn/manager.go

@@ -136,7 +136,7 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 	// download fail
 	if err != nil {
 		downloadSpan.RecordError(err)
-		server.StatSeedFinish(task.TaskID, task.URL, false, err, start.Nanosecond(), time.Now().Nanosecond(), 0, 0)
+		server.StatSeedFinish(task.TaskID, task.URL, false, err, start, time.Now(), 0, 0)
 		seedTask.UpdateStatus(types.TaskInfoCdnStatusSourceError)
 		return seedTask, err
 	}
@@ -146,13 +146,13 @@ func (cm *Manager) TriggerCDN(ctx context.Context, task *types.SeedTask) (seedTa
 	// forth: write to storage
 	downloadMetadata, err := cm.writer.startWriter(ctx, reader, task, detectResult)
 	if err != nil {
-		server.StatSeedFinish(task.TaskID, task.URL, false, err, start.Nanosecond(), time.Now().Nanosecond(), downloadMetadata.backSourceLength,
+		server.StatSeedFinish(task.TaskID, task.URL, false, err, start, time.Now(), downloadMetadata.backSourceLength,
 			downloadMetadata.realSourceFileLength)
 		task.Log().Errorf("failed to write for task: %v", err)
 		seedTask.UpdateStatus(types.TaskInfoCdnStatusFailed)
 		return seedTask, err
 	}
-	server.StatSeedFinish(task.TaskID, task.URL, true, nil, start.Nanosecond(), time.Now().Nanosecond(), downloadMetadata.backSourceLength,
+	server.StatSeedFinish(task.TaskID, task.URL, true, nil, start, time.Now(), downloadMetadata.backSourceLength,
 		downloadMetadata.realSourceFileLength)
 	sourceDigest := reader.Digest()
 	// fifth: handle CDN result