replace hub servers with valid ones

change ITTT metrics
fix tracing for http store
remove unused launch params

cmd/blobcache.go

@@ -21,15 +21,8 @@ func init() {
 		Run:   blobcacheCmd,
 	}
 
-	cmd.Flags().IntVar(&tcpPeerPort, "tcp-peer-port", 5567, "The port reflector will distribute content from for the TCP (LBRY) protocol")
-	cmd.Flags().IntVar(&http3PeerPort, "http3-peer-port", 5568, "The port reflector will distribute content from over HTTP3 protocol")
-	cmd.Flags().IntVar(&httpPeerPort, "http-peer-port", 5569, "The port reflector will distribute content from over HTTP protocol")
-	cmd.Flags().IntVar(&receiverPort, "receiver-port", 5566, "The port reflector will receive content from")
 	cmd.Flags().IntVar(&metricsPort, "metrics-port", 2112, "The port reflector will use for prometheus metrics")
-
 	cmd.Flags().BoolVar(&disableBlocklist, "disable-blocklist", false, "Disable blocklist watching/updating")
-	cmd.Flags().IntVar(&requestQueueSize, "request-queue-size", 200, "How many concurrent requests from downstream should be handled at once (the rest will wait)")
-	cmd.Flags().StringVar(&upstreamEdgeToken, "upstream-edge-token", "", "token used to retrieve/authenticate protected content")
 
 	rootCmd.AddCommand(cmd)
 }

internal/metrics/metrics.go

@@ -69,6 +69,7 @@ const (
 	DirectionDownload = "download" // from reflector
 
 	LabelCacheType = "cache_type"
+	LabelOrigin    = "origin"
 	LabelComponent = "component"
 	LabelSource    = "source"
 
@@ -135,18 +136,12 @@ var (
 		Name:      "hit_total",
 		Help:      "Total number of blobs retrieved from the cache storage",
 	}, []string{LabelCacheType, LabelComponent})
-	ThisHitCount = promauto.NewCounter(prometheus.CounterOpts{
+	ItttHitCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: ns,
 		Subsystem: subsystemITTT,
-		Name:      "this_hit_total",
-		Help:      "Total number of blobs retrieved from the this storage",
-	})
-	ThatHitCount = promauto.NewCounter(prometheus.CounterOpts{
-		Namespace: ns,
-		Subsystem: subsystemITTT,
-		Name:      "that_hit_total",
-		Help:      "Total number of blobs retrieved from the that storage",
-	})
+		Name:      "hits_total",
+		Help:      "Total number of blobs retrieved from the this/that storage",
+	}, []string{LabelOrigin})
 	CacheMissCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: ns,
 		Subsystem: subsystemCache,
@@ -262,6 +257,11 @@ func CacheLabels(name, component string) prometheus.Labels {
 		LabelComponent: component,
 	}
 }
