Add a storetheindex delegated provider

head/head.go

@@ -150,6 +150,15 @@ func NewHead(ctx context.Context, options ...opts.Option) (*Head, chan Bootstrap
 		}
 		providerStore = hproviders.CombineProviders(providerStore, hproviders.AddProviderNotSupported(delegateProvider))
 	}
+	if cfg.StoreTheIndexAddr != "" {
+		log.Infof("will delegate to %v with timeout %v", cfg.StoreTheIndexAddr, cfg.DelegateTimeout)
+		stiProvider, err := hproviders.StoreTheIndexProvider(cfg.StoreTheIndexAddr, cfg.DelegateTimeout)
+		if err != nil {
+			return nil, nil, fmt.Errorf("failed to instantiate delegation client (%w)", err)
+		}
+		providerStore = hproviders.CombineProviders(providerStore, hproviders.AddProviderNotSupported(stiProvider))
+
+	}
 	dhtOpts = append(dhtOpts, dht.ProviderStore(providerStore))
 
 	dhtNode, err := dht.New(ctx, node, dhtOpts...)

head/opts/options.go

@@ -25,6 +25,7 @@ type Options struct {
 	Peerstore            peerstore.Peerstore
 	ProviderStoreBuilder ProviderStoreBuilderFunc
 	DelegateAddr         string
+	StoreTheIndexAddr    string
 	DelegateTimeout      time.Duration
 	RoutingTable         *kbucket.RoutingTable
 	EnableRelay          bool
@@ -101,6 +102,15 @@ func DelegateAddr(addr string) Option {
 	}
 }
 
