get-batch: add Prometheus metrics

* x-moss:
  - start tracking plain objects vs archived files, and
  - throughput (counts + sizes), and
  - operational health (throttling, soft/hard errors)
* remove internal stats but keep CLI observability via CtlMsg/xact.Snap
* tests: fix to override max-soft-errors default
-------
* stats package:
  - group target metrics into 4 categories
  - add get-batch (num-requests, object and file counts/sizes, and more)

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

ais/test/moss_test.go

@@ -11,6 +11,7 @@ import (
 	"io"
 	"os"
 	"path"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -39,6 +40,7 @@ import (
 const (
 	mossMissingPrefix = "missing-"
 	mossMissingSuffix = ".nonexistent"
+	mossMissingRatio  = 3 // as in: every 3rd
 )
 
 type mossConfig struct {
@@ -190,6 +192,8 @@ func TestMoss(t *testing.T) {
 
 	t.Cleanup(stopMossJobs) // in re: ErrLimitedCoexistence
 
+	oconfig := tools.GetClusterConfig(t)
+
 	for _, test := range tests {
 		t.Run(test.name(), func(t *testing.T) {
 			m := ioContext{
@@ -209,6 +213,15 @@ func TestMoss(t *testing.T) {
 			tools.CreateBucket(t, proxyURL, m.bck, nil, true /*cleanup*/)
 			m.init(true /*cleanup*/)
 
+			if test.withMissing {
+				s := strconv.Itoa(numPlainObjs)
+				tools.SetClusterConfig(t, cos.StrKVs{"get_batch.max_soft_errs": s})
+				t.Cleanup(func() {
+					s := strconv.Itoa(oconfig.GetBatch.MaxSoftErrs)
+					tools.SetClusterConfig(t, cos.StrKVs{"get_batch.max_soft_errs": s})
+				})
+			}
+
 			if test.inputFormat == "" {
 				testMossPlainObjects(t, &m, &test, numPlainObjs)
 			} else {
@@ -250,11 +263,11 @@ func testMossPlainObjects(t *testing.T, m *ioContext, test *mossConfig, numObjs
 	// Inject missing objects if requested
 	if test.withMissing {
 		originalEntries := mossIn
-		mossIn = make([]apc.MossIn, 0, len(originalEntries)+len(originalEntries)/3)
+		mossIn = make([]apc.MossIn, 0, len(originalEntries)+len(originalEntries)/mossMissingRatio)
 
 		for i, entry := range originalEntries {
 			mossIn = append(mossIn, entry)
-			if i%3 == 0 {
+			if i%mossMissingRatio == 0 {
 				missingEntry := apc.MossIn{
 					ObjName: mossMissingPrefix + trand.String(8),
 				}
@@ -342,7 +355,7 @@ func testMossArchives(t *testing.T, m *ioContext, test *mossConfig, numArchives,
 	// Inject missing archive paths if requested
 	if test.withMissing {
 		for i := range mossIn {
-			if i%3 == 0 {
+			if i%mossMissingRatio == 0 {
 				mossIn[i].ArchPath = trand.String(8) + mossMissingSuffix
 			}
 		}

cmn/config.go

@@ -2547,7 +2547,7 @@ const (
 	numWarmupWorkersDisabled = -1
 	numWarmupWorkersDflt     = 2
 
-	getBatchSoftErrs = 6
+	GetBatchSoftErrsDflt = 6
 )
 
 func (c *GetBatchConf) WarmupWorkers() int {
@@ -2571,7 +2571,7 @@ func (c *GetBatchConf) Validate() error {
 		}
 	}
 	if c.MaxSoftErrs == 0 {
-		c.MaxSoftErrs = getBatchSoftErrs
+		c.MaxSoftErrs = GetBatchSoftErrsDflt
 	} else if c.MaxSoftErrs < 0 {
 		return fmt.Errorf("invalid get_batch.max_soft_errs=%d (expecting non-negative integer)", c.MaxSoftErrs)
 	}

stats/target_stats.go

@@ -31,24 +31,33 @@ import (
 //
 // all error counters must have "err_" prefix (see `errPrefix`)
 
+//
+// ais target metrics: groups 1 through 4 =====================
+//
+
+// 1. datapath (counters, sizes, latencies) and common errors
 const (
-	// KindCounter & KindSize - always incremented
+	// KindThroughput
+	GetThroughput = "get.bps" // bytes per second
+	PutThroughput = "put.bps" // ditto
 
-	LruEvictCount = "lru.evict.n"
-	LruEvictSize  = "lru.evict.size"
+	// same as above via `.cumulative`
+	GetSize = "get.size"
+	PutSize = "put.size"
 
-	CleanupStoreCount = "cleanup.store.n"
-	CleanupStoreSize  = "cleanup.store.size"
+	// common latencies
+	AppendLatency    = "append.ns"
+	GetRedirLatency  = "get.redir.ns"
+	PutRedirLatency  = "put.redir.ns"
+	HeadLatencyTotal = "head.ns.total"
 
+	// out-of-band
 	VerChangeCount = "ver.change.n"
 	VerChangeSize  = "ver.change.size"
 
-	// errors
+	// errors (note common prefix convention)
 	ErrPutCksumCount = errPrefix + "put.cksum.n"
-
-	ErrFSHCCount = errPrefix + "fshc.n"
-
-	ErrDloadCount = errPrefix + "dl.n"
+	ErrFSHCCount     = errPrefix + "fshc.n"
 
 	// IO errors (must have ioErrPrefix)
 	IOErrGetCount    = ioErrPrefix + "get.n"
@@ -69,50 +78,77 @@ const (
 	RatelimPutRetryCount        = "ratelim.retry.put.n"
 	RatelimPutRetryLatencyTotal = "ratelim.retry.put.ns.total"
 
-	AppendLatency     = "append.ns"
-	GetRedirLatency   = "get.redir.ns"
-	PutRedirLatency   = "put.redir.ns"
-	DloadLatencyTotal = "dl.ns.total"
-	HeadLatencyTotal  = "head.ns.total"
+	// compare w/ common `DeleteCount`
+	RemoteDeletedDelCount = core.RemoteDeletedDelCount
+)
+
+// 2. object metadata in memory
+const (
+	LcacheCollisionCount = core.LcacheCollisionCount
+	LcacheEvictedCount   = core.LcacheEvictedCount
+	LcacheErrCount       = core.LcacheErrCount
+	LcacheFlushColdCount = core.LcacheFlushColdCount
+)
+
+// 3. xactions (jobs)
+const (
+	// blob downloader
+	GetBlobSize = "getblob.size"
+
+	// LRU eviction
+	LruEvictCount = "lru.evict.n"
+	LruEvictSize  = "lru.evict.size"
 
-	// Dsort
+	// space cleanup
+	CleanupStoreCount = "cleanup.store.n"
+	CleanupStoreSize  = "cleanup.store.size"
+
+	// distributed sort (ext/dsort)
 	DsortCreationReqCount    = "dsort.creation.req.n"
 	DsortCreationRespCount   = "dsort.creation.resp.n"
 	DsortCreationRespLatency = "dsort.creation.resp.ns"
 	DsortExtractShardDskCnt  = "dsort.extract.shard.dsk.n"
 	DsortExtractShardMemCnt  = "dsort.extract.shard.mem.n"
 	DsortExtractShardSize    = "dsort.extract.shard.size" // uncompressed
 
-	// ETL
+	// ETL (ext/etl)
 	ETLInlineCount         = "etl.inline.n"
 	ETLInlineLatencyTotal  = "etl.inline.ns.total"
 	ETLInlineSize          = "etl.inline.size"
 	ETLOfflineCount        = "etl.offline.n"
 	ETLOfflineLatencyTotal = "etl.offline.ns.total"
 	ETLOfflineSize         = "etl.offline.size"
 
-	// Downloader
-	DloadSize = "dl.size"
-
-	// KindThroughput
-	GetThroughput = "get.bps" // bytes per second
-	PutThroughput = "put.bps" // ditto
+	// downloader (ext/dload)
+	// (not to confuse with blob downloader)
+	DloadSize         = "dl.size"
+	DloadLatencyTotal = "dl.ns.total"
+	ErrDloadCount     = errPrefix + "dl.n"
 
-	// same as above via `.cumulative`
-	GetSize = "get.size"
-	PutSize = "put.size"
+	// get-batch (x-moss)
+	GetBatchCount     = "getbatch.n"
+	GetBatchObjCount  = "getbatch.obj.n"
+	GetBatchFileCount = "getbatch.file.n"
+	GetBatchObjSize   = "getbatch.obj.size"
+	GetBatchFileSize  = "getbatch.file.size"
 
-	GetBlobSize = "getblob.size"
+	GetBatchRxWaitTotal   = "getbatch.rxwait.ns"
+	GetBatchThrottleTotal = "getbatch.throttle.ns"
 
-	// core
-	RemoteDeletedDelCount = core.RemoteDeletedDelCount // compare w/ common `DeleteCount`
+	ErrGetBatchCount     = errPrefix + "getbatch.n"
+	GetBatchSoftErrCount = errPrefix + "soft.getbatch.n"
+)
 
-	LcacheCollisionCount = core.LcacheCollisionCount
-	LcacheEvictedCount   = core.LcacheEvictedCount
-	LcacheErrCount       = core.LcacheErrCount
-	LcacheFlushColdCount = core.LcacheFlushColdCount
+// 4, streams (peer-to-peer long-lived connections)
+const (
+	_ = cos.StreamsOutObjCount
+	_ = cos.StreamsOutObjSize
+	_ = cos.StreamsInObjCount
+	_ = cos.StreamsInObjSize
+)
 
-	// variable label used for prometheus disk metrics
+// variable label used for prometheus disk metrics
+const (
 	diskMetricLabel = "disk"
 )
 
@@ -420,7 +456,7 @@ func (r *Trunner) RegMetrics(snode *meta.Snode) {
 		},
 	)
 
-	// streams
+	// streams: peer-to-peer long-lived connections
 	r.reg(snode, cos.StreamsOutObjCount, KindCounter,
 		&Extra{
 			Help: "intra-cluster streaming communications: number of sent objects",
@@ -442,6 +478,7 @@ func (r *Trunner) RegMetrics(snode *meta.Snode) {
 		},
 	)
 
+	// downloader (ext/dload)
 	r.reg(snode, DloadSize, KindSize,
 		&Extra{
 			Help:    "total downloaded size (bytes)",
@@ -569,6 +606,53 @@ func (r *Trunner) RegMetrics(snode *meta.Snode) {
 			Help: "number of times a LOM from cache was written to stable storage (core, internal)",
 		},
 	)
+
+	// get-batch (x-moss)
+	r.reg(snode, GetBatchCount, KindCounter,
+		&Extra{
+			Help: "total number of get-batch requests (work items)",
+		},
+	)
+	r.reg(snode, GetBatchObjCount, KindCounter,
+		&Extra{
+			Help: "get-batch: total number of whole objects retrieved and delivered via output archive",
+		},
+	)
+	r.reg(snode, GetBatchFileCount, KindCounter,
+		&Extra{
+			Help: "get-batch: total number of files extracted from shards and delivered via output archive",
+		},
+	)
+	r.reg(snode, GetBatchObjSize, KindSize,
+		&Extra{
+			Help: "get-batch: total cumulative size (bytes) of whole objects",
+		},
+	)
+	r.reg(snode, GetBatchFileSize, KindSize,
+		&Extra{
+			Help: "get-batch: total cumulative size (bytes) of archived files extracted from shards",
+		},
+	)
+	r.reg(snode, GetBatchRxWaitTotal, KindTotal,
+		&Extra{
+			Help: "get-batch: total cumulative time (nanoseconds) spent waiting to receive entries from peer targets",
+		},
+	)
+	r.reg(snode, GetBatchThrottleTotal, KindTotal,
+		&Extra{
+			Help: "get-batch: total cumulative time (nanoseconds) slept due to resource pressure",
+		},
+	)
+	r.reg(snode, GetBatchSoftErrCount, KindCounter,
+		&Extra{
+			Help: "get-batch: number of transient errors (retryable failures under configured limit)",
+		},
+	)
+	r.reg(snode, ErrGetBatchCount, KindCounter,
+		&Extra{
+			Help: "get-batch: number of hard errors including request failures and 429 rejections",
+		},
+	)
 }
 
 func (r *Trunner) RegDiskMetrics(snode *meta.Snode, disk string) {