Repository: honeycombio/refinery
Pull request #491 (2022-08-29 17:09:42)

Added PeerManagement.Timeout config option
## Which problem is this PR solving?
This PR adds a new config option called `PeerManagement.Timeout` which will be used when communicating with peer discovery systems implementing the `Peers` interface. This PR also add support for `context.Context` when first connecting to the `Peers` implementation.

## Short description of the changes
- `NewPeers()` now accepts a `context.Context`
- Moved `NewPeers()` initialization after logging setup to take advantage of logs during any initial connect attempts. 
- Added `//go:build all || race` which is the new 1.18+ format for build tags (I can revert if not needed, my IDE auto fixed it 😄 )
- Fixed minor memory leak in `Router` where the `cancel()` function is never called.

## Note
I'm noting here that `context.Context` passed to `NewPeer()` is not as useful to redis because the redis dial function already has a retry and timeout loop. However, it will be useful for other concrete implementations of the `Peers` interface. If this PR is accepted, I wish to follow up with a PR that uses https://github.com/hashicorp/memberlist for peer discovery.


MikeGoldsmith: This will also need a rebase.
Baliedge: > This will also need a rebase to resolve conflicts.

Done.

Patch:
diff --git a/.gitignore b/.gitignore
index 91efd2d..5ad34c0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -17,3 +17,6 @@ test_redimem
 !/cmd/test_redimem
 
 dockerize*
+
+# IDE configs
+.idea/
diff --git a/cmd/refinery/main.go b/cmd/refinery/main.go
index 532929f..b2f6da9 100644
--- a/cmd/refinery/main.go
+++ b/cmd/refinery/main.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"net/http"
@@ -85,13 +86,6 @@ func main() {
 		os.Exit(1)
 	}
 
-	peers, err := peer.NewPeers(c)
-
-	if err != nil {
-		fmt.Printf("unable to load peers: %+v\n", err)
-		os.Exit(1)
-	}
-
 	// get desired implementation for each dependency to inject
 	lgr := logger.GetLoggerImplementation(c)
 	collector := collect.GetCollectorImplementation(c)