+// StoreTheIndexAddr configures the Hydra Head to delegate routing also to this storetheindex addr.
+// Defaults to empty string which indicates no delegation.
+func StoreTheIndexAddr(addr string) Option {
+	return func(o *Options) error {
+		o.StoreTheIndexAddr = addr
+		return nil
+	}
+}
+
 // DelegateTimeout configures the Hydra Head timeout for delegate routing requests.
 func DelegateTimeout(timeout time.Duration) Option {
 	return func(o *Options) error {

hydra/hydra.go

@@ -61,6 +61,7 @@ type Options struct {
 	ProviderStore     string
 	DelegateAddr      string
 	DelegateTimeout   time.Duration
+	StoreTheIndexAddr string
 	GetPort           func() int
 	NHeads            int
 	ProtocolPrefix    protocol.ID
@@ -150,6 +151,7 @@ func NewHydra(ctx context.Context, options Options) (*Hydra, error) {
 			opts.BootstrapPeers(options.BootstrapPeers),
 			opts.DelegateAddr(options.DelegateAddr),
 			opts.DelegateTimeout(options.DelegateTimeout),
+			opts.StoreTheIndexAddr(options.StoreTheIndexAddr),
 		}
 		if options.EnableRelay {
 			hdOpts = append(hdOpts, opts.EnableRelay())

k8s/alasybil.yaml

@@ -49,7 +49,7 @@ spec:
             - name: HYDRA_DELEGATED_ROUTING_ADDR
               value: "http://127.0.0.1:9999/" # must be a valid URL
             - name: HYDRA_DELEGATED_ROUTING_TIMEOUT
-              value: "2"
+              value: "1000"
           image: libp2p/hydra-booster:master
           resources:
             limits:

main.go

@@ -44,6 +44,7 @@ func main() {
 	httpAPIAddr := flag.String("httpapi-addr", defaultHTTPAPIAddr, "Specify an IP and port to run the HTTP API server on")
 	delegateAddr := flag.String("delegate-addr", "", "API endpoint for delegated routing")
 	delegateTimeout := flag.Int("delegate-timeout", 0, "Timeout for delegated routing in seconds")
+	stiAddr := flag.String("store-the-index-addr", "", "StoreTheIndex API endpoint for delegated routing")
 	inmem := flag.Bool("mem", false, "Use an in-memory database. This overrides the -db option")
 	metricsAddr := flag.String("metrics-addr", defaultMetricsAddr, "Specify an IP and port to run Prometheus metrics and pprof HTTP server on")
 	enableRelay := flag.Bool("enable-relay", false, "Enable libp2p circuit relaying for this node (default false).")
@@ -119,6 +120,9 @@ func main() {
 	if *delegateTimeout == 0 {
 		*delegateTimeout = mustGetEnvInt("HYDRA_DELEGATED_ROUTING_TIMEOUT", 0)
 	}
+	if *stiAddr == "" {
+		*stiAddr = os.Getenv("HYDRA_STORE_THE_INDEX_ADDR")
+	}
 
 	// Allow short keys. Otherwise, we'll refuse connections from the bootsrappers and break the network.
 	// TODO: Remove this when we shut those bootstrappers down.
@@ -161,7 +165,8 @@ func main() {
 		PeerstorePath:     *pstorePath,
 		ProviderStore:     *providerStore,
 		DelegateAddr:      *delegateAddr,
-		DelegateTimeout:   time.Second * time.Duration(*delegateTimeout),
+		DelegateTimeout:   time.Millisecond * time.Duration(*delegateTimeout),
+		StoreTheIndexAddr: *stiAddr,
 		EnableRelay:       *enableRelay,
 		ProtocolPrefix:    protocol.ID(*protocolPrefix),
 		BucketSize:        *bucketSize,

metrics/definitions.go

@@ -55,6 +55,9 @@ var (
 	DelegatedFindProvs         = stats.Int64("delegated_find_provs_total", "Total delegated find provider attempts that were found locally, or not found locally and succeeded, failed or were discarded", stats.UnitDimensionless)
 	DelegatedFindProvsDuration = stats.Float64("delegated_find_provs_duration", "The time it took delegated find provider attempts from the network to succeed or fail because of timeout or completion", stats.UnitMilliseconds)
 
+	STIFindProvs         = stats.Int64("sti_find_provs_total", "Total store the index find provider attempts that were found locally, or not found locally and succeeded, failed or were discarded", stats.UnitDimensionless)
+	STIFindProvsDuration = stats.Float64("sti_find_provs_duration_nanoseconds", "The time it took storetheindex finds from the network to succeed or fail because of timeout or completion", stats.UnitSeconds)
+
 	AWSRequests              = stats.Int64("aws_requests", "Requests made to AWS", stats.UnitDimensionless)
 	AWSRequestDurationMillis = stats.Float64("aws_request_duration", "The time it took to make an AWS request and receive a response", stats.UnitMilliseconds)
 	AWSRequestRetries        = stats.Int64("aws_retries", "Retried requests to AWS", stats.UnitDimensionless)
@@ -157,6 +160,16 @@ var (
 		TagKeys:     []tag.Key{KeyName},
 		Aggregation: view.Sum(),
 	}
+	STIFindProvsView = &view.View{
+		Measure:     STIFindProvs,
+		TagKeys:     []tag.Key{KeyName, KeyStatus},
+		Aggregation: view.Sum(),
+	}
+	STIFindProvsDurationView = &view.View{
+		Measure:     STIFindProvsDuration,
+		TagKeys:     []tag.Key{KeyName, KeyStatus},
+		Aggregation: view.Sum(),
+	}
 	// DHT views
 	ReceivedMessagesView = &view.View{
 		Measure:     dhtmetrics.ReceivedMessages,
@@ -219,6 +232,8 @@ var DefaultViews = []*view.View{
 	UniquePeersView,
 	RoutingTableSizeView,
 	ProviderRecordsView,
+	STIFindProvsView,
+	STIFindProvsDurationView,
 	ProviderRecordsPerKeyView,
 	PrefetchesView,
 	PrefetchDurationMillisView,

providers/combined.go

@@ -80,7 +80,9 @@ func mergeAddrInfos(infos []peer.AddrInfo) []peer.AddrInfo {
 	}
 	var r []peer.AddrInfo
 	for k, v := range m {
-		r = append(r, peer.AddrInfo{ID: k, Addrs: v})
+		if k.Validate() == nil {
+			r = append(r, peer.AddrInfo{ID: k, Addrs: v})
+		}
 	}
 	return r
 }

providers/storetheindex.go

@@ -0,0 +1,52 @@
+package providers
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-kad-dht/providers"
+	"github.com/libp2p/hydra-booster/metrics"
+	"github.com/libp2p/hydra-booster/providers/storetheindex"
+	"github.com/multiformats/go-multihash"
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
+)
+
+func StoreTheIndexProvider(endpointURL string, timeout time.Duration) (providers.ProviderStore, error) {
+	c, err := storetheindex.New(endpointURL, storetheindex.WithHTTPClient(&http.Client{Timeout: timeout}))
+	if err != nil {
+		return nil, err
+	}
+	return &StoreTheIndexProviderStore{c: c}, nil
+}
+
+type StoreTheIndexProviderStore struct {
+	c storetheindex.Client
+}
+
+func (s *StoreTheIndexProviderStore) AddProvider(ctx context.Context, key []byte, prov peer.AddrInfo) error {
+	return fmt.Errorf("adding providers not supported")
+}
+
+func (s *StoreTheIndexProviderStore) GetProviders(ctx context.Context, key []byte) ([]peer.AddrInfo, error) {
+	h, err := multihash.Cast(key)
+	if err != nil {
+		return nil, err
+	}
+	t0 := time.Now()
+	infos, err := s.c.FindProviders(ctx, h)
+	dur := time.Now().Sub(t0)
+	recordSTIFindProvsComplete(ctx, statusFromErr(err), metrics.STIFindProvsDuration.M(float64(dur)))
+	return infos, err
+}
+
+func recordSTIFindProvsComplete(ctx context.Context, status string, extraMeasures ...stats.Measurement) {
+	stats.RecordWithTags(
+		ctx,
+		[]tag.Mutator{tag.Upsert(metrics.KeyStatus, status)},
+		append([]stats.Measurement{metrics.STIFindProvs.M(1)}, extraMeasures...)...,
+	)
+}

providers/storetheindex/client.go

@@ -0,0 +1,42 @@
+package storetheindex
+
+import (
+	"context"
+	"net/http"
+	"net/url"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multihash"
+)
+
+type Client interface {
+	FindProviders(ctx context.Context, mh multihash.Multihash) ([]peer.AddrInfo, error)
+}
+
+type Option func(*client) error
+
+type client struct {
+	client   *http.Client
+	endpoint *url.URL
+}
+
+func WithHTTPClient(hc *http.Client) Option {
+	return func(c *client) error {
+		c.client = hc
+		return nil
+	}
+}
+
+func New(endpoint string, opts ...Option) (*client, error) {
+	u, err := url.Parse(endpoint)
+	if err != nil {
+		return nil, err
+	}
+	c := &client{endpoint: u, client: http.DefaultClient}
+	for _, o := range opts {
+		if err := o(c); err != nil {
+			return nil, err
+		}
+	}
+	return c, nil
+}

providers/storetheindex/findproviders.go

@@ -0,0 +1,72 @@
+package storetheindex
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+
+	logging "github.com/ipfs/go-log"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/multiformats/go-multihash"
+)
+
+var logger = logging.Logger("sti/client")
+
+func (c *client) FindProviders(ctx context.Context, mh multihash.Multihash) ([]peer.AddrInfo, error) {
+	// encode request in URL
+	u := fmt.Sprint(c.endpoint.String(), "/", mh.B58String())
+	httpReq, err := http.NewRequestWithContext(ctx, "GET", u, nil)
+	if err != nil {
+		return nil, err
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.client.Do(httpReq)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		if resp.StatusCode == http.StatusNotFound {
+			return []peer.AddrInfo{}, nil
+		}
+		return nil, fmt.Errorf("find query failed: %v", http.StatusText(resp.StatusCode))
+	}
+
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	parsedResponse := indexFindResponse{}
+	if err := json.Unmarshal(body, &parsedResponse); err != nil {
+		return nil, err
+	}
+
+	if len(parsedResponse.MultihashResults) != 1 {
+		return nil, fmt.Errorf("unexpected number of responses")
+	}
+	result := make([]peer.AddrInfo, len(parsedResponse.MultihashResults[0].ProviderResults))
+	for _, m := range parsedResponse.MultihashResults[0].ProviderResults {
+		result = append(result, m.Provider)
+	}
+
+	return result, nil
+}
+
+type indexFindResponse struct {
+	MultihashResults []indexMultihashResult
+}
+
+type indexMultihashResult struct {
+	Multihash       multihash.Multihash
+	ProviderResults []indexProviderResult
+}
+
+type indexProviderResult struct {
+	ContextID []byte
+	Metadata  json.RawMessage
+	Provider  peer.AddrInfo
+}