feat: modify pool tests

caboose.go

@@ -55,6 +55,9 @@ type Config struct {
 	// PoolRefresh is the interval at which we refresh the pool of upstreams from the orchestrator.
 	PoolRefresh time.Duration
 
+	// PoolTargetSize is a baseline size for the pool - the pool will accept decrements in performance to reach maintain at least this size.
+	PoolTargetSize int
+
 	// MirrorFraction is what fraction of requests will be mirrored to another random node in order to track metrics / determine the current best nodes.
 	MirrorFraction float64
 
@@ -90,6 +93,7 @@ const defaultMirrorFraction = 0.01
 
 const DefaultOrchestratorEndpoint = "https://orchestrator.strn.pl/nodes/nearby?count=200"
 const DefaultPoolRefreshInterval = 5 * time.Minute
+const DefaultPoolTargetSize = 30
 
 // we cool off sending requests for a cid for a certain duration
 // if we've seen a certain number of failures for it already in a given duration.
@@ -131,6 +135,9 @@ func NewCaboose(config *Config) (*Caboose, error) {
 	if override := os.Getenv(BackendOverrideKey); len(override) > 0 {
 		config.OrchestratorOverride = strings.Split(override, ",")
 	}
+	if config.PoolTargetSize == 0 {
+		config.PoolTargetSize = DefaultPoolTargetSize
+	}
 
 	logger := newLogger(config)
 	c := Caboose{
@@ -188,14 +195,14 @@ func (c *Caboose) Fetch(ctx context.Context, path string, cb DataCallback) error
 	ctx, span := spanTrace(ctx, "Fetch", trace.WithAttributes(attribute.String("path", path)))
 	defer span.End()
 
-	return c.pool.fetchResourceWith(ctx, path, cb, c.getAffinity(ctx))
+	return c.pool.fetchResourceWith(ctx, path, cb, c.GetAffinity(ctx))
 }
 
 func (c *Caboose) Has(ctx context.Context, it cid.Cid) (bool, error) {
 	ctx, span := spanTrace(ctx, "Has", trace.WithAttributes(attribute.Stringer("cid", it)))
 	defer span.End()
 
-	blk, err := c.pool.fetchBlockWith(ctx, it, c.getAffinity(ctx))
+	blk, err := c.pool.fetchBlockWith(ctx, it, c.GetAffinity(ctx))
 	if err != nil {
 		return false, err
 	}
@@ -206,7 +213,7 @@ func (c *Caboose) Get(ctx context.Context, it cid.Cid) (blocks.Block, error) {
 	ctx, span := spanTrace(ctx, "Get", trace.WithAttributes(attribute.Stringer("cid", it)))
 	defer span.End()
 
-	blk, err := c.pool.fetchBlockWith(ctx, it, c.getAffinity(ctx))
+	blk, err := c.pool.fetchBlockWith(ctx, it, c.GetAffinity(ctx))
 	if err != nil {
 		return nil, err
 	}
@@ -218,14 +225,14 @@ func (c *Caboose) GetSize(ctx context.Context, it cid.Cid) (int, error) {
 	ctx, span := spanTrace(ctx, "GetSize", trace.WithAttributes(attribute.Stringer("cid", it)))
 	defer span.End()
 
-	blk, err := c.pool.fetchBlockWith(ctx, it, c.getAffinity(ctx))
+	blk, err := c.pool.fetchBlockWith(ctx, it, c.GetAffinity(ctx))
 	if err != nil {
 		return 0, err
 	}
 	return len(blk.RawData()), nil
 }
 
-func (c *Caboose) getAffinity(ctx context.Context) string {
+func (c *Caboose) GetAffinity(ctx context.Context) string {
 	// https://github.com/ipfs/bifrost-gateway/issues/53#issuecomment-1442732865
 	if affG := ctx.Value(gateway.ContentPathKey); affG != nil {
 		contentPath := affG.(ipath.Path).String()

fetcher.go

@@ -49,7 +49,6 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 	}
 	// if the context is already cancelled, there's nothing we can do here.
 	if ce := ctx.Err(); ce != nil {
-		fetchRequestContextErrorTotalMetric.WithLabelValues(resourceType, fmt.Sprintf("%t", errors.Is(ce, context.Canceled)), "fetchResource-init").Add(1)
 		return ce
 	}
 
@@ -117,19 +116,14 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 			ttfbMs = fb.Sub(start).Milliseconds()
 
 			cacheStatus := getCacheStatus(isCacheHit)
-			if isBlockRequest {
-				fetchSizeBlockMetric.Observe(float64(received))
-			} else {
+			if !isBlockRequest {
 				fetchSizeCarMetric.WithLabelValues("success").Observe(float64(received))
 			}
 			durationMs := response_success_end.Sub(start).Milliseconds()
 			fetchSpeedPerPeerSuccessMetric.WithLabelValues(resourceType, cacheStatus).Observe(float64(received) / float64(durationMs))
 			fetchCacheCountSuccessTotalMetric.WithLabelValues(resourceType, cacheStatus).Add(1)
 			// track individual block metrics separately
-			if isBlockRequest {
-				fetchTTFBPerBlockPerPeerSuccessMetric.WithLabelValues(cacheStatus).Observe(float64(ttfbMs))
-				fetchDurationPerBlockPerPeerSuccessMetric.WithLabelValues(cacheStatus).Observe(float64(response_success_end.Sub(start).Milliseconds()))
-			} else {
+			if !isBlockRequest {
 				ci := 0
 				for index, value := range carSizes {
 					if float64(received) < value {
@@ -142,20 +136,11 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 				fetchTTFBPerCARPerPeerSuccessMetric.WithLabelValues(cacheStatus, carSizeStr).Observe(float64(ttfbMs))
 				fetchDurationPerCarPerPeerSuccessMetric.WithLabelValues(cacheStatus).Observe(float64(response_success_end.Sub(start).Milliseconds()))
 			}
-
-			// update L1 server timings
-			updateSuccessServerTimingMetrics(respHeader.Values(servertiming.HeaderKey), resourceType, isCacheHit, durationMs, ttfbMs, received)
 		} else {
-			if isBlockRequest {
-				fetchDurationPerBlockPerPeerFailureMetric.Observe(float64(time.Since(start).Milliseconds()))
-			} else {
+			if !isBlockRequest {
 				fetchDurationPerCarPerPeerFailureMetric.Observe(float64(time.Since(start).Milliseconds()))
 				fetchSizeCarMetric.WithLabelValues("failure").Observe(float64(received))
 			}
-
-			if code == http.StatusBadGateway || code == http.StatusGatewayTimeout {
-				updateLassie5xxTime(respHeader.Values(servertiming.HeaderKey), resourceType)
-			}
 		}
 
 		if err == nil || !errors.Is(err, context.Canceled) {
@@ -195,7 +180,7 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 	defer cancel()
 	req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, reqUrl, nil)
 	if err != nil {
-		if recordIfContextErr(resourceType, reqCtx, "build-http-request") {
+		if isCtxError(reqCtx) {
 			return reqCtx.Err()
 		}
 		return err
@@ -219,11 +204,10 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 
 	var resp *http.Response
 	saturnCallsTotalMetric.WithLabelValues(resourceType).Add(1)
-	startReq := time.Now()
 
 	resp, err = p.config.Client.Do(req)
 	if err != nil {
-		if recordIfContextErr(resourceType, reqCtx, "send-http-request") {
+		if isCtxError(reqCtx) {
 			if errors.Is(err, context.Canceled) {
 				return reqCtx.Err()
 			}
@@ -239,7 +223,6 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 	}
 
 	respHeader = resp.Header
-	headerTTFBPerPeerMetric.WithLabelValues(resourceType, getCacheStatus(respHeader.Get(saturnCacheHitKey) == saturnCacheHit)).Observe(float64(time.Since(startReq).Milliseconds()))
 
 	defer resp.Body.Close()
 
@@ -303,7 +286,7 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 	_, _ = io.Copy(io.Discard, resp.Body)
 
 	if err != nil {
-		if recordIfContextErr(resourceType, reqCtx, "read-http-response") {
+		if isCtxError(reqCtx) {
 			if errors.Is(err, context.Canceled) {
 				return reqCtx.Err()
 			}
@@ -324,6 +307,7 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 
 	// trace-metrics
 	// request life-cycle metrics
+	saturnCallsSuccessTotalMetric.WithLabelValues(resourceType, getCacheStatus(isCacheHit)).Add(1)
 
 	fetchRequestSuccessTimeTraceMetric.WithLabelValues(resourceType, getCacheStatus(isCacheHit), "tcp_connection").Observe(float64(result.TCPConnection.Milliseconds()))
 	fetchRequestSuccessTimeTraceMetric.WithLabelValues(resourceType, getCacheStatus(isCacheHit), "tls_handshake").Observe(float64(result.TLSHandshake.Milliseconds()))
@@ -338,78 +322,18 @@ func (p *pool) fetchResource(ctx context.Context, from *Node, resource string, m
 	fetchRequestSuccessTimeTraceMetric.WithLabelValues(resourceType, getCacheStatus(isCacheHit), "start_transfer").
 		Observe(float64(result.StartTransfer.Milliseconds()))
 
-	saturnCallsSuccessTotalMetric.WithLabelValues(resourceType, getCacheStatus(isCacheHit)).Add(1)
-
 	from.RecordSuccess(start, float64(wrapped.firstByte.Sub(start).Milliseconds()), float64(received)/float64(response_success_end.Sub(start).Milliseconds()))
 
 	return nil
 }
 
-func recordIfContextErr(resourceType string, ctx context.Context, requestState string) bool {
+func isCtxError(ctx context.Context) bool {
 	if ce := ctx.Err(); ce != nil {
-		fetchRequestContextErrorTotalMetric.WithLabelValues(resourceType, fmt.Sprintf("%t", errors.Is(ce, context.Canceled)), requestState).Add(1)
 		return true
 	}
 	return false
 }
 
-func updateLassie5xxTime(timingHeaders []string, resourceType string) {
-	if len(timingHeaders) == 0 {
-		goLogger.Debug("no timing headers in request response.")
-		return
-	}
-
-	for _, th := range timingHeaders {
-		if subReqTiming, err := servertiming.ParseHeader(th); err == nil {
-			for _, m := range subReqTiming.Metrics {
-				switch m.Name {
-				case "shim_lassie_headers":
-					if m.Duration.Milliseconds() != 0 {
-						lassie5XXTimeMetric.WithLabelValues(resourceType).Observe(float64(m.Duration.Milliseconds()))
-					}
-					return
-				default:
-				}
-			}
-		}
-	}
-}
-
-// todo: refactor for dryness
-func updateSuccessServerTimingMetrics(timingHeaders []string, resourceType string, isCacheHit bool, totalTimeMs, ttfbMs int64, recieved int) {
-	if len(timingHeaders) == 0 {
-		goLogger.Debug("no timing headers in request response.")
-		return
-	}
-
-	for _, th := range timingHeaders {
-		if subReqTiming, err := servertiming.ParseHeader(th); err == nil {
-			for _, m := range subReqTiming.Metrics {
-				switch m.Name {
-				case "shim_lassie_headers":
-					if m.Duration.Milliseconds() != 0 && !isCacheHit {
-						fetchDurationPerPeerSuccessCacheMissTotalLassieMetric.WithLabelValues(resourceType).Observe(float64(m.Duration.Milliseconds()))
-					}
-
-				case "nginx":
-					// sanity checks
-					if totalTimeMs != 0 && ttfbMs != 0 && m.Duration.Milliseconds() != 0 {
-						fetchDurationPerPeerSuccessTotalL1NodeMetric.WithLabelValues(resourceType, getCacheStatus(isCacheHit)).Observe(float64(m.Duration.Milliseconds()))
-						networkTimeMs := totalTimeMs - m.Duration.Milliseconds()
-						if networkTimeMs > 0 {
-							s := float64(recieved) / float64(networkTimeMs)
-							fetchNetworkSpeedPerPeerSuccessMetric.WithLabelValues(resourceType).Observe(s)
-						}
-						networkLatencyMs := ttfbMs - m.Duration.Milliseconds()
-						fetchNetworkLatencyPeerSuccessMetric.WithLabelValues(resourceType).Observe(float64(networkLatencyMs))
-					}
-				default:
-				}
-			}
-		}
-	}
-}
-
 func getCacheStatus(isCacheHit bool) string {
 	if isCacheHit {
 		return "Cache-hit"

internal/util/harness.go

@@ -63,7 +63,7 @@ func BuildCabooseHarness(t *testing.T, n int, maxRetries int, opts ...HarnessOpt
 
 		Client:               saturnClient,
 		DoValidation:         false,
-		PoolRefresh:          time.Millisecond * 50,
+		PoolRefresh:          time.Second * 50,
 		MaxRetrievalAttempts: maxRetries,
 		Harness:              &state.State{},
 	}
@@ -78,6 +78,8 @@ func BuildCabooseHarness(t *testing.T, n int, maxRetries int, opts ...HarnessOpt
 	ch.Caboose = bs
 	ch.CabooseActiveNodes = conf.Harness.ActiveNodes.(*caboose.NodeRing)
 	ch.CabooseAllNodes = conf.Harness.AllNodes.(*caboose.NodeHeap)
+	ch.CaboosePool = conf.Harness.PoolController
+	ch.Config = conf
 	return ch
 }
 
@@ -87,11 +89,19 @@ type CabooseHarness struct {
 
 	CabooseActiveNodes *caboose.NodeRing
 	CabooseAllNodes    *caboose.NodeHeap
+	CaboosePool        state.PoolController
+	Config             *caboose.Config
 
 	gol      sync.Mutex
 	goodOrch bool
 }
 
+type NodeStats struct {
+	Start   time.Time
+	Latency float64
+	Size    float64
+}
+
 func (ch *CabooseHarness) RunFetchesForRandCids(n int) {
 	for i := 0; i < n; i++ {
 		randCid, _ := cid.V1Builder{Codec: uint64(multicodec.Raw), MhType: uint64(multicodec.Sha2_256)}.Sum([]byte{uint8(i)})
@@ -122,6 +132,23 @@ func (ch *CabooseHarness) FetchAndAssertSuccess(t *testing.T, ctx context.Contex
 	require.NotEmpty(t, blk)
 }
 
+func (ch *CabooseHarness) RecordSuccesses(t *testing.T, nodes []*caboose.Node, s NodeStats, n int) {
+	for _, node := range nodes {
+		s.Start = time.Now().Add(-time.Second * 5)
+		for i := 0; i < n; i++ {
+			node.RecordSuccess(s.Start, s.Latency, s.Size)
+		}
+	}
+}
+
+func (ch *CabooseHarness) RecordFailures(t *testing.T, nodes []*caboose.Node, n int) {
+	for _, node := range nodes {
+		for i := 0; i < n; i++ {
+			node.RecordFailure()
+		}
+	}
+}
+
 func (ch *CabooseHarness) FailNodesWithCode(t *testing.T, selectorF func(ep *Endpoint) bool, code int) {
 	for _, n := range ch.Endpoints {
 		if selectorF(n) {