Repository: dragonflyoss/Dragonfly2
Pull request #1559 (2022-08-12 06:24:23)

add application-level statistics
<!--- 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)
- [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.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.



Patch:
diff --git a/client/config/dfget.go b/client/config/dfget.go
index c567ef02..3c40ccc7 100644
--- a/client/config/dfget.go
+++ b/client/config/dfget.go
@@ -67,8 +67,8 @@ type ClientOption struct {
 	// Tag identify download task, it is available merely when md5 param not exist.
 	Tag string `yaml:"tag,omitempty" mapstructure:"tag,omitempty"`
 
-	// CallSystem system name that executes dfget.
-	CallSystem string `yaml:"callSystem,omitempty" mapstructure:"callSystem,omitempty"`
+	// Application application name that executes dfget.
+	Application string `yaml:"application,omitempty" mapstructure:"application,omitempty"`
 
 	// DaemonSock is daemon download socket path.
 	DaemonSock string `yaml:"daemonSock,omitempty" mapstructure:"daemon-sock,omitempty"`
diff --git a/client/config/dfget_darwin.go b/client/config/dfget_darwin.go
index 09cbc20d..f5a1347f 100644
--- a/client/config/dfget_darwin.go
+++ b/client/config/dfget_darwin.go
@@ -38,7 +38,7 @@ var dfgetConfig = ClientOption{
 	DigestMethod:      "",
 	DigestValue:       "",
 	Tag:               "",
-	CallSystem:        "",
+	Application:       "",
 	Pattern:           "",
 	Cacerts:           nil,
 	Filter:            "",
diff --git a/client/config/dfget_linux.go b/client/config/dfget_linux.go
index bc8c6493..2fb7f88e 100644
--- a/client/config/dfget_linux.go
+++ b/client/config/dfget_linux.go
@@ -36,7 +36,7 @@ var dfgetConfig = ClientOption{
 	DigestMethod:      "",
 	DigestValue:       "",
 	Tag:               "",
-	CallSystem:        "",
+	Application:       "",
 	Pattern:           "",
 	Cacerts:           nil,
 	Filter:            "",
diff --git a/client/config/headers.go b/client/config/headers.go
index 6b8ace04..9a27296d 100644
--- a/client/config/headers.go
+++ b/client/config/headers.go
@@ -21,7 +21,10 @@ const (
 	HeaderDragonflyPeer   = "X-Dragonfly-Peer"
 	HeaderDragonflyTask   = "X-Dragonfly-Task"
 	HeaderDragonflyRange  = "X-Dragonfly-Range"
-	HeaderDragonflyTag    = "X-Dragonfly-Tag"
+	// HeaderDragonflyTag different HeaderDragonflyTag for the same url will be divided into different P2P overlay
+	HeaderDragonflyTag = "X-Dragonfly-Tag"
+	// HeaderDragonflyApplication is used for statistics and traffic control
+	HeaderDragonflyApplication = "X-Dragonfly-Application"
 	// HeaderDragonflyRegistry is used for dynamic registry mirrors.
 	HeaderDragonflyRegistry = "X-Dragonfly-Registry"
 	// HeaderDragonflyObjectMetaDigest is used for digest of object storage.
diff --git a/client/config/peerhost.go b/client/config/peerhost.go
index 5834e2ce..6096477e 100644
--- a/client/config/peerhost.go
+++ b/client/config/peerhost.go
@@ -269,16 +269,17 @@ type ConcurrentOption struct {
 
 type ProxyOption struct {
 	// WARNING: when add more option, please update ProxyOption.unmarshal function
-	ListenOption    `mapstructure:",squash" yaml:",inline"`
-	BasicAuth       *BasicAuth      `mapstructure:"basicAuth" yaml:"basicAuth"`
-	DefaultFilter   string          `mapstructure:"defaultFilter" yaml:"defaultFilter"`
-	DefaultTag      string          `mapstructure:"defaultTag" yaml:"defaultTag"`
-	MaxConcurrency  int64           `mapstructure:"maxConcurrency" yaml:"maxConcurrency"`
-	RegistryMirror  *RegistryMirror `mapstructure:"registryMirror" yaml:"registryMirror"`
-	WhiteList       []*WhiteList    `mapstructure:"whiteList" yaml:"whiteList"`
-	ProxyRules      []*ProxyRule    `mapstructure:"proxies" yaml:"proxies"`
-	HijackHTTPS     *HijackConfig   `mapstructure:"hijackHTTPS" yaml:"hijackHTTPS"`
-	DumpHTTPContent bool            `mapstructure:"dumpHTTPContent" yaml:"dumpHTTPContent"`
+	ListenOption       `mapstructure:",squash" yaml:",inline"`
+	BasicAuth          *BasicAuth      `mapstructure:"basicAuth" yaml:"basicAuth"`
+	DefaultFilter      string          `mapstructure:"defaultFilter" yaml:"defaultFilter"`
+	DefaultTag         string          `mapstructure:"defaultTag" yaml:"defaultTag"`
+	DefaultApplication string          `mapstructure:"defaultApplication" yaml:"defaultApplication"`
+	MaxConcurrency     int64           `mapstructure:"maxConcurrency" yaml:"maxConcurrency"`
+	RegistryMirror     *RegistryMirror `mapstructure:"registryMirror" yaml:"registryMirror"`
+	WhiteList          []*WhiteList    `mapstructure:"whiteList" yaml:"whiteList"`
+	ProxyRules         []*ProxyRule    `mapstructure:"proxies" yaml:"proxies"`
+	HijackHTTPS        *HijackConfig   `mapstructure:"hijackHTTPS" yaml:"hijackHTTPS"`
+	DumpHTTPContent    bool            `mapstructure:"dumpHTTPContent" yaml:"dumpHTTPContent"`
 	// ExtraRegistryMirrors add more mirror for different ports
 	ExtraRegistryMirrors []*RegistryMirror `mapstructure:"extraRegistryMirrors" yaml:"extraRegistryMirrors"`
 }
@@ -361,6 +362,7 @@ func (p *ProxyOption) unmarshal(unmarshal func(in []byte, out any) (err error),
 		BasicAuth            *BasicAuth        `mapstructure:"basicAuth" yaml:"basicAuth"`
 		DefaultFilter        string            `mapstructure:"defaultFilter" yaml:"defaultFilter"`
 		DefaultTag           string            `mapstructure:"defaultTag" yaml:"defaultTag"`
+		DefaultApplication   string            `mapstructure:"defaultApplication" yaml:"defaultApplication"`
 		MaxConcurrency       int64             `mapstructure:"maxConcurrency" yaml:"maxConcurrency"`
 		RegistryMirror       *RegistryMirror   `mapstructure:"registryMirror" yaml:"registryMirror"`
 		WhiteList            []*WhiteList      `mapstructure:"whiteList" yaml:"whiteList"`
@@ -382,6 +384,7 @@ func (p *ProxyOption) unmarshal(unmarshal func(in []byte, out any) (err error),
 	p.MaxConcurrency = pt.MaxConcurrency
 	p.DefaultFilter = pt.DefaultFilter
 	p.DefaultTag = pt.DefaultTag
+	p.DefaultApplication = pt.DefaultApplication
 	p.BasicAuth = pt.BasicAuth
 	p.DumpHTTPContent = pt.DumpHTTPContent
 
diff --git a/client/daemon/proxy/proxy.go b/client/daemon/proxy/proxy.go
index 5a1d474e..8cb8267e 100644
--- a/client/daemon/proxy/proxy.go
+++ b/client/daemon/proxy/proxy.go
@@ -100,6 +100,9 @@ type Proxy struct {
 	// defaultTag is used when http request without X-Dragonfly-Tag Header
 	defaultTag string
 
+	// defaultApplication is used when http request without X-Dragonfly-Application Header
+	defaultApplication string
+
 	// defaultFilter is used for registering steam task
 	defaultPattern commonv1.Pattern
 
@@ -226,6 +229,14 @@ func WithDefaultTag(t string) Option {
 	}
 }
 
+// WithDefaultApplication sets default tag for http requests without X-Dragonfly-Application Header
+func WithDefaultApplication(t string) Option {
+	return func(p *Proxy) *Proxy {
+		p.defaultApplication = t
+		return p
+	}
+}
+
 // WithDefaultPattern sets default pattern for downloading
 func WithDefaultPattern(pattern commonv1.Pattern) Option {
 	return func(p *Proxy) *Proxy {
@@ -504,6 +515,7 @@ func (proxy *Proxy) newTransport(tlsConfig *tls.Config) http.RoundTripper {
 		transport.WithDefaultFilter(proxy.defaultFilter),
 		transport.WithDefaultPattern(proxy.defaultPattern),
 		transport.WithDefaultTag(proxy.defaultTag),
+		transport.WithDefaultApplication(proxy.defaultApplication),
 		transport.WithDumpHTTPContent(proxy.dumpHTTPContent),
 	)
 	return rt
@@ -518,6 +530,7 @@ func (proxy *Proxy) mirrorRegistry(w http.ResponseWriter, r *http.Request) {
 		transport.WithCondition(proxy.shouldUseDragonflyForMirror),
 		transport.WithDefaultFilter(proxy.defaultFilter),
 		transport.WithDefaultTag(proxy.defaultTag),
+		transport.WithDefaultApplication(proxy.defaultApplication),
 		transport.WithDumpHTTPContent(proxy.dumpHTTPContent),
 	)
 	if err != nil {
diff --git a/client/daemon/proxy/proxy_manager.go b/client/daemon/proxy/proxy_manager.go
index 4528a3f5..7d4faef6 100644
--- a/client/daemon/proxy/proxy_manager.go
+++ b/client/daemon/proxy/proxy_manager.go
@@ -80,6 +80,7 @@ func NewProxyManager(peerHost *schedulerv1.PeerHost, peerTaskManager peer.TaskMa
 		WithMaxConcurrency(proxyOption.MaxConcurrency),
 		WithDefaultFilter(proxyOption.DefaultFilter),
 		WithDefaultTag(proxyOption.DefaultTag),
+		WithDefaultApplication(proxyOption.DefaultApplication),
 		WithDefaultPattern(defaultPattern),
 		WithBasicAuth(proxyOption.BasicAuth),
 		WithDumpHTTPContent(proxyOption.DumpHTTPContent),
diff --git a/client/daemon/transport/transport.go b/client/daemon/transport/transport.go
index b2df5896..290ad226 100644
--- a/client/daemon/transport/transport.go
+++ b/client/daemon/transport/transport.go
@@ -76,6 +76,9 @@ type transport struct {
 	// defaultTag is used when http request without X-Dragonfly-Tag Header
 	defaultTag string
 
+	// defaultTag is used when http request without X-Dragonfly-Tag Header
+	defaultApplication string
+
 	// dumpHTTPContent indicates to dump http request header and response header
 	dumpHTTPContent bool
 
@@ -141,6 +144,14 @@ func WithDefaultTag(b string) Option {
 	}
 }
 
+// WithDefaultApplication sets default Application for http requests with X-Dragonfly-Application Header
+func WithDefaultApplication(b string) Option {
+	return func(rt *transport) *transport {
+		rt.defaultApplication = b
+		return rt
+	}
+}
+
 func WithDumpHTTPContent(b bool) Option {
 	return func(rt *transport) *transport {
 		rt.dumpHTTPContent = b
@@ -239,6 +250,7 @@ func (rt *transport) download(ctx context.Context, req *http.Request) (*http.Res
 	// Pick header's parameters
 	filter := nethttp.PickHeader(req.Header, config.HeaderDragonflyFilter, rt.defaultFilter)
 	tag := nethttp.PickHeader(req.Header, config.HeaderDragonflyTag, rt.defaultTag)
+	application := nethttp.PickHeader(req.Header, config.HeaderDragonflyApplication, rt.defaultApplication)
 
 	// Delete hop-by-hop headers
 	delHopHeaders(req.Header)
@@ -246,6 +258,7 @@ func (rt *transport) download(ctx context.Context, req *http.Request) (*http.Res
 	meta.Header = nethttp.HeaderToMap(req.Header)
 	meta.Tag = tag
 	meta.Filter = filter
+	meta.Application = application
 
 	body, attr, err := rt.peerTaskManager.StartStreamTask(
 		ctx,
diff --git a/client/dfget/dfget.go b/client/dfget/dfget.go
index bd5b505e..53450004 100644
--- a/client/dfget/dfget.go
+++ b/client/dfget/dfget.go
@@ -235,14 +235,14 @@ func newDownRequest(cfg *config.DfgetConfig, hdr map[string]string) *dfdaemonv1.
 		Limit:             float64(cfg.RateLimit.Limit),
 		DisableBackSource: cfg.DisableBackSource,
 		UrlMeta: &commonv1.UrlMeta{
-			Digest: cfg.Digest,
-			Tag:    cfg.Tag,
-			Range:  rg,
-			Filter: cfg.Filter,
-			Header: hdr,
+			Digest:      cfg.Digest,
+			Tag:         cfg.Tag,
+			Range:       rg,
+			Filter:      cfg.Filter,
+			Header:      hdr,
+			Application: cfg.Application,
 		},
 		Pattern:            cfg.Pattern,
-		Callsystem:         cfg.CallSystem,
 		Uid:                int64(basic.UserID),
 		Gid:                int64(basic.UserGroup),
 		KeepOriginalOffset: cfg.KeepOriginalOffset,
diff --git a/cmd/dfget/cmd/root.go b/cmd/dfget/cmd/root.go
index 777e175f..53dc2054 100644
--- a/cmd/dfget/cmd/root.go
+++ b/cmd/dfget/cmd/root.go
@@ -162,7 +162,7 @@ func init() {
 
 	flagSet.BoolP("show-progress", "b", dfgetConfig.ShowProgress, "Show progress bar, it conflicts with --console")
 
-	flagSet.String("callsystem", dfgetConfig.CallSystem, "The caller name which is mainly used for statistics and access control")
+	flagSet.String("application", dfgetConfig.Application, "The caller name which is mainly used for statistics and access control")
 
 	flagSet.String("daemon-sock", dfgetConfig.DaemonSock, "Download socket path of daemon. In linux, default value is /var/run/dfdaemon.sock, in macos(just for testing), default value is /tmp/dfdaemon.sock")
 
diff --git a/go.mod b/go.mod
index cd8c2b5a..ff918fa4 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.1
+	d7y.io/api v1.0.3
 	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 c29d4ca2..a9bf20c9 100644
--- a/go.sum
+++ b/go.sum
@@ -69,8 +69,8 @@ cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RX
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
 cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3fOKtUw0Xmo=
 cloud.google.com/go/storage v1.22.1/go.mod h1:S8N1cAStu7BOeFfE8KAQzmyyLkK8p/vmRq6kuBTW58Y=
-d7y.io/api v1.0.1 h1:FCtyOacd7hBk3H6TFyVBLW9cAlFaS8YyQ7LPcYdrBBY=
-d7y.io/api v1.0.1/go.mod h1:GFnWPZFe4DUW70aOQikRZF0pvXpbUwAsGSCAZFFitPo=
+d7y.io/api v1.0.3 h1:+TUP/AAPkC7+CIGYmJQpsTX/xMfq0oiM1uOcVeQpA6o=
+d7y.io/api v1.0.3/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/pkg/idgen/task_id.go b/pkg/idgen/task_id.go
index 7e39f5a8..0ebe7dc8 100644
--- a/pkg/idgen/task_id.go
+++ b/pkg/idgen/task_id.go
@@ -73,6 +73,10 @@ func taskID(url string, meta *commonv1.UrlMeta, ignoreRange bool) string {
 		data = append(data, meta.Tag)
 	}
 
+	if meta.Application != "" {
+		data = append(data, meta.Application)
+	}
+
 	return digest.SHA256FromStrings(data...)
 }
 
diff --git a/scheduler/metrics/metrics.go b/scheduler/metrics/metrics.go
index ba372252..fa9bf999 100644
--- a/scheduler/metrics/metrics.go
+++ b/scheduler/metrics/metrics.go
@@ -56,28 +56,28 @@ var (
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "register_peer_task_total",
 		Help:      "Counter of the number of the register peer task.",
-	}, []string{"tag"})
+	}, []string{"tag", "app"})
 
 	RegisterPeerTaskFailureCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "register_peer_task_failure_total",
 		Help:      "Counter of the number of failed of the register peer task.",
-	}, []string{"tag"})
+	}, []string{"tag", "app"})
 
 	DownloadCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "download_total",
 		Help:      "Counter of the number of the downloading.",
-	}, []string{"tag"})
+	}, []string{"tag", "app"})
 
 	DownloadFailureCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "download_failure_total",
 		Help:      "Counter of the number of failed of the downloading.",
-	}, []string{"tag", "type"})
+	}, []string{"tag", "app", "type"})
 
 	StatTaskCount = promauto.NewCounter(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
@@ -112,35 +112,35 @@ var (
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "leave_task_total",
 		Help:      "Counter of the number of the leaving task.",
-	}, []string{"tag"})
+	}, []string{"tag", "app"})
 
 	LeaveTaskFailureCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "leave_task_failure_total",
 		Help:      "Counter of the number of failed of the leaving task.",
