Fix some tests

collect/collect_test.go

@@ -93,16 +93,17 @@ func TestAddRootSpan(t *testing.T) {
 	}
 	coll.AddSpan(span)
 
-	time.Sleep(conf.SendTickerVal * 2)
 	// adding one span with no parent ID should:
 	// * create the trace in the cache
 	// * send the trace
 	// * remove the trace from the cache
-	assert.Nil(t, coll.getFromCache(traceID1), "after sending the span, it should be removed from the cache")
-	transmission.Mux.RLock()
-	assert.Equal(t, 1, len(transmission.Events), "adding a root span should send the span")
-	assert.Equal(t, "aoeu", transmission.Events[0].Dataset, "sending a root span should immediately send that span via transmission")
-	transmission.Mux.RUnlock()
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.Nil(c, coll.getFromCache(traceID1), "after sending the span, it should be removed from the cache")
+		transmission.Mux.RLock()
+		assert.Equal(c, 1, len(transmission.Events), "adding a root span should send the span")
+		assert.Equal(c, "aoeu", transmission.Events[0].Dataset, "sending a root span should immediately send that span via transmission")
+		transmission.Mux.RUnlock()
+	}, conf.SendTickerVal*2, 100*time.Millisecond)
 
 	span = &types.Span{
 		TraceID: traceID2,
@@ -112,16 +113,17 @@ func TestAddRootSpan(t *testing.T) {
 		},
 	}
 	coll.AddSpanFromPeer(span)
-	time.Sleep(conf.SendTickerVal * 2)
 	// adding one span with no parent ID should:
 	// * create the trace in the cache
 	// * send the trace
 	// * remove the trace from the cache
-	assert.Nil(t, coll.getFromCache(traceID1), "after sending the span, it should be removed from the cache")
-	transmission.Mux.RLock()
-	assert.Equal(t, 2, len(transmission.Events), "adding another root span should send the span")
-	assert.Equal(t, "aoeu", transmission.Events[1].Dataset, "sending a root span should immediately send that span via transmission")
-	transmission.Mux.RUnlock()
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.Nil(c, coll.getFromCache(traceID1), "after sending the span, it should be removed from the cache")
+		transmission.Mux.RLock()
+		assert.Equal(c, 2, len(transmission.Events), "adding another root span should send the span")
+		assert.Equal(c, "aoeu", transmission.Events[1].Dataset, "sending a root span should immediately send that span via transmission")
+		transmission.Mux.RUnlock()
+	}, conf.SendTickerVal*2, 100*time.Millisecond)
 }
 
 // #490, SampleRate getting stomped could cause confusion if sampling was

config/config_test.go

@@ -4,10 +4,13 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/honeycombio/refinery/config"
+	"github.com/honeycombio/refinery/internal/configwatcher"
+	"github.com/honeycombio/refinery/pubsub"
 	"github.com/stretchr/testify/assert"
 	"gopkg.in/yaml.v3"
 )
@@ -195,6 +198,107 @@ func TestMetricsAPIKeyFallbackEnvVar(t *testing.T) {
 	}
 }
 