+func ItttLabels(orig string) prometheus.Labels {
+	return prometheus.Labels{
+		LabelOrigin: orig,
+	}
+}
 
 func TrackError(direction string, e error) (shouldLog bool) { // shouldLog is a hack, but whatever
 	if e == nil {

readme.md

@@ -96,7 +96,7 @@ Flags:
 ```
 ## Running from Source
 
-This project requires [Go v1.20](https://golang.org/doc/install).
+This project requires [Go v1.23](https://golang.org/doc/install).
 
 On Ubuntu you can install it with `sudo snap install go --classic`
 

reflector/blocklist.go

@@ -22,10 +22,10 @@ const blocklistURL = "https://api.lbry.com/file/list_blocked"
 
 func (s *Server) enableBlocklist(b store.Blocklister) {
 	walletServers := []string{
-		"spv25.lbry.com:50001",
-		"spv26.lbry.com:50001",
-		"spv19.lbry.com:50001",
-		"spv14.lbry.com:50001",
+		"a-hub1.odysee.com:50001",
+		"b-hub1.odysee.com:50001",
+		"c-hub1.odysee.com:50001",
+		"s-hub1.odysee.com:50001",
 	}
 
 	updateBlocklist(b, walletServers, s.grp.Ch())

server/http3/server.go

@@ -11,6 +11,7 @@ import (
 	"math/big"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -157,7 +158,7 @@ func generateTLSConfig() *tls.Config {
 
 func (s *Server) listenAndServe(server *http3.Server) {
 	err := server.ListenAndServe()
-	if err != nil && err != quic.ErrServerClosed {
+	if err != nil && !strings.Contains(err.Error(), "Server closed") {
 		log.Errorln(errors.FullTrace(err))
 	}
 }

store/http.go

@@ -74,23 +74,25 @@ func (c *HttpStore) Has(hash string) (bool, error) {
 }
 
 // Get downloads the blob using the http client
-func (c *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
+func (c *HttpStore) Get(hash string) (b stream.Blob, trace shared.BlobTrace, err error) {
 	log.Debugf("Getting %s from HTTP(s) source", hash[:8])
 	start := time.Now()
+
 	defer func(t time.Time) {
 		log.Debugf("Getting %s from HTTP(s) source took %s", hash[:8], time.Since(t).String())
+		trace = trace.Stack(time.Since(start), c.Name())
 	}(start)
 
 	url := c.endpoint + c.shardedPath(hash)
 	req, err := http.NewRequest(http.MethodGet, url, nil)
 	if err != nil {
-		return nil, shared.NewBlobTrace(time.Since(start), c.Name()), errors.Err(err)
+		return nil, trace, errors.Err(err)
 	}
 	req.Header.Add("User-Agent", "reflector.go/"+meta.Version())
 
 	res, err := c.httpClient.Do(req)
 	if err != nil {
-		return nil, shared.NewBlobTrace(time.Since(start), c.Name()), errors.Err(err)
+		return nil, trace, errors.Err(err)
 	}
 	defer func(Body io.ReadCloser) {
 		err := Body.Close()
@@ -99,21 +101,18 @@ func (c *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 		}
 	}(res.Body)
 
+	// Parse Via header if present
 	viaHeader := res.Header.Get("Via")
-	var trace shared.BlobTrace
 	if viaHeader != "" {
 		parsedTrace, err := shared.Deserialize(viaHeader)
-		if err != nil {
-			return nil, shared.NewBlobTrace(time.Since(start), c.Name()), err
-		}
+		if err == nil {
 			trace = *parsedTrace
-	} else {
-		trace = shared.NewBlobTrace(0, c.Name())
+		}
 	}
 
 	switch res.StatusCode {
 	case http.StatusNotFound:
-		return nil, trace.Stack(time.Since(start), c.Name()), ErrBlobNotFound
+		return nil, trace, ErrBlobNotFound
 	case http.StatusOK:
 		contentLength := res.Header.Get("Content-Length")
 		if contentLength != "" {
@@ -123,7 +122,7 @@ func (c *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 				_, err = io.ReadFull(res.Body, blob)
 				if err == nil {
 					metrics.MtrInBytesHttp.Add(float64(size))
-					return blob, trace.Stack(time.Since(start), c.Name()), nil
+					return blob, trace, nil
 				}
 				log.Warnf("Error reading body with known size: %s", err.Error())
 			}
@@ -132,17 +131,16 @@ func (c *HttpStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 		buffer := getBuffer()
 		defer putBuffer(buffer)
 		if _, err := io.Copy(buffer, res.Body); err != nil {
-			return nil, trace.Stack(time.Since(start), c.Name()), errors.Err(err)
+			return nil, trace, errors.Err(err)
 		}
 		blob := make([]byte, buffer.Len())
 		copy(blob, buffer.Bytes())
 		metrics.MtrInBytesHttp.Add(float64(len(blob)))
-		return blob, trace.Stack(time.Since(start), c.Name()), nil
+		return blob, trace, nil
 	default:
 		body, _ := io.ReadAll(res.Body)
 		log.Warnf("Got status code %d (%s)", res.StatusCode, string(body))
-		return nil, trace.Stack(time.Since(start), c.Name()),
-			errors.Err("upstream error. Status code: %d (%s)", res.StatusCode, string(body))
+		return nil, trace, errors.Err("upstream error. Status code: %d (%s)", res.StatusCode, string(body))
 	}
 }
 

store/ittt.go

@@ -101,7 +101,7 @@ func (c *ITTTStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	start := time.Now()
 	blob, trace, err := c.this.Get(hash)
 	if err == nil {
-		metrics.ThisHitCount.Inc()
+		metrics.ItttHitCount.With(metrics.ItttLabels(c.this.Name())).Inc()
 		return blob, trace.Stack(time.Since(start), c.Name()), err
 	}
 
@@ -109,7 +109,7 @@ func (c *ITTTStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
 	if err != nil {
 		return nil, trace.Stack(time.Since(start), c.Name()), err
 	}
-	metrics.ThatHitCount.Inc()
+	metrics.ItttHitCount.With(metrics.ItttLabels(c.that.Name())).Inc()
 	return blob, trace.Stack(time.Since(start), c.Name()), nil
 }