Skip to content

Commit

Permalink
remove kubo rpc router. move kubo rpc http transport configuration in…
Browse files Browse the repository at this point in the history
…to routing-v1
  • Loading branch information
aschmahmann committed Oct 6, 2023
1 parent 66626c5 commit 0c41709
Show file tree
Hide file tree
Showing 4 changed files with 27 additions and 173 deletions.
2 changes: 2 additions & 0 deletions handlers.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ import (
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

const DefaultKuboRPC = "http://127.0.0.1:5001"

func makeMetricsAndDebuggingHandler() *http.ServeMux {
mux := http.NewServeMux()

Expand Down
1 change: 1 addition & 0 deletions main.go
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ func main() {
RoutingV1: cctx.String("routing"),
KuboRPCURLs: getEnvs(EnvKuboRPC, DefaultKuboRPC),
DHTSharedHost: cctx.Bool("dht-fallback-shared-host"),
DNSCache: newCachedDNS(dnsCacheRefreshInterval),

Check warning on line 96 in main.go

View check run for this annotation

Codecov / codecov/patch

main.go#L94-L96

Added lines #L94 - L96 were not covered by tests
})
if err != nil {
return err
Expand Down
158 changes: 0 additions & 158 deletions routing.go

This file was deleted.

39 changes: 24 additions & 15 deletions setup.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import (
"github.com/libp2p/go-libp2p/p2p/net/connmgr"
"github.com/multiformats/go-multiaddr"
"go.opencensus.io/stats/view"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)

func init() {
Expand Down Expand Up @@ -87,6 +88,7 @@ type Config struct {
KuboRPCURLs []string
DHTSharedHost bool
DHTType DHTType
DNSCache *cachedDNS
}

func Setup(ctx context.Context, cfg *Config) (*Node, error) {
Expand Down Expand Up @@ -143,9 +145,29 @@ func Setup(ctx context.Context, cfg *Config) (*Node, error) {
var vs routing.ValueStore
var cr routing.ContentRouting

// Increase per-host connection pool since we are making lots of concurrent requests.
httpClient := &http.Client{
Transport: otelhttp.NewTransport(
&routingv1client.ResponseBodyLimitedTransport{
RoundTripper: &customTransport{
// Roundtripper with increased defaults than http.Transport such that retrieving
// multiple lookups concurrently is fast.
RoundTripper: &http.Transport{
MaxIdleConns: 1000,
MaxConnsPerHost: 100,
MaxIdleConnsPerHost: 100,
IdleConnTimeout: 90 * time.Second,
DialContext: cfg.DNSCache.dialWithCachedDNS,
ForceAttemptHTTP2: true,
},
},
LimitBytes: 1 << 20,
}),

Check warning on line 165 in setup.go

View check run for this annotation

Codecov / codecov/patch

setup.go#L144-L165

Added lines #L144 - L165 were not covered by tests
}

opts = append(opts, libp2p.Routing(func(h host.Host) (routing.PeerRouting, error) {
if cfg.RoutingV1 != "" {
routingClient, err := delegatedHTTPContentRouter(cfg.RoutingV1, routingv1client.WithStreamResultsRequired())
routingClient, err := delegatedHTTPContentRouter(cfg.RoutingV1, routingv1client.WithStreamResultsRequired(), routingv1client.WithHTTPClient(httpClient))
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -223,7 +245,7 @@ func Setup(ctx context.Context, cfg *Config) (*Node, error) {

// we want to also use the default HTTP routers, so wrap the FullRT client
// in a parallel router that calls them in parallel
httpRouters, err := delegatedHTTPContentRouter(ipniFallbackEndpoint)
httpRouters, err := delegatedHTTPContentRouter(ipniFallbackEndpoint, routingv1client.WithHTTPClient(httpClient))
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -327,22 +349,9 @@ func (b *bundledDHT) Bootstrap(ctx context.Context) error {
var _ routing.Routing = (*bundledDHT)(nil)

func delegatedHTTPContentRouter(endpoint string, rv1Opts ...routingv1client.Option) (routing.Routing, error) {
// Increase per-host connection pool since we are making lots of concurrent requests.
transport := http.DefaultTransport.(*http.Transport).Clone()
transport.MaxIdleConns = 500
transport.MaxIdleConnsPerHost = 100

delegateHTTPClient := &http.Client{
Transport: &routingv1client.ResponseBodyLimitedTransport{
RoundTripper: transport,
LimitBytes: 1 << 20,
},
}

cli, err := routingv1client.New(
endpoint,
append([]routingv1client.Option{
routingv1client.WithHTTPClient(delegateHTTPClient),
routingv1client.WithUserAgent(buildVersion()),
}, rv1Opts...)...,
)
Expand Down

0 comments on commit 0c41709

Please sign in to comment.