+func TestReload(t *testing.T) {
+	cm := makeYAML("General.ConfigurationVersion", 2, "General.ConfigReloadInterval", config.Duration(1*time.Second), "Network.ListenAddr", "0.0.0.0:8080")
+	rm := makeYAML("ConfigVersion", 2)
+	cfg, rules := createTempConfigs(t, cm, rm)
+	defer os.Remove(rules)
+	defer os.Remove(cfg)
+	c, err := getConfig([]string{"--no-validate", "--config", cfg, "--rules_config", rules})
+	assert.NoError(t, err)
+
+	pubsub := &pubsub.LocalPubSub{
+		Config: c,
+	}
+	pubsub.Start()
+	defer pubsub.Stop()
+	watcher := &configwatcher.ConfigWatcher{
+		Config: c,
+		PubSub: pubsub,
+	}
+	watcher.Start()
+	defer watcher.Stop()
+
+	if d, _ := c.GetListenAddr(); d != "0.0.0.0:8080" {
+		t.Error("received", d, "expected", "0.0.0.0:8080")
+	}
+
+	wg := &sync.WaitGroup{}
+
+	ch := make(chan interface{}, 1)
+
+	c.RegisterReloadCallback(func(cfgHash, ruleHash string) {
+		close(ch)
+	})
+
+	// Hey race detector, we're doing some concurrent config reads.
+	// That's cool, right?
+	go func() {
+		tick := time.NewTicker(time.Millisecond)
+		defer tick.Stop()
+		for {
+			c.GetListenAddr()
+			select {
+			case <-ch:
+				return
+			case <-tick.C:
+			}
+		}
+	}()
+
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		select {
+		case <-ch:
+		case <-time.After(5 * time.Second):
+			t.Error("No callback")
+			close(ch)
+		}
+	}()
+
+	if file, err := os.OpenFile(cfg, os.O_RDWR, 0644); err == nil {
+		cm := makeYAML("General.ConfigurationVersion", 2, "General.ConfigReloadInterval", config.Duration(1*time.Second), "Network.ListenAddr", "0.0.0.0:9000")
+		file.WriteString(cm)
+		file.Close()
+	}
+
+	wg.Wait()
+
+	if d, _ := c.GetListenAddr(); d != "0.0.0.0:9000" {
+		t.Error("received", d, "expected", "0.0.0.0:9000")
+	}
+
+}
+
+func TestReloadDisabled(t *testing.T) {
+	cm := makeYAML("General.ConfigurationVersion", 2, "General.ConfigReloadInterval", config.Duration(0*time.Second), "Network.ListenAddr", "0.0.0.0:8080")
+	rm := makeYAML("ConfigVersion", 2)
+	cfg, rules := createTempConfigs(t, cm, rm)
+	defer os.Remove(rules)
+	defer os.Remove(cfg)
+	c, err := getConfig([]string{"--no-validate", "--config", cfg, "--rules_config", rules})
+	assert.NoError(t, err)
+
+	if d, _ := c.GetListenAddr(); d != "0.0.0.0:8080" {
+		t.Error("received", d, "expected", "0.0.0.0:8080")
+	}
+
+	if file, err := os.OpenFile(cfg, os.O_RDWR, 0644); err == nil {
+		// Since we disabled reload checking this should not change anything
+		cm := makeYAML("General.ConfigurationVersion", 2, "General.ConfigReloadInterval", config.Duration(0*time.Second), "Network.ListenAddr", "0.0.0.0:9000")
+		file.WriteString(cm)
+		file.Close()
+	}
+
+	time.Sleep(5 * time.Second)
+
+	if d, _ := c.GetListenAddr(); d != "0.0.0.0:8080" {
+		t.Error("received", d, "expected", "0.0.0.0:8080")
+	}
+}
+
 func TestReadDefaults(t *testing.T) {
 	c, err := getConfig([]string{"--no-validate", "--config", "../config.yaml", "--rules_config", "../rules.yaml"})
 	assert.NoError(t, err)

config/config_test_reload_error_test.go

@@ -1,40 +1,55 @@
-//go:build all || !race
+//xxgo:build all || !race
 
-package config
+package config_test
 
 import (
 	"os"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/honeycombio/refinery/config"
+	"github.com/honeycombio/refinery/internal/configwatcher"
+	"github.com/honeycombio/refinery/pubsub"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestErrorReloading(t *testing.T) {
 	cm := makeYAML(
 		"General.ConfigurationVersion", 2,
-		"General.ConfigReloadInterval", Duration(1*time.Second),
+		"General.ConfigReloadInterval", config.Duration(1*time.Second),
 		"Network.ListenAddr", "0.0.0.0:8080",
 		"HoneycombLogger.APIKey", "SetThisToAHoneycombKey",
 	)
 	rm := makeYAML(
 		"RulesVersion", 2,
 		"Samplers.__default__.DeterministicSampler.SampleRate", 5,
 	)
-	config, rules := createTempConfigs(t, cm, rm)
+	cfg, rules := createTempConfigs(t, cm, rm)
 	defer os.Remove(rules)
-	defer os.Remove(config)
+	defer os.Remove(cfg)
 
-	opts, err := NewCmdEnvOptions([]string{"--config", config, "--rules_config", rules})
+	opts, err := config.NewCmdEnvOptions([]string{"--config", cfg, "--rules_config", rules})
 	assert.NoError(t, err)
 
 	ch := make(chan interface{}, 1)
-	c, err := NewConfig(opts, func(err error) { ch <- 1 })
+	c, err := config.NewConfig(opts, func(err error) { ch <- 1 })
 	assert.NoError(t, err)
 
+	pubsub := &pubsub.LocalPubSub{
+		Config: c,
+	}
+	pubsub.Start()
+	defer pubsub.Stop()
+	watcher := &configwatcher.ConfigWatcher{
+		Config: c,
+		PubSub: pubsub,
+	}
+	watcher.Start()
+	defer watcher.Stop()
+
 	d, name, _ := c.GetSamplerConfigForDestName("dataset5")
-	if _, ok := d.(DeterministicSamplerConfig); ok {
+	if _, ok := d.(config.DeterministicSamplerConfig); ok {
 		t.Error("type received", d, "expected", "DeterministicSampler")
 	}
 	if name != "DeterministicSampler" {
@@ -67,7 +82,7 @@ func TestErrorReloading(t *testing.T) {
 
 	// config should error and not update sampler to invalid type
 	d, _, _ = c.GetSamplerConfigForDestName("dataset5")
-	if _, ok := d.(DeterministicSamplerConfig); ok {
+	if _, ok := d.(config.DeterministicSamplerConfig); ok {
 		t.Error("received", d, "expected", "DeterministicSampler")
 	}
 }