Repository: atlassian/gostatsd
Pull request #290 (2020-02-25 01:45:37)

Cloud provider refactor
The idea is to split `CloudHandler` into two pieces by introducing an interface in the middle.
Currently `CloudHandler` does these things:
1. maintains a (negative) cache of ip->instance info
1. maintains a map of ip->[]metrics that need to be enriched per ip while the lookup it taking place

I'm introducing `CachedInstances` interface to abstract the caching bit (1 in the list above) and leave only 2 in the `CloudHandler`.

The next step/PR is to refactor the k8s cloud provider into an implementation of `CachedInstances`.

This PR is not ready to review in full (tests don't even compile as I haven't touched them yet) but to give you a preview and collect early feedback on the approach and major impl details. WDYT?

mwhittington21: Seems sane to me! Splitting out the caching behaviour from a default cloud provider makes a lot of sense.
ash2k: I'll get back to this PR, hopefully soon.
ash2k: This is ready for review.
tiedotguy: Awesome! I'll try and give it a thorough review tomorrow.
ash2k: @tiedotguy thanks for the review! I know I've done a few too many changes here that could all be separate PRs/commits. I know it's a poor excuse, but I don't have much time to rework this just for the sake of nicer history (which I do like to have too!). If nobody feels strongly about it, can we merge this (assuming everything else is fine)? If there are objections, I'll do more work here.

Regarding putting things into the root package - I don't mind having them there but also don't mind moving them somewhere else if that's what you and others prefer. I like to have api types there because a hypothetical `gostatsd.Metric`/etc reads nice. In the particular case of this PR I moved things because there was some dependency cycle. I do not remember now what it was, sorry, this PR is "spread in time" too much and I'm forgetting "whats" and "whys".

If we go ahead with this, I can squash it before merging.

Next step, which I'm planning to work on, is to refactor the k8s cloud provider into the new interface and do some cleanup around it (e.g. no two sets of defaults will be needed).
tiedotguy: I did track down the dependency cycle, and I agree it was needed. I don't recall specifically what it was now though :)

My main objection to the root is really the physical file placement. The mixing of source files and project files is a mixing of concerns.  Either way, it doesn't need to be handled in this PR

I think this is clear to merge now.
ash2k: Cool! I have rebased and squashed, tick it one more time, please.

Patch:
diff --git a/METRICS.md b/METRICS.md
index 0aea9c5..43ac978 100644
--- a/METRICS.md
+++ b/METRICS.md
@@ -52,7 +52,6 @@ Metrics:
 | cloudprovider.cache_refresh_positive        | gauge (cumulative)  |                              | The cumulative number of positive refreshes
 | cloudprovider.cache_refresh_negative        | gauge (cumulative)  |                              | The cumulative number of refreshes which had an error refreshing and used old data
 | cloudprovider.cache_hit                     | gauge (cumulative)  |                              | The cumulative number of cache hits (host was in the cache)
-| cloudprovider.cache_late_hit                | gauge (cumulative)  |                              | The cumulative number of late cache hits (host was not in the cache, but had a lookup
 |                                             |                     |                              | in progress which completed)
 | cloudprovider.cache_miss                    | gauge (cumulative)  |                              | The cumulative number of cache misses
 | cloudprovider.hosts_queued                  | gauge (flush)       | type                         | The absolute number of hosts waiting to be looked up
