Merge pull request PelicanPlatform#823 from haoming29/add-proj-metric

Add project to Prometheus file transfer metric

docs/pages/prometheus.mdx

@@ -135,6 +135,10 @@ All of the Pelican servers have the following metrics:
 
   Client’s group names in a space-separated list. If no groups are present, the tag variable data is null.
 
+  #### Label: `proj`
+
+  Client’s `User-Agent` header when requesting the file. This is used to label the project name that accesses the file.
+
   #### Label: `type`
 
   Label values:

metrics/xrootd_metrics.go

@@ -62,6 +62,7 @@ type (
 		Role                   string
 		Org                    string
 		Groups                 []string
+		Project                string
 	}
 
 	FileId struct {
@@ -234,17 +235,17 @@ var (
 	TransferReadvSegs = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "xrootd_transfer_readv_segments_count",
 		Help: "Number of segments in readv operations",
-	}, []string{"path", "ap", "dn", "role", "org"})
+	}, []string{"path", "ap", "dn", "role", "org", "proj"})
 
 	TransferOps = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "xrootd_transfer_operations_count",
 		Help: "Number of transfer operations performed",
-	}, []string{"path", "ap", "dn", "role", "org", "type"})
+	}, []string{"path", "ap", "dn", "role", "org", "proj", "type"})
 
 	TransferBytes = promauto.NewCounterVec(prometheus.CounterOpts{
 		Name: "xrootd_transfer_bytes",
 		Help: "Bytes of transfers",
-	}, []string{"path", "ap", "dn", "role", "org", "type"})
+	}, []string{"path", "ap", "dn", "role", "org", "proj", "type"})
 
 	Threads = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Name: "xrootd_sched_thread_count",
@@ -579,6 +580,7 @@ func HandlePacket(packet []byte) error {
 					"dn":   "",
 					"role": "",
 					"org":  "",
+					"proj": "",
 				}
 				var oldReadvSegs uint64 = 0
 				var oldReadOps uint32 = 0
@@ -596,6 +598,7 @@ func HandlePacket(packet []byte) error {
 						labels["dn"] = userRecord.Value().DN
 						labels["role"] = userRecord.Value().Role
 						labels["org"] = userRecord.Value().Org
+						labels["proj"] = userRecord.Value().Project
 					}
 					oldReadvSegs = xferRecord.Value().ReadvSegs
 					oldReadOps = xferRecord.Value().ReadOps
@@ -681,6 +684,7 @@ func HandlePacket(packet []byte) error {
 					"dn":   "",
 					"role": "",
 					"org":  "",
+					"proj": "",
 				}
 
 				if item != nil {
@@ -692,6 +696,7 @@ func HandlePacket(packet []byte) error {
 						labels["dn"] = userRecord.Value().DN
 						labels["role"] = userRecord.Value().Role
 						labels["org"] = userRecord.Value().Org
+						labels["proj"] = userRecord.Value().Project
 					}
 				}
 
@@ -743,8 +748,27 @@ func HandlePacket(packet []byte) error {
 		}
 	case 'g':
 		log.Debug("MonPacket: Received a g-stream packet")
+	case 'i':
+		log.Debug("MonPacket: Received an appinfo packet")
+		infoSize := uint32(header.Plen - 12)
+		log.Warn("Dictid in appinfo is ", dictid)
+		if xrdUserId, appinfo, err := GetSIDRest(packet[12 : 12+infoSize]); err == nil {
+			if userids.Has(xrdUserId) {
+				userId := userids.Get(xrdUserId).Value()
+				if sessions.Has(userId) {
+					existingRec := sessions.Get(userId).Value()
+					existingRec.Project = appinfo
+					sessions.Set(userId, existingRec, ttlcache.DefaultTTL)
+				} else {
+					sessions.Set(userId, UserRecord{Project: appinfo}, ttlcache.DefaultTTL)
+				}
+			}
+		} else {
+			return err
+		}
 	case 'u':
 		log.Debug("MonPacket: Received a user login packet")
+		log.Warn("Dictid in login is ", dictid)
 		infoSize := uint32(header.Plen - 12)
 		if xrdUserId, auth, err := GetSIDRest(packet[12 : 12+infoSize]); err == nil {
 			var record UserRecord
@@ -777,6 +801,7 @@ func HandlePacket(packet []byte) error {
 	case 'T':
 		log.Debug("MonPacket: Received a token info packet")
 		infoSize := uint32(header.Plen - 12)
+		log.Warn("Dictid in token is ", dictid)
 		if _, tokenauth, err := GetSIDRest(packet[12 : 12+infoSize]); err == nil {
 			userId, userRecord, err := ParseTokenAuth(tokenauth)
 			if err != nil {

metrics/xrootd_metrics_test.go

@@ -582,23 +582,23 @@ func TestHandlePacket(t *testing.T) {
 		expectedTransferReadvSegs := `
 		# HELP xrootd_transfer_readv_segments_count Number of segments in readv operations
 		# TYPE xrootd_transfer_readv_segments_count counter
-		xrootd_transfer_readv_segments_count{ap="",dn="",org="",path="/",role=""} 1000
+		xrootd_transfer_readv_segments_count{ap="",dn="",org="",path="/",proj="",role=""} 1000
 		`
 
 		expectedTransferOps := `
 		# HELP xrootd_transfer_operations_count Number of transfer operations performed
 		# TYPE xrootd_transfer_operations_count counter
-		xrootd_transfer_operations_count{ap="",dn="",org="",path="/",role="",type="read"} 120
-		xrootd_transfer_operations_count{ap="",dn="",org="",path="/",role="",type="readv"} 10
-		xrootd_transfer_operations_count{ap="",dn="",org="",path="/",role="",type="write"} 30
+		xrootd_transfer_operations_count{ap="",dn="",org="",path="/",proj="",role="",type="read"} 120
+		xrootd_transfer_operations_count{ap="",dn="",org="",path="/",proj="",role="",type="readv"} 10
+		xrootd_transfer_operations_count{ap="",dn="",org="",path="/",proj="",role="",type="write"} 30
 		`
 
 		expectedTransferBytes := `
 		# HELP xrootd_transfer_bytes Bytes of transfers
 		# TYPE xrootd_transfer_bytes counter
-		xrootd_transfer_bytes{ap="",dn="",org="",path="/",role="",type="read"} 10000
-		xrootd_transfer_bytes{ap="",dn="",org="",path="/",role="",type="readv"} 20000
-		xrootd_transfer_bytes{ap="",dn="",org="",path="/",role="",type="write"} 120
+		xrootd_transfer_bytes{ap="",dn="",org="",path="/",proj="",role="",type="read"} 10000
+		xrootd_transfer_bytes{ap="",dn="",org="",path="/",proj="",role="",type="readv"} 20000
+		xrootd_transfer_bytes{ap="",dn="",org="",path="/",proj="",role="",type="write"} 120
 		`
 
 		expectedTransferReadvSegsReader := strings.NewReader(expectedTransferReadvSegs)