-	}, []string{"tag"})
+	}, []string{"tag", "app"})
 
 	Traffic = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "traffic",
 		Help:      "Counter of the number of traffic.",
-	}, []string{"tag", "type"})
+	}, []string{"tag", "app", "type"})
 
 	PeerHostTraffic = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "peer_host_traffic",
 		Help:      "Counter of the number of per peer host traffic.",
-	}, []string{"tag", "traffic_type", "peer_host_id", "peer_host_ip"})
+	}, []string{"tag", "app", "traffic_type", "peer_host_id", "peer_host_ip"})
 
 	PeerTaskCounter = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: constants.MetricsNamespace,
 		Subsystem: constants.SchedulerMetricsName,
 		Name:      "peer_task_total",
 		Help:      "Counter of the number of peer task.",
-	}, []string{"tag", "type"})
+	}, []string{"tag", "app", "type"})
 
 	PeerTaskDownloadDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
 		Namespace: constants.MetricsNamespace,
@@ -148,7 +148,7 @@ var (
 		Name:      "peer_task_download_duration_milliseconds",
 		Help:      "Histogram of the time each peer task downloading.",
 		Buckets:   []float64{100, 200, 500, 1000, 1500, 2 * 1000, 3 * 1000, 5 * 1000, 10 * 1000, 20 * 1000, 60 * 1000, 120 * 1000, 300 * 1000},
-	}, []string{"tag"})
+	}, []string{"tag", "app"})
 
 	ConcurrentScheduleGauge = promauto.NewGauge(prometheus.GaugeOpts{
 		Namespace: constants.MetricsNamespace,
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 91564f55..06acdd9d 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -40,6 +40,9 @@ const (
 	// Default value of tag.
 	DefaultTag = "unknow"
 
+	//DefaultApplication default value of application
+	DefaultApplication = "unknown"
+
 	// Download tiny file timeout.
 	downloadTinyFileContextTimeout = 30 * time.Second
 )
@@ -110,6 +113,14 @@ func WithTag(tag string) PeerOption {
 	}
 }
 
+// WithApplication sets peer's Application.
+func WithApplication(application string) PeerOption {
+	return func(p *Peer) *Peer {
+		p.Application = application
+		return p
+	}
+}
+
 type Peer struct {
 	// ID is peer id.
 	ID string
@@ -117,6 +128,9 @@ type Peer struct {
 	// Tag is peer tag.
 	Tag string
 
+	// Application is peer application.
+	Application string
+
 	// Pieces is finished piece set.
 	Pieces set.SafeSet[*schedulerv1.PieceResult]
 
@@ -170,6 +184,7 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 	p := &Peer{
 		ID:               id,
 		Tag:              DefaultTag,
+		Application:      DefaultApplication,
 		Pieces:           set.NewSafeSet[*schedulerv1.PieceResult](),
 		FinishedPieces:   &bitset.BitSet{},
 		pieceCosts:       []int64{},
diff --git a/scheduler/resource/seed_peer.go b/scheduler/resource/seed_peer.go
index 8040ccc9..e3bee74a 100644
--- a/scheduler/resource/seed_peer.go
+++ b/scheduler/resource/seed_peer.go
@@ -31,8 +31,10 @@ import (
 )
 
 const (
-	// Default value of biz tag for seed peer.
+	// SeedTag Default value of tag label for seed peer.
 	SeedTag = "d7y/seed"
+	// SeedApplication Default value of application label for seed peer.
+	SeedApplication = "d7y/seed"
 )
 
 const (
@@ -163,7 +165,7 @@ func (s *seedPeer) initSeedPeer(task *Task, ps *cdnsystemv1.PieceSeed) (*Peer, e
 	}
 
 	// New seed peer.
-	peer = NewPeer(ps.PeerId, task, host, WithTag(SeedTag))
+	peer = NewPeer(ps.PeerId, task, host, WithTag(SeedTag), WithApplication(SeedApplication))
 	peer.Log.Info("new seed peer successfully")
 
 	// Store seed peer.
diff --git a/scheduler/rpcserver/rpcserver.go b/scheduler/rpcserver/rpcserver.go
index 3e44605b..1661c324 100644
--- a/scheduler/rpcserver/rpcserver.go
+++ b/scheduler/rpcserver/rpcserver.go
@@ -64,13 +64,18 @@ func (s *Server) RegisterPeerTask(ctx context.Context, req *schedulerv1.PeerTask
 	if req.UrlMeta.Tag != "" {
 		tag = req.UrlMeta.Tag
 	}
-	metrics.RegisterPeerTaskCount.WithLabelValues(tag).Inc()
+
+	application := resource.DefaultApplication
+	if req.UrlMeta.Application != "" {
+		application = req.UrlMeta.Application
+	}
+	metrics.RegisterPeerTaskCount.WithLabelValues(tag, application).Inc()
 
 	resp, err := s.service.RegisterPeerTask(ctx, req)
 	if err != nil {
-		metrics.RegisterPeerTaskFailureCount.WithLabelValues(tag).Inc()
+		metrics.RegisterPeerTaskFailureCount.WithLabelValues(tag, application).Inc()
 	} else {
-		metrics.PeerTaskCounter.WithLabelValues(tag, resp.SizeScope.String()).Inc()
+		metrics.PeerTaskCounter.WithLabelValues(tag, application, resp.SizeScope.String()).Inc()
 	}
 
 	return resp, err
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 99c07971..37b5958d 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -86,7 +86,7 @@ func (s *Service) RegisterPeerTask(ctx context.Context, req *schedulerv1.PeerTas
 		return nil, dferrors.New(commonv1.Code_SchedTaskStatusError, msg)
 	}
 	host := s.registerHost(ctx, req.PeerHost)
-	peer := s.registerPeer(ctx, req.PeerId, task, host, req.UrlMeta.Tag)
+	peer := s.registerPeer(ctx, req.PeerId, task, host, req.UrlMeta.Tag, req.UrlMeta.Application)
 	peer.Log.Infof("register peer task request: %#v %#v %#v", req, req.UrlMeta, req.HostLoad)
 
 	// When the peer registers for the first time and
@@ -321,9 +321,9 @@ func (s *Service) ReportPieceResult(stream schedulerv1.Scheduler_ReportPieceResu
 
 			// Collect peer host traffic metrics.
 			if s.config.Metrics != nil && s.config.Metrics.EnablePeerHost {
-				metrics.PeerHostTraffic.WithLabelValues(peer.Tag, metrics.PeerHostTrafficDownloadType, peer.Host.ID, peer.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
+				metrics.PeerHostTraffic.WithLabelValues(peer.Tag, peer.Application, metrics.PeerHostTrafficDownloadType, peer.Host.ID, peer.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
 				if parent, ok := s.resource.PeerManager().Load(piece.DstPid); ok {
-					metrics.PeerHostTraffic.WithLabelValues(peer.Tag, metrics.PeerHostTrafficUploadType, parent.Host.ID, parent.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
+					metrics.PeerHostTraffic.WithLabelValues(peer.Tag, peer.Application, metrics.PeerHostTrafficUploadType, parent.Host.ID, parent.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
 				} else {
 					peer.Log.Warnf("dst peer %s not found for piece %#v %#v", piece.DstPid, piece, piece.PieceInfo)
 				}
@@ -331,9 +331,9 @@ func (s *Service) ReportPieceResult(stream schedulerv1.Scheduler_ReportPieceResu
 
 			// Collect traffic metrics.
 			if piece.DstPid != "" {
-				metrics.Traffic.WithLabelValues(peer.Tag, metrics.TrafficP2PType).Add(float64(piece.PieceInfo.RangeSize))
+				metrics.Traffic.WithLabelValues(peer.Tag, peer.Application, metrics.TrafficP2PType).Add(float64(piece.PieceInfo.RangeSize))
 			} else {
-				metrics.Traffic.WithLabelValues(peer.Tag, metrics.TrafficBackToSourceType).Add(float64(piece.PieceInfo.RangeSize))
+				metrics.Traffic.WithLabelValues(peer.Tag, peer.Application, metrics.TrafficBackToSourceType).Add(float64(piece.PieceInfo.RangeSize))
 			}
 			continue
 		}
@@ -363,13 +363,13 @@ func (s *Service) ReportPeerResult(ctx context.Context, req *schedulerv1.PeerRes
 		logger.Error(msg)
 		return dferrors.New(commonv1.Code_SchedPeerNotFound, msg)
 	}
-	metrics.DownloadCount.WithLabelValues(peer.Tag).Inc()
+	metrics.DownloadCount.WithLabelValues(peer.Tag, peer.Application).Inc()
 
 	if !req.Success {
 		peer.Log.Errorf("report peer failed result: %s %#v", req.Code, req)
 		if peer.FSM.Is(resource.PeerStateBackToSource) {
 			s.createRecord(peer, storage.PeerStateBackToSourceFailed, req)
-			metrics.DownloadFailureCount.WithLabelValues(peer.Tag, metrics.DownloadFailureBackToSourceType).Inc()
+			metrics.DownloadFailureCount.WithLabelValues(peer.Tag, peer.Application, metrics.DownloadFailureBackToSourceType).Inc()
 
 			s.handleTaskFail(ctx, peer.Task, req.GetSourceError(), nil)
 			s.handlePeerFail(ctx, peer)
@@ -377,12 +377,12 @@ func (s *Service) ReportPeerResult(ctx context.Context, req *schedulerv1.PeerRes
 		}
 
 		s.createRecord(peer, storage.PeerStateFailed, req)
-		metrics.DownloadFailureCount.WithLabelValues(peer.Tag, metrics.DownloadFailureP2PType).Inc()
+		metrics.DownloadFailureCount.WithLabelValues(peer.Tag, peer.Application, metrics.DownloadFailureP2PType).Inc()
 
 		s.handlePeerFail(ctx, peer)
 		return nil
 	}
-	metrics.PeerTaskDownloadDuration.WithLabelValues(peer.Tag).Observe(float64(req.Cost))
+	metrics.PeerTaskDownloadDuration.WithLabelValues(peer.Tag, peer.Application).Observe(float64(req.Cost))
 
 	peer.Log.Infof("report peer result: %#v", req)
 	if peer.FSM.Is(resource.PeerStateBackToSource) {
@@ -426,7 +426,7 @@ func (s *Service) AnnounceTask(ctx context.Context, req *schedulerv1.AnnounceTas
 	task := resource.NewTask(taskID, req.Url, req.TaskType, req.UrlMeta)
 	task, _ = s.resource.TaskManager().LoadOrStore(task)
 	host := s.registerHost(ctx, req.PeerHost)
-	peer := s.registerPeer(ctx, peerID, task, host, req.UrlMeta.Tag)
+	peer := s.registerPeer(ctx, peerID, task, host, req.UrlMeta.Tag, req.UrlMeta.Application)
 	peer.Log.Infof("announce peer task request: %#v %#v %#v %#v", req, req.UrlMeta, req.PeerHost, req.PiecePacket)
 
 	// If the task state is not TaskStateSucceeded,
@@ -496,11 +496,11 @@ func (s *Service) LeaveTask(ctx context.Context, req *schedulerv1.PeerTarget) er
 		logger.Error(msg)
 		return dferrors.New(commonv1.Code_SchedPeerNotFound, msg)
 	}
-	metrics.LeaveTaskCount.WithLabelValues(peer.Tag).Inc()
+	metrics.LeaveTaskCount.WithLabelValues(peer.Tag, peer.Application).Inc()
 
 	peer.Log.Infof("leave task: %#v", req)
 	if err := peer.FSM.Event(resource.PeerEventLeave); err != nil {
-		metrics.LeaveTaskFailureCount.WithLabelValues(peer.Tag).Inc()
+		metrics.LeaveTaskFailureCount.WithLabelValues(peer.Tag, peer.Application).Inc()
 
 		msg := fmt.Sprintf("peer fsm event failed: %s", err.Error())
 		peer.Log.Error(msg)
@@ -572,11 +572,14 @@ func (s *Service) registerHost(ctx context.Context, rawHost *schedulerv1.PeerHos
 }
 
 // registerPeer creates a new peer or reuses a previous peer.
-func (s *Service) registerPeer(ctx context.Context, peerID string, task *resource.Task, host *resource.Host, tag string) *resource.Peer {
+func (s *Service) registerPeer(ctx context.Context, peerID string, task *resource.Task, host *resource.Host, tag, application string) *resource.Peer {
 	var options []resource.PeerOption
 	if tag != "" {
 		options = append(options, resource.WithTag(tag))
 	}
+	if application != "" {
+		options = append(options, resource.WithApplication(application))
+	}
 
 	peer, loaded := s.resource.PeerManager().LoadOrStore(resource.NewPeer(peerID, task, host, options...))
 	if !loaded {


Test patch:
diff --git a/client/config/peerhost_test.go b/client/config/peerhost_test.go
index 72a24c1e..19c50bc2 100644
--- a/client/config/peerhost_test.go
+++ b/client/config/peerhost_test.go
@@ -420,9 +420,10 @@ func TestPeerHostOption_Load(t *testing.T) {
 				Username: "foo",
 				Password: "bar",
 			},
-			DefaultFilter:  "baz",
-			DefaultTag:     "tag",
-			MaxConcurrency: 1,
+			DefaultFilter:      "baz",
+			DefaultTag:         "tag",
+			DefaultApplication: "application",
+			MaxConcurrency:     1,
 			RegistryMirror: &RegistryMirror{
 				Remote: &URL{
 					&url.URL{
diff --git a/client/config/testdata/config/daemon.yaml b/client/config/testdata/config/daemon.yaml
index 48c33faa..3d2d8969 100644
--- a/client/config/testdata/config/daemon.yaml
+++ b/client/config/testdata/config/daemon.yaml
@@ -126,6 +126,7 @@ proxy:
     password: "bar"
   defaultFilter: "baz"
   defaultTag: "tag"
+  defaultApplication: "application"
   maxConcurrency: 1
   security:
     insecure: true
diff --git a/scheduler/resource/peer_test.go b/scheduler/resource/peer_test.go
index b58c1462..6f050b0a 100644
--- a/scheduler/resource/peer_test.go
+++ b/scheduler/resource/peer_test.go
@@ -68,13 +68,14 @@ func TestPeer_NewPeer(t *testing.T) {
 			},
 		},
 		{
-			name:    "new peer with tag",
+			name:    "new peer with tag and application",
 			id:      mockPeerID,
-			options: []PeerOption{WithTag("foo")},
+			options: []PeerOption{WithTag("foo"), WithApplication("bar")},
 			expect: func(t *testing.T, peer *Peer, mockTask *Task, mockHost *Host) {
 				assert := assert.New(t)
 				assert.Equal(peer.ID, mockPeerID)
 				assert.Equal(peer.Tag, "foo")
+				assert.Equal(peer.Application, "bar")
 				assert.Equal(peer.Pieces.Len(), uint(0))
 				assert.Empty(peer.FinishedPieces)
 				assert.Equal(len(peer.PieceCosts()), 0)
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index 891a8fbb..357dac6c 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -2387,7 +2387,7 @@ func TestService_registerPeer(t *testing.T) {
 			mockPeer := resource.NewPeer(mockPeerID, mockTask, mockHost)
 
 			tc.mock(mockPeer, peerManager, res.EXPECT(), peerManager.EXPECT())
-			peer := svc.registerPeer(context.Background(), tc.req.PeerId, mockTask, mockHost, tc.req.UrlMeta.Tag)
+			peer := svc.registerPeer(context.Background(), tc.req.PeerId, mockTask, mockHost, tc.req.UrlMeta.Tag, tc.req.UrlMeta.Application)
 			tc.expect(t, peer)
 		})
 	}