diff --git a/backend.go b/backend.go
index 68809f9..9d880f4 100644
--- a/backend.go
+++ b/backend.go
@@ -17,6 +17,7 @@ type BackendFactory func(config *viper.Viper, pool *transport.TransportPool) (Ba
 type SendCallback func([]error)
 
 // Backend represents a backend.
+// If Backend implements the Runner interface, it's started in a new goroutine at creation.
 type Backend interface {
 	// Name returns the name of the backend.
 	Name() string
diff --git a/cloudprovider.go b/cloudprovider.go
index cf97631..fb209e4 100644
--- a/cloudprovider.go
+++ b/cloudprovider.go
@@ -2,6 +2,7 @@ package gostatsd
 
 import (
 	"context"
+	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
@@ -32,3 +33,30 @@ type CloudProvider interface {
 	// EstimatedTags returns a guess of how many tags are likely to be added by the CloudProvider
 	EstimatedTags() int
 }
+
+type InstanceInfo struct {
+	IP IP
+	// Instance may be nil if the lookup resulted in an error or instance was not found.
+	Instance *Instance
+}
+
+type CachedInstances interface {
+	// Peek fetches instance information from the cache.
+	// The cache is also a negative cache - may be a cache hit but the returned instance is nil.
+	Peek(IP) (*Instance, bool /*is a cache hit*/)
+	// IpSink returns a channel that can be used to supply IP addresses for which information needs
+	// to be fetched and cached.
+	IpSink() chan<- IP
+	// InfoSource returns a channel that can be used to receive information about IPs.
+	InfoSource() <-chan InstanceInfo
+	// EstimatedTags returns a guess for how many tags to pre-allocate
+	EstimatedTags() int
+}
+
+// CacheOptions holds cache behaviour configuration.
+type CacheOptions struct {
+	CacheRefreshPeriod        time.Duration
+	CacheEvictAfterIdlePeriod time.Duration
+	CacheTTL                  time.Duration
+	CacheNegativeTTL          time.Duration
+}
diff --git a/cmd/gostatsd/main.go b/cmd/gostatsd/main.go
index 06cd4ba..9f6cd91 100644
--- a/cmd/gostatsd/main.go
+++ b/cmd/gostatsd/main.go
@@ -13,13 +13,14 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/atlassian/gostatsd/pkg/util"
-
+	"github.com/ash2k/stager"
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/backends"
+	"github.com/atlassian/gostatsd/pkg/cachedinstances"
+	"github.com/atlassian/gostatsd/pkg/cloudproviders"
 	"github.com/atlassian/gostatsd/pkg/statsd"
 	"github.com/atlassian/gostatsd/pkg/transport"
-
+	"github.com/atlassian/gostatsd/pkg/util"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
@@ -66,11 +67,17 @@ func run(v *viper.Viper) error {
 	}
 
 	logrus.Info("Starting server")
-	s, err := constructServer(v)
+	s, runnables, err := constructServer(v)
 	if err != nil {
 		return err
 	}
 
+	stgr := stager.New()
+	defer stgr.Shutdown()
+	for _, runnable := range runnables {
+		stgr.NextStage().StartWithContext(runnable)
+	}
+
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	defer cancelFunc()
 	cancelOnInterrupt(ctx, cancelFunc)
@@ -81,74 +88,93 @@ func run(v *viper.Viper) error {
 	return nil
 }
 
-func constructServer(v *viper.Viper) (*statsd.Server, error) {
+func constructServer(v *viper.Viper) (*statsd.Server, []gostatsd.Runnable, error) {
+	var runnables []gostatsd.Runnable
 	// Logger
 	logger := logrus.StandardLogger()
 
 	// HTTP client pool
 	pool := transport.NewTransportPool(logger, v)
 
-	// Cloud handler factory
-	cloud, err := statsd.NewCloudHandlerFactoryFromViper(v, logger, Version)
-	if err != nil {
-		return nil, err
-	}
-	if cloud != nil {
-		if err := cloud.InitCloudProvider(v); err != nil {
-			return nil, err
+	// Cached instances
+	var cachedInstances gostatsd.CachedInstances
+	selfIP := gostatsd.UnknownIP
+	cloudProviderName := v.GetString(gostatsd.ParamCloudProvider)
+	if cloudProviderName == "" {
+		logger.Info("No cloud provider specified")
+	} else {
+		var err error
+		cloudProvider, err := cloudproviders.Get(logger, cloudProviderName, v, Version)
+		if err != nil {
+			return nil, nil, err
+		}
+		runnables = gostatsd.MaybeAppendRunnable(runnables, cloudProvider)
+		selfIPtmp, err := cloudProvider.SelfIP()
+		if err != nil {
+			logger.WithError(err).Warn("Failed to get self ip")
+		} else {
+			selfIP = selfIPtmp
+		}
+
+		cachedInstances, err = cachedinstances.NewCachedInstancesFromViper(logger, cloudProvider, v)
+		if err != nil {
+			return nil, nil, err
 		}
+		runnables = gostatsd.MaybeAppendRunnable(runnables, cachedInstances)
 	}
 	// Backends
-	backendNames := v.GetStringSlice(statsd.ParamBackends)
-	backendsList := make([]gostatsd.Backend, len(backendNames))
-	for i, backendName := range backendNames {
+	backendNames := v.GetStringSlice(gostatsd.ParamBackends)
+	backendsList := make([]gostatsd.Backend, 0, len(backendNames))
+	for _, backendName := range backendNames {
 		backend, errBackend := backends.InitBackend(backendName, v, pool)
 		if errBackend != nil {
-			return nil, errBackend
+			return nil, nil, errBackend
 		}
-		backendsList[i] = backend
+		backendsList = append(backendsList, backend)
+		runnables = gostatsd.MaybeAppendRunnable(runnables, backend)
 	}
 	// Percentiles
-	pt, err := getPercentiles(v.GetStringSlice(statsd.ParamPercentThreshold))
+	pt, err := getPercentiles(v.GetStringSlice(gostatsd.ParamPercentThreshold))
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	// Create server
 	return &statsd.Server{
 		Backends:            backendsList,
-		CloudHandlerFactory: cloud,
-		InternalTags:        v.GetStringSlice(statsd.ParamInternalTags),
-		InternalNamespace:   v.GetString(statsd.ParamInternalNamespace),
-		DefaultTags:         v.GetStringSlice(statsd.ParamDefaultTags),
-		Hostname:            v.GetString(statsd.ParamHostname),
-		ExpiryInterval:      v.GetDuration(statsd.ParamExpiryInterval),
-		FlushInterval:       v.GetDuration(statsd.ParamFlushInterval),
-		IgnoreHost:          v.GetBool(statsd.ParamIgnoreHost),
-		MaxReaders:          v.GetInt(statsd.ParamMaxReaders),
-		MaxParsers:          v.GetInt(statsd.ParamMaxParsers),
-		MaxWorkers:          v.GetInt(statsd.ParamMaxWorkers),
-		MaxQueueSize:        v.GetInt(statsd.ParamMaxQueueSize),
-		MaxConcurrentEvents: v.GetInt(statsd.ParamMaxConcurrentEvents),
-		EstimatedTags:       v.GetInt(statsd.ParamEstimatedTags),
-		MetricsAddr:         v.GetString(statsd.ParamMetricsAddr),
-		Namespace:           v.GetString(statsd.ParamNamespace),
-		StatserType:         v.GetString(statsd.ParamStatserType),
+		CachedInstances:     cachedInstances,
+		InternalTags:        v.GetStringSlice(gostatsd.ParamInternalTags),
+		InternalNamespace:   v.GetString(gostatsd.ParamInternalNamespace),
+		DefaultTags:         v.GetStringSlice(gostatsd.ParamDefaultTags),
+		Hostname:            v.GetString(gostatsd.ParamHostname),
+		SelfIP:              selfIP,
+		ExpiryInterval:      v.GetDuration(gostatsd.ParamExpiryInterval),
+		FlushInterval:       v.GetDuration(gostatsd.ParamFlushInterval),
+		IgnoreHost:          v.GetBool(gostatsd.ParamIgnoreHost),
+		MaxReaders:          v.GetInt(gostatsd.ParamMaxReaders),
+		MaxParsers:          v.GetInt(gostatsd.ParamMaxParsers),
+		MaxWorkers:          v.GetInt(gostatsd.ParamMaxWorkers),
+		MaxQueueSize:        v.GetInt(gostatsd.ParamMaxQueueSize),
+		MaxConcurrentEvents: v.GetInt(gostatsd.ParamMaxConcurrentEvents),
+		EstimatedTags:       v.GetInt(gostatsd.ParamEstimatedTags),
+		MetricsAddr:         v.GetString(gostatsd.ParamMetricsAddr),
+		Namespace:           v.GetString(gostatsd.ParamNamespace),
+		StatserType:         v.GetString(gostatsd.ParamStatserType),
 		PercentThreshold:    pt,
-		HeartbeatEnabled:    v.GetBool(statsd.ParamHeartbeatEnabled),
-		ReceiveBatchSize:    v.GetInt(statsd.ParamReceiveBatchSize),
-		ConnPerReader:       v.GetBool(statsd.ParamConnPerReader),
-		ServerMode:          v.GetString(statsd.ParamServerMode),
-		LogRawMetric:        v.GetBool(statsd.ParamLogRawMetric),
+		HeartbeatEnabled:    v.GetBool(gostatsd.ParamHeartbeatEnabled),
+		ReceiveBatchSize:    v.GetInt(gostatsd.ParamReceiveBatchSize),
+		ConnPerReader:       v.GetBool(gostatsd.ParamConnPerReader),
+		ServerMode:          v.GetString(gostatsd.ParamServerMode),
+		LogRawMetric:        v.GetBool(gostatsd.ParamLogRawMetric),
 		HeartbeatTags: gostatsd.Tags{
 			fmt.Sprintf("version:%s", Version),
 			fmt.Sprintf("commit:%s", GitCommit),
 		},
 		DisabledSubTypes:          gostatsd.DisabledSubMetrics(v),
-		BadLineRateLimitPerSecond: rate.Limit(v.GetFloat64(statsd.ParamBadLinesPerMinute) / 60.0),
-		HistogramLimit:            v.GetUint32(statsd.ParamTimerHistogramLimit),
+		BadLineRateLimitPerSecond: rate.Limit(v.GetFloat64(gostatsd.ParamBadLinesPerMinute) / 60.0),
+		HistogramLimit:            v.GetUint32(gostatsd.ParamTimerHistogramLimit),
 		Viper:                     v,
 		TransportPool:             pool,
-	}, nil
+	}, runnables, nil
 }
 
 func getPercentiles(s []string) ([]float64, error) {
@@ -191,7 +217,7 @@ func setupConfiguration() (*viper.Viper, bool, error) {
 	cmd.String(ParamProfile, "", "Enable profiler endpoint on the specified address and port")
 	cmd.String(ParamConfigPath, "", "Path to the configuration file")
 
-	statsd.AddFlags(cmd)
+	gostatsd.AddFlags(cmd)
 
 	cmd.VisitAll(func(flag *pflag.Flag) {
 		if err := v.BindPFlag(flag.Name, flag); err != nil {
diff --git a/pkg/statsd/statsd_defaults_and_params.go b/defaults_and_params.go
similarity index 87%
rename from pkg/statsd/statsd_defaults_and_params.go
rename to defaults_and_params.go
index a1f07b8..e175aea 100644
--- a/pkg/statsd/statsd_defaults_and_params.go
+++ b/defaults_and_params.go
@@ -1,15 +1,14 @@
-package statsd
+package gostatsd
 
 import (
 	"math"
+	"os"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 
-	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/cloudproviders/aws"
-	"github.com/atlassian/gostatsd/pkg/cloudproviders/k8s"
-
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 )
 
@@ -29,47 +28,10 @@ var DefaultMaxParsers = runtime.NumCPU()
 var DefaultPercentThreshold = []float64{90}
 
 // DefaultTags is the default list of additional tags.
-var DefaultTags = gostatsd.Tags{}
+var DefaultTags = Tags{}
 
 // DefaultInternalTags is the default list of additional tags on internal metrics
-var DefaultInternalTags = gostatsd.Tags{}
-
-// DefaultCloudProviderCacheValues contains the default cache values for each cloud provider.
-var DefaultCloudProviderCacheValues = map[string]CacheOptions{
-	k8s.ProviderName: {
-		// The refresh period is low for k8s provider since there is no network cost to looking up pods
-		CacheRefreshPeriod:        15 * time.Second,
-		CacheEvictAfterIdlePeriod: DefaultCacheEvictAfterIdlePeriod,
-		// TTLs are low for the k8s provider because there is no network cost to doing a refresh of the data
-		// This means we get fresh data every time the refresh period is up
-		CacheTTL:         1 * time.Millisecond,
-		CacheNegativeTTL: 1 * time.Millisecond,
-	},
-	aws.ProviderName: {
-		CacheRefreshPeriod:        DefaultCacheRefreshPeriod,
-		CacheEvictAfterIdlePeriod: DefaultCacheEvictAfterIdlePeriod,
-		CacheTTL:                  DefaultCacheTTL,
-		CacheNegativeTTL:          DefaultCacheNegativeTTL,
-	},
-}
-
-type LimiterValues struct {
-	MaxCloudRequests   int
-	BurstCloudRequests int
-}
-
-// DefaultCloudProviderLimiterValues contains the default limiter values for each cloud provider.
-var DefaultCloudProviderLimiterValues = map[string]LimiterValues{
-	k8s.ProviderName: {
-		// High limit for k8s since we don't make network requests for each data request
-		MaxCloudRequests:   10000,
-		BurstCloudRequests: 5000,
-	},
-	aws.ProviderName: {
-		MaxCloudRequests:   DefaultMaxCloudRequests,
-		BurstCloudRequests: DefaultBurstCloudRequests,
-	},
-}
+var DefaultInternalTags = Tags{}
 
 const (
 	// StatserInternal is the name used to indicate the use of the internal statser.
@@ -237,3 +199,20 @@ func minInt(a, b int) int {
 	}
 	return b
 }
+
+func toStringSlice(fs []float64) []string {
+	s := make([]string, len(fs))
+	for i, f := range fs {
+		s[i] = strconv.FormatFloat(f, 'f', -1, 64)
+	}
+	return s
+}
+
+func getHost() string {
+	host, err := os.Hostname()
+	if err != nil {
+		logrus.Warnf("Cannot get hostname: %v", err)
+		return ""
+	}
+	return host
+}
diff --git a/pkg/cachedinstances/cloudprovider/cached_cloud_provider.go b/pkg/cachedinstances/cloudprovider/cached_cloud_provider.go
new file mode 100644
index 0000000..ebb75cc
--- /dev/null
+++ b/pkg/cachedinstances/cloudprovider/cached_cloud_provider.go
@@ -0,0 +1,261 @@
+package cloudprovider
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/ash2k/stager/wait"
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/stats"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
+)
+
+func NewCachedCloudProvider(logger logrus.FieldLogger, limiter *rate.Limiter, cloudProvider gostatsd.CloudProvider, cacheOpts gostatsd.CacheOptions) *CachedCloudProvider {
+	return &CachedCloudProvider{
+		logger:         logger,
+		limiter:        limiter,
+		cloudProvider:  cloudProvider,
+		cacheOpts:      cacheOpts,
+		ipSinkSource:   make(chan gostatsd.IP),
+		infoSinkSource: make(chan gostatsd.InstanceInfo),
+		emitChan:       make(chan stats.Statser),
+		cache:          make(map[gostatsd.IP]*instanceHolder),
+	}
+}
+
+type CachedCloudProvider struct {
+	// These fields may only be read or written by the main CachedCloudProvider.Run goroutine
+	statsCacheRefreshPositive uint64 // Cumulative number of positive refreshes (ie, a refresh which succeeded)
+	statsCacheRefreshNegative uint64 // Cumulative number of negative refreshes (ie, a refresh which failed and used old data)
+	statsCachePositive        uint64 // Absolute number of positive entries in cache
+	statsCacheNegative        uint64 // Absolute number of negative entries in cache
+
+	logger         logrus.FieldLogger
+	limiter        *rate.Limiter
+	cloudProvider  gostatsd.CloudProvider
+	cacheOpts      gostatsd.CacheOptions
+	ipSinkSource   chan gostatsd.IP
+	infoSinkSource chan gostatsd.InstanceInfo
+
+	// emitChan triggers a write of all the current stats when it is given a Statser
+	emitChan     chan stats.Statser
+	rw           sync.RWMutex // Protects cache
+	cache        map[gostatsd.IP]*instanceHolder
+	toLookupIPs  []gostatsd.IP
+	toReturnInfo []gostatsd.InstanceInfo
+}
+
+func (ccp *CachedCloudProvider) Run(ctx context.Context) {
+	var (
+		toLookupC     chan<- gostatsd.IP
+		toLookupIP    gostatsd.IP
+		toReturnInfoC chan<- gostatsd.InstanceInfo
+		toReturnInfo  gostatsd.InstanceInfo
+		wg            wait.Group
+	)
+	// this goroutine needs to populate/update the cache so an intermediate InstanceInfo channel is used below that allows
+	// to intercept, update the cache and then push the information through to the cache consumer.
+	ownInfoSource := make(chan gostatsd.InstanceInfo)
+	ld := cloudProviderLookupDispatcher{
+		logger:        ccp.logger,
+		limiter:       ccp.limiter,
+		cloudProvider: ccp.cloudProvider,
+		ipSource:      ccp.ipSinkSource, // our sink is their source
+		infoSink:      ownInfoSource,    // their sink is our source
+	}
+
+	defer wg.Wait() // Wait for cloudProviderLookupDispatcher to stop
+
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel() // Tell cloudProviderLookupDispatcher to stop
+
+	wg.StartWithContext(ctx, ld.run)
+
+	refreshTicker := time.NewTicker(ccp.cacheOpts.CacheRefreshPeriod)
+	defer refreshTicker.Stop()
+	// No locking for ccp.cache READ access required - this goroutine owns the object and only it mutates it.
+	// So reads from the same goroutine are always safe (no concurrent mutations).
+	// When we mutate the cache, we hold the exclusive (write) lock to avoid concurrent reads.
+	// When we read from the cache from other goroutines in the Peek() method, we obtain the read lock.
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case toLookupC <- toLookupIP:
+			toLookupIP = gostatsd.UnknownIP // enable GC
+			toLookupC = nil                 // ip has been sent; if there is nothing to send, the case is disabled
+		case toReturnInfoC <- toReturnInfo:
+			toReturnInfo = gostatsd.InstanceInfo{} // enable GC
+			toReturnInfoC = nil                    // info has been sent; if there is nothing to send, the case is disabled
+		case info := <-ownInfoSource:
+			ccp.handleInstanceInfo(info)
+		case t := <-refreshTicker.C:
+			ccp.doRefresh(t)
+		case statser := <-ccp.emitChan:
+			ccp.emit(statser)
+		}
+		if toLookupC == nil && len(ccp.toLookupIPs) > 0 {
+			last := len(ccp.toLookupIPs) - 1
+			toLookupIP = ccp.toLookupIPs[last]
+			ccp.toLookupIPs[last] = gostatsd.UnknownIP // enable GC
+			ccp.toLookupIPs = ccp.toLookupIPs[:last]
+			toLookupC = ccp.ipSinkSource
+		}
+		if toReturnInfoC == nil && len(ccp.toReturnInfo) > 0 {
+			last := len(ccp.toReturnInfo) - 1
+			toReturnInfo = ccp.toReturnInfo[last]
+			ccp.toReturnInfo[last] = gostatsd.InstanceInfo{} // enable GC
+			ccp.toReturnInfo = ccp.toReturnInfo[:last]
+			toReturnInfoC = ccp.infoSinkSource
+		}
+	}
+}
+
+func (ccp *CachedCloudProvider) Peek(ip gostatsd.IP) (*gostatsd.Instance, bool /*is a cache hit*/) {
+	ccp.rw.RLock()
+	holder, existsInCache := ccp.cache[ip]
+	ccp.rw.RUnlock()
+	if !existsInCache {
+		return nil, false
+	}
+	holder.updateAccess()
+	return holder.instance, true // can be nil, true
+}
+
+func (ccp *CachedCloudProvider) IpSink() chan<- gostatsd.IP {
+	return ccp.ipSinkSource
+}
+
+func (ccp *CachedCloudProvider) InfoSource() <-chan gostatsd.InstanceInfo {
+	return ccp.infoSinkSource
+}
+
+func (ccp *CachedCloudProvider) EstimatedTags() int {
+	return ccp.cloudProvider.EstimatedTags()
+}
+
+func (ccp *CachedCloudProvider) RunMetrics(ctx context.Context, statser stats.Statser) {
+	// All the channels are unbuffered, so no CSWs
+	flushed, unregister := statser.RegisterFlush()
+	defer unregister()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-flushed:
+			ccp.scheduleEmit(ctx, statser)
+		}
+	}
+}
+
+// scheduleEmit is used to push a request to the main goroutine requesting metrics
+// be emitted.  This is done so we can skip atomic operations on most of our metric
+// counters.  In line with the flush notifier, it is fire and forget and won't block
+func (ccp *CachedCloudProvider) scheduleEmit(ctx context.Context, statser stats.Statser) {
+	select {
+	case ccp.emitChan <- statser:
+		// success
+	case <-ctx.Done():
+		// success-ish
+	default:
+		// at least we tried
+	}
+}
+
+func (ccp *CachedCloudProvider) emit(statser stats.Statser) {
+	// regular
+	statser.Gauge("cloudprovider.cache_positive", float64(ccp.statsCachePositive), nil)
+	statser.Gauge("cloudprovider.cache_negative", float64(ccp.statsCacheNegative), nil)
+	statser.Gauge("cloudprovider.cache_refresh_positive", float64(ccp.statsCacheRefreshPositive), nil)
+	statser.Gauge("cloudprovider.cache_refresh_negative", float64(ccp.statsCacheRefreshNegative), nil)
+}
+
+func (ccp *CachedCloudProvider) doRefresh(t time.Time) {
+	var toDelete []gostatsd.IP
+	now := t.UnixNano()
+	idleNano := ccp.cacheOpts.CacheEvictAfterIdlePeriod.Nanoseconds()
+
+	for ip, holder := range ccp.cache {
+		if now-holder.lastAccess() > idleNano {
+			// Entry was not used recently, remove it.
+			toDelete = append(toDelete, ip)
+			if holder.instance == nil {
+				ccp.statsCacheNegative--
+			} else {
+				ccp.statsCachePositive--
+			}
+		} else if t.After(holder.expires) {
+			// Entry needs a refresh.
+			ccp.toLookupIPs = append(ccp.toLookupIPs, ip)
+		}
+	}
+
+	if len(toDelete) > 0 {
+		ccp.rw.Lock()
+		defer ccp.rw.Unlock()
+		for _, ip := range toDelete {
+			delete(ccp.cache, ip)
+		}
+	}
+}
+
+func (ccp *CachedCloudProvider) handleInstanceInfo(info gostatsd.InstanceInfo) {
+	var ttl time.Duration
+	if info.Instance == nil {
+		ttl = ccp.cacheOpts.CacheNegativeTTL
+	} else {
+		ttl = ccp.cacheOpts.CacheTTL
+	}
+	now := time.Now()
+	newHolder := &instanceHolder{
+		expires:  now.Add(ttl),
+		instance: info.Instance,
+	}
+	currentHolder := ccp.cache[info.IP]
+	if currentHolder == nil {
+		// Not in cache, count it
+		if info.Instance == nil {
+			ccp.statsCacheNegative++
+		} else {
+			ccp.statsCachePositive++
+		}
+		newHolder.lastAccessNano = now.UnixNano()
+	} else {
+		// In cache, don't count it
+		newHolder.lastAccessNano = currentHolder.lastAccess()
+		if info.Instance == nil {
+			// Use the old instance if there was a lookup error.
+			newHolder.instance = currentHolder.instance
+			ccp.statsCacheRefreshNegative++
+		} else {
+			if currentHolder.instance == nil && newHolder.instance != nil {
+				// An entry has flipped from invalid to valid
+				ccp.statsCacheNegative--
+				ccp.statsCachePositive++
+			}
+			ccp.statsCacheRefreshPositive++
+		}
+	}
+	ccp.rw.Lock()
+	ccp.cache[info.IP] = newHolder
+	ccp.rw.Unlock()
+	ccp.toReturnInfo = append(ccp.toReturnInfo, info)
+}
+
+type instanceHolder struct {
+	lastAccessNano int64
+	expires        time.Time          // When this record expires.
+	instance       *gostatsd.Instance // Can be nil if the lookup resulted in an error or instance was not found
+}
+
+func (ih *instanceHolder) updateAccess() {
+	atomic.StoreInt64(&ih.lastAccessNano, time.Now().UnixNano())
+}
+
+func (ih *instanceHolder) lastAccess() int64 {
+	return atomic.LoadInt64(&ih.lastAccessNano)
+}
diff --git a/pkg/statsd/handler_cloud_lookup.go b/pkg/cachedinstances/cloudprovider/cached_cloud_provider_lookup.go
similarity index 64%
rename from pkg/statsd/handler_cloud_lookup.go
rename to pkg/cachedinstances/cloudprovider/cached_cloud_provider_lookup.go
index a9b5b93..f631cad 100644
--- a/pkg/statsd/handler_cloud_lookup.go
+++ b/pkg/cachedinstances/cloudprovider/cached_cloud_provider_lookup.go
@@ -1,36 +1,35 @@
-package statsd
+package cloudprovider
 
 import (
 	"context"
 	"time"
 
 	"github.com/atlassian/gostatsd"
-
 	"github.com/sirupsen/logrus"
 	"golang.org/x/time/rate"
 )
 
 const (
-	maxLookupIPs  = 32
 	batchDuration = 10 * time.Millisecond
 )
 
-type lookupDispatcher struct {
-	limiter       *rate.Limiter
-	cloud         gostatsd.CloudProvider // Cloud provider interface
-	toLookup      <-chan gostatsd.IP
-	lookupResults chan<- *lookupResult
+type cloudProviderLookupDispatcher struct {
 	logger        logrus.FieldLogger
+	limiter       *rate.Limiter
+	cloudProvider gostatsd.CloudProvider
+	ipSource      <-chan gostatsd.IP
+	infoSink      chan<- gostatsd.InstanceInfo
 }
 
-func (ld *lookupDispatcher) run(ctx context.Context) {
+func (ld *cloudProviderLookupDispatcher) run(ctx context.Context) {
+	maxLookupIPs := ld.cloudProvider.MaxInstancesBatch()
 	ips := make([]gostatsd.IP, 0, maxLookupIPs)
 	var c <-chan time.Time
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case ip := <-ld.toLookup:
+		case ip := <-ld.ipSource:
 			ips = append(ips, ip)
 			if len(ips) >= maxLookupIPs {
 				break // enough ips, exit select
@@ -38,7 +37,7 @@ func (ld *lookupDispatcher) run(ctx context.Context) {
 			if c == nil {
 				c = time.After(batchDuration)
 			}
-			continue // have some more time to collect ips
+			continue // have some more space-time to collect ips
 		case <-c: // time to do the lookup
 		}
 		c = nil
@@ -59,32 +58,22 @@ func (ld *lookupDispatcher) run(ctx context.Context) {
 	}
 }
 
-func (ld *lookupDispatcher) doLookup(ctx context.Context, ips []gostatsd.IP) {
+func (ld *cloudProviderLookupDispatcher) doLookup(ctx context.Context, ips []gostatsd.IP) {
 	// instances may contain partial result even if err != nil
-	instances, err := ld.cloud.Instance(ctx, ips...)
+	instances, err := ld.cloudProvider.Instance(ctx, ips...)
 	if err != nil {
 		// Something bad happened, but process what we have still
 		ld.logger.Infof("Error retrieving instance details from cloud provider: %v", err)
 	}
 	for _, ip := range ips {
-		instance := instances[ip]
-		var res *lookupResult
-		if instance != nil {
-			res = &lookupResult{
-				ip:       ip,
-				instance: instance,
-			}
-
-		} else {
-			res = &lookupResult{
-				ip:       ip,
-				instance: nil,
-			}
+		res := gostatsd.InstanceInfo{
+			IP:       ip,
+			Instance: instances[ip],
 		}
 		select {
 		case <-ctx.Done():
 			return
-		case ld.lookupResults <- res:
+		case ld.infoSink <- res:
 		}
 	}
 }
diff --git a/pkg/cachedinstances/factory.go b/pkg/cachedinstances/factory.go
new file mode 100644
index 0000000..a4bfeaa
--- /dev/null
+++ b/pkg/cachedinstances/factory.go
@@ -0,0 +1,85 @@
+package cachedinstances
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/cachedinstances/cloudprovider"
+	"github.com/atlassian/gostatsd/pkg/cloudproviders/aws"
+	"github.com/atlassian/gostatsd/pkg/cloudproviders/k8s"
+	"github.com/sirupsen/logrus"
+	"github.com/spf13/viper"
+	"golang.org/x/time/rate"
+)
+
+var (
+	// defaultCloudProviderCacheValues contains the default cache values for each cloud provider.
+	defaultCloudProviderCacheValues = map[string]gostatsd.CacheOptions{
+		k8s.ProviderName: {
+			// The refresh period is low for k8s provider since there is no network cost to looking up pods
+			CacheRefreshPeriod:        15 * time.Second,
+			CacheEvictAfterIdlePeriod: gostatsd.DefaultCacheEvictAfterIdlePeriod,
+			// TTLs are low for the k8s provider because there is no network cost to doing a refresh of the data
+			// This means we get fresh data every time the refresh period is up
+			CacheTTL:         1 * time.Millisecond,
+			CacheNegativeTTL: 1 * time.Millisecond,
+		},
+		aws.ProviderName: {
+			CacheRefreshPeriod:        gostatsd.DefaultCacheRefreshPeriod,
+			CacheEvictAfterIdlePeriod: gostatsd.DefaultCacheEvictAfterIdlePeriod,
+			CacheTTL:                  gostatsd.DefaultCacheTTL,
+			CacheNegativeTTL:          gostatsd.DefaultCacheNegativeTTL,
+		},
+	}
+
+	// defaultCloudProviderLimiterValues contains the default limiter values for each cloud provider.
+	defaultCloudProviderLimiterValues = map[string]limiterValues{
+		k8s.ProviderName: {
+			// High limit for k8s since we don't make network requests for each data request
+			maxCloudRequests:   10000,
+			burstCloudRequests: 5000,
+		},
+		aws.ProviderName: {
+			maxCloudRequests:   gostatsd.DefaultMaxCloudRequests,
+			burstCloudRequests: gostatsd.DefaultBurstCloudRequests,
+		},
+	}
+)
+
+type limiterValues struct {
+	maxCloudRequests   int
+	burstCloudRequests int
+}
+
+// NewCachedInstancesFromViper initialises a new cached instances.
+func NewCachedInstancesFromViper(logger logrus.FieldLogger, cloudProvider gostatsd.CloudProvider, v *viper.Viper) (gostatsd.CachedInstances, error) {
+	// Cloud provider defaults
+	cloudProviderName := cloudProvider.Name()
+	cpDefaultCacheOpts, ok := defaultCloudProviderCacheValues[cloudProviderName]
+	if !ok {
+		return nil, fmt.Errorf("could not find default cache values for cloud provider %q", cloudProviderName)
+	}
+	cpDefaultLimiterOpts, ok := defaultCloudProviderLimiterValues[cloudProviderName]
+	if !ok {
+		return nil, fmt.Errorf("could not find default cache values for cloud provider %q", cloudProviderName)
+	}
+
+	// Set the defaults in Viper based on the cloud provider values before we manipulate things
+	v.SetDefault(gostatsd.ParamCacheRefreshPeriod, cpDefaultCacheOpts.CacheRefreshPeriod)
+	v.SetDefault(gostatsd.ParamCacheEvictAfterIdlePeriod, cpDefaultCacheOpts.CacheEvictAfterIdlePeriod)
+	v.SetDefault(gostatsd.ParamCacheTTL, cpDefaultCacheOpts.CacheTTL)
+	v.SetDefault(gostatsd.ParamCacheNegativeTTL, cpDefaultCacheOpts.CacheNegativeTTL)
+	v.SetDefault(gostatsd.ParamMaxCloudRequests, cpDefaultLimiterOpts.maxCloudRequests)
+	v.SetDefault(gostatsd.ParamBurstCloudRequests, cpDefaultLimiterOpts.burstCloudRequests)
+
+	// Set the used values based on the defaults merged with any overrides
+	cacheOptions := gostatsd.CacheOptions{
+		CacheRefreshPeriod:        v.GetDuration(gostatsd.ParamCacheRefreshPeriod),
+		CacheEvictAfterIdlePeriod: v.GetDuration(gostatsd.ParamCacheEvictAfterIdlePeriod),
+		CacheTTL:                  v.GetDuration(gostatsd.ParamCacheTTL),
+		CacheNegativeTTL:          v.GetDuration(gostatsd.ParamCacheNegativeTTL),
+	}
+	limiter := rate.NewLimiter(rate.Limit(v.GetInt(gostatsd.ParamMaxCloudRequests)), v.GetInt(gostatsd.ParamBurstCloudRequests))
+	return cloudprovider.NewCachedCloudProvider(logger, limiter, cloudProvider, cacheOptions), nil
+}
diff --git a/pkg/cloudproviders/cloudproviders.go b/pkg/cloudproviders/cloudproviders.go
index 9e64366..4d6803c 100644
--- a/pkg/cloudproviders/cloudproviders.go
+++ b/pkg/cloudproviders/cloudproviders.go
@@ -3,13 +3,11 @@ package cloudproviders
 import (
 	"fmt"
 
-	"github.com/spf13/viper"
-
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/cloudproviders/aws"
 	"github.com/atlassian/gostatsd/pkg/cloudproviders/k8s"
-
 	"github.com/sirupsen/logrus"
+	"github.com/spf13/viper"
 )
 
 // All registered cloud providers.
@@ -18,32 +16,11 @@ var providers = map[string]gostatsd.CloudProviderFactory{
 	k8s.ProviderName: k8s.NewProviderFromViper,
 }
 
-// Get creates an instance of the named provider, or nil if
-// the name is not known.  The error return is only used if the named provider
-// was known but failed to initialize.
-func Get(name string, v *viper.Viper, logger logrus.FieldLogger, version string) (gostatsd.CloudProvider, error) {
+// Get creates an instance of the named provider.
+func Get(logger logrus.FieldLogger, name string, v *viper.Viper, version string) (gostatsd.CloudProvider, error) {
 	f, found := providers[name]
 	if !found {
-		return nil, nil
-	}
-	return f(v, logger, version)
-}
-
-// Init creates an instance of the named cloud provider.
-func Init(name string, v *viper.Viper, logger logrus.FieldLogger, version string) (gostatsd.CloudProvider, error) {
-	if name == "" {
-		logrus.Info("No cloud provider specified")
-		return nil, nil
-	}
-
-	provider, err := Get(name, v, logger.WithField("cloud_provider", name), version)
-	if err != nil {
-		return nil, fmt.Errorf("could not init cloud provider %q: %v", name, err)
-	}
-	if provider == nil {
 		return nil, fmt.Errorf("unknown cloud provider %q", name)
 	}
-	logrus.Infof("Initialised cloud provider %q", name)
-
-	return provider, nil
+	return f(v, logger.WithField("cloud_provider", name), version)
 }
diff --git a/pkg/cloudproviders/fakeprovider/fake.go b/pkg/cloudproviders/fakeprovider/fake.go
new file mode 100644
index 0000000..fa2cda9
--- /dev/null
+++ b/pkg/cloudproviders/fakeprovider/fake.go
@@ -0,0 +1,154 @@
+package fakeprovider
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+
+	"github.com/atlassian/gostatsd"
+)
+
+type Counting struct {
+	mu          sync.Mutex
+	ips         []gostatsd.IP
+	invocations uint64
+}
+
+func (fp *Counting) IPs() []gostatsd.IP {
+	fp.mu.Lock()
+	defer fp.mu.Unlock()
+	result := make([]gostatsd.IP, len(fp.ips))
+	copy(result, fp.ips)
+	return result
+}
+
+func (fp *Counting) EstimatedTags() int {
+	return 0
+}
+
+func (fp *Counting) MaxInstancesBatch() int {
+	return 16
+}
+
+func (fp *Counting) Invocations() uint64 {
+	fp.mu.Lock()
+	defer fp.mu.Unlock()
+	return fp.invocations
+}
+
+func (fp *Counting) count(ips ...gostatsd.IP) {
+	fp.mu.Lock()
+	defer fp.mu.Unlock()
+	fp.ips = append(fp.ips, ips...)
+	fp.invocations++
+}
+
+func (fp *Counting) SelfIP() (gostatsd.IP, error) {
+	return gostatsd.UnknownIP, nil
+}
+
+type IP struct {
+	Counting
+	Region string
+	Tags   gostatsd.Tags
+}
+
+func (fp *IP) Name() string {
+	return "FakeProviderIP"
+}
+
+func (fp *IP) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
+	fp.count(ips...)
+	instances := make(map[gostatsd.IP]*gostatsd.Instance, len(ips))
+	for _, ip := range ips {
+		instances[ip] = &gostatsd.Instance{
+			ID:   "i-" + string(ip),
+			Tags: fp.Tags,
+		}
+	}
+	return instances, nil
+}
+
+type NotFound struct {
+	Counting
+}
+
+func (fp *NotFound) Name() string {
+	return "FakeProviderNotFound"
+}
+
+func (fp *NotFound) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
+	fp.count(ips...)
+	return nil, nil
+}
+
+type Failing struct {
+	Counting
+}
+
+func (fp *Failing) Name() string {
+	return "FakeFailingProvider"
+}
+
+func (fp *Failing) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
+	fp.count(ips...)
+	return nil, errors.New("clear skies, no clouds available")
+}
+
+type Transient struct {
+	call        uint64
+	FailureMode []int
+}
+
+func (fpt *Transient) Name() string {
+	return "FakeProviderTransient"
+}
+
+func (fpt *Transient) EstimatedTags() int {
+	return 1
+}
+
+func (fpt *Transient) MaxInstancesBatch() int {
+	return 1
+}
+
+func (fpt *Transient) SelfIP() (gostatsd.IP, error) {
+	return gostatsd.UnknownIP, nil
+}
+
+// Instance emulates a lookup based on the supplied criteria.
+// A failure mode of 0 is a successful lookup
+// A failure mode of 1 is nil instance, no error (lookup failure)
+// A failure mode of 2 is nil instance, with error
+// Repeats the last specified failure mode
+func (fpt *Transient) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
+	r := make(map[gostatsd.IP]*gostatsd.Instance)
+
+	c := atomic.AddUint64(&fpt.call, 1) - 1
+	if c >= uint64(len(fpt.FailureMode)) {
+		c = uint64(len(fpt.FailureMode) - 1)
+	}
+	switch fpt.FailureMode[c] {
+	case 0:
+		for _, ip := range ips {
+			r[ip] = &gostatsd.Instance{
+				ID:   string(ip),
+				Tags: gostatsd.Tags{"tag:value"},
+			}
+		}
+		return r, nil
+	case 1:
+		for _, ip := range ips {
+			r[ip] = nil
+		}
+		return r, nil
+	case 2:
+		for _, ip := range ips {
+			r[ip] = nil
+		}
+		return r, errors.New("failure mode 2")
+	default:
+		panic("fake misuse")
+	}
+}
diff --git a/pkg/statsd/handler_cloud.go b/pkg/statsd/handler_cloud.go
index 3bec00c..aed6c65 100644
--- a/pkg/statsd/handler_cloud.go
+++ b/pkg/statsd/handler_cloud.go
@@ -2,176 +2,52 @@ package statsd
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"sync/atomic"
-	"time"
 
+	"github.com/ash2k/stager/wait"
 	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/cloudproviders"
 	"github.com/atlassian/gostatsd/pkg/stats"
-
-	"github.com/ash2k/stager"
-	"github.com/ash2k/stager/wait"
-	"github.com/sirupsen/logrus"
-	"github.com/spf13/viper"
-	"golang.org/x/time/rate"
 )
 
-type lookupResult struct {
-	ip       gostatsd.IP
-	instance *gostatsd.Instance // Can be nil if lookup failed or instance was not found
-}
-
-type instanceHolder struct {
-	lastAccessNano int64
-	expires        time.Time          // When this record expires.
-	instance       *gostatsd.Instance // Can be nil if the lookup resulted in an error or instance was not found
-}
-
-func (ih *instanceHolder) updateAccess() {
-	atomic.StoreInt64(&ih.lastAccessNano, time.Now().UnixNano())
-}
-
-func (ih *instanceHolder) lastAccess() int64 {
-	return atomic.LoadInt64(&ih.lastAccessNano)
-}
-
-// CacheOptions holds cache behaviour configuration.
-type CacheOptions struct {
-	CacheRefreshPeriod        time.Duration
-	CacheEvictAfterIdlePeriod time.Duration
-	CacheTTL                  time.Duration
-	CacheNegativeTTL          time.Duration
-}
-
-// CloudHandlerFactory creates a CloudHandler based on some predefined options.
-type CloudHandlerFactory struct {
-	cloudProviderName string
-	version           string
-	cloudProvider     gostatsd.CloudProvider
-	logger            logrus.FieldLogger
-	limiter           *rate.Limiter
-	cacheOptions      CacheOptions
-}
-
-// newCloudHandlerFactory initialises a new cloud handler factory
-func newCloudHandlerFactory(cloudProviderName string, logger logrus.FieldLogger, cacheOptions CacheOptions, limiter *rate.Limiter, version string) *CloudHandlerFactory {
-	return &CloudHandlerFactory{
-		cacheOptions:      cacheOptions,
-		logger:            logger,
-		limiter:           limiter,
-		cloudProvider:     nil,
-		cloudProviderName: cloudProviderName,
-		version:           version,
-	}
-}
-
-// NewCloudHandlerFactoryFromViper initialises a new cloud handler factory with defaults based on the name of the cloud
-// provider requested.
-func NewCloudHandlerFactoryFromViper(v *viper.Viper, logger logrus.FieldLogger, version string) (*CloudHandlerFactory, error) {
-	cloudProviderName := v.GetString(ParamCloudProvider)
-	if cloudProviderName == "" {
-		return nil, nil
-	}
-
-	if _, ok := DefaultCloudProviderCacheValues[cloudProviderName]; !ok {
-		return nil, fmt.Errorf("unknown cloud provider %q", cloudProviderName)
-	}
-
-	// Cloud provider defaults
-	cpDefaultCacheOpts := DefaultCloudProviderCacheValues[cloudProviderName]
-	cpDefaultLimiterOpts := DefaultCloudProviderLimiterValues[cloudProviderName]
-
-	// Set the defaults in Viper based on the cloud provider values before we manipulate things
-	v.SetDefault(ParamCacheRefreshPeriod, cpDefaultCacheOpts.CacheRefreshPeriod)
-	v.SetDefault(ParamCacheEvictAfterIdlePeriod, cpDefaultCacheOpts.CacheEvictAfterIdlePeriod)
-	v.SetDefault(ParamCacheTTL, cpDefaultCacheOpts.CacheTTL)
-	v.SetDefault(ParamCacheNegativeTTL, cpDefaultCacheOpts.CacheNegativeTTL)
-	v.SetDefault(ParamMaxCloudRequests, cpDefaultLimiterOpts.MaxCloudRequests)
-	v.SetDefault(ParamBurstCloudRequests, cpDefaultLimiterOpts.BurstCloudRequests)
-
-	// Set the used values based on the defaults merged with any overrides
-	cacheOptions := CacheOptions{
-		CacheRefreshPeriod:        v.GetDuration(ParamCacheRefreshPeriod),
-		CacheEvictAfterIdlePeriod: v.GetDuration(ParamCacheEvictAfterIdlePeriod),
-		CacheTTL:                  v.GetDuration(ParamCacheTTL),
-		CacheNegativeTTL:          v.GetDuration(ParamCacheNegativeTTL),
-	}
-	limiter := rate.NewLimiter(rate.Limit(v.GetInt(ParamMaxCloudRequests)), v.GetInt(ParamBurstCloudRequests))
-	return newCloudHandlerFactory(cloudProviderName, logger, cacheOptions, limiter, version), nil
-}
-
-// InitCloudProvider initialises the cloud provider given some already parsed defaults.
-// This is mainly split out to enable testing the config parsing separately from the cloud providers.
-func (f *CloudHandlerFactory) InitCloudProvider(v *viper.Viper) error {
-	cloudProvider, err := cloudproviders.Init(f.cloudProviderName, v, f.logger, f.version)
-	if err != nil {
-		return err
-	}
-	f.cloudProvider = cloudProvider
-	return nil
-}
-
-// NewCloudHandler creates a new Cloud Handler based on the options set in the CloudHandlerFactory.
-func (f *CloudHandlerFactory) NewCloudHandler(handler gostatsd.PipelineHandler) *CloudHandler {
-	return NewCloudHandler(f.cloudProvider, handler, f.logger, f.limiter, f.cacheOptions)
-}
-
 // CloudHandler enriches metrics and events with additional information fetched from cloud provider.
 type CloudHandler struct {
-	// statsCacheHit is accessed by any go routine, must use atomic ops
-	statsCacheHit uint64 // Cumulative number of cache hits
+	// These fields are accessed by any go routine, must use atomic ops
+	statsCacheHit  uint64 // Cumulative number of cache hits
+	statsCacheMiss uint64 // Cumulative number of cache misses
 
 	// All other stats fields may only be read or written by the main CloudHandler.Run goroutine
-	statsCacheLateHit         uint64 // Cumulative number of late cache hits
-	statsCacheMiss            uint64 // Cumulative number of cache misses
-	statsCachePositive        uint64 // Absolute number of positive entries in cache
-	statsCacheNegative        uint64 // Absolute number of negative entries in cache
-	statsCacheRefreshPositive uint64 // Cumulative number of positive refreshes (ie, a refresh which succeeded)
-	statsCacheRefreshNegative uint64 // Cumulative number of negative refreshes (ie, a refresh which failed and used old data)
-	statsMetricItemsQueued    uint64 // Absolute number of metrics queued, waiting for a CP to respond
-	statsMetricHostsQueued    uint64 // Absolute number of IPs waiting for a CP to respond for metrics
-	statsEventItemsQueued     uint64 // Absolute number of events queued, waiting for a CP to respond
-	statsEventHostsQueued     uint64 // Absolute number of IPs waiting for a CP to respond for events
+	statsMetricItemsQueued uint64 // Absolute number of metrics queued, waiting for a CP to respond
+	statsMetricHostsQueued uint64 // Absolute number of IPs waiting for a CP to respond for metrics
+	statsEventItemsQueued  uint64 // Absolute number of events queued, waiting for a CP to respond
+	statsEventHostsQueued  uint64 // Absolute number of IPs waiting for a CP to respond for events
 
-	// emitChan triggers a write of all the current stats when it is given a Statser
-	emitChan chan stats.Statser
-
-	cacheOpts       CacheOptions
-	cloud           gostatsd.CloudProvider // Cloud provider interface
+	cachedInstances gostatsd.CachedInstances
 	handler         gostatsd.PipelineHandler
-	limiter         *rate.Limiter
-	metricSource    chan []*gostatsd.Metric
-	eventSource     chan *gostatsd.Event
+	incomingMetrics chan []*gostatsd.Metric
+	incomingEvents  chan *gostatsd.Event
+
+	// emitChan triggers a write of all the current stats when it is given a Statser
+	emitChan        chan stats.Statser
 	awaitingEvents  map[gostatsd.IP][]*gostatsd.Event
 	awaitingMetrics map[gostatsd.IP][]*gostatsd.Metric
 	toLookupIPs     []gostatsd.IP
 	wg              sync.WaitGroup
 
-	rw    sync.RWMutex // Protects cache
-	cache map[gostatsd.IP]*instanceHolder
-
-	logger logrus.FieldLogger
-
 	estimatedTags int
 }
 
 // NewCloudHandler initialises a new cloud handler.
-func NewCloudHandler(cloud gostatsd.CloudProvider, handler gostatsd.PipelineHandler, logger logrus.FieldLogger, limiter *rate.Limiter, cacheOptions CacheOptions) *CloudHandler {
+func NewCloudHandler(cachedInstances gostatsd.CachedInstances, handler gostatsd.PipelineHandler) *CloudHandler {
 	return &CloudHandler{
-		cacheOpts:       cacheOptions,
-		cloud:           cloud,
+		cachedInstances: cachedInstances,
 		handler:         handler,
-		limiter:         limiter,
-		metricSource:    make(chan []*gostatsd.Metric),
-		eventSource:     make(chan *gostatsd.Event),
+		incomingMetrics: make(chan []*gostatsd.Metric),
+		incomingEvents:  make(chan *gostatsd.Event),
 		emitChan:        make(chan stats.Statser),
 		awaitingEvents:  make(map[gostatsd.IP][]*gostatsd.Event),
 		awaitingMetrics: make(map[gostatsd.IP][]*gostatsd.Metric),
-		cache:           make(map[gostatsd.IP]*instanceHolder),
-		estimatedTags:   handler.EstimatedTags() + cloud.EstimatedTags(),
-		logger:          logger,
+		estimatedTags:   handler.EstimatedTags() + cachedInstances.EstimatedTags(),
 	}
 }
 
@@ -185,7 +61,6 @@ func (ch *CloudHandler) DispatchMetrics(ctx context.Context, metrics []*gostatsd
 	var toHandle []*gostatsd.Metric
 	for _, m := range metrics {
 		if ch.updateTagsAndHostname(m.SourceIP, &m.Tags, &m.Hostname) {
-			atomic.AddUint64(&ch.statsCacheHit, 1)
 			toDispatch = append(toDispatch, m)
 		} else {
 			toHandle = append(toHandle, m)
@@ -199,7 +74,7 @@ func (ch *CloudHandler) DispatchMetrics(ctx context.Context, metrics []*gostatsd
 	if len(toHandle) > 0 {
 		select {
 		case <-ctx.Done():
-		case ch.metricSource <- toHandle:
+		case ch.incomingMetrics <- toHandle:
 		}
 	}
 }
@@ -214,7 +89,6 @@ func (ch *CloudHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.Metr
 
 func (ch *CloudHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	if ch.updateTagsAndHostname(e.SourceIP, &e.Tags, &e.Hostname) {
-		atomic.AddUint64(&ch.statsCacheHit, 1)
 		ch.handler.DispatchEvent(ctx, e)
 		return
 	}
@@ -222,7 +96,7 @@ func (ch *CloudHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	select {
 	case <-ctx.Done():
 		ch.wg.Done()
-	case ch.eventSource <- e:
+	case ch.incomingEvents <- e:
 	}
 }
 
@@ -233,7 +107,7 @@ func (ch *CloudHandler) WaitForEvents() {
 }
 
 func (ch *CloudHandler) RunMetrics(ctx context.Context, statser stats.Statser) {
-	if me, ok := ch.cloud.(MetricEmitter); ok {
+	if me, ok := ch.cachedInstances.(MetricEmitter); ok {
 		var wg wait.Group
 		defer wg.Wait()
 		wg.Start(func() {
@@ -272,13 +146,7 @@ func (ch *CloudHandler) scheduleEmit(ctx context.Context, statser stats.Statser)
 func (ch *CloudHandler) emit(statser stats.Statser) {
 	// atomic
 	statser.Gauge("cloudprovider.cache_hit", float64(atomic.LoadUint64(&ch.statsCacheHit)), nil)
-	// regular
-	statser.Gauge("cloudprovider.cache_late_hit", float64(ch.statsCacheLateHit), nil)
-	statser.Gauge("cloudprovider.cache_miss", float64(ch.statsCacheMiss), nil)
-	statser.Gauge("cloudprovider.cache_positive", float64(ch.statsCachePositive), nil)
-	statser.Gauge("cloudprovider.cache_negative", float64(ch.statsCacheNegative), nil)
-	statser.Gauge("cloudprovider.cache_refresh_positive", float64(ch.statsCacheRefreshPositive), nil)
-	statser.Gauge("cloudprovider.cache_refresh_negative", float64(ch.statsCacheRefreshNegative), nil)
+	statser.Gauge("cloudprovider.cache_miss", float64(atomic.LoadUint64(&ch.statsCacheMiss)), nil)
 	t := gostatsd.Tags{"type:metric"}
 	statser.Gauge("cloudprovider.hosts_queued", float64(ch.statsMetricHostsQueued), t)
 	statser.Gauge("cloudprovider.items_queued", float64(ch.statsMetricItemsQueued), t)
@@ -288,58 +156,27 @@ func (ch *CloudHandler) emit(statser stats.Statser) {
 }
 
 func (ch *CloudHandler) Run(ctx context.Context) {
-	// IPs to lookup. Can make the channel bigger or smaller but this is the perfect size.
-	toLookup := make(chan gostatsd.IP, ch.cloud.MaxInstancesBatch())
-	var toLookupC chan<- gostatsd.IP
-	var toLookupIP gostatsd.IP
-
-	lookupResults := make(chan *lookupResult)
-
-	ld := lookupDispatcher{
-		limiter:       ch.limiter,
-		cloud:         ch.cloud,
-		toLookup:      toLookup,
-		lookupResults: lookupResults,
-		logger:        ch.logger,
-	}
-
-	// Stager will perform ordered, graceful shutdown. Stage by stage in reverse startup order.
-	stgr := stager.New()
-	defer stgr.Shutdown() // Wait for CloudProvider and lookupDispatcher to stop
-
-	stage := stgr.NextStage()
-
-	// If the cloud provider implements runnable, it has things to do. It must start before anything that
-	// accesses it
-	if ch.cloud != nil {
-		if r, ok := ch.cloud.(gostatsd.Runner); ok {
-			go r.Run(ctx) // fork
-		}
-	}
-
-	stage.StartWithContext(ld.run) // Start the lookup mechanism
-
-	refreshTicker := time.NewTicker(ch.cacheOpts.CacheRefreshPeriod)
-	defer refreshTicker.Stop()
-	// No locking for ch.cache READ access required - this goroutine owns the object and only it mutates it.
-	// So reads from the same goroutine are always safe (no concurrent mutations).
-	// When we mutate the cache, we hold the exclusive (write) lock to avoid concurrent reads.
-	// When we read from the cache from other goroutines, we obtain the read lock.
+	var (
+		toLookupC  chan<- gostatsd.IP
+		toLookupIP gostatsd.IP
+	)
+	infoSource := ch.cachedInstances.InfoSource()
+	ipSink := ch.cachedInstances.IpSink()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case toLookupC <- toLookupIP:
-			toLookupIP = gostatsd.UnknownIP
-			toLookupC = nil // ip has been sent; if there is nothing to send, will block
-		case lr := <-lookupResults:
-			ch.handleLookupResult(ctx, lr)
-		case t := <-refreshTicker.C:
-			ch.doRefresh(ctx, t)
-		case metrics := <-ch.metricSource:
-			ch.handleMetrics(ctx, metrics)
-		case e := <-ch.eventSource:
-			ch.handleEvent(ctx, e)
+			toLookupIP = gostatsd.UnknownIP // Enable GC
+			toLookupC = nil                 // ip has been sent; if there is nothing to send, will block
+		case info := <-infoSource:
+			ch.handleInstanceInfo(ctx, info)
+		case metrics := <-ch.incomingMetrics:
+			// Add metrics to awaitingMetrics, accumulate IPs to lookup
+			ch.handleIncomingMetrics(metrics)
+		case e := <-ch.incomingEvents:
+			// Add event to awaitingEvents, accumulate IPs to lookup
+			ch.handleIncomingEvent(e)
 		case statser := <-ch.emitChan:
 			ch.emit(statser)
 		}
@@ -348,125 +185,50 @@ func (ch *CloudHandler) Run(ctx context.Context) {
 			toLookupIP = ch.toLookupIPs[last]
 			ch.toLookupIPs[last] = gostatsd.UnknownIP // Enable GC
 			ch.toLookupIPs = ch.toLookupIPs[:last]
-			toLookupC = toLookup
-		}
-	}
-}
-
-func (ch *CloudHandler) doRefresh(ctx context.Context, t time.Time) {
-	var toDelete []gostatsd.IP
-	now := t.UnixNano()
-	idleNano := ch.cacheOpts.CacheEvictAfterIdlePeriod.Nanoseconds()
-
-	for ip, holder := range ch.cache {
-		if now-holder.lastAccess() > idleNano {
-			// Entry was not used recently, remove it.
-			toDelete = append(toDelete, ip)
-			if holder.instance == nil {
-				ch.statsCacheNegative--
-			} else {
-				ch.statsCachePositive--
-			}
-		} else if t.After(holder.expires) {
-			// Entry needs a refresh.
-			ch.toLookupIPs = append(ch.toLookupIPs, ip)
-		}
-	}
-
-	if len(toDelete) > 0 {
-		ch.rw.Lock()
-		defer ch.rw.Unlock()
-		for _, ip := range toDelete {
-			delete(ch.cache, ip)
+			toLookupC = ipSink
 		}
 	}
 }
 
-func (ch *CloudHandler) handleLookupResult(ctx context.Context, lr *lookupResult) {
-	var ttl time.Duration
-	if lr.instance == nil {
-		ttl = ch.cacheOpts.CacheNegativeTTL
-	} else {
-		ttl = ch.cacheOpts.CacheTTL
-	}
-	now := time.Now()
-	newHolder := &instanceHolder{
-		expires:  now.Add(ttl),
-		instance: lr.instance,
-	}
-	currentHolder := ch.cache[lr.ip]
-	if currentHolder == nil {
-		// Not in cache, count it
-		if lr.instance == nil {
-			ch.statsCacheNegative++
-		} else {
-			ch.statsCachePositive++
-		}
-		newHolder.lastAccessNano = now.UnixNano()
-	} else {
-		// In cache, don't count it
-		newHolder.lastAccessNano = currentHolder.lastAccess()
-		if lr.instance == nil {
-			// Use the old instance if there was a lookup error.
-			newHolder.instance = currentHolder.instance
-			ch.statsCacheRefreshNegative++
-		} else {
-			if currentHolder.instance == nil && newHolder.instance != nil {
-				// An entry has flipped from invalid to valid
-				ch.statsCacheNegative--
-				ch.statsCachePositive++
-			}
-			ch.statsCacheRefreshPositive++
-		}
-	}
-	func() {
-		ch.rw.Lock()
-		defer ch.rw.Unlock()
-		ch.cache[lr.ip] = newHolder
-	}()
-	metrics := ch.awaitingMetrics[lr.ip]
-	if metrics != nil {
-		delete(ch.awaitingMetrics, lr.ip)
+func (ch *CloudHandler) handleInstanceInfo(ctx context.Context, info gostatsd.InstanceInfo) {
+	metrics := ch.awaitingMetrics[info.IP]
+	if len(metrics) > 0 {
+		delete(ch.awaitingMetrics, info.IP)
 		ch.statsMetricItemsQueued -= uint64(len(metrics))
 		ch.statsMetricHostsQueued--
-		go ch.updateAndDispatchMetrics(ctx, lr.instance, metrics)
+		go ch.updateAndDispatchMetrics(ctx, info.Instance, metrics)
 	}
-	events := ch.awaitingEvents[lr.ip]
-	if events != nil {
-		delete(ch.awaitingEvents, lr.ip)
+	events := ch.awaitingEvents[info.IP]
+	if len(events) > 0 {
+		delete(ch.awaitingEvents, info.IP)
 		ch.statsEventItemsQueued -= uint64(len(events))
 		ch.statsEventHostsQueued--
-		go ch.updateAndDispatchEvents(ctx, lr.instance, events)
+		go ch.updateAndDispatchEvents(ctx, info.Instance, events)
 	}
 }
 
-func (ch *CloudHandler) handleMetrics(ctx context.Context, metrics []*gostatsd.Metric) {
-	var toDispatch []*gostatsd.Metric
+func (ch *CloudHandler) handleIncomingMetrics(metrics []*gostatsd.Metric) {
 	for _, m := range metrics {
-		holder, ok := ch.cache[m.SourceIP]
-		if ok {
-			// While metric was in the queue the cache was primed. Use the value.
-			ch.statsCacheLateHit++
-			holder.updateAccess()
-			updateInplace(&m.Tags, &m.Hostname, holder.instance)
-			toDispatch = append(toDispatch, m)
-		} else {
-			// Still nothing in the cache.
-			queue := ch.awaitingMetrics[m.SourceIP]
-			ch.awaitingMetrics[m.SourceIP] = append(queue, m)
-			if len(queue) == 0 {
-				// This is the first metric in the queue
-				ch.toLookupIPs = append(ch.toLookupIPs, m.SourceIP)
-				ch.statsMetricHostsQueued++
-			}
-			ch.statsMetricItemsQueued++
-			ch.statsCacheMiss++
+		queue := ch.awaitingMetrics[m.SourceIP]
+		ch.awaitingMetrics[m.SourceIP] = append(queue, m)
+		if len(queue) == 0 && len(ch.awaitingEvents[m.SourceIP]) == 0 {
+			// This is the first metric for that IP in the queue. Need to fetch an Instance for this IP.
+			ch.toLookupIPs = append(ch.toLookupIPs, m.SourceIP)
+			ch.statsMetricHostsQueued++
 		}
 	}
+	ch.statsMetricItemsQueued += uint64(len(metrics))
+}
 
-	if len(toDispatch) > 0 {
-		go ch.handler.DispatchMetrics(ctx, toDispatch)
+func (ch *CloudHandler) handleIncomingEvent(e *gostatsd.Event) {
+	queue := ch.awaitingEvents[e.SourceIP]
+	ch.awaitingEvents[e.SourceIP] = append(queue, e)
+	if len(queue) == 0 && len(ch.awaitingMetrics[e.SourceIP]) == 0 {
+		// This is the first event for that IP in the queue. Need to fetch an Instance for this IP.
+		ch.toLookupIPs = append(ch.toLookupIPs, e.SourceIP)
+		ch.statsEventHostsQueued++
 	}
+	ch.statsEventItemsQueued++
 }
 
 func (ch *CloudHandler) updateAndDispatchMetrics(ctx context.Context, instance *gostatsd.Instance, metrics []*gostatsd.Metric) {
@@ -476,27 +238,6 @@ func (ch *CloudHandler) updateAndDispatchMetrics(ctx context.Context, instance *
 	ch.handler.DispatchMetrics(ctx, metrics)
 }
 
-func (ch *CloudHandler) handleEvent(ctx context.Context, e *gostatsd.Event) {
-	holder, ok := ch.cache[e.SourceIP]
-	if ok {
-		// While event was in the queue the cache was primed. Use the value.
-		holder.updateAccess()
-		ch.statsCacheLateHit++
-		go ch.updateAndDispatchEvents(ctx, holder.instance, []*gostatsd.Event{e})
-	} else {
-		// Still nothing in the cache.
-		queue := ch.awaitingEvents[e.SourceIP]
-		ch.awaitingEvents[e.SourceIP] = append(queue, e)
-		if len(queue) == 0 {
-			// This is the first event in the queue
-			ch.toLookupIPs = append(ch.toLookupIPs, e.SourceIP)
-			ch.statsEventHostsQueued++
-		}
-		ch.statsEventItemsQueued++
-		ch.statsCacheMiss++
-	}
-}
-
 func (ch *CloudHandler) updateAndDispatchEvents(ctx context.Context, instance *gostatsd.Instance, events []*gostatsd.Event) {
 	var dispatched int
 	defer func() {
@@ -509,26 +250,25 @@ func (ch *CloudHandler) updateAndDispatchEvents(ctx context.Context, instance *g
 	}
 }
 
-func (ch *CloudHandler) updateTagsAndHostname(ip gostatsd.IP, tags *gostatsd.Tags, hostname *string) bool {
-	instance, ok := ch.getInstance(ip)
-	if ok {
+func (ch *CloudHandler) updateTagsAndHostname(ip gostatsd.IP, tags *gostatsd.Tags, hostname *string) bool /*is a cache hit*/ {
+	instance, cacheHit := ch.getInstance(ip)
+	if cacheHit {
 		updateInplace(tags, hostname, instance)
 	}
-	return ok
+	return cacheHit
 }
 
-func (ch *CloudHandler) getInstance(ip gostatsd.IP) (*gostatsd.Instance, bool) {
+func (ch *CloudHandler) getInstance(ip gostatsd.IP) (*gostatsd.Instance, bool /*is a cache hit*/) {
 	if ip == gostatsd.UnknownIP {
 		return nil, true
 	}
-	ch.rw.RLock()
-	holder, ok := ch.cache[ip]
-	ch.rw.RUnlock()
-	if ok {
-		holder.updateAccess()
-		return holder.instance, true
+	instance, cacheHit := ch.cachedInstances.Peek(ip)
+	if !cacheHit {
+		atomic.AddUint64(&ch.statsCacheMiss, 1)
+		return nil, false
 	}
-	return nil, false
+	atomic.AddUint64(&ch.statsCacheHit, 1)
+	return instance, true
 }
 
 func updateInplace(tags *gostatsd.Tags, hostname *string, instance *gostatsd.Instance) {
diff --git a/pkg/statsd/handler_http_forwarder_v2.go b/pkg/statsd/handler_http_forwarder_v2.go
index b37f559..c500b0f 100644
--- a/pkg/statsd/handler_http_forwarder_v2.go
+++ b/pkg/statsd/handler_http_forwarder_v2.go
@@ -64,7 +64,7 @@ func NewHttpForwarderHandlerV2FromViper(logger logrus.FieldLogger, v *viper.Vipe
 	subViper.SetDefault("api-endpoint", defaultApiEndpoint)
 	subViper.SetDefault("max-requests", defaultMaxRequests)
 	subViper.SetDefault("max-request-elapsed-time", defaultMaxRequestElapsedTime)
-	subViper.SetDefault("consolidator-slots", v.GetInt(ParamMaxParsers))
+	subViper.SetDefault("consolidator-slots", v.GetInt(gostatsd.ParamMaxParsers))
 	subViper.SetDefault("flush-interval", defaultConsolidatorFlushInterval)
 
 	return NewHttpForwarderHandlerV2(
@@ -172,7 +172,7 @@ func (hfh *HttpForwarderHandlerV2) DispatchMetricMap(ctx context.Context, mm *go
 	hfh.consolidator.ReceiveMetricMap(mm)
 }
 
-func (hfh *HttpForwarderHandlerV2) RunMetrics(ctx context.Context) {
+func (hfh *HttpForwarderHandlerV2) RunMetricsContext(ctx context.Context) {
 	statser := stats.FromContext(ctx)
 
 	notify, cancel := statser.RegisterFlush()
diff --git a/pkg/statsd/parser.go b/pkg/statsd/parser.go
index 1bceab3..64bf4cf 100644
--- a/pkg/statsd/parser.go
+++ b/pkg/statsd/parser.go
@@ -62,7 +62,7 @@ func NewDatagramParser(in <-chan []*Datagram, ns string, ignoreHost bool, estima
 	}
 }
 
-func (dp *DatagramParser) RunMetrics(ctx context.Context) {
+func (dp *DatagramParser) RunMetricsContext(ctx context.Context) {
 	statser := stats.FromContext(ctx)
 	flushed, unregister := statser.RegisterFlush()
 	defer unregister()
diff --git a/pkg/statsd/receiver.go b/pkg/statsd/receiver.go
index 61452de..55570ab 100644
--- a/pkg/statsd/receiver.go
+++ b/pkg/statsd/receiver.go
@@ -48,7 +48,7 @@ func NewDatagramReceiver(out chan<- []*Datagram, sf SocketFactory, numReaders, r
 	}
 }
 
-func (dr *DatagramReceiver) RunMetrics(ctx context.Context) {
+func (dr *DatagramReceiver) RunMetricsContext(ctx context.Context) {
 	statser := stats.FromContext(ctx)
 	flushed, unregister := statser.RegisterFlush()
 	defer unregister()
diff --git a/pkg/statsd/statsd.go b/pkg/statsd/statsd.go
index 68e7f58..87a3f4a 100644
--- a/pkg/statsd/statsd.go
+++ b/pkg/statsd/statsd.go
@@ -5,16 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"net"
-	"os"
-	"strconv"
 	"time"
 
+	"github.com/ash2k/stager"
 	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
 	"github.com/atlassian/gostatsd/pkg/web"
-
-	"github.com/ash2k/stager"
 	"github.com/libp2p/go-reuseport"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
@@ -25,7 +22,7 @@ import (
 // the statsd server. These can either be set via command line or directly.
 type Server struct {
 	Backends                  []gostatsd.Backend
-	CloudHandlerFactory       *CloudHandlerFactory
+	CachedInstances           gostatsd.CachedInstances
 	InternalTags              gostatsd.Tags
 	InternalNamespace         string
 	DefaultTags               gostatsd.Tags
@@ -52,6 +49,7 @@ type Server struct {
 	BadLineRateLimitPerSecond rate.Limit
 	ServerMode                string
 	Hostname                  string
+	SelfIP                    gostatsd.IP
 	LogRawMetric              bool
 	Viper                     *viper.Viper
 	TransportPool             *transport.TransportPool
@@ -88,12 +86,6 @@ func socketFactory(metricsAddr string, connPerReader bool) SocketFactory {
 func (s *Server) createStandaloneSink() (gostatsd.PipelineHandler, []gostatsd.Runnable, error) {
 	var runnables []gostatsd.Runnable
 
-	for _, backend := range s.Backends {
-		if r, ok := backend.(gostatsd.Runner); ok {
-			runnables = append(runnables, r.Run)
-		}
-	}
-
 	// Create the backend handler
 	factory := agrFactory{
 		percentThresholds: s.PercentThreshold,
@@ -125,7 +117,7 @@ func (s *Server) createForwarderSink() (gostatsd.PipelineHandler, []gostatsd.Run
 	// Create a Flusher, this is primarily for all the periodic metrics which are emitted.
 	flusher := NewMetricFlusher(s.FlushInterval, nil, s.Backends)
 
-	return forwarderHandler, []gostatsd.Runnable{forwarderHandler.Run, forwarderHandler.RunMetrics, flusher.Run}, nil
+	return forwarderHandler, []gostatsd.Runnable{forwarderHandler.Run, forwarderHandler.RunMetricsContext, flusher.Run}, nil
 }
 
 func (s *Server) createFinalSink() (gostatsd.PipelineHandler, []gostatsd.Runnable, error) {
@@ -149,23 +141,16 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 	handler = NewTagHandlerFromViper(s.Viper, handler, s.DefaultTags)
 
 	// Create the cloud handler
-	ip := gostatsd.UnknownIP
-	if s.CloudHandlerFactory != nil {
-		cloudHandler := s.CloudHandlerFactory.NewCloudHandler(handler)
-		runnables = append(runnables, cloudHandler.Run)
+	if s.CachedInstances != nil {
+		cloudHandler := NewCloudHandler(s.CachedInstances, handler)
+		runnables = gostatsd.MaybeAppendRunnable(runnables, cloudHandler)
 		handler = cloudHandler
-		selfIP, err2 := cloudHandler.cloud.SelfIP()
-		if err2 != nil {
-			log.Warnf("Failed to get self ip: %v", err2)
-		} else {
-			ip = selfIP
-		}
 	}
 
 	// Create the heartbeater
 	if s.HeartbeatEnabled {
 		hb := stats.NewHeartBeater("heartbeat", s.HeartbeatTags)
-		runnables = append(runnables, hb.Run)
+		runnables = gostatsd.MaybeAppendRunnable(runnables, hb)
 	}
 
 	// Open receiver <-> parser chan
@@ -173,22 +158,19 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 
 	// Create the Parser
 	parser := NewDatagramParser(datagrams, s.Namespace, s.IgnoreHost, s.EstimatedTags, handler, s.BadLineRateLimitPerSecond, s.LogRawMetric)
-	runnables = append(runnables, parser.RunMetrics)
+	runnables = append(runnables, parser.RunMetricsContext)
 	for i := 0; i < s.MaxParsers; i++ {
 		runnables = append(runnables, parser.Run)
 	}
 
 	// Create the Receiver
 	receiver := NewDatagramReceiver(datagrams, sf, s.MaxReaders, s.ReceiveBatchSize)
-	runnables = append(runnables, receiver.RunMetrics)
-	runnables = append(runnables, receiver.Run) // loop is contained in Run to keep additional logic contained
+	runnables = gostatsd.MaybeAppendRunnable(runnables, receiver)
 
 	// Create the Statser
 	hostname := s.Hostname
 	statser := s.createStatser(hostname, handler)
-	if runner, ok := statser.(gostatsd.Runner); ok {
-		runnables = append(runnables, runner.Run)
-	}
+	runnables = gostatsd.MaybeAppendRunnable(runnables, statser)
 
 	// Create any http servers
 	httpServers, err := web.NewHttpServersFromViper(s.Viper, log.StandardLogger(), handler)
@@ -196,7 +178,7 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 		return err
 	}
 	for _, server := range httpServers {
-		runnables = append(runnables, server.Run)
+		runnables = gostatsd.MaybeAppendRunnable(runnables, server)
 	}
 
 	// Start the world!
@@ -209,8 +191,8 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 
 	// Send events on start and on stop
 	// TODO: Push these in to statser
-	defer sendStopEvent(handler, ip, hostname)
-	sendStartEvent(runCtx, handler, ip, hostname)
+	defer sendStopEvent(handler, s.SelfIP, hostname)
+	sendStartEvent(runCtx, handler, s.SelfIP, hostname)
 
 	// Listen until done
 	<-ctx.Done()
@@ -219,9 +201,9 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 
 func (s *Server) createStatser(hostname string, handler gostatsd.PipelineHandler) stats.Statser {
 	switch s.StatserType {
-	case StatserNull:
+	case gostatsd.StatserNull:
 		return stats.NewNullStatser()
-	case StatserLogging:
+	case gostatsd.StatserLogging:
 		return stats.NewLoggingStatser(s.InternalTags, log.NewEntry(log.New()))
 	default:
 		namespace := s.Namespace
@@ -261,15 +243,6 @@ func sendStopEvent(handler gostatsd.PipelineHandler, selfIP gostatsd.IP, hostnam
 	handler.WaitForEvents()
 }
 
-func getHost() string {
-	host, err := os.Hostname()
-	if err != nil {
-		log.Warnf("Cannot get hostname: %v", err)
-		return ""
-	}
-	return host
-}
-
 type agrFactory struct {
 	percentThresholds []float64
 	expiryInterval    time.Duration
@@ -280,11 +253,3 @@ type agrFactory struct {
 func (af *agrFactory) Create() Aggregator {
 	return NewMetricAggregator(af.percentThresholds, af.expiryInterval, af.disabledSubtypes, af.histogramLimit)
 }
-
-func toStringSlice(fs []float64) []string {
-	s := make([]string, len(fs))
-	for i, f := range fs {
-		s[i] = strconv.FormatFloat(f, 'f', -1, 64)
-	}
-	return s
-}
diff --git a/pkg/web/http_receiver_v2.go b/pkg/web/http_receiver_v2.go
index 42dfabc..4d31ac8 100644
--- a/pkg/web/http_receiver_v2.go
+++ b/pkg/web/http_receiver_v2.go
@@ -37,7 +37,7 @@ func newRawHttpHandlerV2(logger logrus.FieldLogger, serverName string, handler g
 	}
 }
 
-func (rhh *rawHttpHandlerV2) RunMetrics(ctx context.Context) {
+func (rhh *rawHttpHandlerV2) RunMetricsContext(ctx context.Context) {
 	statser := stats.FromContext(ctx).WithTags([]string{"server-name:" + rhh.serverName})
 
 	notify, cancel := statser.RegisterFlush()
diff --git a/pkg/web/httpservers.go b/pkg/web/httpservers.go
index 8478ec7..0155727 100644
--- a/pkg/web/httpservers.go
+++ b/pkg/web/httpservers.go
@@ -192,7 +192,7 @@ func (hs *httpServer) Run(ctx context.Context) {
 	if hs.rawMetricsV2 != nil {
 		var wg wait.Group
 		defer wg.Wait()
-		wg.StartWithContext(ctx, hs.rawMetricsV2.RunMetrics)
+		wg.StartWithContext(ctx, hs.rawMetricsV2.RunMetricsContext)
 	}
 
 	server := &http.Server{
diff --git a/types.go b/types.go
index 19ab3dd..29a2ac2 100644
--- a/types.go
+++ b/types.go
@@ -48,11 +48,25 @@ type TimerSubtypes struct {
 }
 
 // Runnable is a long running function intended to be launched in a goroutine.
-type Runnable func(ctx context.Context)
+type Runnable func(context.Context)
 
 // Runner exposes a Runnable through an interface
 type Runner interface {
-	Run(ctx context.Context)
+	Run(context.Context)
+}
+
+type MetricsRunner interface {
+	RunMetricsContext(context.Context)
+}
+
+func MaybeAppendRunnable(runnables []Runnable, maybeRunner interface{}) []Runnable {
+	if r, ok := maybeRunner.(Runner); ok {
+		runnables = append(runnables, r.Run)
+	}
+	if r, ok := maybeRunner.(MetricsRunner); ok {
+		runnables = append(runnables, r.RunMetricsContext)
+	}
+	return runnables
 }
 
 // RawMetricHandler is an interface that accepts a Metric for processing.  Raw refers to pre-aggregation, not
@@ -68,7 +82,7 @@ type PipelineHandler interface {
 	// EstimatedTags returns a guess for how many tags to pre-allocate
 	EstimatedTags() int
 	// DispatchEvent dispatches event to the next step in a pipeline.
-	DispatchEvent(ctx context.Context, e *Event)
+	DispatchEvent(context.Context, *Event)
 	// WaitForEvents waits for all event-dispatching goroutines to finish.
 	WaitForEvents()
 }


Test patch:
diff --git a/cmd/tester/main.go b/cmd/tester/main.go
index 9c9ec4f..002a421 100644
--- a/cmd/tester/main.go
+++ b/cmd/tester/main.go
@@ -8,9 +8,9 @@ import (
 	"runtime/pprof"
 	"time"
 
+	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/fakesocket"
 	"github.com/atlassian/gostatsd/pkg/statsd"
-
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
@@ -37,14 +37,14 @@ func main() {
 	}
 	if s.Benchmark != 0 {
 		server := statsd.Server{
-			DefaultTags:      statsd.DefaultTags,
-			ExpiryInterval:   statsd.DefaultExpiryInterval,
-			FlushInterval:    statsd.DefaultFlushInterval,
-			MaxReaders:       statsd.DefaultMaxReaders,
-			MaxWorkers:       statsd.DefaultMaxWorkers,
-			MaxQueueSize:     statsd.DefaultMaxQueueSize,
-			PercentThreshold: statsd.DefaultPercentThreshold,
-			ReceiveBatchSize: statsd.DefaultReceiveBatchSize,
+			DefaultTags:      gostatsd.DefaultTags,
+			ExpiryInterval:   gostatsd.DefaultExpiryInterval,
+			FlushInterval:    gostatsd.DefaultFlushInterval,
+			MaxReaders:       gostatsd.DefaultMaxReaders,
+			MaxWorkers:       gostatsd.DefaultMaxWorkers,
+			MaxQueueSize:     gostatsd.DefaultMaxQueueSize,
+			PercentThreshold: gostatsd.DefaultPercentThreshold,
+			ReceiveBatchSize: gostatsd.DefaultReceiveBatchSize,
 			Viper:            viper.New(),
 		}
 		ctx, cancelFunc := context.WithTimeout(context.Background(), time.Duration(s.Benchmark)*time.Second)
diff --git a/pkg/cachedinstances/cloudprovider/cached_cloud_provider_test.go b/pkg/cachedinstances/cloudprovider/cached_cloud_provider_test.go
new file mode 100644
index 0000000..146e893
--- /dev/null
+++ b/pkg/cachedinstances/cloudprovider/cached_cloud_provider_test.go
@@ -0,0 +1,67 @@
+package cloudprovider
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/ash2k/stager/wait"
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/cloudproviders/fakeprovider"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/time/rate"
+)
+
+func TestCachedCloudProviderExpirationAndRefresh(t *testing.T) {
+	// These still use a real clock, which means they're more susceptible to
+	// CPU load triggering a race condition, therefore there's no t.Parallel()
+	fp := &fakeprovider.IP{}
+	/*
+		Note: lookup reads in to a batch for up to 10ms
+
+		T+0: IP is sent to lookup
+		T+10: lookup is performed, cached with eviction time = T+10+50=60, refresh time = T+10+10=20
+		T+11: refresh loop, nothing to do
+		T+20: cache entry passes refresh time
+		T+22: refresh loop, cache item is dispatched for refreshing
+		T+32: cache lookup is performed, eviction time is unchanged, refresh time = T+32+10=42
+		T+33: refresh loop, nothing to do
+		T+42: cache entry passes refresh time
+		T+44: refresh loop, cache item is dispatched for refreshing
+		T+54: cache lookup is performed, eviction time is unchanged, refresh time = T+54+10=64
+		T+55: refresh loop, nothing to do
+		T+60: cache entry passes expiry time
+		T+64: cache entry passes refresh time
+		T+66: refresh loop, entry is expired
+		T+70: sleep completes
+	*/
+	ci := NewCachedCloudProvider(logrus.StandardLogger(), rate.NewLimiter(100, 120), fp, gostatsd.CacheOptions{
+		CacheRefreshPeriod:        11 * time.Millisecond,
+		CacheEvictAfterIdlePeriod: 50 * time.Millisecond,
+		CacheTTL:                  10 * time.Millisecond,
+		CacheNegativeTTL:          100 * time.Millisecond,
+	})
+	var wg wait.Group
+	defer wg.Wait()
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	wg.StartWithContext(ctx, ci.Run)
+	const peekIp gostatsd.IP = "1.2.3.4"
+	instance, exists := ci.Peek(peekIp)
+	require.Nil(t, instance)
+	require.False(t, exists)
+	ci.IpSink() <- peekIp
+	time.Sleep(70 * time.Millisecond) // Should be refreshed couple of times and evicted.
+
+	cancelFunc()
+	wg.Wait()
+
+	// Cache might refresh multiple times, ensure it only refreshed with the expected IP
+	for _, ip := range fp.IPs() {
+		assert.Equal(t, peekIp, ip)
+	}
+	assert.GreaterOrEqual(t, len(fp.IPs()), 2) // Ensure it does at least 1 lookup + 1 refresh
+	assert.Zero(t, len(ci.cache))              // Ensure it eventually expired
+}
diff --git a/pkg/statsd/handler_cloud_test.go b/pkg/statsd/handler_cloud_test.go
index d806093..d68c58d 100644
--- a/pkg/statsd/handler_cloud_test.go
+++ b/pkg/statsd/handler_cloud_test.go
@@ -2,20 +2,15 @@ package statsd
 
 import (
 	"context"
-	"errors"
 	"sort"
-	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
-	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/cloudproviders/aws"
-	"github.com/atlassian/gostatsd/pkg/cloudproviders/k8s"
-
 	"github.com/ash2k/stager/wait"
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/cachedinstances/cloudprovider"
+	"github.com/atlassian/gostatsd/pkg/cloudproviders/fakeprovider"
 	"github.com/sirupsen/logrus"
-	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/time/rate"
 )
@@ -23,17 +18,19 @@ import (
 // BenchmarkCloudHandlerDispatchMetric is a benchmark intended to (manually) test
 // the impact of the CloudHandler.statsCacheHit field.
 func BenchmarkCloudHandlerDispatchMetric(b *testing.B) {
-	fp := &fakeProviderIP{}
+	fp := &fakeprovider.IP{}
 	nh := &nopHandler{}
-	ch := NewCloudHandler(fp, nh, logrus.StandardLogger(), rate.NewLimiter(100, 120), CacheOptions{
+	ci := cloudprovider.NewCachedCloudProvider(logrus.StandardLogger(), rate.NewLimiter(100, 120), fp, gostatsd.CacheOptions{
 		CacheRefreshPeriod:        100 * time.Millisecond,
 		CacheEvictAfterIdlePeriod: 700 * time.Millisecond,
 		CacheTTL:                  500 * time.Millisecond,
 		CacheNegativeTTL:          500 * time.Millisecond,
 	})
+	ch := NewCloudHandler(ci, nh)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	go ci.Run(ctx)
 	go ch.Run(ctx)
 
 	b.ReportAllocs()
@@ -50,8 +47,8 @@ func BenchmarkCloudHandlerDispatchMetric(b *testing.B) {
 
 func TestTransientInstanceFailure(t *testing.T) {
 	t.Parallel()
-	fpt := &fakeProviderTransient{
-		failureMode: []int{
+	fpt := &fakeprovider.Transient{
+		FailureMode: []int{
 			0, // success, prime the cache
 			1, // soft failure, data should remain in cache
 		},
@@ -59,12 +56,13 @@ func TestTransientInstanceFailure(t *testing.T) {
 
 	counting := &countingHandler{}
 
-	ch := NewCloudHandler(fpt, counting, logrus.StandardLogger(), rate.NewLimiter(100, 120), CacheOptions{
+	ci := cloudprovider.NewCachedCloudProvider(logrus.StandardLogger(), rate.NewLimiter(100, 120), fpt, gostatsd.CacheOptions{
 		CacheRefreshPeriod:        50 * time.Millisecond,
 		CacheEvictAfterIdlePeriod: 1 * time.Minute,
 		CacheTTL:                  1 * time.Millisecond,
 		CacheNegativeTTL:          1 * time.Millisecond,
 	})
+	ch := NewCloudHandler(ci, counting)
 
 	// t+0: instance is queried, goes in cache
 	// t+50ms: instance refreshed (failure)
@@ -75,6 +73,7 @@ func TestTransientInstanceFailure(t *testing.T) {
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	defer cancelFunc()
 	wg.StartWithContext(ctx, ch.Run)
+	wg.StartWithContext(ctx, ci.Run)
 	m1 := sm1()
 	m2 := sm1()
 
@@ -100,76 +99,12 @@ func TestTransientInstanceFailure(t *testing.T) {
 	expectedMetrics[1].Tags = gostatsd.Tags{"a1", "tag:value"}
 	expectedMetrics[1].Hostname = "1.2.3.4"
 
-	assert.Equal(t, expectedMetrics, counting.metrics)
-}
-
-func TestCloudHandlerExpirationAndRefresh(t *testing.T) {
-	// These still use a real clock, which means they're more susceptible to
-	// CPU load triggering a race condition, therefore there's no t.Parallel()
-	t.Run("4.3.2.1", func(t *testing.T) {
-		testExpireAndRefresh(t, "4.3.2.1", func(h *CloudHandler) {
-			e := se1()
-			h.DispatchEvent(context.Background(), &e)
-		})
-	})
-	t.Run("1.2.3.4", func(t *testing.T) {
-		testExpireAndRefresh(t, "1.2.3.4", func(h *CloudHandler) {
-			m := sm1()
-			h.DispatchMetrics(context.Background(), []*gostatsd.Metric{&m})
-		})
-	})
-}
-
-func testExpireAndRefresh(t *testing.T, expectedIp gostatsd.IP, f func(*CloudHandler)) {
-	fp := &fakeProviderIP{}
-	counting := &countingHandler{}
-	/*
-		Note: lookup reads in to a batch for up to 10ms
-
-		T+0: IP is sent to lookup
-		T+10: lookup is performed, cached with eviction time = T+10+50=60, refresh time = T+10+10=20
-		T+11: refresh loop, nothing to do
-		T+20: cache entry passes refresh time
-		T+22: refresh loop, cache item is dispatched for refreshing
-		T+32: cache lookup is performed, eviction time is unchanged, refresh time = T+32+10=42
-		T+33: refresh loop, nothing to do
-		T+42: cache entry passes refresh time
-		T+44: refresh loop, cache item is dispatched for refreshing
-		T+54: cache lookup is performed, eviction time is unchanged, refresh time = T+54+10=64
-		T+55: refresh loop, nothing to do
-		T+60: cache entry passes expiry time
-		T+64: cache entry passes refresh time
-		T+66: refresh loop, entry is expired
-		T+70: sleep completes
-	*/
-	ch := NewCloudHandler(fp, counting, logrus.StandardLogger(), rate.NewLimiter(100, 120), CacheOptions{
-		CacheRefreshPeriod:        11 * time.Millisecond,
-		CacheEvictAfterIdlePeriod: 50 * time.Millisecond,
-		CacheTTL:                  10 * time.Millisecond,
-		CacheNegativeTTL:          100 * time.Millisecond,
-	})
-	var wg wait.Group
-	defer wg.Wait()
-	ctx, cancelFunc := context.WithCancel(context.Background())
-	defer cancelFunc()
-	wg.StartWithContext(ctx, ch.Run)
-	f(ch)
-	time.Sleep(70 * time.Millisecond) // Should be refreshed couple of times and evicted.
-
-	cancelFunc()
-	wg.Wait()
-
-	// Cache might refresh multiple times, ensure it only refreshed with the expected IP
-	for _, ip := range fp.ips {
-		assert.Equal(t, expectedIp, ip)
-	}
-	assert.GreaterOrEqual(t, len(fp.ips), 2) // Ensure it does at least 1 lookup + 1 refresh
-	assert.Zero(t, len(ch.cache))            // Ensure it eventually expired
+	assert.Equal(t, expectedMetrics, counting.Metrics())
 }
 
 func TestCloudHandlerDispatch(t *testing.T) {
 	t.Parallel()
-	fp := &fakeProviderIP{
+	fp := &fakeprovider.IP{
 		Tags: gostatsd.Tags{"region:us-west-3", "tag1", "tag2:234"},
 	}
 
@@ -208,12 +143,12 @@ func TestCloudHandlerDispatch(t *testing.T) {
 			SourceIP: "4.3.2.1",
 		},
 	}
-	doCheck(t, fp, sm1(), se1(), sm2(), se2(), &fp.ips, expectedIps, expectedMetrics, expectedEvents)
+	doCheck(t, fp, sm1(), se1(), sm2(), se2(), fp.IPs, expectedIps, expectedMetrics, expectedEvents)
 }
 
 func TestCloudHandlerInstanceNotFound(t *testing.T) {
 	t.Parallel()
-	fp := &fakeProviderNotFound{}
+	fp := &fakeprovider.NotFound{}
 	expectedIps := []gostatsd.IP{"1.2.3.4", "4.3.2.1"}
 	expectedMetrics := []gostatsd.Metric{
 		sm1(),
@@ -223,12 +158,12 @@ func TestCloudHandlerInstanceNotFound(t *testing.T) {
 		se1(),
 		se2(),
 	}
-	doCheck(t, fp, sm1(), se1(), sm2(), se2(), &fp.ips, expectedIps, expectedMetrics, expectedEvents)
+	doCheck(t, fp, sm1(), se1(), sm2(), se2(), fp.IPs, expectedIps, expectedMetrics, expectedEvents)
 }
 
 func TestCloudHandlerFailingProvider(t *testing.T) {
 	t.Parallel()
-	fp := &fakeFailingProvider{}
+	fp := &fakeprovider.Failing{}
 	expectedIps := []gostatsd.IP{"1.2.3.4", "4.3.2.1"}
 	expectedMetrics := []gostatsd.Metric{
 		sm1(),
@@ -238,77 +173,42 @@ func TestCloudHandlerFailingProvider(t *testing.T) {
 		se1(),
 		se2(),
 	}
-	doCheck(t, fp, sm1(), se1(), sm2(), se2(), &fp.ips, expectedIps, expectedMetrics, expectedEvents)
-}
-
-func TestConstructCloudHandlerFactoryFromViper(t *testing.T) {
-	t.Parallel()
-	v := viper.New()
-	logger := logrus.StandardLogger()
-
-	// Test no cloud handler
-	factory, err := NewCloudHandlerFactoryFromViper(v, logger, "test")
-	assert.NoError(t, err)
-	assert.Nil(t, factory)
-
-	// Test unknown cloud handler - unsupported
-	v.Set(ParamCloudProvider, "unknown")
-	_, err = NewCloudHandlerFactoryFromViper(v, logger, "test")
-	assert.Error(t, err)
-
-	// Test known cloud provider defaults
-	cloudProvidersToTest := []string{aws.ProviderName, k8s.ProviderName}
-	for _, cpName := range cloudProvidersToTest {
-		v.Set(ParamCloudProvider, cpName)
-		factory, err = NewCloudHandlerFactoryFromViper(v, logger, "test")
-		assert.NoError(t, err)
-		assert.EqualValues(t, DefaultCloudProviderCacheValues[cpName], factory.cacheOptions)
-		assert.Equal(t, rate.Limit(DefaultCloudProviderLimiterValues[cpName].MaxCloudRequests), factory.limiter.Limit())
-		assert.Equal(t, DefaultCloudProviderLimiterValues[cpName].BurstCloudRequests, factory.limiter.Burst())
-	}
-}
-
-func TestInitCloudHandlerFactory(t *testing.T) {
-	t.Parallel()
-	v := viper.New()
-	logger := logrus.StandardLogger()
-
-	factory, err := NewCloudHandlerFactoryFromViper(v, logger, "test")
-	assert.NoError(t, err)
-	assert.Nil(t, factory)
-
-	// We don't test the path for specific cloud providers as they have logic that isn't easily mockable
+	doCheck(t, fp, sm1(), se1(), sm2(), se2(), fp.IPs, expectedIps, expectedMetrics, expectedEvents)
 }
 
-func doCheck(t *testing.T, cloud gostatsd.CloudProvider, m1 gostatsd.Metric, e1 gostatsd.Event, m2 gostatsd.Metric, e2 gostatsd.Event, ips *[]gostatsd.IP, expectedIps []gostatsd.IP, expectedM []gostatsd.Metric, expectedE gostatsd.Events) {
+func doCheck(t *testing.T, cloud CountingProvider, m1 gostatsd.Metric, e1 gostatsd.Event, m2 gostatsd.Metric, e2 gostatsd.Event, ipsFunc func() []gostatsd.IP, expectedIps []gostatsd.IP, expectedM []gostatsd.Metric, expectedE gostatsd.Events) {
 	counting := &countingHandler{}
-	ch := NewCloudHandler(cloud, counting, logrus.StandardLogger(), rate.NewLimiter(100, 120), CacheOptions{
-		CacheRefreshPeriod:        DefaultCacheRefreshPeriod,
-		CacheEvictAfterIdlePeriod: DefaultCacheEvictAfterIdlePeriod,
-		CacheTTL:                  DefaultCacheTTL,
-		CacheNegativeTTL:          DefaultCacheNegativeTTL,
+	ci := cloudprovider.NewCachedCloudProvider(logrus.StandardLogger(), rate.NewLimiter(100, 120), cloud, gostatsd.CacheOptions{
+		CacheRefreshPeriod:        gostatsd.DefaultCacheRefreshPeriod,
+		CacheEvictAfterIdlePeriod: gostatsd.DefaultCacheEvictAfterIdlePeriod,
+		CacheTTL:                  gostatsd.DefaultCacheTTL,
+		CacheNegativeTTL:          gostatsd.DefaultCacheNegativeTTL,
 	})
+	ch := NewCloudHandler(ci, counting)
+
 	var wg wait.Group
 	defer wg.Wait()
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	defer cancelFunc()
 	wg.StartWithContext(ctx, ch.Run)
+	wg.StartWithContext(ctx, ci.Run)
 	ch.DispatchMetrics(ctx, []*gostatsd.Metric{&m1})
 	ch.DispatchEvent(ctx, &e1)
 	time.Sleep(1 * time.Second)
 	ch.DispatchMetrics(ctx, []*gostatsd.Metric{&m2})
 	ch.DispatchEvent(ctx, &e2)
-	time.Sleep(1 * time.Second)
+	time.Sleep(10 * time.Second)
 	cancelFunc()
 	wg.Wait()
 
-	sort.Slice(*ips, func(i, j int) bool {
-		return (*ips)[i] < (*ips)[j]
+	ips := ipsFunc()
+	sort.Slice(ips, func(i, j int) bool {
+		return ips[i] < ips[j]
 	})
-	assert.Equal(t, expectedIps, *ips)
-	assert.Equal(t, expectedM, counting.metrics)
-	assert.Equal(t, expectedE, counting.events)
-	assert.EqualValues(t, 1, cloud.(CountingProvider).Invocations())
+	assert.Equal(t, expectedIps, ips)
+	assert.Equal(t, expectedM, counting.Metrics())
+	assert.Equal(t, expectedE, counting.Events())
+	assert.EqualValues(t, 1, cloud.Invocations())
 }
 
 func sm1() gostatsd.Metric {
@@ -354,140 +254,6 @@ func se2() gostatsd.Event {
 }
 
 type CountingProvider interface {
+	gostatsd.CloudProvider
 	Invocations() uint64
 }
-
-type fakeCountingProvider struct {
-	mu          sync.Mutex
-	ips         []gostatsd.IP
-	invocations uint64
-}
-
-func (fp *fakeCountingProvider) EstimatedTags() int {
-	return 0
-}
-
-func (fp *fakeCountingProvider) MaxInstancesBatch() int {
-	return 16
-}
-
-func (fp *fakeCountingProvider) Invocations() uint64 {
-	fp.mu.Lock()
-	defer fp.mu.Unlock()
-	return fp.invocations
-}
-
-func (fp *fakeCountingProvider) count(ips ...gostatsd.IP) {
-	fp.mu.Lock()
-	defer fp.mu.Unlock()
-	fp.ips = append(fp.ips, ips...)
-	fp.invocations++
-}
-
-func (fp *fakeCountingProvider) SelfIP() (gostatsd.IP, error) {
-	return gostatsd.UnknownIP, nil
-}
-
-type fakeProviderIP struct {
-	fakeCountingProvider
-	Region string
-	Tags   gostatsd.Tags
-}
-
-func (fp *fakeProviderIP) Name() string {
-	return "fakeProviderIP"
-}
-
-func (fp *fakeProviderIP) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
-	fp.count(ips...)
-	instances := make(map[gostatsd.IP]*gostatsd.Instance, len(ips))
-	for _, ip := range ips {
-		instances[ip] = &gostatsd.Instance{
-			ID:   "i-" + string(ip),
-			Tags: fp.Tags,
-		}
-	}
-	return instances, nil
-}
-
-type fakeProviderNotFound struct {
-	fakeCountingProvider
-}
-
-func (fp *fakeProviderNotFound) Name() string {
-	return "fakeProviderNotFound"
-}
-
-func (fp *fakeProviderNotFound) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
-	fp.count(ips...)
-	return nil, nil
-}
-
-type fakeFailingProvider struct {
-	fakeCountingProvider
-}
-
-func (fp *fakeFailingProvider) Name() string {
-	return "fakeFailingProvider"
-}
-
-func (fp *fakeFailingProvider) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
-	fp.count(ips...)
-	return nil, errors.New("clear skies, no clouds available")
-}
-
-type fakeProviderTransient struct {
-	call        uint64
-	failureMode []int
-}
-
-func (fpt *fakeProviderTransient) Name() string {
-	return "fakeProviderTransient"
-}
-
-func (fpt *fakeProviderTransient) EstimatedTags() int {
-	return 1
-}
-
-func (fpt *fakeProviderTransient) MaxInstancesBatch() int {
-	return 1
-}
-
-func (fpt *fakeProviderTransient) SelfIP() (gostatsd.IP, error) {
-	return gostatsd.UnknownIP, nil
-}
-
-// Instance emulates a lookup based on the supplied criteria.
-// A failure mode of 0 is a successful lookup
-// A failure mode of 1 is nil instance, no error (lookup failure)
-// A failure mode of 2 is nil instance, with error
-// Repeats the last specified failure mode
-func (fpt *fakeProviderTransient) Instance(ctx context.Context, ips ...gostatsd.IP) (map[gostatsd.IP]*gostatsd.Instance, error) {
-	r := make(map[gostatsd.IP]*gostatsd.Instance)
-
-	c := atomic.AddUint64(&fpt.call, 1) - 1
-	if c >= uint64(len(fpt.failureMode)) {
-		c = uint64(len(fpt.failureMode) - 1)
-	}
-	switch fpt.failureMode[c] {
-	case 0:
-		for _, ip := range ips {
-			r[ip] = &gostatsd.Instance{
-				ID:   string(ip),
-				Tags: gostatsd.Tags{"tag:value"},
-			}
-		}
-		return r, nil
-	case 1:
-		for _, ip := range ips {
-			r[ip] = nil
-		}
-		return r, nil
-	case 2:
-		for _, ip := range ips {
-			r[ip] = nil
-		}
-		return r, errors.New("failure mode 2")
-	}
-	return nil, nil
-}
diff --git a/pkg/statsd/handler_fixtures_test.go b/pkg/statsd/handler_fixtures_test.go
index 0b79faf..92725ea 100644
--- a/pkg/statsd/handler_fixtures_test.go
+++ b/pkg/statsd/handler_fixtures_test.go
@@ -56,6 +56,22 @@ type countingHandler struct {
 	events  gostatsd.Events
 }
 
+func (ch *countingHandler) Metrics() []gostatsd.Metric {
+	ch.mu.Lock()
+	defer ch.mu.Unlock()
+	result := make([]gostatsd.Metric, len(ch.metrics))
+	copy(result, ch.metrics)
+	return result
+}
+
+func (ch *countingHandler) Events() gostatsd.Events {
+	ch.mu.Lock()
+	defer ch.mu.Unlock()
+	result := make(gostatsd.Events, len(ch.events))
+	copy(result, ch.events)
+	return result
+}
+
 func (ch *countingHandler) EstimatedTags() int {
 	return 0
 }
diff --git a/pkg/statsd/receiver_test.go b/pkg/statsd/receiver_test.go
index 76ba11f..3199493 100644
--- a/pkg/statsd/receiver_test.go
+++ b/pkg/statsd/receiver_test.go
@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/atlassian/gostatsd"
 	"github.com/atlassian/gostatsd/pkg/fakesocket"
 	"github.com/magiconair/properties/assert"
 	"github.com/stretchr/testify/require"
@@ -19,7 +20,7 @@ func BenchmarkReceive(b *testing.B) {
 	//
 	// ... so this is pretty arbitrary.
 	ch := make(chan []*Datagram, 5000)
-	mr := NewDatagramReceiver(ch, nil, 0, DefaultReceiveBatchSize)
+	mr := NewDatagramReceiver(ch, nil, 0, gostatsd.DefaultReceiveBatchSize)
 	c, done := fakesocket.NewCountedFakePacketConn(uint64(b.N))
 
 	var wg sync.WaitGroup
diff --git a/pkg/statsd/statsd_test.go b/pkg/statsd/statsd_test.go
index cf4452e..a28c36c 100644
--- a/pkg/statsd/statsd_test.go
+++ b/pkg/statsd/statsd_test.go
@@ -8,9 +8,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ash2k/stager/wait"
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/cachedinstances/cloudprovider"
 	"github.com/atlassian/gostatsd/pkg/fakesocket"
-
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"golang.org/x/time/rate"
@@ -23,44 +24,44 @@ func TestStatsdThroughput(t *testing.T) {
 	var memStatsStart, memStatsFinish runtime.MemStats
 	runtime.ReadMemStats(&memStatsStart)
 	backend := &countingBackend{}
-
-	cloudHandlerFactory := newCloudHandlerFactory(
-		"",
-		logrus.StandardLogger(),
-		CacheOptions{
-			CacheRefreshPeriod:        DefaultCacheRefreshPeriod,
-			CacheEvictAfterIdlePeriod: DefaultCacheEvictAfterIdlePeriod,
-			CacheTTL:                  DefaultCacheTTL,
-			CacheNegativeTTL:          DefaultCacheNegativeTTL,
-		},
-		rate.NewLimiter(DefaultMaxCloudRequests, DefaultBurstCloudRequests),
-		"test")
-	// Inject the mock provider
-	cloudHandlerFactory.cloudProvider = &fakeProvider{
+	cloudProvider := &fakeProvider{
 		instance: &gostatsd.Instance{
 			ID:   "i-13123123",
 			Tags: gostatsd.Tags{"region:us-west-3", "tag1", "tag2:234"},
 		},
 	}
-
+	cachedInstances := cloudprovider.NewCachedCloudProvider(
+		logrus.StandardLogger(),
+		rate.NewLimiter(gostatsd.DefaultMaxCloudRequests, gostatsd.DefaultBurstCloudRequests),
+		cloudProvider,
+		gostatsd.CacheOptions{
+			CacheRefreshPeriod:        gostatsd.DefaultCacheRefreshPeriod,
+			CacheEvictAfterIdlePeriod: gostatsd.DefaultCacheEvictAfterIdlePeriod,
+			CacheTTL:                  gostatsd.DefaultCacheTTL,
+			CacheNegativeTTL:          gostatsd.DefaultCacheNegativeTTL,
+		})
 	s := Server{
 		Backends:            []gostatsd.Backend{backend},
-		CloudHandlerFactory: cloudHandlerFactory,
-		DefaultTags:         DefaultTags,
-		ExpiryInterval:      DefaultExpiryInterval,
-		FlushInterval:       DefaultFlushInterval,
-		MaxReaders:          DefaultMaxReaders,
-		MaxParsers:          DefaultMaxParsers,
-		MaxWorkers:          DefaultMaxWorkers,
-		MaxQueueSize:        DefaultMaxQueueSize,
+		CachedInstances:     cachedInstances,
+		DefaultTags:         gostatsd.DefaultTags,
+		ExpiryInterval:      gostatsd.DefaultExpiryInterval,
+		FlushInterval:       gostatsd.DefaultFlushInterval,
+		MaxReaders:          gostatsd.DefaultMaxReaders,
+		MaxParsers:          gostatsd.DefaultMaxParsers,
+		MaxWorkers:          gostatsd.DefaultMaxWorkers,
+		MaxQueueSize:        gostatsd.DefaultMaxQueueSize,
 		EstimatedTags:       1, // Travis has limited memory
-		PercentThreshold:    DefaultPercentThreshold,
-		HeartbeatEnabled:    DefaultHeartbeatEnabled,
-		ReceiveBatchSize:    DefaultReceiveBatchSize,
+		PercentThreshold:    gostatsd.DefaultPercentThreshold,
+		HeartbeatEnabled:    gostatsd.DefaultHeartbeatEnabled,
+		ReceiveBatchSize:    gostatsd.DefaultReceiveBatchSize,
 		MaxConcurrentEvents: 2,
 		ServerMode:          "standalone",
 		Viper:               viper.New(),
 	}
+	ctxCached, cancelCached := context.WithCancel(context.Background())
+	defer cancelCached()
+	var wg wait.Group
+	wg.StartWithContext(ctxCached, cachedInstances.Run)
 	ctx, cancelFunc := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancelFunc()
 	start := time.Now()
@@ -68,6 +69,8 @@ func TestStatsdThroughput(t *testing.T) {
 	if err != nil && err != context.Canceled && err != context.DeadlineExceeded {
 		t.Errorf("statsd run failed: %v", err)
 	}
+	cancelCached()
+	wg.Wait()
 	duration := float64(time.Since(start)) / float64(time.Second)
 
 	runtime.ReadMemStats(&memStatsFinish)