Repository: honeycombio/refinery
Pull request #594 (2023-01-31 18:07:18)

feat: Stress Relief system
# Stress Relief

The problem this is designed to address is that refinery can become unstable under heavy load, and it goes from stable to unstable very quickly. A larger refinery cluster can be very hard to bring up -- we've seen what are effectively contagious restarts, where a given node crashes, wakes up, and then another node crashes, etc. 

The general idea here is "let's make refinery be more stable by allowing it to shed load quickly under stress". This PR implements a "StressRelief" system that, when activated, deterministically samples each span based only on its trace ID, without attempting to direct traces to the appropriate shard first. The configuration determines the sample rate used.

There is configuration to set this up, and it allows the "mode" to be configured as "never" (it's not active), "monitor" (it's active when the stress indicator is on), or "always" (always active). 

I tested this system with mode set to never and established a baseline instability level for a cluster of 3 refineries at a given "load index" (an arbitrary value indicating the amount of traffic). I found that nodes began restarting once the load index went over 600. When I set the mode to "always", the cluster was quite happy with a load of 1500, and there was no indication of stress -- it felt like it could go much higher, but I haven't yet tried higher numbers. 

The next problem was to get an accurate measurement of stress so that we can automatically turn this system on. Many hours of watching refinery crash led to the realization that the internal queues seem to be critical, as does memory usage. 

The desired strategy is to measure some key metrics and compare them with some defined threshold values on an appropriately adjusted scale from 0 to 1. We can then watch to see when any of them starts to rise too close to the threshold, and turn on the stress relief (and also give the user some indication of which metric is the biggest problem)

The StressRelief system uses a hysteresis model where it turns on at a much higher level than it turns off -- the current defaults are 50 and 20 (user-adjustable, of course). 

The set of particular metrics to watch is not currently user-adjustable, although they're coded to make that possible without too much difficulty. But it's a lot of added complexity in the config for probably little value, since most people won't have a good idea which things they should worry about.

## Memory

The existing cache system tries to manage memory, particularly with the newer "impact" cache overrun strategy. But we can also use excessive RAM usage as a signal for stress.

## Queues
There are multiple queues that are tracked in metrics:

* "collector_peer_queue_length"
* "collector_incoming_queue_length"
* "libhoney_peer_queue_length"
* "libhoney_upstream_queue_length"

Once any of these queue lengths rise above zero, it means that refinery is not processing data as quickly as it is arriving, and in a high-volume situation, it doesn't take long for that to overwhelm refinery and crash it. But the configured length of the queues is definitely a factor.





kentquirk: I'll resolve conflicts etc after #593 is merged.

Patch:
diff --git a/cmd/refinery/main.go b/cmd/refinery/main.go
index 17272cd..49556f0 100644
--- a/cmd/refinery/main.go
+++ b/cmd/refinery/main.go
@@ -184,6 +184,8 @@ func main() {
 		os.Exit(1)
 	}
 
