Function TestStopStreaming

pkg/acquisition/modules/loki/loki_test.go:370–416

(t *testing.T)

}

func TestStopStreaming(t *testing.T) {
	cstest.SetAWSTestEnv(t)

	ctx := t.Context()

	config := `
mode: tail
source: loki
url: http://127.0.0.1:3100
headers:
  x-scope-orgid: "1234"
query: >
  {server="demo"}
`
	logger := log.New()
	subLogger := logger.WithField("type", loki.ModuleName)
	title := time.Now().String()
	lokiSource := loki.Source{}

	err := lokiSource.Configure(ctx, []byte(config), subLogger, metrics.AcquisitionMetricsLevelNone)
	if err != nil {
		t.Fatalf("Unexpected error : %s", err)
	}

	out := make(chan pipeline.Event)

	lokiTomb := &tomb.Tomb{}

	err = lokiSource.StreamingAcquisition(ctx, out, lokiTomb)
	if err != nil {
		t.Fatalf("Unexpected error : %s", err)
	}

	time.Sleep(time.Second * 2)

	err = feedLoki(ctx, subLogger, 1, title)
	if err != nil {
		t.Fatalf("Unexpected error : %s", err)
	}

	lokiTomb.Kill(nil)

	err = lokiTomb.Wait()
	if err != nil {
		t.Fatalf("Unexpected error : %s", err)
	}
}

type LogStreams struct {
	Streams []LogStream `json:"streams"`