@@ -110,6 +104,15 @@ func main() {
 		os.Exit(1)
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), c.GetPeerTimeout())
+	defer cancel()
+	peers, err := peer.NewPeers(ctx, c)
+
+	if err != nil {
+		fmt.Printf("unable to load peers: %+v\n", err)
+		os.Exit(1)
+	}
+
 	// upstreamTransport is the http transport used to send things on to Honeycomb
 	upstreamTransport := &http.Transport{
 		Proxy: http.ProxyFromEnvironment,
diff --git a/config/config.go b/config/config.go
index 60f8989..29e5c0e 100644
--- a/config/config.go
+++ b/config/config.go
@@ -159,4 +159,6 @@ type Config interface {
 	GetGRPCTime() time.Duration
 
 	GetGRPCTimeout() time.Duration
+
+	GetPeerTimeout() time.Duration
 }
diff --git a/config/file_config.go b/config/file_config.go
index 8853830..74f210c 100644
--- a/config/file_config.go
+++ b/config/file_config.go
@@ -94,6 +94,7 @@ type PeerManagementConfig struct {
 	IdentifierInterfaceName string
 	UseIPV6Identifier       bool
 	RedisIdentifier         string
+	Timeout                 time.Duration
 }
 
 // GRPCServerParameters allow you to configure the GRPC ServerParameters used
@@ -127,6 +128,7 @@ func NewConfig(config, rules string, errorCallback func(error)) (Config, error)
 	c.SetDefault("PeerManagement.UseTLS", false)
 	c.SetDefault("PeerManagement.UseTLSInsecure", false)
 	c.SetDefault("PeerManagement.UseIPV6Identifier", false)
+	c.SetDefault("PeerManagement.Timeout", 5*time.Second)
 	c.SetDefault("HoneycombAPI", "https://api.honeycomb.io")
 	c.SetDefault("Logger", "logrus")
 	c.SetDefault("LoggingLevel", "debug")
@@ -852,3 +854,10 @@ func (f *fileConfig) GetGRPCTimeout() time.Duration {
 
 	return f.conf.GRPCServerParameters.Timeout
 }
+
+func (f *fileConfig) GetPeerTimeout() time.Duration {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return f.conf.PeerManagement.Timeout
+}
diff --git a/config/mock.go b/config/mock.go
index aa90acc..e17c5da 100644
--- a/config/mock.go
+++ b/config/mock.go
@@ -80,6 +80,7 @@ type MockConfig struct {
 	GRPCMaxConnectionAgeGrace     time.Duration
 	GRPCTime                      time.Duration
 	GRPCTimeout                   time.Duration
+	PeerTimeout                   time.Duration
 
 	Mux sync.RWMutex
 }
@@ -427,3 +428,10 @@ func (f *MockConfig) GetGRPCTimeout() time.Duration {
 
 	return f.GRPCTimeout
 }
+
+func (f *MockConfig) GetPeerTimeout() time.Duration {
+	f.Mux.RLock()
+	defer f.Mux.RUnlock()
+
+	return f.PeerTimeout
+}
diff --git a/config_complete.toml b/config_complete.toml
index 2fd4c02..fe9e240 100644
--- a/config_complete.toml
+++ b/config_complete.toml
@@ -212,6 +212,10 @@ Metrics = "honeycomb"
 # Not eligible for live reload. Overrides IdentifierInterfaceName, if both are set.
 # RedisIdentifier = "192.168.1.1"
 
+# Timeout is optional. By default, when using RedisHost, Refinery will timeout
+# after 5s when communicating with Redis.
+# Timeout = "5s"
+
 #########################
 ## In-Memory Collector ##
 #########################
diff --git a/go.mod b/go.mod
index 82d01e1..de45166 100644
--- a/go.mod
+++ b/go.mod
@@ -12,7 +12,7 @@ require (
 	github.com/gorilla/mux v1.8.0
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/honeycombio/dynsampler-go v0.2.1
-	github.com/honeycombio/husky v0.10.6
+	github.com/honeycombio/husky v0.15.0
 	github.com/honeycombio/libhoney-go v1.16.0
 	github.com/jessevdk/go-flags v1.5.0
 	github.com/json-iterator/go v1.1.12
@@ -26,7 +26,6 @@ require (
 	github.com/stretchr/testify v1.8.0
 	github.com/tidwall/gjson v1.14.3
 	github.com/vmihailenco/msgpack/v4 v4.3.11
-	go.opentelemetry.io/proto/otlp v0.11.0
 	golang.org/x/net v0.0.0-20220826154423-83b083e8dc8b // indirect
 	google.golang.org/grpc v1.49.0
 	google.golang.org/protobuf v1.28.1
@@ -44,7 +43,7 @@ require (
 	github.com/go-playground/locales v0.13.0 // indirect
 	github.com/go-playground/universal-translator v0.17.0 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.11.1 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/leodido/go-urn v1.2.0 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
diff --git a/go.sum b/go.sum
index ef04f2a..13ceeb8 100644
--- a/go.sum
+++ b/go.sum
@@ -45,7 +45,6 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
-github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -61,10 +60,6 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
-github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
-github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
-github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
-github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -73,7 +68,6 @@ github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.m
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.9.7/go.mod h1:cwu0lG7PUMfa9snN8LXBig5ynNVH9qI8YYLbd1fK2po=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a h1:yDWHCSQ40h88yih2JAcL6Ls/kVkSE8GFACTGVnMPruw=
 github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a/go.mod h1:7Ga40egUymuWXxAe151lTNnCv97MddSOVsjpPPkityA=
@@ -96,7 +90,6 @@ github.com/facebookgo/subset v0.0.0-20200203212716-c811ad88dec4/go.mod h1:5tD+ne
 github.com/frankban/quicktest v1.14.3 h1:FJKSZTDHjyhriyC81FLQ0LY93eSai0ZyR/ZIkd3ZUKE=
 github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI=
 github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
-github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
@@ -115,6 +108,7 @@ github.com/go-playground/validator v9.31.0+incompatible/go.mod h1:yrEkQXlcI+Pugk
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
+github.com/golang/glog v1.0.0 h1:nfP3RFugxnNRyKgeWd4oI1nYvXpxrx8ck8ZrcizshdQ=
 github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -178,8 +172,8 @@ github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5m
 github.com/googleapis/google-cloud-go-testing v0.0.0-20200911160855-bcd43fbb19e8/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
-github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4M0+kPpLofRdBo=
-github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.11.1 h1:/sDbPb60SusIXjiJGYLUoS/rAQurQmvGWmwn2bBPM9c=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.11.1/go.mod h1:G+WkljZi4mflcqVxYSgvt8MNctRQHjEH8ubKtt1Ka3w=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
@@ -188,8 +182,8 @@ github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/honeycombio/dynsampler-go v0.2.1 h1:IbhjbdB0IbLSZn7xVYuk6jjk/ZDk/EO+DJ5OXFZliv8=
 github.com/honeycombio/dynsampler-go v0.2.1/go.mod h1:BOeTUPT6fCRH5X/+QqF6Kza3IyLp9uSq/rWgEtI4aZI=
-github.com/honeycombio/husky v0.10.6 h1:jU/lXqo7Qz6e9eUJErIH3Lst2gjKWSJ4oAXYjFSXkn0=
-github.com/honeycombio/husky v0.10.6/go.mod h1:i69+3xApIcsQn9PPeFRndTOOw6edNG66TxTTd+L6oq0=
+github.com/honeycombio/husky v0.15.0 h1:RuYmKC1jHSB6y3C7pSXf889w8AaSzZ1JkIkLaiXlC5o=
+github.com/honeycombio/husky v0.15.0/go.mod h1:ZO2AzJnnJkUoXbIQ5qLCvQuWdQOG/BZEup8pf2y5ua0=
 github.com/honeycombio/libhoney-go v1.16.0 h1:kPpqoz6vbOzgp7jC6SR7SkNj7rua7rgxvznI6M3KdHc=
 github.com/honeycombio/libhoney-go v1.16.0/go.mod h1:izP4fbREuZ3vqC4HlCAmPrcPT9gxyxejRjGtCYpmBn0=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
@@ -271,7 +265,6 @@ github.com/prometheus/procfs v0.7.3 h1:4jVXhlkAyzOScmCkXBTOLRLTz8EeU+eyjrwB/EPq0
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 h1:MkV+77GLUNo5oJ0jf870itWm3D0Sjh7+Za9gazKc5LQ=
 github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
-github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
@@ -327,9 +320,6 @@ go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
-go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
-go.opentelemetry.io/proto/otlp v0.11.0 h1:cLDgIBTf4lLOlztkhzAEdQsJ4Lj+i5Wc9k6Nn0K1VyU=
-go.opentelemetry.io/proto/otlp v0.11.0/go.mod h1:QpEjXPrNQzrFDZgoTo49dgHR9RYRSrg3NAKnUGl9YpQ=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -590,7 +580,6 @@ google.golang.org/genproto v0.0.0-20200312145019-da6875a35672/go.mod h1:55QSHmfG
 google.golang.org/genproto v0.0.0-20200331122359-1ee6d9798940/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200430143042-b979b6f78d84/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200511104702-f5ebc3bea380/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
-google.golang.org/genproto v0.0.0-20200513103714-09dca8ec2884/go.mod h1:55QSHmfGQM9UVYDPBsyGGes0y52j32PQ3BqQfXhyH3c=
 google.golang.org/genproto v0.0.0-20200515170657-fc4c6c6a6587/go.mod h1:YsZOwe1myG/8QRHRsmBRE1LrgQY60beZKjly0O1fX9U=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
 google.golang.org/genproto v0.0.0-20200618031413-b414f8b61790/go.mod h1:jDfRM7FcilCzHH/e9qn6dsT145K34l5v+OpcnNgKAAA=
@@ -619,12 +608,9 @@ google.golang.org/grpc v1.29.1/go.mod h1:itym6AZVZYACWQqET3MqgPpjcuV5QH3BxFS3Iji
 google.golang.org/grpc v1.30.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.31.0/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
 google.golang.org/grpc v1.31.1/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM8pak=
-google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTpR3n0=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
 google.golang.org/grpc v1.34.0/go.mod h1:WotjhfgOW/POjDeRt8vscBtXq+2VjORFy659qA51WJ8=
 google.golang.org/grpc v1.35.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.42.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
 google.golang.org/grpc v1.49.0 h1:WTLtQzmQori5FUH25Pq4WT22oCsv8USpQ+F6rqtsmxw=
 google.golang.org/grpc v1.49.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
@@ -639,7 +625,6 @@ google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGj
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
 google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
@@ -656,7 +641,6 @@ gopkg.in/ini.v1 v1.66.4 h1:SsAcf+mM7mRZo2nJNGt8mZCjG8ZRaNGMURJw7BsIST4=
 gopkg.in/ini.v1 v1.66.4/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/internal/peer/peers.go b/internal/peer/peers.go
index ff5a261..c17000e 100644
--- a/internal/peer/peers.go
+++ b/internal/peer/peers.go
@@ -1,8 +1,8 @@
 package peer
 
 import (
+	"context"
 	"errors"
-
 	"github.com/honeycombio/refinery/config"
 )
 
@@ -13,7 +13,7 @@ type Peers interface {
 	RegisterUpdatedPeersCallback(callback func())
 }
 
-func NewPeers(c config.Config) (Peers, error) {
+func NewPeers(ctx context.Context, c config.Config) (Peers, error) {
 	t, err := c.GetPeerManagementType()
 
 	if err != nil {
@@ -24,8 +24,8 @@ func NewPeers(c config.Config) (Peers, error) {
 	case "file":
 		return newFilePeers(c), nil
 	case "redis":
-		return newRedisPeers(c)
+		return newRedisPeers(ctx, c)
 	default:
-		return nil, errors.New("Invalid PeerManagement Type")
+		return nil, errors.New("invalid config option 'PeerManagement.Type'")
 	}
 }
diff --git a/internal/peer/redis.go b/internal/peer/redis.go
index 4507e17..041433a 100644
--- a/internal/peer/redis.go
+++ b/internal/peer/redis.go
@@ -45,7 +45,7 @@ type redisPeers struct {
 }
 
 // NewRedisPeers returns a peers collection backed by redis
-func newRedisPeers(c config.Config) (Peers, error) {
+func newRedisPeers(ctx context.Context, c config.Config) (Peers, error) {
 	redisHost, _ := c.GetRedisHost()
 
 	if redisHost == "" {
@@ -101,9 +101,9 @@ func newRedisPeers(c config.Config) (Peers, error) {
 	}
 
 	// register myself once
-	err = peers.store.Register(context.TODO(), address, peerEntryTimeout)
+	err = peers.store.Register(ctx, address, peerEntryTimeout)
 	if err != nil {
-		logrus.WithError(err).Errorf("failed to register self with peer store")
+		logrus.WithError(err).Errorf("failed to register self with redis peer store")
 		return nil, err
 	}
 
@@ -139,15 +139,26 @@ func (p *redisPeers) RegisterUpdatedPeersCallback(cb func()) {
 func (p *redisPeers) registerSelf() {
 	tk := time.NewTicker(refreshCacheInterval)
 	for range tk.C {
-		// every 5 seconds, insert a 30sec timeout record
-		p.store.Register(context.TODO(), p.publicAddr, peerEntryTimeout)
+		ctx, cancel := context.WithTimeout(context.Background(), p.c.GetPeerTimeout())
+		// every 5 seconds, insert a 30sec timeout record. we ignore the error
+		// here since Register() logs the error for us.
+		p.store.Register(ctx, p.publicAddr, peerEntryTimeout)
+		cancel()
 	}
 }
 
 func (p *redisPeers) updatePeerListOnce() {
-	currentPeers, err := p.store.GetMembers(context.TODO())
+	ctx, cancel := context.WithTimeout(context.Background(), p.c.GetPeerTimeout())
+	defer cancel()
+
+	currentPeers, err := p.store.GetMembers(ctx)
 	if err != nil {
-		// TODO maybe do something better here?
+		logrus.WithError(err).
+			WithFields(logrus.Fields{
+				"name":    p.publicAddr,
+				"timeout": p.c.GetPeerTimeout().String(),
+			}).
+			Error("get members failed")
 		return
 	}
 	sort.Strings(currentPeers)
@@ -163,11 +174,21 @@ func (p *redisPeers) watchPeers() {
 	tk := time.NewTicker(refreshCacheInterval)
 
 	for range tk.C {
-		currentPeers, err := p.store.GetMembers(context.TODO())
+		ctx, cancel := context.WithTimeout(context.Background(), p.c.GetPeerTimeout())
+		currentPeers, err := p.store.GetMembers(ctx)
+		cancel()
+
 		if err != nil {
-			// TODO maybe do something better here?
+			logrus.WithError(err).
+				WithFields(logrus.Fields{
+					"name":     p.publicAddr,
+					"timeout":  p.c.GetPeerTimeout().String(),
+					"oldPeers": oldPeerList,
+				}).
+				Error("get members failed during watch")
 			continue
 		}
+
 		sort.Strings(currentPeers)
 		if !equal(oldPeerList, currentPeers) {
 			// update peer list and trigger callbacks saying the peer list has changed
diff --git a/route/otlp_trace.go b/route/otlp_trace.go
index 6b75b29..f4e86c4 100644
--- a/route/otlp_trace.go
+++ b/route/otlp_trace.go
@@ -8,7 +8,7 @@ import (
 	huskyotlp "github.com/honeycombio/husky/otlp"
 	"github.com/honeycombio/refinery/types"
 
-	collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	collectortrace "github.com/honeycombio/husky/proto/otlp/collector/trace/v1"
 )
 
 func (router *Router) postOTLP(w http.ResponseWriter, req *http.Request) {
diff --git a/route/route.go b/route/route.go
index 18af0b0..8727ac6 100644
--- a/route/route.go
+++ b/route/route.go
@@ -39,7 +39,7 @@ import (
 	"github.com/honeycombio/refinery/transmit"
 	"github.com/honeycombio/refinery/types"
 
-	collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	collectortrace "github.com/honeycombio/husky/proto/otlp/collector/trace/v1"
 )
 
 const (
@@ -248,6 +248,7 @@ func (r *Router) LnS(incomingOrPeer string) {
 func (r *Router) Stop() error {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
 	defer cancel()
+
 	err := r.server.Shutdown(ctx)
 	if err != nil {
 		return err


Test patch:
diff --git a/app/app_test.go b/app/app_test.go
index 401bf5a..8b17176 100644
--- a/app/app_test.go
+++ b/app/app_test.go
@@ -1,3 +1,6 @@
+//go:build all || race
+// +build all race
+
 package app
 
 import (
@@ -122,7 +125,7 @@ func newStartedApp(
 
 	var err error
 	if peers == nil {
-		peers, err = peer.NewPeers(c)
+		peers, err = peer.NewPeers(context.Background(), c)
 		assert.NoError(t, err)
 	}
 
diff --git a/internal/peer/peers_test.go b/internal/peer/peers_test.go
index b411477..5e6917e 100644
--- a/internal/peer/peers_test.go
+++ b/internal/peer/peers_test.go
@@ -1,20 +1,27 @@
+//go:build all || race
+// +build all race
+
 package peer
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/honeycombio/refinery/config"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestNewPeers(t *testing.T) {
 	c := &config.MockConfig{
 		PeerManagementType: "file",
+		PeerTimeout:        5 * time.Second,
 	}
 
-	p, err := NewPeers(c)
-
-	assert.Equal(t, nil, err)
+	p, err := NewPeers(context.Background(), c)
+	assert.NoError(t, err)
+	require.NotNil(t, p)
 
 	switch i := p.(type) {
 	case *filePeers:
@@ -25,11 +32,12 @@ func TestNewPeers(t *testing.T) {
 	c = &config.MockConfig{
 		GetPeerListenAddrVal: "0.0.0.0:8081",
 		PeerManagementType:   "redis",
+		PeerTimeout:          5 * time.Second,
 	}
 
-	p, err = NewPeers(c)
-
-	assert.Equal(t, nil, err)
+	p, err = NewPeers(context.Background(), c)
+	assert.NoError(t, err)
+	require.NotNil(t, p)
 
 	switch i := p.(type) {
 	case *redisPeers:
diff --git a/route/otlp_trace_test.go b/route/otlp_trace_test.go
index 1e7e875..bd1adde 100644
--- a/route/otlp_trace_test.go
+++ b/route/otlp_trace_test.go
@@ -12,17 +12,18 @@ import (
 	"time"
 
 	huskyotlp "github.com/honeycombio/husky/otlp"
+	collectortrace "github.com/honeycombio/husky/proto/otlp/collector/trace/v1"
+	common "github.com/honeycombio/husky/proto/otlp/common/v1"
+	resource "github.com/honeycombio/husky/proto/otlp/resource/v1"
+	trace "github.com/honeycombio/husky/proto/otlp/trace/v1"
 	"github.com/honeycombio/refinery/config"
 	"github.com/honeycombio/refinery/logger"
 	"github.com/honeycombio/refinery/metrics"
 	"github.com/honeycombio/refinery/transmit"
 	"github.com/klauspost/compress/zstd"
 	"github.com/stretchr/testify/assert"
-	collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1"
-	common "go.opentelemetry.io/proto/otlp/common/v1"
-	resource "go.opentelemetry.io/proto/otlp/resource/v1"
-	trace "go.opentelemetry.io/proto/otlp/trace/v1"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -67,7 +68,7 @@ func TestOTLPHandler(t *testing.T) {
 	t.Run("span with status", func(t *testing.T) {
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: helperOTLPRequestSpansWithStatus(),
 				}},
 			}},
@@ -83,7 +84,7 @@ func TestOTLPHandler(t *testing.T) {
 	t.Run("span without status", func(t *testing.T) {
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: helperOTLPRequestSpansWithoutStatus(),
 				}},
 			}},
@@ -105,7 +106,7 @@ func TestOTLPHandler(t *testing.T) {
 		spanID := []byte{1, 0, 0, 0, 0}
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: []*trace.Span{{
 						TraceId: traceID,
 						SpanId:  spanID,
@@ -153,7 +154,7 @@ func TestOTLPHandler(t *testing.T) {
 
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: []*trace.Span{{
 						Name:    "span_with_link",
 						TraceId: traceID,
@@ -191,7 +192,7 @@ func TestOTLPHandler(t *testing.T) {
 	t.Run("can receive OTLP over HTTP/protobuf", func(t *testing.T) {
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: helperOTLPRequestSpansWithStatus(),
 				}},
 			}},
@@ -218,7 +219,7 @@ func TestOTLPHandler(t *testing.T) {
 	t.Run("can receive OTLP over HTTP/protobuf with gzip encoding", func(t *testing.T) {
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: helperOTLPRequestSpansWithStatus(),
 				}},
 			}},
@@ -254,7 +255,7 @@ func TestOTLPHandler(t *testing.T) {
 	t.Run("can receive OTLP over HTTP/protobuf with zstd encoding", func(t *testing.T) {
 		req := &collectortrace.ExportTraceServiceRequest{
 			ResourceSpans: []*trace.ResourceSpans{{
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: helperOTLPRequestSpansWithStatus(),
 				}},
 			}},
@@ -290,8 +291,20 @@ func TestOTLPHandler(t *testing.T) {
 		mockTransmission.Flush()
 	})
 
-	t.Run("rejects OTLP over HTTP/JSON ", func(t *testing.T) {
-		request, _ := http.NewRequest("POST", "/v1/traces", strings.NewReader("{}"))
+	t.Run("accepts OTLP over HTTP/JSON ", func(t *testing.T) {
+		req := &collectortrace.ExportTraceServiceRequest{
+			ResourceSpans: []*trace.ResourceSpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
+					Spans: helperOTLPRequestSpansWithStatus(),
+				}},
+			}},
+		}
+		body, err := protojson.Marshal(req)
+		if err != nil {
+			t.Error(err)
+		}
+
+		request, _ := http.NewRequest("POST", "/v1/traces", bytes.NewReader(body))
 		request.Header = http.Header{}
 		request.Header.Set("content-type", "application/json")
 		request.Header.Set("x-honeycomb-team", legacyAPIKey)
@@ -299,10 +312,10 @@ func TestOTLPHandler(t *testing.T) {
 
 		w := httptest.NewRecorder()
 		router.postOTLP(w, request)
-		assert.Equal(t, w.Code, http.StatusNotImplemented)
-		assert.Equal(t, `{"source":"refinery","error":"invalid content-type - only 'application/protobuf' is supported"}`, string(w.Body.String()))
+		assert.Equal(t, w.Code, http.StatusOK)
+		assert.Equal(t, "", w.Body.String())
 
-		assert.Equal(t, 0, len(mockTransmission.Events))
+		assert.Equal(t, 2, len(mockTransmission.Events))
 		mockTransmission.Flush()
 	})
 
@@ -317,7 +330,7 @@ func TestOTLPHandler(t *testing.T) {
 						{Key: "service.name", Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "my-service"}}},
 					},
 				},
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: []*trace.Span{{
 						Name: "my-span",
 					}},
@@ -350,7 +363,7 @@ func TestOTLPHandler(t *testing.T) {
 						{Key: "service.name", Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "my-service"}}},
 					},
 				},
-				InstrumentationLibrarySpans: []*trace.InstrumentationLibrarySpans{{
+				ScopeSpans: []*trace.ScopeSpans{{
 					Spans: []*trace.Span{{
 						Name: "my-span",
 					}},
diff --git a/sharder/deterministic_test.go b/sharder/deterministic_test.go
index ab99d11..8d32287 100644
--- a/sharder/deterministic_test.go
+++ b/sharder/deterministic_test.go
@@ -1,6 +1,7 @@
 package sharder
 
 import (
+	"context"
 	"testing"
 
 	"github.com/honeycombio/refinery/config"
@@ -25,7 +26,7 @@ func TestWhichShard(t *testing.T) {
 		GetPeersVal:          peers,
 		PeerManagementType:   "file",
 	}
-	filePeers, err := peer.NewPeers(config)
+	filePeers, err := peer.NewPeers(context.Background(), config)
 	assert.Equal(t, nil, err)
 	sharder := DeterministicSharder{
 		Config: config,
@@ -66,7 +67,7 @@ func TestWhichShardAtEdge(t *testing.T) {
 		GetPeersVal:          peers,
 		PeerManagementType:   "file",
 	}
-	filePeers, err := peer.NewPeers(config)
+	filePeers, err := peer.NewPeers(context.Background(), config)
 	assert.Equal(t, nil, err)
 	sharder := DeterministicSharder{
 		Config: config,