add integration test for applyconfig

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.4.3
 	github.com/google/dnsmasq_exporter v0.0.0-00010101000000-000000000000
+	github.com/google/go-cmp v0.5.4
 	github.com/gorilla/mux v1.8.0
 	github.com/grafana/loki v1.6.2-0.20210205130758-59a34f9867ce
 	github.com/hashicorp/consul/api v1.8.1

pkg/loki/loki_test.go

@@ -0,0 +1,130 @@
+package loki
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/go-kit/kit/log"
+	"github.com/grafana/agent/pkg/util"
+	"github.com/grafana/loki/pkg/distributor"
+	"github.com/grafana/loki/pkg/logproto"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v2"
+)
+
+func TestLoki(t *testing.T) {
+	//
+	// Create a temporary file to tail
+	//
+	positionsDir, err := ioutil.TempDir(os.TempDir(), "positions-*")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, os.RemoveAll(positionsDir))
+	})
+
+	tmpFile, err := ioutil.TempFile(os.TempDir(), "*.log")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, os.RemoveAll(tmpFile.Name()))
+	})
+
+	//
+	// Listen for push requests and pass them through to a channel
+	//
+	pushes := make(chan *logproto.PushRequest)
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, lis.Close())
+	})
+	go http.Serve(lis, http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+		req, err := distributor.ParseRequest(r)
+		require.NoError(t, err)
+
+		pushes <- req
+		_, _ = rw.Write(nil)
+	}))
+
+	//
+	// Launch Loki so it starts tailing the file and writes to our server.
+	//
+	cfgText := util.Untab(fmt.Sprintf(`
+positions_directory: %s
+configs:
+- name: default
+  clients:
+  - url: http://%s/loki/api/v1/push
+		batchwait: 50ms
+		batchsize: 1
+  scrape_configs:
+  - job_name: system
+    static_configs:
+    - targets: [localhost]
+      labels:
+        job: test
+        __path__: %s
+	`, positionsDir, lis.Addr().String(), tmpFile.Name()))
+
+	var cfg Config
+	dec := yaml.NewDecoder(strings.NewReader(cfgText))
+	dec.SetStrict(true)
+	require.NoError(t, dec.Decode(&cfg))
+
+	l, err := New(prometheus.NewRegistry(), cfg, log.NewNopLogger())
+	require.NoError(t, err)
+	defer l.Stop()
+
+	//
+	// Write a log line and wait for it to come through.
+	//
+	fmt.Fprintf(tmpFile, "Hello, world!\n")
+	select {
+	case <-time.After(time.Second * 30):
+		require.FailNow(t, "timed out waiting for data to be pushed")
+	case req := <-pushes:
+		require.Equal(t, "Hello, world!", req.Streams[0].Entries[0].Line)
+	}
+
+	//
+	// Apply a new config and write a new line.
+	//
+	cfgText = util.Untab(fmt.Sprintf(`
+positions_directory: %s
+configs:
+- name: default
+  clients:
+  - url: http://%s/loki/api/v1/push
+		batchwait: 50ms
+		batchsize: 5
+  scrape_configs:
+  - job_name: system
+    static_configs:
+    - targets: [localhost]
+      labels:
+        job: test-2
+        __path__: %s
+	`, positionsDir, lis.Addr().String(), tmpFile.Name()))
+
+	var newCfg Config
+	dec = yaml.NewDecoder(strings.NewReader(cfgText))
+	dec.SetStrict(true)
+	require.NoError(t, dec.Decode(&newCfg))
+
+	require.NoError(t, l.ApplyConfig(newCfg))
+
+	fmt.Fprintf(tmpFile, "Hello again!\n")
+	select {
+	case <-time.After(time.Second * 30):
+		require.FailNow(t, "timed out waiting for data to be pushed")
+	case req := <-pushes:
+		require.Equal(t, "Hello again!", req.Streams[0].Entries[0].Line)
+	}
+}

pkg/prom/agent.go

@@ -212,6 +212,20 @@ func newAgent(reg prometheus.Registerer, cfg Config, logger log.Logger, fact ins
 	return a, nil
 }
 
+// ApplyConfig applies a new Config to an Agent.
+func (a *Agent) ApplyConfig(cfg Config) error {
+	// We need to update cfg.InstanceRestartBackoff in our basic manager.
+	// If the InstanceMode changed, we need to stop all instances and create a group manager.
+	// What do we need to do with counting manager?
+
+	// Apply all configs against the new config manager chain.
+
+	// If HA mode changed, we need to either start or stop it.
+	// If HA mode is already running, we need to pass the latest config through to it.
+
+	return nil
+}
+
 // newInstance creates a new Instance given a config.
 func (a *Agent) newInstance(c instance.Config) (instance.ManagedInstance, error) {
 	// Controls the label

pkg/util/untab.go

@@ -0,0 +1,10 @@
+package util
+
+import "strings"
+
+// Untab is a utility function for tests to make it easier
+// to write YAML tests, where some editors will insert tabs
+// into strings by default.
+func Untab(s string) string {
+	return strings.ReplaceAll(s, "\t", "  ")
+}

vendor/modules.txt

@@ -262,6 +262,7 @@ github.com/google/btree
 ## explicit
 github.com/google/dnsmasq_exporter/collector
 # github.com/google/go-cmp v0.5.4
+## explicit
 github.com/google/go-cmp/cmp
 github.com/google/go-cmp/cmp/internal/diff
 github.com/google/go-cmp/cmp/internal/flags