+	stressRelief := &collect.StressRelief{Done: done}
+
 	var g inject.Graph
 	if opts.Debug {
 		g.Logger = graphLogger{}
@@ -204,6 +206,7 @@ func main() {
 		&inject.Object{Value: peerMetricsRecorder, Name: "peerMetrics"},
 		&inject.Object{Value: version, Name: "version"},
 		&inject.Object{Value: samplerFactory},
+		&inject.Object{Value: stressRelief, Name: "stressRelief"},
 		&inject.Object{Value: &a},
 	)
 	if err != nil {
@@ -237,6 +240,10 @@ func main() {
 		os.Exit(1)
 	}
 
+	// these have to be done after the injection (of metrics)
+	metricsSingleton.Store("UPSTREAM_BUFFER_SIZE", float64(c.GetUpstreamBufferSize()))
+	metricsSingleton.Store("PEER_BUFFER_SIZE", float64(c.GetPeerBufferSize()))
+
 	// set up signal channel to exit
 	sigsToExit := make(chan os.Signal, 1)
 	signal.Notify(sigsToExit, syscall.SIGINT, syscall.SIGTERM)
diff --git a/collect/collect.go b/collect/collect.go
index dc798f4..c9e74e1 100644
--- a/collect/collect.go
+++ b/collect/collect.go
@@ -27,6 +27,9 @@ type Collector interface {
 	// scheduled for transmission.
 	AddSpan(*types.Span) error
 	AddSpanFromPeer(*types.Span) error
+	Stressed() bool
+	GetStressedSampleRate(traceID string) (rate uint, keep bool, reason string)
+	ProcessSpanImmediately(sp *types.Span, keep bool, sampleRate uint, reason string)
 }
 
 func GetCollectorImplementation(c config.Config) Collector {
@@ -61,6 +64,7 @@ type InMemCollector struct {
 	Transmission   transmit.Transmission  `inject:"upstreamTransmission"`
 	Metrics        metrics.Metrics        `inject:"genericMetrics"`
 	SamplerFactory *sample.SamplerFactory `inject:""`
+	StressRelief   StressReliever         `inject:"stressRelief"`
 
 	// For test use only
 	BlockOnAddSpan bool
@@ -89,6 +93,7 @@ func (i *InMemCollector) Start() error {
 		return err
 	}
 	i.cache = cache.NewInMemCache(imcConfig.CacheCapacity, i.Metrics, i.Logger)
+	i.StressRelief.UpdateFromConfig(i.Config.GetStressReliefConfig())
 
 	// listen for config reloads
 	i.Config.RegisterReloadCallback(i.sendReloadSignal)
@@ -99,6 +104,8 @@ func (i *InMemCollector) Start() error {
 	i.Metrics.Register("collector_peer_queue_length", "gauge")
 	i.Metrics.Register("collector_incoming_queue_length", "gauge")
 	i.Metrics.Register("collector_peer_queue", "histogram")
+	i.Metrics.Register("stress_level", "gauge")
+	i.Metrics.Register("stress_relief_activated", "gauge")
 	i.Metrics.Register("collector_cache_size", "gauge")
 	i.Metrics.Register("memory_heap_allocation", "gauge")
 	i.Metrics.Register("span_received", "counter")
@@ -136,6 +143,8 @@ func (i *InMemCollector) Start() error {
 
 	i.incoming = make(chan *types.Span, imcConfig.CacheCapacity*3)
 	i.fromPeer = make(chan *types.Span, imcConfig.CacheCapacity*3)
+	i.Metrics.Store("INCOMING_CAP", float64(cap(i.incoming)))
+	i.Metrics.Store("PEER_CAP", float64(cap(i.fromPeer)))
 	i.reload = make(chan struct{}, 1)
 	i.datasetSamplers = make(map[string]sample.Sampler)
 
@@ -191,6 +200,8 @@ func (i *InMemCollector) reloadConfigs() {
 		i.Logger.Error().WithField("cache", i.cache.(*cache.DefaultInMemCache)).Logf("skipping reloading the cache on config reload because it's not an in-memory cache")
 	}
 
+	i.StressRelief.UpdateFromConfig(i.Config.GetStressReliefConfig())
+
 	// clear out any samplers that we have previously created
 	// so that the new configuration will be propagated
 	i.datasetSamplers = make(map[string]sample.Sampler)
@@ -256,6 +267,7 @@ func (i *InMemCollector) oldCheckAlloc() {
 
 func (i *InMemCollector) newCheckAlloc() {
 	inMemConfig, err := i.Config.GetInMemCollectorCacheCapacity()
+	i.Metrics.Store("MEMORY_MAX_ALLOC", float64(inMemConfig.MaxAlloc))
 
 	var mem runtime.MemStats
 	runtime.ReadMemStats(&mem)
@@ -334,6 +346,15 @@ func (i *InMemCollector) AddSpanFromPeer(sp *types.Span) error {
 	return i.add(sp, i.fromPeer)
 }
 
+// Stressed returns true if the collector is undergoing significant stress
+func (i *InMemCollector) Stressed() bool {
+	return i.StressRelief.Stressed()
+}
+
+func (i *InMemCollector) GetStressedSampleRate(traceID string) (rate uint, keep bool, reason string) {
+	return i.StressRelief.GetSampleRate(traceID)
+}
+
 func (i *InMemCollector) add(sp *types.Span, ch chan<- *types.Span) error {
 	if i.BlockOnAddSpan {
 		ch <- sp
@@ -373,6 +394,12 @@ func (i *InMemCollector) collect() {
 		i.Metrics.Histogram("collector_peer_queue", float64(len(i.fromPeer)))
 		i.Metrics.Gauge("collector_incoming_queue_length", float64(len(i.incoming)))
 		i.Metrics.Gauge("collector_peer_queue_length", float64(len(i.fromPeer)))
+		i.Metrics.Gauge("stress_level", float64(i.StressRelief.StressLevel()))
+		if i.StressRelief.Stressed() {
+			i.Metrics.Gauge("stress_relief_activated", 1)
+		} else {
+			i.Metrics.Gauge("stress_relief_activated", 0)
+		}
 
 		// Always drain peer channel before doing anything else. By processing peer
 		// traffic preferentially we avoid the situation where the cluster essentially
@@ -500,6 +527,40 @@ func (i *InMemCollector) processSpan(sp *types.Span) {
 	}
 }
 
+// ProcessSpanImmediately is an escape hatch used under stressful conditions --
+// it submits a span for immediate transmission without enqueuing it for normal
+// processing. This means it ignores dry run mode and doesn't build a complete
+// trace context or cache the trace in the active trace buffer. It only gets
+// called on the first span for a trace under stressful conditions; we got here
+// because the StressRelief system detected that this is a new trace AND that it
+// is being sampled. Therefore, we also put the traceID into the sent traces
+// cache as "kept".
+// It doesn't do any logging and barely touches metrics; this is about as
+// minimal as we can make it.
+func (i *InMemCollector) ProcessSpanImmediately(sp *types.Span, keep bool, sampleRate uint, reason string) {
+	now := time.Now()
+	trace := &types.Trace{
+		APIHost:     sp.APIHost,
+		APIKey:      sp.APIKey,
+		Dataset:     sp.Dataset,
+		TraceID:     sp.TraceID,
+		ArrivalTime: now,
+		SendBy:      now,
+	}
+	// we do want a record of how we disposed of traces in case more come in after we've
+	// turned off stress relief (if stress relief is on we'll keep making the same decisions)
+	i.sampleTraceCache.Record(trace, keep)
+	if !keep {
+		i.Metrics.Increment("dropped_from_stress")
+		return
+	}
+	// ok, we're sending it, so decorate it first
+	sp.Event.Data["meta.stressed"] = true
+	sp.Event.Data["meta.refinery.reason"] = reason
+	mergeTraceAndSpanSampleRates(sp, sampleRate)
+	i.Transmission.EnqueueSpan(sp)
+}
+
 // dealWithSentTrace handles a span that has arrived after the sampling decision
 // on the trace has already been made, and it obeys that decision by either
 // sending the span immediately or dropping it.
diff --git a/collect/stressRelief.go b/collect/stressRelief.go
new file mode 100644
index 0000000..eef780c
--- /dev/null
+++ b/collect/stressRelief.go
@@ -0,0 +1,313 @@
+package collect
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/dgryski/go-wyhash"
+	"github.com/honeycombio/refinery/config"
+	"github.com/honeycombio/refinery/logger"
+	"github.com/honeycombio/refinery/metrics"
+)
+
+type StressReliever interface {
+	Start() error
+	UpdateFromConfig(cfg config.StressReliefConfig) error
+	Recalc()
+	StressLevel() uint
+	Stressed() bool
+	GetSampleRate(traceID string) (rate uint, keep bool, reason string)
+}
+
+type MockStressReliever struct{}
+
+func (m *MockStressReliever) Start() error                                         { return nil }
+func (m *MockStressReliever) UpdateFromConfig(cfg config.StressReliefConfig) error { return nil }
+func (m *MockStressReliever) Recalc()                                              {}
+func (m *MockStressReliever) StressLevel() uint                                    { return 0 }
+func (m *MockStressReliever) Stressed() bool                                       { return false }
+func (m *MockStressReliever) GetSampleRate(traceID string) (rate uint, keep bool, reason string) {
+	return 1, false, ""
+}
+
+// hashSeed is a random value to seed the hash generator for the sampler.
+// We want it to be a constant that's the same across all nodes so that they
+// all make the same sampling decisions during stress relief.
+const hashSeed = 34527861234
+
+type StressReliefMode int
+
+const (
+	Never StressReliefMode = iota
+	Monitor
+	Always
+)
+
+type StressRelief struct {
+	mode            StressReliefMode
+	activateLevel   uint
+	deactivateLevel uint
+	sampleRate      uint64
+	upperBound      uint64
+	stressLevel     uint
+	reason          string
+	stressed        bool
+	belowMin        bool
+	minDuration     time.Duration
+	RefineryMetrics metrics.Metrics `inject:"metrics"`
+	Logger          logger.Logger   `inject:""`
+	Done            chan struct{}
+
+	algorithms map[string]func(string, string) float64
+	calcs      []StressReliefCalculation
+	lock       sync.RWMutex
+}
+
+func (s *StressRelief) Start() error {
+	s.Logger.Debug().Logf("Starting StressRelief system")
+	defer func() { s.Logger.Debug().Logf("Finished starting StressRelief system") }()
+
+	// We use an algorithms map so that we can name these algorithms, which makes it easier for several things:
+	// - change our mind about which algorithm to use
+	// - logging the algorithm actually used
+	// - making it easier to make them configurable
+	// At the moment, we are not permitting these to be configurable, but we might change our minds on this.
+	// Thus, we're also including a couple of algorithms we don't currently use for convenience.
+	s.algorithms = map[string]func(string, string) float64{
+		"linear":  s.linear,  // just use the ratio
+		"sqrt":    s.sqrt,    // small values are inflated
+		"square":  s.square,  // big values are deflated
+		"sigmoid": s.sigmoid, // don't worry about small stuff, but if we cross the midline, start worrying quickly
+	}
+
+	s.calcs = []StressReliefCalculation{
+		{Numerator: "collector_peer_queue_length", Denominator: "PEER_CAP", Algorithm: "sqrt", Reason: "CacheCapacity (peer)"},
+		{Numerator: "collector_incoming_queue_length", Denominator: "INCOMING_CAP", Algorithm: "sqrt", Reason: "CacheCapacity (incoming)"},
+		{Numerator: "libhoney_peer_queue_length", Denominator: "PEER_BUFFER_SIZE", Algorithm: "sqrt", Reason: "PeerBufferSize"},
+		{Numerator: "libhoney_upstream_queue_length", Denominator: "UPSTREAM_BUFFER_SIZE", Algorithm: "sqrt", Reason: "UpstreamBufferSize"},
+		{Numerator: "memory_heap_allocation", Denominator: "MEMORY_MAX_ALLOC", Algorithm: "sigmoid", Reason: "MaxAlloc"},
+	}
+
+	// start our monitor goroutine that periodically calls recalc
+	go func(d *StressRelief) {
+		tick := time.NewTicker(100 * time.Millisecond)
+		defer tick.Stop()
+		for {
+			select {
+			case <-tick.C:
+				d.Recalc()
+			case <-d.Done:
+				d.Logger.Debug().Logf("Stopping StressRelief system")
+				return
+			}
+		}
+	}(s)
+	return nil
+}
+
+func (s *StressRelief) UpdateFromConfig(cfg config.StressReliefConfig) error {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	switch cfg.Mode {
+	case "never", "":
+		s.mode = Never
+	case "monitor":
+		s.mode = Monitor
+	case "always":
+		s.mode = Always
+	default: // validation shouldn't let this happen but we'll be safe...
+		s.mode = Never
+		s.Logger.Error().Logf("StressRelief mode is '%s' which shouldn't happen", cfg.Mode)
+	}
+	s.Logger.Debug().WithField("mode", s.mode).Logf("setting StressRelief mode")
+
+	s.activateLevel = cfg.ActivationLevel
+	s.deactivateLevel = cfg.DeactivationLevel
+	s.sampleRate = cfg.StressSamplingRate
+	if s.sampleRate == 0 {
+		s.sampleRate = 1
+	}
+	s.minDuration = cfg.MinimumActivationDuration
+	s.Logger.Debug().
+		WithField("activation_level", s.activateLevel).
+		WithField("deactivation_level", s.deactivateLevel).
+		WithField("sampling_rate", s.sampleRate).
+		WithField("min_duration", s.minDuration).
+		Logf("StressRelief parameters")
+
+	// Get the actual upper bound - the largest possible value divided by
+	// the sample rate. In the case where the sample rate is 1, this should
+	// sample every value.
+	s.upperBound = math.MaxUint64 / s.sampleRate
+
+	return nil
+}
+
+func clamp(f float64, min float64, max float64) float64 {
+	if f < min {
+		return min
+	}
+	if f > max {
+		return max
+	}
+	return f
+}
+
+// ratio is a function that returns the ratio of two values looked up in the metrics,
+// clamped between 0 and 1. Since we know this is the range, we know that
+// sqrt has the effect of making small values larger, and square has the
+// effect of making large values smaller. We can use these functions to bias the
+// weighting of our calculations.
+func (s *StressRelief) ratio(num, denom string) float64 {
+	numerator, ok := s.RefineryMetrics.Get(num)
+	if !ok {
+		s.Logger.Debug().Logf("stress recalc: missing numerator %s", num)
+		return 0
+	}
+	denominator, ok := s.RefineryMetrics.Get(denom)
+	if !ok {
+		s.Logger.Debug().Logf("stress recalc: missing denominator %s", denom)
+		return 0
+	}
+	if denominator != 0 {
+		stress := clamp(numerator/denominator, 0, 1)
+		s.Logger.Debug().
+			WithField("numerator_name", num).
+			WithField("numerator_value", numerator).
+			WithField("denominator_name", denom).
+			WithField("denominator_value", denominator).
+			WithField("unscaled_result", stress).
+			Logf("stress recalc: detail")
+		return stress
+	}
+	return 0
+}
+
+// linear simply returns the value it calculates
+func (s *StressRelief) linear(num, denom string) float64 {
+	stress := s.ratio(num, denom)
+	s.Logger.Debug().
+		WithField("algorithm", "linear").
+		WithField("result", stress).
+		Logf("stress recalc: result")
+	return stress
+}
+
+// sqrt returns the square root of the value calculated, which (in the range [0-1])
+// inflates them a bit without affecting the ends of the range.
+func (s *StressRelief) sqrt(num, denom string) float64 {
+	stress := math.Sqrt(s.ratio(num, denom))
+	s.Logger.Debug().
+		WithField("algorithm", "sqrt").
+		WithField("result", stress).
+		Logf("stress recalc: result")
+	return stress
+}
+
+// square returns the square of the value calculated, which (in the range [0-1])
+// deflates them a bit without affecting the ends of the range.
+func (s *StressRelief) square(num, denom string) float64 {
+	r := s.ratio(num, denom)
+	stress := r * r
+	s.Logger.Debug().
+		WithField("algorithm", "square").
+		WithField("result", stress).
+		Logf("stress recalc: result")
+	return stress
+}
+
+// sigmoid returns a value along a sigmoid (s-shaped) curve of the value
+// calculated, which (in the range [0-1]) deflates low values and inflates high
+// values without affecting the ends of the range. We use this one for memory pressure,
+// under the presumption that if we're using less than half of RAM,
+func (s *StressRelief) sigmoid(num, denom string) float64 {
+	r := s.ratio(num, denom)
+	// this is an S curve from 0 to 1, centered around 0.5 -- determined
+	// by messing around with a graphing calculator
+	stress := .395*math.Atan(6*(r-0.5)) + 0.5
+	s.Logger.Debug().
+		WithField("algorithm", "sigmoid").
+		WithField("result", stress).
+		Logf("stress recalc: result")
+	return stress
+}
+
+type StressReliefCalculation struct {
+	Numerator   string
+	Denominator string
+	Algorithm   string
+	Reason      string
+}
+
+// We want to calculate the stress from various values around the system. Each key value
+// can be reported as a key-value.
+// This should be called periodically.
+func (s *StressRelief) Recalc() {
+	// we have multiple queues to watch, and for each we calculate a stress level for that queue, which is
+	// 100 * the fraction of its capacity in use. Our overall stress level is the max of those values.
+	// We track the config value that is under stress as "reason".
+
+	var level float64
+	var reason string
+	for _, c := range s.calcs {
+		stress := 100 * s.algorithms[c.Algorithm](c.Numerator, c.Denominator)
+		if stress > level {
+			level = stress
+			reason = c.Reason
+		}
+	}
+	s.Logger.Debug().WithField("stress_level", level).WithField("reason", reason).Logf("calculated stress level")
+
+	s.lock.Lock()
+	s.stressLevel = uint(level)
+	s.reason = reason
+	s.lock.Unlock()
+}
+
+func (s *StressRelief) StressLevel() uint {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.stressLevel
+}
+
+// Stressed() indicates whether the system should act as if it's stressed.
+// Note that the stress_level metric is independent of mode.
+func (s *StressRelief) Stressed() bool {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	switch s.mode {
+	case Never:
+		s.stressed = false
+	case Always:
+		s.stressed = true
+	case Monitor:
+		if !s.stressed && s.stressLevel >= s.activateLevel {
+			s.stressed = true
+			// we want make sure that stress relief is on for a minimum time
+			s.belowMin = true
+			time.AfterFunc(s.minDuration, func() {
+				s.lock.Lock()
+				s.belowMin = false
+				s.lock.Unlock()
+			})
+			s.Logger.Info().WithField("stress_level", s.stressLevel).WithField("reason", s.reason).Logf("StressRelief has been activated")
+		}
+		if s.stressed && !s.belowMin && s.stressLevel < s.deactivateLevel {
+			s.stressed = false
+			s.Logger.Info().WithField("stress_level", s.stressLevel).Logf("StressRelief has been deactivated")
+		}
+	}
+	return s.stressed
+}
+
+func (s *StressRelief) GetSampleRate(traceID string) (rate uint, keep bool, reason string) {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	if s.sampleRate <= 1 {
+		return 1, true, "stress_relief/always"
+	}
+	hash := wyhash.Hash([]byte(traceID), hashSeed)
+	return uint(s.sampleRate), hash <= s.upperBound, "stress_relief/deterministic"
+}
diff --git a/config/config.go b/config/config.go
index dec7c50..f3c087c 100644
--- a/config/config.go
+++ b/config/config.go
@@ -178,6 +178,8 @@ type Config interface {
 	GetConfigMetadata() []ConfigMetadata
 
 	GetSampleCacheConfig() SampleCacheConfig
+
+	GetStressReliefConfig() StressReliefConfig
 }
 
 type ConfigMetadata struct {
diff --git a/config/file_config.go b/config/file_config.go
index 8240e00..3fe40c1 100644
--- a/config/file_config.go
+++ b/config/file_config.go
@@ -62,7 +62,8 @@ type configContents struct {
 	AdditionalErrorFields     []string
 	AddSpanCountToRoot        bool
 	CacheOverrunStrategy      string
-	SampleCache               SampleCacheConfig `validate:"required"`
+	SampleCache               SampleCacheConfig  `validate:"required"`
+	StressRelief              StressReliefConfig `validate:"required"`
 }
 
 type InMemoryCollectorCacheCapacity struct {
@@ -126,6 +127,14 @@ type GRPCServerParameters struct {
 	Timeout               time.Duration
 }
 
+type StressReliefConfig struct {
+	Mode                      string `validate:"required,oneof= always never monitor"`
+	ActivationLevel           uint
+	DeactivationLevel         uint
+	StressSamplingRate        uint64
+	MinimumActivationDuration time.Duration
+}
+
 // NewConfig creates a new config struct
 func NewConfig(config, rules string, errorCallback func(error)) (Config, error) {
 	c := viper.New()
@@ -178,6 +187,11 @@ func NewConfig(config, rules string, errorCallback func(error)) (Config, error)
 	c.SetDefault("SampleCache.KeptSize", 10_000)
 	c.SetDefault("SampleCache.DroppedSize", 1_000_000)
 	c.SetDefault("SampleCache.SizeCheckInterval", 10*time.Second)
+	c.SetDefault("StressRelief.Mode", "never")
+	c.SetDefault("StressRelief.ActivationLevel", 75)
+	c.SetDefault("StressRelief.DeactivationLevel", 25)
+	c.SetDefault("StressRelief.StressSamplingRate", 100)
+	c.SetDefault("StressRelief.MinimumActivationDuration", 10*time.Second)
 
 	c.SetConfigFile(config)
 	err := c.ReadInConfig()
@@ -942,6 +956,13 @@ func (f *fileConfig) GetSampleCacheConfig() SampleCacheConfig {
 	return f.conf.SampleCache
 }
 
+func (f *fileConfig) GetStressReliefConfig() StressReliefConfig {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return f.conf.StressRelief
+}
+
 // calculates an MD5 sum for a file that returns the same result as the md5sum command
 func calcMD5For(filename string) string {
 	f, err := os.Open(filename)
diff --git a/config/mock.go b/config/mock.go
index 6d561f6..078c9c2 100644
--- a/config/mock.go
+++ b/config/mock.go
@@ -87,6 +87,7 @@ type MockConfig struct {
 	AddSpanCountToRoot            bool
 	CacheOverrunStrategy          string
 	SampleCache                   SampleCacheConfig
+	StressRelief                  StressReliefConfig
 	CfgMetadata                   []ConfigMetadata
 
 	Mux sync.RWMutex
@@ -485,6 +486,13 @@ func (f *MockConfig) GetSampleCacheConfig() SampleCacheConfig {
 	return f.SampleCache
 }
 
+func (f *MockConfig) GetStressReliefConfig() StressReliefConfig {
+	f.Mux.RLock()
+	defer f.Mux.RUnlock()
+
+	return f.StressRelief
+}
+
 func (f *MockConfig) GetConfigMetadata() []ConfigMetadata {
 	f.Mux.RLock()
 	defer f.Mux.RUnlock()
diff --git a/config_complete.toml b/config_complete.toml
index 129848f..5cc531a 100644
--- a/config_complete.toml
+++ b/config_complete.toml
@@ -474,3 +474,70 @@ MetricsReportingInterval = 3
 # Does not apply to the "legacy" type of cache.
 # Eligible for live reload.
 # SizeCheckInterval = "10s"
+
+
+#################################
+## Stress Relief Configuration ##
+#################################
+
+[StressRelief]
+
+# Controls the parameters of the stress relief system. There is a metric called
+# stress_level that is emitted as part of refinery metrics. It is a measure of
+# refinery's throughput rate relative to its processing rate, combined with the
+# amount of room in its internal queues, and ranges from 0 to 100. It is
+# generally expected to be 0 except under heavy load. When stress levels reach
+# 100, there is an increased chance that refinery will become unstable.
+#
+# To avoid this problem, the Stress Relief system can do deterministic sampling
+# on new trace traffic based solely on TraceID, without having to store traces
+# in the cache or take the time processing sampling rules. Existing traces in
+# flight will be processed normally, but when Stress Relief is active, trace
+# decisions are made deterministically on a per-span basis; all spans will be
+# sampled according to the SamplingRate specified here.
+#
+# Once Stress Relief activates (by exceeding the ActivationLevel), it will not
+# deactivate until stress_level falls below the DeactivationLevel. When it
+# deactivates, normal trace decisions are made -- and any additional spans that
+# arrive for traces that were active during Stress Relief will respect those
+# decisions.
+#
+# The measurement of stress is a lagging indicator and is highly dependent on
+# Refinery configuration and scaling. Other configuration values should be well
+# tuned first, before adjusting the Stress Relief Activation parameters.
+
+# Mode is a string indicating how to use Stress Relief. Options are:
+# - "never" means that Stress Relief will never activate
+# - "monitor" is the recommended setting, and means that Stress Relief will monitor
+#     the status of refinery and activate according to the levels set below.
+# - "always" means that Stress Relief is always on, which may be useful in an
+#     emergency situation.
+# Default is "never".
+# Eligible for live reload.
+# Mode = "monitor"
+
+# ActivationLevel is the stress_level (from 0-100) at which Stress Relief is
+# triggered.
+# Default = 75
+# Eligible for live reload.
+# ActivationLevel = 75
+
+# DeactivationLevel is the stress_level (from 0-100) at which Stress Relief is
+# turned off (subject to MinimumActivationDuration). Under normal circumstances,
+# it should be well below ActivationLevel to avoid oscillations.
+# Default = 25
+# Eligible for live reload.
+# DeactivationLevel = 25
+
+# StressSamplingRate is the sampling rate to use when Stress Relief is
+# activated. All new traces will be deterministically sampled at this rate based
+# only on the traceID.
+# Default = 100
+# Eligible for live reload.
+# StressSamplingRate = 100
+
+# MinimumActivationDuration is the minimum time that stress relief will stay
+# enabled, once activated. This prevents oscillations.
+# Default = 10s
+# Eligible for live reload.
+# MinimumActivationDuration = 10s
diff --git a/metrics/honeycomb.go b/metrics/honeycomb.go
index b657ddc..73370ec 100644
--- a/metrics/honeycomb.go
+++ b/metrics/honeycomb.go
@@ -302,25 +302,6 @@ func (h *HoneycombMetrics) Get(name string) (float64, bool) {
 	return 0, false
 }
 
-func (h *HoneycombMetrics) GetAllNames() []string {
-	h.lock.RLock()
-	defer h.lock.RUnlock()
-	names := []string{}
-	for name := range h.counters {
-		names = append(names, name)
-	}
-	for name := range h.updowns {
-		names = append(names, name)
-	}
-	for name := range h.gauges {
-		names = append(names, name)
-	}
-	for name := range h.constants {
-		names = append(names, name)
-	}
-	return names
-}
-
 func average(vals []float64) float64 {
 	var total float64
 	for _, val := range vals {
diff --git a/route/route.go b/route/route.go
index d616539..5d324b6 100644
--- a/route/route.go
+++ b/route/route.go
@@ -499,7 +499,7 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 		WithString("dataset", ev.Dataset).
 		WithString("environment", ev.Environment)
 
-	// extract trace ID, route to self or peer, pass on to collector
+	// extract trace ID
 	// TODO make trace ID field configurable
 	var traceID string
 	if trID, ok := ev.Data["trace.trace_id"]; ok {
@@ -518,7 +518,22 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 	}
 	debugLog = debugLog.WithString("trace_id", traceID)
 
-	// ok, we're a span. Figure out if we should handle locally or pass on to a peer
+	span := &types.Span{
+		Event:   *ev,
+		TraceID: traceID,
+	}
+
+	// we know we're a span, but we need to check if we're in Stress Relief mode;
+	// if we are, then we want to make an immediate, deterministic trace decision
+	// and either drop or send the trace without even trying to cache or forward it.
+	if r.Collector.Stressed() {
+		rate, keep, reason := r.Collector.GetStressedSampleRate(traceID)
+
+		r.Collector.ProcessSpanImmediately(span, keep, rate, reason)
+		return nil
+	}
+
+	// Figure out if we should handle this span locally or pass on to a peer
 	targetShard := r.Sharder.WhichShard(traceID)
 	if r.incomingOrPeer == "incoming" && !targetShard.Equals(r.Sharder.MyShard()) {
 		r.Metrics.Increment(r.incomingOrPeer + "_router_peer")
@@ -533,12 +548,8 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 		return nil
 	}
 
-	// we're supposed to handle it
+	// we're supposed to handle it normally
 	var err error
-	span := &types.Span{
-		Event:   *ev,
-		TraceID: traceID,
-	}
 	if r.incomingOrPeer == "incoming" {
 		err = r.Collector.AddSpan(span)
 	} else {
@@ -551,7 +562,8 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 	}
 
 	r.Metrics.Increment(r.incomingOrPeer + "_router_span")
-	debugLog.Logf("Accepting span from batch for collection into a trace")
+
+	debugLog.WithField("source", r.incomingOrPeer).Logf("Accepting span from batch for collection into a trace")
 	return nil
 }
 


Test patch:
diff --git a/app/app_test.go b/app/app_test.go
index 056497b..6c4f21a 100644
--- a/app/app_test.go
+++ b/app/app_test.go
@@ -195,6 +195,7 @@ func newStartedApp(
 		&inject.Object{Value: metricsr, Name: "peerMetrics"},
 		&inject.Object{Value: "test", Name: "version"},
 		&inject.Object{Value: samplerFactory},
+		&inject.Object{Value: &collect.MockStressReliever{}, Name: "stressRelief"},
 		&inject.Object{Value: &a},
 	)
 	assert.NoError(t, err)
diff --git a/collect/collect_test.go b/collect/collect_test.go
index 28d0cb3..86bdb34 100644
--- a/collect/collect_test.go
+++ b/collect/collect_test.go
@@ -39,6 +39,7 @@ func TestAddRootSpan(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -117,6 +118,7 @@ func TestOriginalSampleRateIsNotedInMetaField(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -175,6 +177,7 @@ func TestTransmittedSpansShouldHaveASampleRateOfAtLeastOne(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -237,6 +240,7 @@ func TestAddSpan(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -313,6 +317,7 @@ func TestDryRunMode(t *testing.T) {
 		Logger:         &logger.NullLogger{},
 		Transmission:   transmission,
 		Metrics:        &metrics.NullMetrics{},
+		StressRelief:   &MockStressReliever{},
 		SamplerFactory: samplerFactory,
 	}
 	c := cache.NewInMemCache(3, &metrics.NullMetrics{}, &logger.NullLogger{})
@@ -430,6 +435,7 @@ func TestCacheSizeReload(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -504,6 +510,7 @@ func TestSampleConfigReload(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -577,6 +584,7 @@ func TestOldMaxAlloc(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -671,6 +679,7 @@ func TestStableMaxAlloc(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -769,6 +778,7 @@ func TestAddSpanNoBlock(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -815,6 +825,7 @@ func TestDependencyInjection(t *testing.T) {
 		&inject.Object{Value: &transmit.MockTransmission{}, Name: "upstreamTransmission"},
 		&inject.Object{Value: &metrics.NullMetrics{}, Name: "genericMetrics"},
 		&inject.Object{Value: &sample.SamplerFactory{}},
+		&inject.Object{Value: &MockStressReliever{}, Name: "stressRelief"},
 	)
 	if err != nil {
 		t.Error(err)
@@ -841,6 +852,7 @@ func TestAddSpanCount(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
@@ -910,6 +922,7 @@ func TestLateRootGetsSpanCount(t *testing.T) {
 		Logger:       &logger.NullLogger{},
 		Transmission: transmission,
 		Metrics:      &metrics.NullMetrics{},
+		StressRelief: &MockStressReliever{},
 		SamplerFactory: &sample.SamplerFactory{
 			Config: conf,
 			Logger: &logger.NullLogger{},
diff --git a/route/route_test.go b/route/route_test.go
index c11f8f4..d8a7048 100644
--- a/route/route_test.go
+++ b/route/route_test.go
@@ -428,6 +428,7 @@ func TestDependencyInjection(t *testing.T) {
 		&inject.Object{Value: &collect.InMemCollector{}},
 		&inject.Object{Value: &metrics.NullMetrics{}, Name: "metrics"},
 		&inject.Object{Value: &metrics.NullMetrics{}, Name: "genericMetrics"},
+		&inject.Object{Value: &collect.MockStressReliever{}, Name: "stressRelief"},
 	)
 	if err != nil {
 		t.Error(err)